Ratatoskr/polit-scraper
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

web_check.py: Implementierung der technischen Relevanz-Schranke (Threshold) und detailliertes Debug-Logging der Begründungen

Browse Source
This commit is contained in:
Ratatoskr
2026-03-05 21:56:20 +01:00
parent edea7f1319
commit a383d29478
1 changed files with 104 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files

176
web_check.py
View File

@@ -5,39 +5,19 @@ import sqlite3
import subprocess import subprocess
import difflib import difflib
import sys import sys
import re
from bs4 import BeautifulSoup from bs4 import BeautifulSoup
from urllib.parse import urljoin from urllib.parse import urljoin, urlparse
from datetime import datetime from datetime import datetime
from dotenv import load_dotenv 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 --- # --- KONFIGURATION LADEN ---
load_dotenv() load_dotenv()
ADMIN_USER = os.getenv("ADMIN_USER") ADMIN_USER = os.getenv("ADMIN_USER")
ADMIN_PW = os.getenv("ADMIN_PW") ADMIN_PW = os.getenv("ADMIN_PW")
NTFY_BASE = os.getenv("NTFY_BASE") 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_SESSION = "1" GEMINI_SESSION = "1"
TOPIC_PUBLIC = "polit-scraper-public" TOPIC_PUBLIC = "polit-scraper-public"
@@ -49,25 +29,51 @@ scraper = cloudscraper.create_scraper(
browser={"browser": "chrome", "platform": "windows", "desktop": True} browser={"browser": "chrome", "platform": "windows", "desktop": True}
) )
def analysiere_mit_gemini(text, link): def debug(nachricht):
"""KI-Analyse: Sendet Text und den EINEN passenden Link an Gemini (Isoliert).""" if DEBUG_LOG:
anweisung = f"WICHTIG: Analysiere NUR den folgenden Text. Nutze für das Feld 'direct_link' exakt diese URL: {link}\n\nTEXT:\n{text}" 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 Panda-Regeln (V8).\n\nTEXT:\n{sicherer_text}"
befehl = ["gemini", "-m", "gemini-2.5-flash", "-r", GEMINI_SESSION, "-p", anweisung] befehl = ["gemini", "-m", "gemini-2.5-flash", "-r", GEMINI_SESSION, "-p", anweisung]
try: try:
prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8") prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8", timeout=300)
antwort_roh = prozess.stdout.strip() antwort_roh = prozess.stdout.strip()
start = antwort_roh.find("{") start = antwort_roh.find("{")
ende = antwort_roh.rfind("}") + 1 ende = antwort_roh.rfind("}") + 1
if start != -1 and ende != -1: if start != -1 and ende != -1:
return json.loads(antwort_roh[start:ende]) return json.loads(antwort_roh[start:ende])
return {"relevant": False} return {"relevant": False, "relevanz_score": 0, "begruendung_score": "Fehler: Kein JSON gefunden"}
except Exception as e: except Exception as e:
print(f" [!] KI-Fehler: {e}") print(f" [!] KI-Fehler: {e}")
return {"relevant": False} return {"relevant": False, "relevanz_score": 0, "begruendung_score": f"Systemfehler: {e}"}
def sende_public(titel, nachricht, klick_url, category="info", priority=3): def sende_public(titel, nachricht, klick_url, category="info", priority=3):
"""Sendet Push-Nachricht an ntfy (Public).""" """Sendet Push-Nachricht an ntfy."""
if not NTFY_BASE: return if not NTFY_BASE: return
url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" url = f"{NTFY_BASE}/{TOPIC_PUBLIC}"
emoji_mapping = { emoji_mapping = {
@@ -82,56 +88,63 @@ def sende_public(titel, nachricht, klick_url, category="info", priority=3):
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: try:
import requests 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=10)
except Exception as e: except Exception as e:
print(f" [!] Fehler Public-Versand: {e}") print(f" [!] Fehler Public-Versand: {e}")
def sende_admin(titel, nachricht): def sende_admin(titel, nachricht):
"""Sendet Fehlermeldung an ntfy (Admin)."""
if not NTFY_BASE: return if not NTFY_BASE: return
url = f"{NTFY_BASE}/{TOPIC_ADMIN}" 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: try:
import requests 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=10)
except Exception as e: except Exception as e:
print(f" [!] Fehler Admin-Versand: {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): def extrahiere_links(soup, basis_url):
"""Extrahiert alle absoluten Links aus einem HTML-Objekt."""
links = [] 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): for a in soup.find_all("a", href=True):
links.append(urljoin(basis_url, a["href"])) 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)) return list(set(links))
def hole_artikel_text(url): def hole_artikel_text(url):
"""Lädt Unterseite und extrahiert Hauptinhalt."""
try: try:
antwort = scraper.get(url, timeout=10) antwort = scraper.get(url, timeout=10)
unter_soup = BeautifulSoup(antwort.text, "html.parser") 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
return bereich.get_text(separator=" ", strip=True) text = bereich.get_text(separator=" ", strip=True)
if len(text) > 12000: return text[:12000]
return text
except Exception as e: except Exception as e:
return f"Fehler beim Laden des Artikels!: {e}" return f"Fehler beim Laden: {e}"
def finde_link_fuer_text(text_snippet, soup, basis_url): def finde_link_fuer_text(text_snippet, soup, basis_url):
"""Sucht im HTML nach einem Link in der Nähe des Textes.""" schnipsel = text_snippet.strip().lower()
for element in soup.find_all(string=True): if len(schnipsel) < 10: return None
if text_snippet in element: for a in soup.find_all("a", href=True):
parent = element.find_parent("a") link_text = a.get_text(separator=" ", strip=True).lower()
if parent and parent.has_attr("href"): if schnipsel in link_text or link_text in schnipsel:
return urljoin(basis_url, parent["href"]) return bereinige_url(a["href"], basis_url)
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 return None
def verarbeite_website(url): def verarbeite_website(url):
"""Hauptlogik für eine Webseite unter Nutzung von SQL-Speicherung.""" """Hauptlogik mit Scoring-Logik und detaillierter Begründung im Debug-Modus."""
try: try:
antwort = scraper.get(url, timeout=15) antwort = scraper.get(url, timeout=15)
antwort.encoding = "utf-8"
if antwort.status_code != 200: 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 return
@@ -146,7 +159,6 @@ def verarbeite_website(url):
aktuelle_links = extrahiere_links(hauptbereich, url) aktuelle_links = extrahiere_links(hauptbereich, url)
# 3. Filtere neue Links (Eindeutigkeit garantieren)
neue_artikel_links = [] neue_artikel_links = []
for l in aktuelle_links: for l in aktuelle_links:
cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (l,)) cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (l,))
@@ -156,47 +168,67 @@ def verarbeite_website(url):
if neue_artikel_links or (alter_text != neuer_text): if neue_artikel_links or (alter_text != neuer_text):
print(f" [NEU] Aktivität auf {url} erkannt.") print(f" [NEU] Aktivität auf {url} erkannt.")
# FALL A: Neue Deep-Links vorhanden # FALL A: Neue Deep-Links
for link in neue_artikel_links[:3]: for link in neue_artikel_links[:5]:
print(f" [..] Bearbeite Deep-Link: {link}") print(f" [..] Analysiere neuen Artikel: {link}")
artikel_inhalt = hole_artikel_text(link) artikel_inhalt = hole_artikel_text(link)
analyse = analysiere_mit_gemini(artikel_inhalt, link) analyse = analysiere_mit_gemini(artikel_inhalt)
if analyse.get("relevant"): score = analyse.get("relevanz_score", 0)
final_url = analyse.get("direct_link", link) begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden")
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")) if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE:
print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}")
debug(f"Begründung: {begruendung}")
sende_public(
titel=analyse.get("titel", "Polit-Update"),
nachricht=analyse.get("nachricht", "Neu auf der Seite"),
klick_url=link,
category=analyse.get("category", "info"),
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")) cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (link, "erledigt"))
conn.commit() conn.commit()
# FALL B: Nur Textänderung auf Hauptseite (Filter gegen Duplikate) # FALL B: Textänderung auf Hauptseite
if not neue_artikel_links and (alter_text != neuer_text): if alter_text != neuer_text:
alt_z = alter_text.splitlines() diff = [z[2:] for z in difflib.ndiff(alter_text.splitlines(), neuer_text.splitlines()) if z.startswith("+ ")]
neu_z = neuer_text.splitlines()
diff = [z[2:] for z in difflib.ndiff(alt_z, neu_z) if z.startswith("+ ")]
for line in diff: for line in diff:
if len(line.strip()) < 20: continue if len(line.strip()) < 20: continue
passender_link = finde_link_fuer_text(line[:30], hauptbereich, url) passender_link = finde_link_fuer_text(line[:40], hauptbereich, url)
target_url = passender_link if passender_link else 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 cursor.fetchone() is None:
analyse = analysiere_mit_gemini(line, target_url) if passender_link:
if analyse.get("relevant"): 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")
if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE:
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")) cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (target_url, "erledigt"))
conn.commit() 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() conn.commit()
else: else:
print(f" [OK] Keine Änderungen auf {url}.") debug(f"[OK] Keine Änderungen auf {url}.")
except Exception as e: except Exception as e:
print(f" ❌ Fehler bei {url}: {e}") print(f" ❌ Fehler bei {url}: {e}")
sende_admin("Programmfehler", f"{url}: {e}")
# --- HAUPTPROGRAMM --- # --- HAUPTPROGRAMM ---
with open("./urls.txt", "r", encoding="utf-8") as datei: with open("./urls.txt", "r", encoding="utf-8") as datei: