import cloudscraper import json import os 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 # --- 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] 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" ) 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 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", } 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), } 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, ) 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 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, ) 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.""" 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.""" 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 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" 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", "") 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 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) # 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 # 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}" 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, } else: print(f" [OK] Keine Änderungen auf {url}.") except Exception as e: 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 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!")