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

chore: web-check.py endgültig aus Repo entfernt

Browse Source
This commit is contained in:
Ratatoskr
2026-04-04 15:17:23 +02:00
parent 86fc7817cc
commit 991b05c5ac
1 changed files with 0 additions and 466 deletions
Download Patch File Download Diff File Expand all files Collapse all files

466
web_check.py
View File

@@ -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.")