117 lines
4.1 KiB
Python
117 lines
4.1 KiB
Python
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()
|