From 7d5f75813b3d1ad5814bc23a0e05c85c94fbed0e Mon Sep 17 00:00:00 2001 From: Ratatoskr Date: Wed, 25 Mar 2026 12:33:02 +0100 Subject: [PATCH] =?UTF-8?q?GEMINI=5FMODEL=20hinzugef=C3=BCgt,=20zum=20schn?= =?UTF-8?q?elleren=20wechseln?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- web_check.py | 172 ++++++++++++--------------------------------------- 1 file changed, 41 insertions(+), 131 deletions(-) diff --git a/web_check.py b/web_check.py index 7bcbc40..c53c3cf 100644 --- a/web_check.py +++ b/web_check.py @@ -10,6 +10,7 @@ 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() @@ -18,6 +19,7 @@ ADMIN_PW = os.getenv("ADMIN_PW") NTFY_BASE = os.getenv("NTFY_BASE") DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true" MIN_RELEVANZ_SCORE = int(os.getenv("MIN_RELEVANZ_SCORE", "30")) +GEMINI_MODEL "gemini-3.1-flash-lite-preview" GEMINI_SESSION = "1" TOPIC_PUBLIC = "polit-scraper-public" @@ -29,13 +31,11 @@ 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() @@ -56,48 +56,27 @@ CREATE TABLE IF NOT EXISTS artikel ( """) 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 Panda-Regeln (V8).\n\nTEXT:\n{sicherer_text}" - befehl = [ - "gemini", - "-m", - "gemini-3.1-flash-lite-preview", - "-r", - GEMINI_SESSION, - "-p", - anweisung, - ] - + 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 - ) + prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8", timeout=300) 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, - "relevanz_score": 0, - "begruendung_score": "Fehler: Kein JSON gefunden", - } + return {"relevant": False, "relevanz_score": 0, "begruendung_score": "Fehler: Kein JSON gefunden"} except Exception as e: print(f" [!] KI-Fehler: {e}") - return { - "relevant": False, - "relevanz_score": 0, - "begruendung_score": f"Systemfehler: {e}", - } - + return {"relevant": False, "relevanz_score": 0, "begruendung_score": f"Systemfehler: {e}"} def sende_public(titel, nachricht, klick_url, category="info", priority=3): """Sendet Push-Nachricht an ntfy.""" - if not NTFY_BASE: - return + if not NTFY_BASE: return url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" emoji_mapping = { "demo": "rotating_light,loudspeaker", @@ -105,55 +84,32 @@ def sende_public(titel, nachricht, klick_url, category="info", priority=3): "polizei": "police_car,warning", "stadt": "cityscape,newspaper", "solidaritaet": "fist,heart", - "info": "newspaper", + "info": "newspaper" } tags = emoji_mapping.get(category.lower(), "newspaper") - 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 - - requests.post( - url, - data=nachricht.encode("utf-8"), - headers=headers, - auth=(ADMIN_USER, ADMIN_PW), - timeout=10, - ) + 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 + 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, - ) + 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() + 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") @@ -161,48 +117,38 @@ def extrahiere_links(soup, basis_url): 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 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 - ) + 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] + 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 + 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}." - ) + sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.") return soup = BeautifulSoup(antwort.text, "html.parser") @@ -214,7 +160,7 @@ def verarbeite_website(url): 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,)) @@ -229,12 +175,10 @@ def verarbeite_website(url): print(f" [..] Analysiere neuen Artikel: {link}") artikel_inhalt = hole_artikel_text(link) analyse = analysiere_mit_gemini(artikel_inhalt) - + score = analyse.get("relevanz_score", 0) - begruendung = analyse.get( - "begruendung_score", "Keine Begründung vorhanden" - ) - + begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden") + if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}") debug(f"Begründung: {begruendung}") @@ -243,76 +187,44 @@ def verarbeite_website(url): nachricht=analyse.get("nachricht", "Neu auf der Seite"), klick_url=link, category=analyse.get("category", "info"), - priority=analyse.get("priority", 3), + priority=analyse.get("priority", 3) ) 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"), - ) + 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("+ ") - ] + 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 + 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,) - ) + + 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) - + score = analyse.get("relevanz_score", 0) - begruendung = analyse.get( - "begruendung_score", "Keine Begründung vorhanden" - ) - + begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden") + if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: - print( - f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}" - ) + print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}") debug(f"Begründung: {begruendung}") - sende_public( - analyse.get("titel"), - analyse.get("nachricht"), - target_url, - analyse.get("category"), - analyse.get("priority"), - ) + sende_public(analyse.get("titel"), analyse.get("nachricht"), target_url, analyse.get("category"), analyse.get("priority")) 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"), - ) + 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())), - ) + 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}.") @@ -320,13 +232,11 @@ def verarbeite_website(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 + if not adresse or adresse.startswith("#"): continue print(f"Ich prüfe: {adresse}") verarbeite_website(adresse)