diff --git a/web_check.py b/web_check.py deleted file mode 100644 index 4556768..0000000 --- a/web_check.py +++ /dev/null @@ -1,466 +0,0 @@ -import cloudscraper -import json -import os -import sqlite3 -import subprocess -import difflib -import sys -import re -from bs4 import BeautifulSoup -from urllib.parse import urljoin, urlparse -from datetime import datetime -from dotenv import load_dotenv -from requests import models - -# --- KONFIGURATION LADEN --- -load_dotenv() -ADMIN_USER = os.getenv("ADMIN_USER") -ADMIN_PW = os.getenv("ADMIN_PW") -NTFY_BASE = os.getenv("NTFY_BASE") -SIGNAL_NUMBER = os.getenv("SIGNAL_NUMBER") -SIGNAL_GROUPE_ID = os.getenv("SIGNAL_GROUPE_ID") -DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true" -MIN_RELEVANZ_SCORE = int(os.getenv("MIN_RELEVANZ_SCORE", "30")) -GEMINI_MODEL = "gemini-2.5-flash-lite" -GEMINI_SESSION = "1" -NTFY = "true" -SIGNAL = "true" -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 debug(nachricht): - if DEBUG_LOG: - zeit = datetime.now().strftime("%H:%M:%S") - print(f" [DEBUG {zeit}] {nachricht}") - - -# --- DATENBANK SETUP --- -conn = sqlite3.connect("polit_scraper.db", timeout=20) -cursor = conn.cursor() - -cursor.execute(""" -CREATE TABLE IF NOT EXISTS seiten_stand ( - url TEXT PRIMARY KEY, - inhalt TEXT, - zeit TEXT -) -""") - -cursor.execute(""" -CREATE TABLE IF NOT EXISTS artikel ( - artikel_url TEXT PRIMARY KEY, - status TEXT -) -""") -conn.commit() - - -def analysiere_mit_gemini(text): - """KI-Analyse: Bewertet NUR den Text laut V8 (inkl. Begründung).""" - sicherer_text = text[:5000] - anweisung = f"WICHTIG: Analysiere NUR den folgenden Text laut deinen Regeln @gemini_instructions.md .\n\nTEXT:\n{sicherer_text}" - befehl = ["gemini", "-m", GEMINI_MODEL, "-r", GEMINI_SESSION, "-p", anweisung] - - try: - prozess = subprocess.run( - befehl, capture_output=True, text=True, encoding="utf-8", timeout=300 - ) - antwort_roh = prozess.stdout.strip() - if DEBUG_LOG: - # Logge die rohe Ausgabe vom gemini CLI, um das Problem einzugrenzen - debug(f"Raw KI output: {antwort_roh}") - - start = antwort_roh.find("{") - ende = antwort_roh.rfind("}") + 1 - if start != -1 and ende != -1: - parsed_json = json.loads(antwort_roh[start:ende]) - if DEBUG_LOG: - # Logge das geparste JSON-Dictionary - debug(f"Parsed KI result: {parsed_json}") - return parsed_json - else: - # Falls keine gültige JSON-Struktur gefunden wird, logge dies und gib ein Fehler-Dictionary zurück - if DEBUG_LOG: - debug( - f"No valid JSON structure found in KI output. Raw output: {antwort_roh}" - ) - return { - "relevant": False, - "relevanz_score": 0, - "begruendung_score": "Fehler: Kein JSON gefunden", - } - except Exception as e: - print(f" [!] KI-Fehler: {e}") - if DEBUG_LOG: - # Logge die Exception für Debugging - debug(f"Exception during KI analysis: {e}") - return { - "relevant": False, - "relevanz_score": 0, - "begruendung_score": f"Systemfehler: {e}", - } - - -def sende_signal(titel, nachricht, klick_url, signal_number, signal_group_id): - """Sendet eine Benachrichtigung über Signal direkt vom lokalen System.""" - if not signal_number or not signal_group_id: - print( - " [!] Signal-Benachrichtigung kann nicht gesendet werden: SIGNAL_NUMBER oder SIGNAL_GROUPE_ID nicht gesetzt." - ) - return - - # Nachricht formatieren: Titel in der ersten Zeile, Nachricht in der nächsten - message_content = f"{titel}\n{nachricht}\n{klick_url}" - - # Befehl für die direkte Ausführung lokal - # Der Config-Pfad muss hier dem Pfad entsprechen, den Sie bei der Registrierung verwendet haben - cmd = [ - "signal-cli", - "--config", - "/root/.local/share/signal-cli", - "-u", - signal_number, - "send", - "-m", - message_content, - "-g", - signal_group_id, - ] - - debug(f"Executing Signal command locally: {' '.join(cmd)}") - - try: - process_result = subprocess.run( - cmd, capture_output=True, text=True, encoding="utf-8", timeout=30 - ) - - if process_result.returncode == 0: - print(" [SUCCESS] Signal-Nachricht erfolgreich gesendet.") - else: - print( - f" [!] Fehler beim Senden der Signal-Nachricht: {process_result.stderr}" - ) - sende_admin( - "Signal Sende-Fehler", - f"Fehler beim Senden der Signal-Nachricht: {process_result.stderr}", - ) - except Exception as e: - print(f" [!] Unerwarteter Fehler beim Senden der Signal-Nachricht: {e}") - - -def sende_ntfy(titel, nachricht, klick_url, tags, priority=3): - """Sendet Push-Nachricht an ntfy.""" - if NTFY == "true": - if not NTFY_BASE: - return - url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" - headers = { - "Title": titel.encode("utf-8"), - "Tags": tags.encode("utf-8"), - "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=10, - ) - except Exception as e: - print(f" [!] Fehler Public-Versand: {e}") - - -def sende_admin(titel, nachricht): - 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=10, - ) - except Exception as e: - print(f" [!] Fehler Admin-Versand: {e}") - - -def bereinige_url(url_roh, basis_url): - clean = url_roh.replace("”", "").replace('"', "").replace("'", "").strip() - if clean.lower().startswith("http"): - return clean - return urljoin(basis_url, clean) - - -def extrahiere_links(soup, basis_url): - links = [] - ignore_ext = (".pdf", ".jpg", ".jpeg", ".png", ".gif", ".zip", ".docx") - basis_domain = urlparse(basis_url).netloc - - for a in soup.find_all("a", href=True): - full_url = bereinige_url(a["href"], basis_url) - if full_url.lower().endswith(ignore_ext): - continue - if urlparse(full_url).netloc == basis_domain: - if len(full_url) > len(basis_url.rstrip("/")) + 1: - links.append(full_url) - return list(set(links)) - - -def hole_artikel_text(url): - 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 - ) - text = bereich.get_text(separator=" ", strip=True) - if len(text) > 12000: - return text[:12000] - return text - except Exception as e: - return f"Fehler beim Laden: {e}" - - -def finde_link_fuer_text(text_snippet, soup, basis_url): - schnipsel = text_snippet.strip().lower() - if len(schnipsel) < 10: - return None - for a in soup.find_all("a", href=True): - link_text = a.get_text(separator=" ", strip=True).lower() - if schnipsel in link_text or link_text in schnipsel: - return bereinige_url(a["href"], basis_url) - return None - - -def verarbeite_website(url): - """Hauptlogik mit Scoring-Logik und detaillierter Begründung im Debug-Modus.""" - try: - antwort = scraper.get(url, timeout=15) - 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) - - 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 - for link in neue_artikel_links[:5]: - print(f" [..] Analysiere neuen Artikel: {link}") - artikel_inhalt = hole_artikel_text(link) - analyse = analysiere_mit_gemini(artikel_inhalt) - - # Prüfen, ob die KI-Analyse einen kritischen Fehler meldet - # Ein Fehler wird angenommen, wenn 'begruendung_score' mit "Fehler:" beginnt - # oder wenn der Score 0 ist und eine Fehlermeldung im 'begruendung_score' vorhanden ist. - is_analysis_error = analyse.get("begruendung_score", "").startswith( - "Fehler:" - ) or ( - analyse.get("relevanz_score", 0) == 0 - and "Fehler:" in analyse.get("begruendung_score", "") - ) - - if is_analysis_error: - print( - f" [!] Analyse fehlgeschlagen für {link}: {analyse.get('begruendung_score')}" - ) - debug( - f"Skipping article processing and not marking as processed due to analysis error: {link}" - ) - continue # Nächsten Artikel im Loop bearbeiten, diesen nicht als erledigt markieren - - # Wenn die Analyse erfolgreich war (auch wenn der Score niedrig ist): - score = analyse.get("relevanz_score", 0) - begruendung = analyse.get( - "begruendung_score", "Keine Begründung vorhanden" - ) - - if score >= MIN_RELEVANZ_SCORE: - if DEBUG_LOG: - debug( - f"KI-Analyse-Ergebnis (JSON): {json.dumps(analyse, indent=2, ensure_ascii=False)}" - ) - - # Sende NTFY Benachrichtigung, wenn konfiguriert - if NTFY == "true": - sende_ntfy( - titel=analyse.get("titel", "Polit-Update"), - nachricht=analyse.get("nachricht", "Neu auf der Seite"), - klick_url=link, - priority=analyse.get("priority", 3), - tags=analyse.get("tags", "panda_face"), - ) - - # Sende Signal Benachrichtigung, wenn konfiguriert - if SIGNAL == "true": - sende_signal( - titel=analyse.get("titel", "Polit-Update"), - nachricht=analyse.get("nachricht", "Neu auf der Seite"), - klick_url=link, - signal_number=SIGNAL_NUMBER, - signal_group_id=SIGNAL_GROUPE_ID, - ) - - else: - debug( - f"Überspringe Artikel (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}" - ) - - cursor.execute( - "INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", - (link, "erledigt"), - ) - conn.commit() - - # FALL B: Textänderung auf Hauptseite - if alter_text != neuer_text: - diff = [ - z[2:] - for z in difflib.ndiff( - alter_text.splitlines(), neuer_text.splitlines() - ) - if z.startswith("+ ") - ] - for line in diff: - if len(line.strip()) < 20: - continue - passender_link = finde_link_fuer_text(line[:40], 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: - if passender_link: - artikel_inhalt = hole_artikel_text(passender_link) - analyse = analysiere_mit_gemini(artikel_inhalt) - else: - analyse = analysiere_mit_gemini(line) - - # Prüfen, ob die KI-Analyse einen kritischen Fehler meldet - is_analysis_error = analyse.get( - "begruendung_score", "" - ).startswith("Fehler:") or ( - analyse.get("relevanz_score", 0) == 0 - and "Fehler:" in analyse.get("begruendung_score", "") - ) - - if is_analysis_error: - print( - f" [!] Analyse fehlgeschlagen für {target_url}: {analyse.get('begruendung_score')}" - ) - debug( - f"Skipping article processing and not marking as processed due to analysis error: {target_url}" - ) - continue # Nächste Zeile im Diff bearbeiten, diesen nicht als erledigt markieren - - # Wenn die Analyse erfolgreich war (auch wenn der Score niedrig ist): - score = analyse.get("relevanz_score", 0) - begruendung = analyse.get( - "begruendung_score", "Keine Begründung vorhanden" - ) - - if score >= MIN_RELEVANZ_SCORE: - print( - f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}" - ) - if DEBUG_LOG: - debug( - f"KI-Analyse-Ergebnis (JSON): {json.dumps(analyse, indent=2, ensure_ascii=False)}" - ) - - # Sende NTFY Benachrichtigung, wenn konfiguriert - if NTFY == "true": - sende_ntfy( - titel=analyse.get("titel", "Polit-Update"), - nachricht=analyse.get( - "nachricht", "Neu auf der Seite" - ), - klick_url=target_url, - tags=analyse.get("tags"), - priority=analyse.get("priority"), - ) - - # Sende Signal Benachrichtigung, wenn konfiguriert - if SIGNAL == "true": - sende_signal( - titel=analyse.get("titel", "Polit-Update"), - nachricht=analyse.get( - "nachricht", "Neu auf der Seite" - ), - klick_url=link, - signal_number=SIGNAL_NUMBER, - signal_group_id=SIGNAL_GROUPE_ID, - ) - else: - debug( - f"Überspringe Änderung (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}" - ) - - 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: - debug(f"[OK] Keine Änderungen auf {url}.") - - except Exception as e: - print(f" ❌ Fehler bei {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(" ") - print( - "----------------------------------------------------------------------------------------------------" - ) - print(" ") - print(f"Ich prüfe: {adresse}") - verarbeite_website(adresse) - -conn.close() -print("\nBleibt wachsam und solidarisch! Bis zum nächsten Lauf.")