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 # --- 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") GEMINI_SESSION = "1" TOPIC_PUBLIC = "polit-scraper-public" TOPIC_ADMIN = "polit-scraper-admin" sys.stdout.reconfigure(encoding="utf-8") 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}" 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") 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} except Exception as e: print(f" [!] KI-Fehler: {e}") return {"relevant": False} def sende_public(titel, nachricht, klick_url, category="info", priority=3): """Sendet Push-Nachricht an ntfy (Public).""" if not NTFY_BASE: return url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" 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") 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) 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) except Exception as e: print(f" [!] Fehler Admin-Versand: {e}") def extrahiere_links(soup, basis_url): """Extrahiert alle absoluten Links aus einem HTML-Objekt.""" links = [] for a in soup.find_all("a", href=True): links.append(urljoin(basis_url, a["href"])) 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) except Exception as e: 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): """Hauptlogik für eine Webseite unter Nutzung von SQL-Speicherung.""" 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 soup = BeautifulSoup(antwort.text, "html.parser") hauptbereich = soup.find("main") or soup.find("article") or soup.body or soup neuer_text = hauptbereich.get_text(separator=" ", strip=True) cursor.execute("SELECT inhalt FROM seiten_stand WHERE url = ?", (url,)) zeile = cursor.fetchone() alter_text = zeile[0] if zeile else "" 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) 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() 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() 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}.") 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: for zeile in datei: adresse = zeile.strip() if not adresse or adresse.startswith("#"): continue print(f"Ich prüfe: {adresse}") verarbeite_website(adresse) conn.close() print("\nFertig. Gute Nacht!")