467 lines
17 KiB
Python
467 lines
17 KiB
Python
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.")
|