import sqlite3 import json import os import subprocess import time import requests import logging import sys import re 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 = [ "/opt/signal-cli-0.13.21/bin/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, env=os.environ) 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: # Entferne ungültige Steuerzeichen und versuche das JSON zu parsen clean_json = re.sub(r"[\x00-\x1f\x7f]", "", result_json) analyse = json.loads(clean_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()