270 lines
10 KiB
Python
270 lines
10 KiB
Python
import cloudscraper
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import difflib
|
|
import sys
|
|
from bs4 import BeautifulSoup
|
|
from urllib.parse import urljoin
|
|
from datetime import datetime
|
|
from dotenv import load_dotenv, main
|
|
|
|
# --- KONFIGURATION LADEN ---
|
|
# Sucht nach einer Datei namens .env und lädt deren Variablen in das System
|
|
load_dotenv()
|
|
|
|
# os.getenv liest eine Variable aus dem System (die zuvor aus der .env geladen wurde)
|
|
ADMIN_USER = os.getenv("ADMIN_USER")
|
|
ADMIN_PW = os.getenv("ADMIN_PW")
|
|
NTFY_BASE = os.getenv("NTFY_BASE")
|
|
|
|
# Name der Datei, in der die alten Webseiten-Stände gespeichert werden
|
|
JSON_DATEI = "./results.json"
|
|
# Die ID der Gemini-Sitzung, die für die Analyse genutzt werden soll
|
|
GEMINI_SESSION = "1"
|
|
|
|
# Namen der ntfy-Topics für öffentliche News und Admin-Fehler
|
|
TOPIC_PUBLIC = "polit-scraper-public"
|
|
TOPIC_ADMIN = "polit-scraper-admin"
|
|
|
|
# Stellt sicher, dass das Skript Umlaute (ä, ö, ü) korrekt im Terminal ausgibt
|
|
sys.stdout.reconfigure(encoding="utf-8")
|
|
|
|
# Erstellt ein Scraper-Objekt, das sich als Chrome-Browser unter Windows ausgibt
|
|
# Dies dient dazu, Sicherheitsabfragen wie Cloudflare zu umgehen
|
|
scraper = cloudscraper.create_scraper(
|
|
browser={"browser": "chrome", "platform": "windows", "desktop": True}
|
|
)
|
|
|
|
|
|
def analysiere_mit_gemini(text, links):
|
|
"""KI-Analyse: Sendet den Text und die gefundenen Links an Gemini."""
|
|
|
|
# Erklärung der folgenden Zeile (Ternärer Operator):
|
|
# Wenn die Liste 'links' nicht leer ist, wird ein String mit der Überschrift erstellt
|
|
# und alle Links mit Kommas getrennt (", ".join) dahintergehängt.
|
|
# Wenn 'links' leer ist (else), wird einfach ein leerer String ("") zugewiesen.
|
|
links_text = "\nRelevante Links: " + ", ".join(links) if links else ""
|
|
|
|
# Baut die endgültige Frage (Prompt) für die KI zusammen
|
|
anweisung = f"Analysiere diesen Text: {text}{links_text}"
|
|
|
|
# Befehl für das Terminal: gemini -r [Sitzung] -p [Frage]
|
|
befehl = ["gemini", "-r", GEMINI_SESSION, "-p", anweisung]
|
|
|
|
try:
|
|
# Führt den Terminal-Befehl aus und speichert die Antwort
|
|
# capture_output=True fängt den Text ab, text=True macht daraus einen Python-String
|
|
prozess = subprocess.run(
|
|
befehl, capture_output=True, text=True, encoding="utf-8"
|
|
)
|
|
antwort_roh = prozess.stdout.strip()
|
|
|
|
# Sucht den Anfang ({) und das Ende (}) des JSON-Objekts in der Antwort
|
|
# Das ist nötig, falls die KI zusätzlichen Text um das JSON herum schreibt
|
|
start = antwort_roh.find("{")
|
|
ende = antwort_roh.rfind("}") + 1
|
|
|
|
if start != -1 and ende != -1:
|
|
# json.loads macht aus dem Text ein echtes Python-Dictionary (Objekt)
|
|
return json.loads(antwort_roh[start:ende])
|
|
|
|
# Falls kein JSON gefunden wurde, wird die Nachricht als nicht relevant eingestuft
|
|
return {"relevant": False}
|
|
except Exception as e:
|
|
print(f" [!] KI-Fehler: {e}")
|
|
return {"relevant": False}
|
|
|
|
|
|
def sende_public(titel, nachricht, klick_url, category="info", priority=3):
|
|
"""Sendet eine Push-Nachricht an den öffentlichen ntfy-Kanal."""
|
|
if not NTFY_BASE:
|
|
return
|
|
|
|
# Baut die Ziel-Adresse für ntfy zusammen
|
|
url = f"{NTFY_BASE}/{TOPIC_PUBLIC}"
|
|
|
|
# Zuordnung von Kategorien zu Emojis für die Anzeige auf dem Handy
|
|
emoji_mapping = {
|
|
"demo": "rotating_light,loudspeaker",
|
|
"socialmedia": "detective,no_entry",
|
|
"polizei": "police_car,warning",
|
|
"stadt": "cityscape,newspaper",
|
|
"solidaritaet": "fist,heart",
|
|
"info": "newspaper",
|
|
}
|
|
tags = emoji_mapping.get(category.lower(), "newspaper")
|
|
|
|
# ntfy-Header: Metadaten der Nachricht (Titel, Icons, Klick-Aktion, Wichtigkeit)
|
|
headers = {
|
|
"Title": titel.encode("utf-8"),
|
|
"Tags": tags,
|
|
"Click": klick_url,
|
|
"Priority": str(priority),
|
|
}
|
|
try:
|
|
import requests
|
|
|
|
# POST-Anfrage sendet die Daten an den ntfy-Server
|
|
requests.post(
|
|
url,
|
|
data=nachricht.encode("utf-8"),
|
|
headers=headers,
|
|
auth=(ADMIN_USER, ADMIN_PW),
|
|
timeout=5,
|
|
)
|
|
except Exception as e:
|
|
print(f" [!] Fehler Public-Versand: {e}")
|
|
|
|
|
|
def sende_admin(titel, nachricht):
|
|
"""Sendet Fehlermeldungen an den passwortgeschützten Admin-Kanal."""
|
|
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=5,
|
|
)
|
|
except Exception as e:
|
|
print(f" [!] Fehler Admin-Versand: {e}")
|
|
|
|
|
|
def lade_gedaechtnis():
|
|
"""Liest das results.json-File ein und gibt den Inhalt als Dictionary zurück."""
|
|
if not os.path.exists(JSON_DATEI):
|
|
return {}
|
|
with open(JSON_DATEI, "r", encoding="utf-8") as f:
|
|
try:
|
|
return json.load(f)
|
|
except:
|
|
return {}
|
|
|
|
|
|
def speichere_gedaechtnis(daten):
|
|
"""Speichert das aktuelle Wissen (daten) im JSON-Format auf die Festplatte."""
|
|
with open(JSON_DATEI, "w", encoding="utf-8") as f:
|
|
# indent=4 sorgt für eine lesbare Struktur (Einrückung) der Datei
|
|
json.dump(daten, f, indent=4, ensure_ascii=False)
|
|
|
|
|
|
def extrahiere_links(soup, basis_url):
|
|
"""Sucht alle <a>-Tags im HTML und macht daraus absolute Internetadressen."""
|
|
links = []
|
|
# find_all('a') sucht alle Hyperlinks im HTML-Dokument
|
|
for a in soup.find_all("a", href=True):
|
|
# urljoin kombiniert die Basis-URL (z.B. pnp.de) mit dem Link (z.B. /artikel1)
|
|
links.append(urljoin(basis_url, a["href"]))
|
|
# list(set(links)) entfernt doppelte Einträge
|
|
return list(set(links))
|
|
|
|
|
|
def hole_artikel_text(url):
|
|
"""Rufe eine Unterseite auf und extrahiert den Haupttext."""
|
|
try:
|
|
# Scraper um die Unterseite zu laden
|
|
antwort = scraper.get(url, timeout=10)
|
|
# Verwandle die Antwort in durchsuchbares HTML
|
|
unter_soup = BeautifulSoup(antwort.text, "htpm.parser")
|
|
# Suche nach Hauptbereich (wie im Haptscript)
|
|
bereich = (
|
|
unter_soup.find("main") or unter_soup.find("article") or unter_soup.body
|
|
)
|
|
# Gibt den reinen Text ohne Leerzeichen-Salat zurück
|
|
return bereich.get_text(separator=" ", strip=True)
|
|
except Exception as e:
|
|
# Fals etwas schiefgeht, gib einen Hinweis zurück
|
|
return f"Fehler beim Laden des Artikels!: {e}"
|
|
|
|
|
|
def verarbeite_website(url, gedaechtnis):
|
|
"""Kern-Logik für eine einzelne Webseite."""
|
|
try:
|
|
# Lädt die Webseite mit dem Cloudscraper
|
|
antwort = scraper.get(url, timeout=15)
|
|
antwort.encoding = "utf-8"
|
|
status = antwort.status_code
|
|
|
|
# Bei Server-Fehlern (z.B. 404 oder 500) wird der Admin informiert
|
|
if status != 200:
|
|
print(f" [!] Seite fehlerhaft: {url} (Status: {status})")
|
|
sende_admin("Fehler bei Webseite", f"{url} lieferte Status {status}.")
|
|
return
|
|
|
|
# Verwandelt den rohen HTML-Text in ein BeautifulSoup-Objekt zum Suchen
|
|
soup = BeautifulSoup(antwort.text, "html.parser")
|
|
# Sucht nach dem Inhaltsbereich (main, article oder body)
|
|
hauptbereich = soup.find("main") or soup.find("article") or soup.body or soup
|
|
# Extrahiert den reinen Text ohne HTML-Tags
|
|
neuer_text = hauptbereich.get_text(separator=" ", strip=True)
|
|
|
|
# Holt den alten Text dieser URL aus dem Gedächtnis
|
|
alter_eintrag = gedaechtnis.get(url, {})
|
|
alter_text = alter_eintrag.get("inhalt", "")
|
|
|
|
# Prüft, ob sich der Text seit dem letzten Lauf verändert hat
|
|
if alter_text != neuer_text:
|
|
print(f" [NEU] Änderung auf {url} gefunden. Ich frage die KI...")
|
|
# Zerlegt den Text in Zeilen
|
|
alt_z = alter_text.splitlines()
|
|
neu_z = neuer_text.splitlines()
|
|
# Findet nur die Zeilen, die neu dazugekommen sind (+ )
|
|
diff = [z[2:] for z in difflib.ndiff(alt_z, neu_z) if z.startswith("+ ")]
|
|
text_fuer_ki = "\n".join(diff)
|
|
|
|
# Sammelt alle Links aus dem Bereich für die KI
|
|
alle_links = extrahiere_links(hauptbereich, url)
|
|
|
|
# Nur wenn auch wirklich neuer Text vorhanden ist
|
|
if text_fuer_ki.strip():
|
|
analyse = analysiere_mit_gemini(text_fuer_ki, alle_links)
|
|
# Falls die KI die Änderung als politisch wichtig einstuft
|
|
if analyse.get("relevant") == True:
|
|
print(f" [!] KI bestätigt Relevanz: {analyse.get('titel')}")
|
|
# Nutzt den direkten Link der KI oder die Haupt-URL
|
|
klick_url = analyse.get("direct_link", url)
|
|
sende_public(
|
|
titel=analyse.get("titel", "Polit-Update"),
|
|
nachricht=analyse.get("nachricht", "Neu auf der Seite"),
|
|
klick_url=klick_url,
|
|
category=analyse.get("category", "info"),
|
|
priority=analyse.get("priority", 3),
|
|
)
|
|
|
|
# Speichert den neuen Stand dieser Webseite im lokalen Gedächtnis
|
|
gedaechtnis[url] = {
|
|
"inhalt": neuer_text,
|
|
"zeit": str(datetime.now()),
|
|
"status": status,
|
|
}
|
|
else:
|
|
print(f" [OK] Keine Änderungen auf {url}.")
|
|
|
|
except Exception as e:
|
|
print(f" ❌ Fehler bei {url}: {e}")
|
|
sende_admin("Programmfehler", f"{url}: {e}")
|
|
|
|
|
|
# --- HAUPTPROGRAMM ---
|
|
# Startet den Prozess: Gedächtnis laden -> URLs prüfen -> Gedächtnis speichern
|
|
gedaechtnis = lade_gedaechtnis()
|
|
with open("./urls.txt", "r", encoding="utf-8") as datei:
|
|
for zeile in datei:
|
|
# Entfernt Leerzeichen und ignoriert Kommentarzeilen (#)
|
|
adresse = zeile.strip()
|
|
if not adresse or adresse.startswith("#"):
|
|
continue
|
|
print(f"Ich prüfe: {adresse}")
|
|
verarbeite_website(adresse, gedaechtnis)
|
|
|
|
# Speichert das aktualisierte Wissen für den nächsten Cronjob-Lauf
|
|
speichere_gedaechtnis(gedaechtnis)
|
|
print("\nFertig. Gute Nacht!")
|