diff --git a/analyzer.py b/analyzer.py new file mode 100644 index 0000000..fe32a91 --- /dev/null +++ b/analyzer.py @@ -0,0 +1,132 @@ +import sqlite3 +import json +import os +import subprocess +import time +import sys +import logging +import re +from datetime import datetime +from dotenv import load_dotenv + +# --- KONFIGURATION & LOGGING --- +load_dotenv() +BASE_DIR = os.path.dirname(__file__) +DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true" +LOG_LEVEL = logging.DEBUG if DEBUG_LOG else logging.INFO +MY_PID = os.getpid() + +logging.basicConfig( + level=LOG_LEVEL, + format='%(asctime)s [%(levelname)s] [%(process)d] %(message)s', + handlers=[ + logging.FileHandler(os.path.join(BASE_DIR, "analyzer.log"), encoding='utf-8'), + logging.StreamHandler(sys.stdout) + ] +) +logger = logging.getLogger(f"analyzer_{MY_PID}") + +GEMINI_MODEL = "gemini-2.5-flash-lite" +DB_PATH = os.path.join(BASE_DIR, "polit_scraper.db") +INSTRUCTION_PATH = os.path.join(BASE_DIR, "gemini_instructions.md") + +def extract_json(text): + start = text.find("{") + ende = text.rfind("}") + 1 + if start != -1 and ende > start: + json_str = text[start:ende] + json_str = re.sub(r'```json\s*', '', json_str) + json_str = re.sub(r'\s*```', '', json_str) + return json_str.strip() + return None + +def analysiere_mit_gemini(text): + sicherer_text = text[:8000] + anweisung = f"Analysiere diesen Text.\n\nTEXT:\n{sicherer_text}" + + env = os.environ.copy() + env["GEMINI_SYSTEM_MD"] = INSTRUCTION_PATH + env["NODE_ENV"] = "production" + env["CI"] = "true" + env["FORCE_COLOR"] = "0" + + befehl = ["gemini", "-m", GEMINI_MODEL, "-p", anweisung] + + try: + prozess = subprocess.run( + befehl, + capture_output=True, + text=True, + encoding="utf-8", + timeout=300, + env=env, + stdin=subprocess.DEVNULL + ) + stdout = prozess.stdout.strip() + stderr = prozess.stderr.strip() + + if DEBUG_LOG: + logger.debug(f"DEBUG_STDOUT: {stdout}") + if stderr: logger.debug(f"DEBUG_STDERR: {stderr}") + + if "Error: setRawMode EIO" in stderr: + logger.error(f"TTY_ERROR detected in stderr: {stderr}") + return "TTY_ERROR" + + if "exhausted your capacity" in stderr or "exhausted your capacity" in stdout: + return "RATE_LIMIT" + + json_str = extract_json(stdout) + if json_str: + return json_str + + logger.error(f"PARSING_FAILED. Roh-Output war: {stdout}") + return None + except Exception as e: + logger.error(f"SUBPROCESS_CRASH: {e}") + return None + +def main(): + logger.info(f"Worker {MY_PID} bereit.") + while True: + try: + conn = sqlite3.connect(DB_PATH, timeout=60) + cursor = conn.cursor() + + cursor.execute(""" + UPDATE analysis_queue + SET status='PROCESSING', worker_id=?, last_update=CURRENT_TIMESTAMP + WHERE id=(SELECT id FROM analysis_queue WHERE status='PENDING' ORDER BY id ASC LIMIT 1) + """, (MY_PID,)) + conn.commit() + + cursor.execute("SELECT id, url, raw_text FROM analysis_queue WHERE status='PROCESSING' AND worker_id=?", (MY_PID,)) + job = cursor.fetchone() + + if job: + job_id, url, text = job + logger.info(f"Analysiere Job {job_id}: {url}") + ergebnis = analysiere_mit_gemini(text) + + if ergebnis == "RATE_LIMIT" or ergebnis == "TTY_ERROR": + logger.warning(f"Job {job_id} ({url}) returned {ergebnis}. Re-queueing and waiting.") + cursor.execute("UPDATE analysis_queue SET status='PENDING', worker_id=NULL WHERE id=?", (job_id,)) + conn.commit() + time.sleep(60) + elif ergebnis: + cursor.execute("UPDATE analysis_queue SET status='ANALYZED', result_json=?, worker_id=NULL WHERE id=?", (ergebnis, job_id)) + logger.info(f"Job {job_id} ANALYZED.") + conn.commit() + else: + cursor.execute("UPDATE analysis_queue SET status='FAILED', worker_id=NULL WHERE id=?", (job_id,)) + logger.error(f"Job {job_id} FAILED (Parser returned None).") + conn.commit() + conn.close() + time.sleep(1) + except sqlite3.OperationalError: + time.sleep(5) + except Exception as e: + logger.error(f"Main Loop Fehler: {e}") + time.sleep(10) + +if __name__ == "__main__": main() diff --git a/crawler.py b/crawler.py new file mode 100644 index 0000000..0e8110d --- /dev/null +++ b/crawler.py @@ -0,0 +1,181 @@ +import cloudscraper +import sqlite3 +import os +import difflib +import sys +import subprocess +import logging +from bs4 import BeautifulSoup +from urllib.parse import urljoin, urlparse +from datetime import datetime +from dotenv import load_dotenv + +# --- KONFIGURATION & LOGGING --- +load_dotenv() +DEBUG_LOG = os.getenv("DEBUG_LOG", "false").lower() == "true" +LOG_LEVEL = logging.DEBUG if DEBUG_LOG else logging.INFO +BASE_DIR = os.path.dirname(__file__) + +logging.basicConfig( + level=LOG_LEVEL, + format='%(asctime)s [%(levelname)s] %(message)s', + handlers=[ + logging.FileHandler(os.path.join(BASE_DIR, "crawler.log"), encoding='utf-8'), + logging.StreamHandler(sys.stdout) + ] +) +logger = logging.getLogger("crawler") +sys.stdout.reconfigure(encoding="utf-8") + +scraper = cloudscraper.create_scraper( + browser={"browser": "chrome", "platform": "windows", "desktop": True} +) + +# --- DATENBANK SETUP --- +DB_PATH = os.path.join(BASE_DIR, "polit_scraper.db") + +def setup_db(): + conn = sqlite3.connect(DB_PATH, 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)") + + # Tabelle direkt mit allen Spalten erstellen, ohne ALTER TABLE + cursor.execute(""" + CREATE TABLE IF NOT EXISTS analysis_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT, + raw_text TEXT, + source_url TEXT, + status TEXT DEFAULT 'PENDING', + result_json TEXT, + worker_id INTEGER, + last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.commit() + conn.close() + +def queue_article(url, text, source_url): + conn = sqlite3.connect(DB_PATH, timeout=20) + cursor = conn.cursor() + cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (url,)) + if cursor.fetchone() is None: + logger.debug(f"Queue: {url}") + cursor.execute( + "INSERT OR IGNORE INTO analysis_queue (url, raw_text, source_url) VALUES (?, ?, ?)", + (url, text[:12000], source_url) + ) + cursor.execute("INSERT OR IGNORE INTO artikel (artikel_url, status) VALUES (?, ?)", (url, "gequeued")) + conn.commit() + conn.close() + +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 + return bereich.get_text(separator=" ", strip=True) + except Exception as e: + logger.error(f"Fehler bei {url}: {e}") + return f"Fehler: {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): + logger.info(f"Prüfe: {url}") + try: + conn = sqlite3.connect(DB_PATH, timeout=20) + cursor = conn.cursor() + antwort = scraper.get(url, timeout=15) + if antwort.status_code != 200: 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) + for link in aktuelle_links: + cursor.execute("SELECT 1 FROM artikel WHERE artikel_url = ?", (link,)) + if cursor.fetchone() is None: + logger.info(f"Neu: {link}") + artikel_inhalt = hole_artikel_text(link) + queue_article(link, artikel_inhalt, url) + + if alter_text != neuer_text: + diff = [z[2:] for z in difflib.ndiff(alter_text.splitlines(), neuer_text.splitlines()) if z.startswith("+ ") and len(z[2:].strip()) > 30] + for line in diff: + 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: + logger.info(f"Änderung: {target_url}") + text_zu_analysieren = hole_artikel_text(passender_link) if passender_link else line + queue_article(target_url, text_zu_analysieren, url) + + cursor.execute("INSERT OR REPLACE INTO seiten_stand (url, inhalt, zeit) VALUES (?, ?, ?)", (url, neuer_text, str(datetime.now()))) + conn.commit() + conn.close() + except Exception as e: + logger.error(f"Fehler bei {url}: {e}") + +def ensure_workers_running(): + try: + max_workers = int(os.getenv("MAX_ANALYZER_WORKERS", "5")) + output = subprocess.check_output(["ps", "-ef"]).decode() + running_analyzers = output.count("analyzer.py") + running_notifiers = output.count("notifier.py") + + base_dir = os.path.dirname(__file__) + if running_analyzers < max_workers: + needs = max_workers - running_analyzers + logger.info(f"Starte {needs} Analyzer (Ziel: {max_workers})...") + for _ in range(needs): + subprocess.Popen([sys.executable, os.path.join(base_dir, "analyzer.py")], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + if running_notifiers < 1: + logger.info("Starte Notifier...") + subprocess.Popen([sys.executable, os.path.join(base_dir, "notifier.py")], + stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) + except Exception as e: + logger.error(f"Worker-Check Fehler: {e}") + +# --- HAUPTPROGRAMM --- +logger.info("Crawler Start.") +setup_db() +urls_file = os.path.join(BASE_DIR, "urls.txt") +with open(urls_file, "r", encoding="utf-8") as datei: + for zeile in datei: + adresse = zeile.strip() + if not adresse or adresse.startswith("#"): continue + verarbeite_website(adresse) + +ensure_workers_running() +logger.info("Crawler Ende.") diff --git a/monitor.py b/monitor.py new file mode 100644 index 0000000..95c9f0e --- /dev/null +++ b/monitor.py @@ -0,0 +1,70 @@ +import sqlite3 +import os +import time +from datetime import datetime + +# Wir suchen die DB im selben Ordner wie das Skript +DB_PATH = os.path.join(os.path.dirname(__file__), "polit_scraper.db") + +def get_stats(): + try: + conn = sqlite3.connect(DB_PATH) + cursor = conn.cursor() + + # Statistiken + cursor.execute("SELECT status, count(*) FROM analysis_queue GROUP BY status") + stats = dict(cursor.fetchall()) + + # Laufende Jobs (PROCESSING) + cursor.execute("SELECT id, url, worker_id, last_update FROM analysis_queue WHERE status='PROCESSING' ORDER BY last_update DESC") + processing = cursor.fetchall() + + # Letzte 10 Einträge (allgemein) + cursor.execute("SELECT id, status, url, created_at FROM analysis_queue ORDER BY id DESC LIMIT 10") + recent = cursor.fetchall() + + conn.close() + return stats, processing, recent + except Exception as e: + return {}, [], [] + +def main(): + while True: + try: + stats, processing, recent = get_stats() + os.system('clear') + print(f"--- Polit-Scraper Live Monitor --- {datetime.now().strftime('%H:%M:%S')}") + print("-" * 80) + + # Stats Anzeige + all_statuses = ['PENDING', 'PROCESSING', 'ANALYZED', 'COMPLETED', 'FAILED'] + stat_line = " | ".join([f"{s}: {stats.get(s, 0)}" for s in all_statuses]) + print(stat_line) + print("-" * 80) + + # PROCESSING Jobs (aktiv) + print(f"{'AKTIV (PROCESSING)':<80}") + print(f"{'ID':<4} | {'Worker':<8} | {'URL (Beginn)'}") + print("-" * 80) + for r_id, url, worker_id, ts in processing: + url_display = (url[:60] + '..') if len(url) > 60 else url + print(f"{r_id:<4} | {str(worker_id):<8} | {url_display}") + print("-" * 80) + + # LETZTE 10 Jobs + print(f"{'LETZTE 10 JOB-STATUS':<80}") + print(f"{'ID':<4} | {'Status':<11} | {'URL (Ende)'}") + print("-" * 80) + for r_id, status, url, ts in recent: + url_display = (url[:60] + '..') if len(url) > 60 else url + print(f"{r_id:<4} | {status:<11} | {url_display}") + + time.sleep(2) + except KeyboardInterrupt: + break + except Exception as e: + print(f"Fehler: {e}") + time.sleep(5) + +if __name__ == "__main__": + main() diff --git a/notifier.py b/notifier.py new file mode 100644 index 0000000..5b1c8a1 --- /dev/null +++ b/notifier.py @@ -0,0 +1,116 @@ +import sqlite3 +import json +import os +import subprocess +import time +import requests +import logging +import sys +from dotenv import load_dotenv + +# --- KONFIGURATION & LOGGING --- +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")) + +NTFY_ENABLED = os.getenv("NTFY", "true").lower() == "true" +SIGNAL_ENABLED = os.getenv("SIGNAL", "true").lower() == "true" +TOPIC_PUBLIC = "polit-scraper-public" + +BASE_DIR = os.path.dirname(__file__) +DB_PATH = os.path.join(BASE_DIR, "polit_scraper.db") + +logging.basicConfig( + level=logging.DEBUG if DEBUG_LOG else logging.INFO, + format='%(asctime)s [%(levelname)s] %(message)s', + handlers=[ + logging.FileHandler(os.path.join(BASE_DIR, "notifier.log"), encoding='utf-8'), + logging.StreamHandler(sys.stdout) + ] +) +logger = logging.getLogger("notifier") + +def ensure_tables(): + conn = sqlite3.connect(DB_PATH, timeout=60) + cursor = conn.cursor() + cursor.execute(""" + CREATE TABLE IF NOT EXISTS analysis_queue ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + url TEXT, + raw_text TEXT, + source_url TEXT, + status TEXT DEFAULT 'PENDING', + result_json TEXT, + worker_id INTEGER, + last_update TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """) + conn.commit() + conn.close() + +def sende_signal(titel, nachricht, klick_url): + if not SIGNAL_ENABLED: return + message_content = f"{titel}\n{nachricht}\n{klick_url}" + cmd = [ + "signal-cli", "--config", "/root/.local/share/signal-cli", + "-u", SIGNAL_NUMBER, "send", "-m", message_content, "-g", SIGNAL_GROUPE_ID, + ] + try: + subprocess.run(cmd, capture_output=True, text=True, timeout=30) + except Exception as e: + logger.error(f"Signal-Fehler: {e}") + +def sende_ntfy(titel, nachricht, klick_url, priority=3): + if not NTFY_ENABLED or not NTFY_BASE: return + url = f"{NTFY_BASE}/{TOPIC_PUBLIC}" + headers = {"Title": titel.encode("utf-8"), "Click": klick_url, "Priority": str(priority)} + try: + requests.post(url, data=nachricht.encode("utf-8"), headers=headers, auth=(ADMIN_USER, ADMIN_PW), timeout=10) + except Exception as e: + logger.error(f"NTFY-Fehler: {e}") + +def main(): + logger.info("Notifier startet.") + ensure_tables() + while True: + try: + conn = sqlite3.connect(DB_PATH, timeout=60) + conn.execute("PRAGMA journal_mode=WAL") # Parallele Zugriffe erlauben + cursor = conn.cursor() + + cursor.execute("SELECT id, url, result_json FROM analysis_queue WHERE status = 'ANALYZED' ORDER BY id ASC LIMIT 1") + job = cursor.fetchone() + + if job: + job_id, url, result_json = job + try: + analyse = json.loads(result_json) + score = analyse.get("relevanz_score", 0) + + if score >= MIN_RELEVANZ_SCORE: + sende_ntfy(analyse.get("titel", "Polit-Update"), analyse.get("nachricht", ""), url, analyse.get("priority", 3)) + sende_signal(analyse.get("titel", "Polit-Update"), analyse.get("nachricht", ""), url) + + cursor.execute("UPDATE analysis_queue SET status = 'COMPLETED' WHERE id = ?", (job_id,)) + conn.commit() + logger.info(f"Job {job_id} verarbeitet.") + except Exception as e: + logger.error(f"Fehler Job {job_id}: {e}") + cursor.execute("UPDATE analysis_queue SET status = 'FAILED' WHERE id = ?", (job_id,)) + conn.commit() + + conn.close() + time.sleep(5) # Etwas mehr Zeit lassen + except sqlite3.OperationalError: + time.sleep(5) + except Exception as e: + logger.error(f"Notifier Loop: {e}") + time.sleep(10) + +if __name__ == "__main__": main()