diff --git a/web_check.py b/web_check.py index 7e7e436..fc5b76c 100644 --- a/web_check.py +++ b/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, links): - """KI-Analyse: Sendet den Text und die gefundenen Links an Gemini.""" - - # 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] - +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}" + befehl = ["gemini", "-m", "gemini-2.5-flash", "-r", GEMINI_SESSION, "-p", anweisung] + try: - # Führt den Terminal-Befehl aus und speichert die Antwort - # 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" - ) + 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.""" - if not NTFY_BASE: - return - - # Baut die Ziel-Adresse für ntfy zusammen + """Sendet Push-Nachricht an ntfy (Public).""" + if not NTFY_BASE: return 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") - - # ntfy-Header: Metadaten der Nachricht (Titel, Icons, Klick-Aktion, Wichtigkeit) - 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 - - # 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, - ) + 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.""" - if not NTFY_BASE: - return + """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=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 -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 - ) - # Gibt den reinen Text ohne Leerzeichen-Salat zurück + bereich = unter_soup.find("main") or unter_soup.find("article") or unter_soup.body 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): - """Kern-Logik für eine einzelne Webseite.""" +def verarbeite_website(url): + """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 - - # 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}.") + if antwort.status_code != 200: + sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.") 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 - alter_eintrag = gedaechtnis.get(url, {}) - alter_text = alter_eintrag.get("inhalt", "") - bekannte_links = alter_eintrag.get("links", []) + cursor.execute("SELECT inhalt FROM seiten_stand WHERE url = ?", (url,)) + zeile = cursor.fetchone() + alter_text = zeile[0] if zeile else "" - # Sammelt alle Links aus dem Bereich für die KI 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,)) + if cursor.fetchone() is None and l not in neue_artikel_links: + neue_artikel_links.append(l) - # Prüft, ob sich der Text seit dem letzten Lauf verändert hat - if alter_text != neuer_text or aktuelle_links != bekannte_links: - print(f" [NEU] Änderung auf {url} gefunden. Ich frage die KI...") - # Zerlegt den Text in Zeilen - 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) + if neue_artikel_links or (alter_text != neuer_text): + print(f" [NEU] Aktivität auf {url} erkannt.") - # Nur wenn auch wirklich neuer Text vorhanden ist - if text_fuer_ki.strip(): - # --- NEU: Neue Links verfolgen und Artikeltexte abrufen --- - 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 + # 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() - # Wir begrenzen das auf max. 3 neue Links, um Zeit und Tokens zu sparen - for link in neue_links[:3]: - print(f" [..] Lade Artikelinhalt von: {link}") - artikel_inhalt = hole_artikel_text(link) - # Wir hängen den Inhalt direkt an den Text für die KI an - text_fuer_ki += f"\n\n--- ZUSÄTZLICHER ARTIKELKONTEXT ({link}) ---\n{artikel_inhalt}" + # 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("+ ")] + + for line in diff: + if len(line.strip()) < 20: continue + passender_link = finde_link_fuer_text(line[:30], 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"): + 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) - # Falls die KI die Änderung als politisch wichtig einstuft - 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, - } + 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}.") @@ -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("#"): - continue + if not adresse or adresse.startswith("#"): 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 -speichere_gedaechtnis(gedaechtnis) +conn.close() print("\nFertig. Gute Nacht!")