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 ist_sinnvoller_text(text): # Ignoriere Texte, die sehr kurz sind (wahrscheinlich Navigations-Links) if len(text.strip()) < 300: return False # Prüfe auf zu viele generische Wörter, die auf Navigations-Seiten hindeuten nav_keywords = ["startseite", "lokales", "aktuelles", "veranstaltungen", "kontakt", "impressum", "baustelle", "meldungen"] text_lower = text.lower() match_count = sum(text_lower.count(k) for k in nav_keywords) # Wenn mehr als 5 Navigations-Begriffe vorkommen, ist es vermutlich eine Nav-Seite if match_count > 5: return False return True def queue_article(url, text, source_url): if not ist_sinnvoller_text(text): logger.debug(f"Überspringe Müll-Text: {url}") return 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) soup = BeautifulSoup(antwort.text, "html.parser") # Müll entfernen for element in soup(["header", "footer", "nav", "script", "style", "aside"]): element.decompose() # Gezielter suchen bereich = soup.find("main") or soup.find("article") or soup.find("div", class_="content") if not bereich: return None text = bereich.get_text(separator=" ", strip=True) return text[:12000] except Exception as e: logger.error(f"Fehler bei {url}: {e}") return None 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() import requests from urllib.parse import urljoin, urlparse # ... def sende_admin(titel, nachricht): admin_url = f"{os.getenv('NTFY_BASE')}/polit-scraper-admin" try: requests.post( admin_url, data=nachricht.encode("utf-8"), headers={"Title": titel.encode("utf-8"), "Priority": "4"}, auth=(os.getenv("ADMIN_USER"), os.getenv("ADMIN_PW")), timeout=5 ) except: pass # ... in verarbeite_website ... antwort = scraper.get(url, timeout=15) if antwort.status_code != 200: logger.error(f"Website Fehler: {url} Status {antwort.status_code}") sende_admin("Crawler Fehler", f"URL {url} lieferte Status {antwort.status_code}.") return soup = BeautifulSoup(antwort.text, "html.parser") # Müll entfernen for element in soup(["header", "footer", "nav", "script", "style", "aside"]): element.decompose() # Gezielter suchen hauptbereich = soup.find("main") or soup.find("article") or soup.find("div", class_="content") if not hauptbereich: logger.debug(f"Überspringe Seite ohne Inhalts-Struktur: {url}") return 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(soup, 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) if artikel_inhalt: 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], soup, 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 if text_zu_analysieren: 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.")