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

GEMINI_MODEL hinzugefügt, zum schnelleren wechseln

Browse Source
This commit is contained in:
Ratatoskr
2026-03-25 12:33:02 +01:00
parent 570d3dc3d6
commit 7d5f75813b
1 changed files with 41 additions and 131 deletions
Download Patch File Download Diff File Expand all files Collapse all files

172
web_check.py
View File

@@ -10,6 +10,7 @@ from bs4 import BeautifulSoup
from urllib.parse import urljoin, urlparse from urllib.parse import urljoin, urlparse
from datetime import datetime from datetime import datetime
from dotenv import load_dotenv from dotenv import load_dotenv
from requests import models
# --- KONFIGURATION LADEN --- # --- KONFIGURATION LADEN ---
load_dotenv() load_dotenv()
@@ -18,6 +19,7 @@ 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" DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true"
MIN_RELEVANZ_SCORE = int(os.getenv("MIN_RELEVANZ_SCORE", "30")) MIN_RELEVANZ_SCORE = int(os.getenv("MIN_RELEVANZ_SCORE", "30"))
GEMINI_MODEL "gemini-3.1-flash-lite-preview"
GEMINI_SESSION = "1" GEMINI_SESSION = "1"
TOPIC_PUBLIC = "polit-scraper-public" TOPIC_PUBLIC = "polit-scraper-public"
@@ -29,13 +31,11 @@ scraper = cloudscraper.create_scraper(
browser={"browser": "chrome", "platform": "windows", "desktop": True} browser={"browser": "chrome", "platform": "windows", "desktop": True}
) )
def debug(nachricht): def debug(nachricht):
if DEBUG_LOG: if DEBUG_LOG:
zeit = datetime.now().strftime("%H:%M:%S") zeit = datetime.now().strftime("%H:%M:%S")
print(f" [DEBUG {zeit}] {nachricht}") print(f" [DEBUG {zeit}] {nachricht}")
# --- DATENBANK SETUP --- # --- DATENBANK SETUP ---
conn = sqlite3.connect("polit_scraper.db", timeout=20) conn = sqlite3.connect("polit_scraper.db", timeout=20)
cursor = conn.cursor() cursor = conn.cursor()
@@ -56,48 +56,27 @@ CREATE TABLE IF NOT EXISTS artikel (
""") """)
conn.commit() conn.commit()
def analysiere_mit_gemini(text): def analysiere_mit_gemini(text):
"""KI-Analyse: Bewertet NUR den Text laut V8 (inkl. Begründung).""" """KI-Analyse: Bewertet NUR den Text laut V8 (inkl. Begründung)."""
sicherer_text = text[:5000] sicherer_text = text[:5000]
anweisung = f"WICHTIG: Analysiere NUR den folgenden Text laut deinen Panda-Regeln (V8).\n\nTEXT:\n{sicherer_text}" anweisung = f"WICHTIG: Analysiere NUR den folgenden Text laut deinen Panda-Regeln (V8).\n\nTEXT:\n{sicherer_text}"
befehl = [ befehl = ["gemini", "-m", GEMINI_MODEL, "-r", GEMINI_SESSION, "-p", anweisung]
"gemini",
"-m",
"gemini-3.1-flash-lite-preview",
"-r",
GEMINI_SESSION,
"-p",
anweisung,
]
try: try:
prozess = subprocess.run( prozess = subprocess.run(befehl, capture_output=True, text=True, encoding="utf-8", timeout=300)
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 { return {"relevant": False, "relevanz_score": 0, "begruendung_score": "Fehler: Kein JSON gefunden"}
"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 { return {"relevant": False, "relevanz_score": 0, "begruendung_score": f"Systemfehler: {e}"}
"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.""" """Sendet Push-Nachricht an ntfy."""
if not NTFY_BASE: if not NTFY_BASE: return
return
url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" url = f"{NTFY_BASE}/{TOPIC_PUBLIC}"
emoji_mapping = { emoji_mapping = {
"demo": "rotating_light,loudspeaker", "demo": "rotating_light,loudspeaker",
@@ -105,55 +84,32 @@ def sende_public(titel, nachricht, klick_url, category="info", priority=3):
"polizei": "police_car,warning", "polizei": "police_car,warning",
"stadt": "cityscape,newspaper", "stadt": "cityscape,newspaper",
"solidaritaet": "fist,heart", "solidaritaet": "fist,heart",
"info": "newspaper", "info": "newspaper"
} }
tags = emoji_mapping.get(category.lower(), "newspaper") tags = emoji_mapping.get(category.lower(), "newspaper")
headers = { headers = {"Title": titel.encode("utf-8"), "Tags": tags, "Click": klick_url, "Priority": str(priority)}
"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=10)
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):
if not NTFY_BASE: if not NTFY_BASE: return
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=10)
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): 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"): if clean.lower().startswith("http"):
return clean return clean
return urljoin(basis_url, clean) return urljoin(basis_url, clean)
def extrahiere_links(soup, basis_url): def extrahiere_links(soup, basis_url):
links = [] links = []
ignore_ext = (".pdf", ".jpg", ".jpeg", ".png", ".gif", ".zip", ".docx") 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): for a in soup.find_all("a", href=True):
full_url = bereinige_url(a["href"], basis_url) full_url = bereinige_url(a["href"], basis_url)
if full_url.lower().endswith(ignore_ext): if full_url.lower().endswith(ignore_ext): continue
continue
if urlparse(full_url).netloc == basis_domain: if urlparse(full_url).netloc == basis_domain:
if len(full_url) > len(basis_url.rstrip("/")) + 1: if len(full_url) > len(basis_url.rstrip("/")) + 1:
links.append(full_url) links.append(full_url)
return list(set(links)) return list(set(links))
def hole_artikel_text(url): def hole_artikel_text(url):
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 = ( bereich = unter_soup.find("main") or unter_soup.find("article") or unter_soup.body
unter_soup.find("main") or unter_soup.find("article") or unter_soup.body
)
text = bereich.get_text(separator=" ", strip=True) text = bereich.get_text(separator=" ", strip=True)
if len(text) > 12000: if len(text) > 12000: return text[:12000]
return text[:12000]
return text return text
except Exception as e: except Exception as e:
return f"Fehler beim Laden: {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):
schnipsel = text_snippet.strip().lower() schnipsel = text_snippet.strip().lower()
if len(schnipsel) < 10: if len(schnipsel) < 10: return None
return None
for a in soup.find_all("a", href=True): for a in soup.find_all("a", href=True):
link_text = a.get_text(separator=" ", strip=True).lower() link_text = a.get_text(separator=" ", strip=True).lower()
if schnipsel in link_text or link_text in schnipsel: if schnipsel in link_text or link_text in schnipsel:
return bereinige_url(a["href"], basis_url) return bereinige_url(a["href"], basis_url)
return None return None
def verarbeite_website(url): def verarbeite_website(url):
"""Hauptlogik mit Scoring-Logik und detaillierter Begründung im Debug-Modus.""" """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)
if antwort.status_code != 200: if antwort.status_code != 200:
sende_admin( sende_admin("Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}.")
"Fehler bei Webseite", f"{url} lieferte Status {antwort.status_code}."
)
return return
soup = BeautifulSoup(antwort.text, "html.parser") soup = BeautifulSoup(antwort.text, "html.parser")
@@ -214,7 +160,7 @@ def verarbeite_website(url):
alter_text = zeile[0] if zeile else "" alter_text = zeile[0] if zeile else ""
aktuelle_links = extrahiere_links(hauptbereich, url) aktuelle_links = extrahiere_links(hauptbereich, url)
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,))
@@ -229,12 +175,10 @@ def verarbeite_website(url):
print(f" [..] Analysiere neuen Artikel: {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) analyse = analysiere_mit_gemini(artikel_inhalt)
score = analyse.get("relevanz_score", 0) score = analyse.get("relevanz_score", 0)
begruendung = analyse.get( begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden")
"begruendung_score", "Keine Begründung vorhanden"
)
if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: 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}") debug(f"Begründung: {begruendung}")
@@ -243,76 +187,44 @@ def verarbeite_website(url):
nachricht=analyse.get("nachricht", "Neu auf der Seite"), nachricht=analyse.get("nachricht", "Neu auf der Seite"),
klick_url=link, klick_url=link,
category=analyse.get("category", "info"), category=analyse.get("category", "info"),
priority=analyse.get("priority", 3), priority=analyse.get("priority", 3)
) )
else: else:
debug( debug(f"Überspringe Artikel (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}")
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: Textänderung auf Hauptseite # FALL B: Textänderung auf Hauptseite
if alter_text != neuer_text: if alter_text != neuer_text:
diff = [ diff = [z[2:] for z in difflib.ndiff(alter_text.splitlines(), neuer_text.splitlines()) if z.startswith("+ ")]
z[2:]
for z in difflib.ndiff(
alter_text.splitlines(), neuer_text.splitlines()
)
if z.startswith("+ ")
]
for line in diff: for line in diff:
if len(line.strip()) < 20: if len(line.strip()) < 20: continue
continue
passender_link = finde_link_fuer_text(line[:40], 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( cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (target_url,))
"SELECT 1 FROM artikel WHERE artikel_url = ?", (target_url,)
)
if cursor.fetchone() is None: if cursor.fetchone() is None:
if passender_link: if passender_link:
artikel_inhalt = hole_artikel_text(passender_link) artikel_inhalt = hole_artikel_text(passender_link)
analyse = analysiere_mit_gemini(artikel_inhalt) analyse = analysiere_mit_gemini(artikel_inhalt)
else: else:
analyse = analysiere_mit_gemini(line) analyse = analysiere_mit_gemini(line)
score = analyse.get("relevanz_score", 0) score = analyse.get("relevanz_score", 0)
begruendung = analyse.get( begruendung = analyse.get("begruendung_score", "Keine Begründung vorhanden")
"begruendung_score", "Keine Begründung vorhanden"
)
if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE: if analyse.get("relevant") and score >= MIN_RELEVANZ_SCORE:
print( print(f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}")
f" [!] RELEVANT (Score: {score}): {analyse.get('titel')}"
)
debug(f"Begründung: {begruendung}") debug(f"Begründung: {begruendung}")
sende_public( sende_public(analyse.get("titel"), analyse.get("nachricht"), target_url, analyse.get("category"), analyse.get("priority"))
analyse.get("titel"),
analyse.get("nachricht"),
target_url,
analyse.get("category"),
analyse.get("priority"),
)
else: else:
debug( debug(f"Überspringe Änderung (Score {score} < {MIN_RELEVANZ_SCORE}). Begründung: {begruendung}")
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( cursor.execute("INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", (url, neuer_text, str(datetime.now())))
"INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)",
(url, neuer_text, str(datetime.now())),
)
conn.commit() conn.commit()
else: else:
debug(f"[OK] Keine Änderungen auf {url}.") debug(f"[OK] Keine Änderungen auf {url}.")
@@ -320,13 +232,11 @@ def verarbeite_website(url):
except Exception as e: except Exception as e:
print(f" ❌ Fehler bei {url}: {e}") print(f" ❌ Fehler bei {url}: {e}")
# --- HAUPTPROGRAMM --- # --- HAUPTPROGRAMM ---
with open("./urls.txt", "r", encoding="utf-8") as datei: with open("./urls.txt", "r", encoding="utf-8") as datei:
for zeile in datei: for zeile in datei:
adresse = zeile.strip() adresse = zeile.strip()
if not adresse or adresse.startswith("#"): if not adresse or adresse.startswith("#"): continue
continue
print(f"Ich prüfe: {adresse}") print(f"Ich prüfe: {adresse}")
verarbeite_website(adresse) verarbeite_website(adresse)