web_check.py: Umstellung auf SQLite-Datenbank, Implementierung der Deep-Analysis-Schleife (Einzelfallprüfung) und ntfy-Prioritäten

web_check.py

@@ -1,263 +1,196 @@
 import cloudscraper
 import json
 import os
+import sqlite3
 import subprocess
 import difflib
 import sys
 from bs4 import BeautifulSoup
 from urllib.parse import urljoin
 from datetime import datetime
-from dotenv import load_dotenv, main
+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 ---
-# Sucht nach einer Datei namens .env und lädt deren Variablen in das System
 load_dotenv()
-
-# os.getenv liest eine Variable aus dem System (die zuvor aus der .env geladen wurde)
 ADMIN_USER = os.getenv("ADMIN_USER")
 ADMIN_PW = os.getenv("ADMIN_PW")
 NTFY_BASE = os.getenv("NTFY_BASE")
-
-# Name der Datei, in der die alten Webseiten-Stände gespeichert werden
-JSON_DATEI = "./results.json"
-# Die ID der Gemini-Sitzung, die für die Analyse genutzt werden soll
 GEMINI_SESSION = "1"
 
-# Namen der ntfy-Topics für öffentliche News und Admin-Fehler
 TOPIC_PUBLIC = "polit-scraper-public"
 TOPIC_ADMIN = "polit-scraper-admin"
 
-# Stellt sicher, dass das Skript Umlaute (ä, ö, ü) korrekt im Terminal ausgibt
 sys.stdout.reconfigure(encoding="utf-8")
 
-# Erstellt ein Scraper-Objekt, das sich als Chrome-Browser unter Windows ausgibt
-# Dies dient dazu, Sicherheitsabfragen wie Cloudflare zu umgehen
 scraper = cloudscraper.create_scraper(
     browser={"browser": "chrome", "platform": "windows", "desktop": True}
 )
 
-
+def analysiere_mit_gemini(text, link):
-def analysiere_mit_gemini(text, links):
+    """KI-Analyse: Sendet Text und den EINEN passenden Link an Gemini (Isoliert)."""
-    """KI-Analyse: Sendet den Text und die gefundenen Links an Gemini."""
+    anweisung = f"WICHTIG: Analysiere NUR den folgenden Text. Nutze für das Feld 'direct_link' exakt diese URL: {link}\n\nTEXT:\n{text}"
-
+    befehl = ["gemini", "-m", "gemini-2.5-flash", "-r", GEMINI_SESSION, "-p", anweisung]
-    # Erklärung der folgenden Zeile (Ternärer Operator):
-    # Wenn die Liste 'links' nicht leer ist, wird ein String mit der Überschrift erstellt
-    # und alle Links mit Kommas getrennt (", ".join) dahintergehängt.
-    # Wenn 'links' leer ist (else), wird einfach ein leerer String ("") zugewiesen.
-    links_text = "\nRelevante Links: " + ", ".join(links) if links else ""
-
-    # Baut die endgültige Frage (Prompt) für die KI zusammen
-    anweisung = f"Analysiere diesen Text: {text}{links_text}"
-
-    # Befehl für das Terminal: gemini -r [Sitzung] -p [Frage]
-    befehl = ["gemini", "-r", GEMINI_SESSION, "-p", anweisung]
     
     try:
-        # Führt den Terminal-Befehl aus und speichert die Antwort
+        prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8")
-        # capture_output=True fängt den Text ab, text=True macht daraus einen Python-String
-        prozess = subprocess.run(
-            befehl, capture_output=True, text=True, encoding="utf-8"
-        )
         antwort_roh = prozess.stdout.strip()
-
-        # Sucht den Anfang ({) und das Ende (}) des JSON-Objekts in der Antwort
-        # Das ist nötig, falls die KI zusätzlichen Text um das JSON herum schreibt
         start = antwort_roh.find("{")
         ende = antwort_roh.rfind("}") + 1
-
         if start != -1 and ende != -1:
-            # json.loads macht aus dem Text ein echtes Python-Dictionary (Objekt)
             return json.loads(antwort_roh[start:ende])
-
-        # Falls kein JSON gefunden wurde, wird die Nachricht als nicht relevant eingestuft
         return {"relevant": False}
     except Exception as e:
         print(f"  [!] KI-Fehler: {e}")
         return {"relevant": False}
 
-
 def sende_public(titel, nachricht, klick_url, category="info", priority=3):
-    """Sendet eine Push-Nachricht an den öffentlichen ntfy-Kanal."""
+    """Sendet Push-Nachricht an ntfy (Public)."""
-    if not NTFY_BASE:
+    if not NTFY_BASE: return
-        return
-
-    # Baut die Ziel-Adresse für ntfy zusammen
     url = f"{NTFY_BASE}/{TOPIC_PUBLIC}"
-
-    # Zuordnung von Kategorien zu Emojis für die Anzeige auf dem Handy
     emoji_mapping = {
         "demo": "rotating_light,loudspeaker",
         "socialmedia": "detective,no_entry",
         "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)}
-    # ntfy-Header: Metadaten der Nachricht (Titel, Icons, Klick-Aktion, Wichtigkeit)
-    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)
-        # POST-Anfrage sendet die Daten an den ntfy-Server
-        requests.post(
-            url,
-            data=nachricht.encode("utf-8"),
-            headers=headers,
-            auth=(ADMIN_USER, ADMIN_PW),
-            timeout=5,
-        )
     except Exception as e:
         print(f"  [!] Fehler Public-Versand: {e}")
 
-
 def sende_admin(titel, nachricht):
-    """Sendet Fehlermeldungen an den passwortgeschützten Admin-Kanal."""
+    """Sendet Fehlermeldung an ntfy (Admin)."""
-    if not NTFY_BASE:
+    if not NTFY_BASE: return
-        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=5,
-        )
     except Exception as e:
         print(f"  [!] Fehler Admin-Versand: {e}")
 
-
-def lade_gedaechtnis():
-    """Liest das results.json-File ein und gibt den Inhalt als Dictionary zurück."""
-    if not os.path.exists(JSON_DATEI):
-        return {}
-    with open(JSON_DATEI, "r", encoding="utf-8") as f:
-        try:
-            return json.load(f)
-        except:
-            return {}
-
-
-def speichere_gedaechtnis(daten):
-    """Speichert das aktuelle Wissen (daten) im JSON-Format auf die Festplatte."""
-    with open(JSON_DATEI, "w", encoding="utf-8") as f:
-        # indent=4 sorgt für eine lesbare Struktur (Einrückung) der Datei
-        json.dump(daten, f, indent=4, ensure_ascii=False)
-
-
 def extrahiere_links(soup, basis_url):
-    """Sucht alle <a>-Tags im HTML und macht daraus absolute Internetadressen."""
+    """Extrahiert alle absoluten Links aus einem HTML-Objekt."""
     links = []
-    # find_all('a') sucht alle Hyperlinks im HTML-Dokument
     for a in soup.find_all("a", href=True):
-        # urljoin kombiniert die Basis-URL (z.B. pnp.de) mit dem Link (z.B. /artikel1)
         links.append(urljoin(basis_url, a["href"]))
-    # list(set(links)) entfernt doppelte Einträge
     return list(set(links))
 
-
 def hole_artikel_text(url):
-    """Rufe eine Unterseite auf und extrahiert den Haupttext."""
+    """Lädt Unterseite und extrahiert Hauptinhalt."""
     try:
-        # Scraper um die Unterseite zu laden
         antwort = scraper.get(url, timeout=10)
-        # Verwandle die Antwort in durchsuchbares HTML
         unter_soup = BeautifulSoup(antwort.text, "html.parser")
-        # Suche nach Hauptbereich (wie im Haptscript)
+        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
-        )
-        # Gibt den reinen Text ohne Leerzeichen-Salat zurück
         return bereich.get_text(separator=" ", strip=True)
     except Exception as e:
-        # Fals etwas schiefgeht, gib einen Hinweis zurück
         return f"Fehler beim Laden des Artikels!: {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"])
+    return None
 
-def verarbeite_website(url, gedaechtnis):
+def verarbeite_website(url):
-    """Kern-Logik für eine einzelne Webseite."""
+    """Hauptlogik für eine Webseite unter Nutzung von SQL-Speicherung."""
     try:
-        # Lädt die Webseite mit dem Cloudscraper
         antwort = scraper.get(url, timeout=15)
         antwort.encoding = "utf-8"
-        status = antwort.status_code
+        if antwort.status_code != 200:
-
+            sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.")
-        # Bei Server-Fehlern (z.B. 404 oder 500) wird der Admin informiert
-        if status != 200:
-            print(f"  [!] Seite fehlerhaft: {url} (Status: {status})")
-            sende_admin("Fehler bei Webseite", f"{url} lieferte Status {status}.")
             return
 
-        # Verwandelt den rohen HTML-Text in ein BeautifulSoup-Objekt zum Suchen
         soup = BeautifulSoup(antwort.text, "html.parser")
-        # Sucht nach dem Inhaltsbereich (main, article oder body)
         hauptbereich = soup.find("main") or soup.find("article") or soup.body or soup
-        # Extrahiert den reinen Text ohne HTML-Tags
         neuer_text = hauptbereich.get_text(separator=" ", strip=True)
 
-        # Holt den alten Text dieser URL aus dem Gedächtnis
+        cursor.execute("SELECT inhalt FROM seiten_stand WHERE url = ?", (url,))
-        alter_eintrag = gedaechtnis.get(url, {})
+        zeile = cursor.fetchone()
-        alter_text = alter_eintrag.get("inhalt", "")
+        alter_text = zeile[0] if zeile else ""
-        bekannte_links = alter_eintrag.get("links", [])
 
-        # Sammelt alle Links aus dem Bereich für die KI
         aktuelle_links = extrahiere_links(hauptbereich, url)
         
-        # Prüft, ob sich der Text seit dem letzten Lauf verändert hat
+        # 3. Filtere neue Links (Eindeutigkeit garantieren)
-        if alter_text != neuer_text or aktuelle_links != bekannte_links:
+        neue_artikel_links = []
-            print(f"  [NEU] Änderung auf {url} gefunden. Ich frage die KI...")
+        for l in aktuelle_links:
-            # Zerlegt den Text in Zeilen
+            cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (l,))
+            if cursor.fetchone() is None and l not in neue_artikel_links:
+                neue_artikel_links.append(l)
+
+        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}")
+                artikel_inhalt = hole_artikel_text(link)
+                analyse = analysiere_mit_gemini(artikel_inhalt, link)
+                
+                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"))
+                
+                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()
-            # Findet nur die Zeilen, die neu dazugekommen sind (+ )
                 diff = [z[2:] for z in difflib.ndiff(alt_z, neu_z) if z.startswith("+ ")]
-            text_fuer_ki = "\n".join(diff)
                 
-            # Nur wenn auch wirklich neuer Text vorhanden ist
+                for line in diff:
-            if text_fuer_ki.strip():
+                    if len(line.strip()) < 20: continue
-                # --- NEU: Neue Links verfolgen und Artikeltexte abrufen ---
+                    passender_link = finde_link_fuer_text(line[:30], hauptbereich, url)
-                neue_links = [l for l in aktuelle_links if l not in bekannte_links]
-                # Fallback, falls Gemini den direct_link vergisst: Nimm den ersten neuen Link oder die Haupt-URL
-                backup_link = neue_links[0] if neue_links else url
                     
-                # Wir begrenzen das auf max. 3 neue Links, um Zeit und Tokens zu sparen
+                    target_url = passender_link if passender_link else url
-                for link in neue_links[:3]:
+                    cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (target_url,))
-                    print(f"  [..] Lade Artikelinhalt von: {link}")
+                    if cursor.fetchone() is None:
-                    artikel_inhalt = hole_artikel_text(link)
+                        analyse = analysiere_mit_gemini(line, target_url)
-                    # Wir hängen den Inhalt direkt an den Text für die KI an
+                        if analyse.get("relevant"):
-                    text_fuer_ki += f"\n\n--- ZUSÄTZLICHER ARTIKELKONTEXT ({link}) ---\n{artikel_inhalt}"
+                            sende_public(analyse.get("titel"), analyse.get("nachricht"), target_url, analyse.get("category"), analyse.get("priority"))
+                        cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (target_url, "erledigt"))
+                        conn.commit()
 
-                analyse = analysiere_mit_gemini(text_fuer_ki, aktuelle_links)
+            cursor.execute("INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", (url, neuer_text, str(datetime.now())))
-                # Falls die KI die Änderung als politisch wichtig einstuft
+            conn.commit()
-                if analyse.get("relevant") == True:
-                    print(f"  [!] KI bestätigt Relevanz: {analyse.get('titel')}")
-                    # Nutzt den direkten Link der KI oder den Backup-Link
-                    klick_url = analyse.get("direct_link", backup_link)
-                    sende_public(
-                        titel=analyse.get("titel", "Polit-Update"),
-                        nachricht=analyse.get("nachricht", "Neu auf der Seite"),
-                        klick_url=klick_url,
-                        category=analyse.get("category", "info"),
-                        priority=analyse.get("priority", 3),
-                    )
-
-            # Speichert den neuen Stand dieser Webseite im lokalen Gedächtnis
-            gedaechtnis[url] = {
-                "inhalt": neuer_text,
-                "links": aktuelle_links,
-                "zeit": str(datetime.now()),
-                "status": status,
-            }
         else:
             print(f"  [OK] Keine Änderungen auf {url}.")
 
@@ -265,19 +198,13 @@ def verarbeite_website(url, gedaechtnis):
         print(f"  ❌ Fehler bei {url}: {e}")
         sende_admin("Programmfehler", f"{url}: {e}")
 
-
 # --- HAUPTPROGRAMM ---
-# Startet den Prozess: Gedächtnis laden -> URLs prüfen -> Gedächtnis speichern
-gedaechtnis = lade_gedaechtnis()
 with open("./urls.txt", "r", encoding="utf-8") as datei:
     for zeile in datei:
-        # Entfernt Leerzeichen und ignoriert Kommentarzeilen (#)
         adresse = zeile.strip()
-        if not adresse or adresse.startswith("#"):
+        if not adresse or adresse.startswith("#"): continue
-            continue
         print(f"Ich prüfe: {adresse}")
-        verarbeite_website(adresse, gedaechtnis)
+        verarbeite_website(adresse)
 
-# Speichert das aktualisierte Wissen für den nächsten Cronjob-Lauf
+conn.close()
-speichere_gedaechtnis(gedaechtnis)
 print("\nFertig. Gute Nacht!")