diff --git a/web_check.py b/web_check.py index 917ee60..7bcbc40 100644 --- a/web_check.py +++ b/web_check.py @@ -29,11 +29,13 @@ scraper = cloudscraper.create_scraper( browser={"browser": "chrome", "platform": "windows", "desktop": True} ) + def debug(nachricht): if DEBUG_LOG: zeit = datetime.now().strftime("%H:%M:%S") print(f" [DEBUG {zeit}] {nachricht}") + # --- DATENBANK SETUP --- conn = sqlite3.connect("polit_scraper.db", timeout=20) cursor = conn.cursor() @@ -54,27 +56,48 @@ CREATE TABLE IF NOT EXISTS artikel ( """) conn.commit() + def analysiere_mit_gemini(text): """KI-Analyse: Bewertet NUR den Text laut V8 (inkl. Begründung).""" sicherer_text = text[:5000] anweisung = f"WICHTIG: Analysiere NUR den folgenden Text laut deinen Panda-Regeln (V8).\n\nTEXT:\n{sicherer_text}" - befehl = ["gemini", "-m", "gemini-2.5-flash", "-r", GEMINI_SESSION, "-p", anweisung] - + befehl = [ + "gemini", + "-m", + "gemini-3.1-flash-lite-preview", + "-r", + GEMINI_SESSION, + "-p", + anweisung, + ] + try: - prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8", timeout=300) + prozess = subprocess.run( + befehl, capture_output=True, text=True, encoding="utf-8", timeout=300 + ) antwort_roh = prozess.stdout.strip() start = antwort_roh.find("{") ende = antwort_roh.rfind("}") + 1 if start != -1 and ende != -1: return json.loads(antwort_roh[start:ende]) - return {"relevant": False, "relevanz_score": 0, "begruendung_score": "Fehler: Kein JSON gefunden"} + return { + "relevant": False, + "relevanz_score": 0, + "begruendung_score": "Fehler: Kein JSON gefunden", + } except Exception as e: print(f" [!] KI-Fehler: {e}") - return {"relevant": False, "relevanz_score": 0, "begruendung_score": f"Systemfehler: {e}"} + return { + "relevant": False, + "relevanz_score": 0, + "begruendung_score": f"Systemfehler: {e}", + } + def sende_public(titel, nachricht, klick_url, category="info", priority=3): """Sendet Push-Nachricht an ntfy.""" - if not NTFY_BASE: return + if not NTFY_BASE: + return url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" emoji_mapping = { "demo": "rotating_light,loudspeaker", @@ -82,32 +105,55 @@ def sende_public(titel, nachricht, klick_url, category="info", priority=3): "polizei": "police_car,warning", "stadt": "cityscape,newspaper", "solidaritaet": "fist,heart", - "info": "newspaper" + "info": "newspaper", } tags = emoji_mapping.get(category.lower(), "newspaper") - headers = {"Title": titel.encode("utf-8"), "Tags": tags, "Click": klick_url, "Priority": str(priority)} + headers = { + "Title": titel.encode("utf-8"), + "Tags": tags, + "Click": klick_url, + "Priority": str(priority), + } try: import requests - requests.post(url, data=nachricht.encode("utf-8"), headers=headers, auth=(ADMIN_USER, ADMIN_PW), timeout=10) + + requests.post( + url, + data=nachricht.encode("utf-8"), + headers=headers, + auth=(ADMIN_USER, ADMIN_PW), + timeout=10, + ) except Exception as e: print(f" [!] Fehler Public-Versand: {e}") + def sende_admin(titel, nachricht): - if not NTFY_BASE: return + if not NTFY_BASE: + return url = f"{NTFY_BASE}/{TOPIC_ADMIN}" headers = {"Title": titel.encode("utf-8"), "Tags": "warning,skull", "Priority": "4"} try: import requests - requests.post(url, data=nachricht.encode("utf-8"), headers=headers, auth=(ADMIN_USER, ADMIN_PW), timeout=10) + + requests.post( + url, + data=nachricht.encode("utf-8"), + headers=headers, + auth=(ADMIN_USER, ADMIN_PW), + timeout=10, + ) except Exception as e: print(f" [!] Fehler Admin-Versand: {e}") + def bereinige_url(url_roh, basis_url): - clean = url_roh.replace('”', '').replace('"', '').replace("'", "").strip() + clean = url_roh.replace("”", "").replace('"', "").replace("'", "").strip() if clean.lower().startswith("http"): return clean return urljoin(basis_url, clean) + def extrahiere_links(soup, basis_url): links = [] ignore_ext = (".pdf", ".jpg", ".jpeg", ".png", ".gif", ".zip", ".docx") @@ -115,38 +161,48 @@ def extrahiere_links(soup, basis_url): for a in soup.find_all("a", href=True): full_url = bereinige_url(a["href"], basis_url) - if full_url.lower().endswith(ignore_ext): continue + if full_url.lower().endswith(ignore_ext): + continue if urlparse(full_url).netloc == basis_domain: if len(full_url) > len(basis_url.rstrip("/")) + 1: links.append(full_url) return list(set(links)) + def hole_artikel_text(url): try: antwort = scraper.get(url, timeout=10) unter_soup = BeautifulSoup(antwort.text, "html.parser") - bereich = unter_soup.find("main") or unter_soup.find("article") or unter_soup.body + bereich = ( + unter_soup.find("main") or unter_soup.find("article") or unter_soup.body + ) text = bereich.get_text(separator=" ", strip=True) - if len(text) > 12000: return text[:12000] + if len(text) > 12000: + return text[:12000] return text except Exception as e: return f"Fehler beim Laden: {e}" + def finde_link_fuer_text(text_snippet, soup, basis_url): schnipsel = text_snippet.strip().lower() - if len(schnipsel) < 10: return None + if len(schnipsel) < 10: + return None for a in soup.find_all("a", href=True): link_text = a.get_text(separator=" ", strip=True).lower() if schnipsel in link_text or link_text in schnipsel: return bereinige_url(a["href"], basis_url) return None + def verarbeite_website(url): """Hauptlogik mit Scoring-Logik und detaillierter Begründung im Debug-Modus.""" try: antwort = scraper.get(url, timeout=15) if antwort.status_code != 200: - sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.") + sende_admin( + "Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}." + ) return soup = BeautifulSoup(antwort.text, "html.parser") @@ -158,7 +214,7 @@ def verarbeite_website(url): alter_text = zeile[0] if zeile else "" aktuelle_links = extrahiere_links(hauptbereich, url) - + neue_artikel_links = [] for l in aktuelle_links: cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (l,)) @@ -173,10 +229,12 @@ def verarbeite_website(url): print(f" [..] Analysiere neuen Artikel: {link}") artikel_inhalt = hole_artikel_text(link) analyse = analysiere_mit_gemini(artikel_inhalt) - + score = analyse.get("relevanz_score", 0) - begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden") - + begruendung = analyse.get( + "begruendung_score", "Keine Begründung vorhanden" + ) + if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}") debug(f"Begründung: {begruendung}") @@ -185,44 +243,76 @@ def verarbeite_website(url): nachricht=analyse.get("nachricht", "Neu auf der Seite"), klick_url=link, category=analyse.get("category", "info"), - priority=analyse.get("priority", 3) + priority=analyse.get("priority", 3), ) else: - debug(f"Überspringe Artikel (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}") - - cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (link, "erledigt")) + debug( + f"Überspringe Artikel (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}" + ) + + cursor.execute( + "INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", + (link, "erledigt"), + ) conn.commit() # FALL B: Textänderung auf Hauptseite if alter_text != neuer_text: - diff = [z[2:] for z in difflib.ndiff(alter_text.splitlines(), neuer_text.splitlines()) if z.startswith("+ ")] + diff = [ + z[2:] + for z in difflib.ndiff( + alter_text.splitlines(), neuer_text.splitlines() + ) + if z.startswith("+ ") + ] for line in diff: - if len(line.strip()) < 20: continue + if len(line.strip()) < 20: + continue passender_link = finde_link_fuer_text(line[:40], hauptbereich, url) target_url = passender_link if passender_link else url - - cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (target_url,)) + + cursor.execute( + "SELECT 1 FROM artikel WHERE artikel_url = ?", (target_url,) + ) if cursor.fetchone() is None: if passender_link: artikel_inhalt = hole_artikel_text(passender_link) analyse = analysiere_mit_gemini(artikel_inhalt) else: analyse = analysiere_mit_gemini(line) - + score = analyse.get("relevanz_score", 0) - begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden") - + begruendung = analyse.get( + "begruendung_score", "Keine Begründung vorhanden" + ) + if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: - print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}") + print( + f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}" + ) debug(f"Begründung: {begruendung}") - sende_public(analyse.get("titel"), analyse.get("nachricht"), target_url, analyse.get("category"), analyse.get("priority")) + sende_public( + analyse.get("titel"), + analyse.get("nachricht"), + target_url, + analyse.get("category"), + analyse.get("priority"), + ) else: - debug(f"Überspringe Änderung (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}") - - cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (target_url, "erledigt")) + debug( + f"Überspringe Änderung (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}" + ) + + cursor.execute( + "INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", + (target_url, "erledigt"), + ) conn.commit() - cursor.execute("INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", (url, neuer_text, str(datetime.now()))) + cursor.execute( + "INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", + (url, neuer_text, str(datetime.now())), + ) conn.commit() else: debug(f"[OK] Keine Änderungen auf {url}.") @@ -230,11 +320,13 @@ def verarbeite_website(url): except Exception as e: print(f" ❌ Fehler bei {url}: {e}") + # --- HAUPTPROGRAMM --- with open("./urls.txt", "r", encoding="utf-8") as datei: for zeile in datei: adresse = zeile.strip() - if not adresse or adresse.startswith("#"): continue + if not adresse or adresse.startswith("#"): + continue print(f"Ich prüfe: {adresse}") verarbeite_website(adresse)