Ratatoskr/polit-scraper
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

die alte web_check-py ist nun modular in analyzer.py crawler.py und notifier.py aufgeteilt. monitor.py bietet status über analyzer

Browse Source
This commit is contained in:
Ratatoskr
2026-04-04 14:32:24 +02:00
parent ee625f40d5
commit 86fc7817cc
4 changed files with 499 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

132
analyzer.py Normal file
View File

@@ -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()

181
crawler.py Normal file
View File

@@ -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.")

70
monitor.py Normal file
View File

@@ -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()

116
notifier.py Normal file
View File

@@ -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()