polit-scraper/notifier.py

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,
        "--trust-new-identities",
        "always",
        "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()