diff --git a/web_check.py b/web_check.py index fc5b76c..917ee60 100644 --- a/web_check.py +++ b/web_check.py @@ -5,39 +5,19 @@ import sqlite3 import subprocess import difflib import sys +import re from bs4 import BeautifulSoup -from urllib.parse import urljoin +from urllib.parse import urljoin, urlparse from datetime import datetime from dotenv import load_dotenv -# --- DATENBANK SETUP --- -# Verbindung zur lokalen SQLite-Datenbank herstellen (mit Timeout gegen Sperren) -conn = sqlite3.connect("polit_scraper.db", timeout=20) -cursor = conn.cursor() - -# Tabelle für den Stand der Hauptseiten -cursor.execute(""" -CREATE TABLE IF NOT EXISTS seiten_stand ( - url TEXT PRIMARY KEY, - inhalt TEXT, - zeit TEXT -) -""") - -# Tabelle für bereits verarbeitete Unterartikel -cursor.execute(""" -CREATE TABLE IF NOT EXISTS artikel ( - artikel_url TEXT PRIMARY KEY, - status TEXT -) -""") -conn.commit() - # --- KONFIGURATION LADEN --- load_dotenv() ADMIN_USER = os.getenv("ADMIN_USER") ADMIN_PW = os.getenv("ADMIN_PW") NTFY_BASE = os.getenv("NTFY_BASE") +DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true" +MIN_RELEVANZ_SCORE = int(os.getenv("MIN_RELEVANZ_SCORE", "30")) GEMINI_SESSION = "1" TOPIC_PUBLIC = "polit-scraper-public" @@ -49,25 +29,51 @@ scraper = cloudscraper.create_scraper( browser={"browser": "chrome", "platform": "windows", "desktop": True} ) -def analysiere_mit_gemini(text, link): - """KI-Analyse: Sendet Text und den EINEN passenden Link an Gemini (Isoliert).""" - anweisung = f"WICHTIG: Analysiere NUR den folgenden Text. Nutze für das Feld 'direct_link' exakt diese URL: {link}\n\nTEXT:\n{text}" +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() + +cursor.execute(""" +CREATE TABLE IF NOT EXISTS seiten_stand ( + url TEXT PRIMARY KEY, + inhalt TEXT, + zeit TEXT +) +""") + +cursor.execute(""" +CREATE TABLE IF NOT EXISTS artikel ( + artikel_url TEXT PRIMARY KEY, + status TEXT +) +""") +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] try: - prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8") + 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} + return {"relevant": False, "relevanz_score": 0, "begruendung_score": "Fehler: Kein JSON gefunden"} except Exception as e: print(f" [!] KI-Fehler: {e}") - return {"relevant": False} + 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 (Public).""" + """Sendet Push-Nachricht an ntfy.""" if not NTFY_BASE: return url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" emoji_mapping = { @@ -82,56 +88,63 @@ def sende_public(titel, nachricht, klick_url, category="info", priority=3): 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=5) + 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): - """Sendet Fehlermeldung an ntfy (Admin).""" 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=5) + 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() + if clean.lower().startswith("http"): + return clean + return urljoin(basis_url, clean) + def extrahiere_links(soup, basis_url): - """Extrahiert alle absoluten Links aus einem HTML-Objekt.""" links = [] + ignore_ext = (".pdf", ".jpg", ".jpeg", ".png", ".gif", ".zip", ".docx") + basis_domain = urlparse(basis_url).netloc + for a in soup.find_all("a", href=True): - links.append(urljoin(basis_url, a["href"])) + full_url = bereinige_url(a["href"], basis_url) + 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): - """Lädt Unterseite und extrahiert Hauptinhalt.""" 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 - return bereich.get_text(separator=" ", strip=True) + text = bereich.get_text(separator=" ", strip=True) + if len(text) > 12000: return text[:12000] + return text except Exception as e: - return f"Fehler beim Laden des Artikels!: {e}" + return f"Fehler beim Laden: {e}" def finde_link_fuer_text(text_snippet, soup, basis_url): - """Sucht im HTML nach einem Link in der Nähe des Textes.""" - for element in soup.find_all(string=True): - if text_snippet in element: - parent = element.find_parent("a") - if parent and parent.has_attr("href"): - return urljoin(basis_url, parent["href"]) - container = element.find_parent(["div", "li", "article", "section"]) - if container: - link = container.find("a", href=True) - if link: return urljoin(basis_url, link["href"]) + schnipsel = text_snippet.strip().lower() + 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 für eine Webseite unter Nutzung von SQL-Speicherung.""" + """Hauptlogik mit Scoring-Logik und detaillierter Begründung im Debug-Modus.""" try: antwort = scraper.get(url, timeout=15) - antwort.encoding = "utf-8" if antwort.status_code != 200: sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.") return @@ -146,7 +159,6 @@ def verarbeite_website(url): aktuelle_links = extrahiere_links(hauptbereich, url) - # 3. Filtere neue Links (Eindeutigkeit garantieren) neue_artikel_links = [] for l in aktuelle_links: cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (l,)) @@ -156,47 +168,67 @@ def verarbeite_website(url): if neue_artikel_links or (alter_text != neuer_text): print(f" [NEU] Aktivität auf {url} erkannt.") - # FALL A: Neue Deep-Links vorhanden - for link in neue_artikel_links[:3]: - print(f" [..] Bearbeite Deep-Link: {link}") + # FALL A: Neue Deep-Links + for link in neue_artikel_links[:5]: + print(f" [..] Analysiere neuen Artikel: {link}") artikel_inhalt = hole_artikel_text(link) - analyse = analysiere_mit_gemini(artikel_inhalt, link) + analyse = analysiere_mit_gemini(artikel_inhalt) - if analyse.get("relevant"): - final_url = analyse.get("direct_link", link) - if not final_url or "http" not in final_url: final_url = link - sende_public(analyse.get("titel"), analyse.get("nachricht"), final_url, analyse.get("category"), analyse.get("priority")) + score = analyse.get("relevanz_score", 0) + 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}") + sende_public( + titel=analyse.get("titel", "Polit-Update"), + nachricht=analyse.get("nachricht", "Neu auf der Seite"), + klick_url=link, + category=analyse.get("category", "info"), + 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")) conn.commit() - # FALL B: Nur Textänderung auf Hauptseite (Filter gegen Duplikate) - if not neue_artikel_links and (alter_text != neuer_text): - alt_z = alter_text.splitlines() - neu_z = neuer_text.splitlines() - diff = [z[2:] for z in difflib.ndiff(alt_z, neu_z) if z.startswith("+ ")] - + # 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("+ ")] for line in diff: if len(line.strip()) < 20: continue - passender_link = finde_link_fuer_text(line[:30], hauptbereich, url) - + 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,)) if cursor.fetchone() is None: - analyse = analysiere_mit_gemini(line, target_url) - if analyse.get("relevant"): + 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") + + if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: + 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")) + 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")) conn.commit() cursor.execute("INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", (url, neuer_text, str(datetime.now()))) conn.commit() else: - print(f" [OK] Keine Änderungen auf {url}.") + debug(f"[OK] Keine Änderungen auf {url}.") except Exception as e: print(f" ❌ Fehler bei {url}: {e}") - sende_admin("Programmfehler", f"{url}: {e}") # --- HAUPTPROGRAMM --- with open("./urls.txt", "r", encoding="utf-8") as datei: