From 92fd25239c09de5879ecef7150b2dcbab183c42f Mon Sep 17 00:00:00 2001 From: Michaelis Date: Mon, 2 Mar 2026 18:51:13 +0000 Subject: [PATCH] aktualisiert --- web_check.py | 182 ++++++++++++++++++++++++++++++++++++++++----------- 1 file changed, 144 insertions(+), 38 deletions(-) diff --git a/web_check.py b/web_check.py index 63b8ff1..e4f192c 100644 --- a/web_check.py +++ b/web_check.py @@ -7,142 +7,243 @@ import sys from bs4 import BeautifulSoup from urllib.parse import urljoin from datetime import datetime -from dotenv import load_dotenv +from dotenv import load_dotenv, main -# --- KONFIGURATION --- -# Laden der Umgebungsvariablen aus .env Datei +# --- 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" -# ntfy-Themen (Öffentlich) +# Namen der ntfy-Topics für öffentliche News und Admin-Fehler TOPIC_PUBLIC = "polit-scraper-public" TOPIC_ADMIN = "polit-scraper-admin" -# UTF-8 für Konsole erzwingen -sys.stdout.reconfigure(encoding='utf-8') +# Stellt sicher, dass das Skript Umlaute (ä, ö, ü) korrekt im Terminal ausgibt +sys.stdout.reconfigure(encoding="utf-8") -# Cloudscraper initialisieren (Chrome-Emulation) +# 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} + browser={"browser": "chrome", "platform": "windows", "desktop": True} ) + def analysiere_mit_gemini(text, links): - """KI-Analyse: Text und Links an Gemini senden, JSON extrahieren.""" + """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] + try: - prozess = subprocess.run(befehl, capture_output=True, text=True, encoding='utf-8') + # 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" + ) antwort_roh = prozess.stdout.strip() - start = antwort_roh.find('{') - ende = antwort_roh.rfind('}') + 1 + + # 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): - """Öffentlicher ntfy-Versand mit Priorität und Klick-URL.""" - if not NTFY_BASE: return + """Sendet eine Push-Nachricht an den öffentlichen ntfy-Kanal.""" + if not NTFY_BASE: + 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") + + # ntfy-Header: Metadaten der Nachricht (Titel, Icons, Klick-Aktion, Wichtigkeit) headers = { - "Title": titel.encode('utf-8'), + "Title": titel.encode("utf-8"), "Tags": tags, "Click": klick_url, - "Priority": str(priority) + "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): - """Admin-Benachrichtigung bei technischen Fehlern.""" - if not NTFY_BASE: return + """Sendet Fehlermeldungen an den passwortgeschützten Admin-Kanal.""" + if not NTFY_BASE: + return url = f"{NTFY_BASE}/{TOPIC_ADMIN}" - headers = {"Title": titel.encode('utf-8'), "Tags": "warning,skull", "Priority": "4"} + 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(): - 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 {} + """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): - with open(JSON_DATEI, "w", encoding='utf-8') as f: + """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.""" links = [] - for a in soup.find_all('a', href=True): - links.append(urljoin(basis_url, a['href'])) + # 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 verarbeite_website(url, gedaechtnis): + +def hole_artikel_text(url): + """Rufe eine Unterseite auf und extrahiert den Haupttext.""" try: + # Scraper um die Unterseite zu laden + antwort = scraper.get(url, timeout=10) + # Verwandle die Antwort in durchsuchbares HTML + unter_soup = BeautifulSoup(antwort.text, "htpm.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 + 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 verarbeite_website(url, gedaechtnis): + """Kern-Logik für eine einzelne Webseite.""" + try: + # Lädt die Webseite mit dem Cloudscraper antwort = scraper.get(url, timeout=15) - antwort.encoding = 'utf-8' + 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}.") 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", "") + # Prüft, ob sich der Text seit dem letzten Lauf verändert hat if alter_text != neuer_text: 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) + + # Sammelt alle Links aus dem Bereich für die KI alle_links = extrahiere_links(hauptbereich, url) + # Nur wenn auch wirklich neuer Text vorhanden ist if text_fuer_ki.strip(): analyse = analysiere_mit_gemini(text_fuer_ki, alle_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 die Haupt-URL klick_url = analyse.get("direct_link", url) 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) + priority=analyse.get("priority", 3), ) - - gedaechtnis[url] = {"inhalt": neuer_text, "zeit": str(datetime.now()), "status": status} + + # Speichert den neuen Stand dieser Webseite im lokalen Gedächtnis + gedaechtnis[url] = { + "inhalt": neuer_text, + "zeit": str(datetime.now()), + "status": status, + } else: print(f" [OK] Keine Änderungen auf {url}.") @@ -150,14 +251,19 @@ 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: +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) +# Speichert das aktualisierte Wissen für den nächsten Cronjob-Lauf speichere_gedaechtnis(gedaechtnis) print("\nFertig. Gute Nacht!")