"""Grundquellen-Verwaltung und Kundenquellen-Übersicht.""" import json import logging import hashlib import os import re import uuid from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, HTTPException, Request, UploadFile, status from pydantic import BaseModel, Field from typing import Optional import aiosqlite from auth import get_current_admin from database import db_dependency, get_db from audit import log_action, get_client_ip from source_meta import get_meta from health_run import new_state, try_start, mark_finished, mark_failed from config import DB_PATH from shared.source_rules import ( discover_source, discover_all_feeds, evaluate_feeds_with_claude, domain_to_display_name, ) from shared.services.source_classifier import ( bulk_classify, classify_source, ALIGNMENT_VALUES, POLITICAL_VALUES, MEDIA_TYPE_VALUES, RELIABILITY_VALUES, ) from shared.services.external_reputation import ( apply_reputation_overrides, sync_all as sync_external_reputation, ) logger = logging.getLogger("verwaltung.sources") router = APIRouter(prefix="/api/sources", tags=["sources"]) SOURCE_UPDATE_COLUMNS = { "name", "url", "domain", "source_type", "category", "status", "notes", "language", "bias", "fetch_strategy", "political_orientation", "media_type", "reliability", "state_affiliated", "country_code", } SOURCE_CLASSIFICATION_FIELDS = { "political_orientation", "media_type", "reliability", "state_affiliated", "country_code", } async def _load_alignments_for(db: aiosqlite.Connection, source_ids: list[int]) -> dict[int, list[str]]: if not source_ids: return {} placeholders = ",".join("?" for _ in source_ids) cursor = await db.execute( f"SELECT source_id, alignment FROM source_alignments WHERE source_id IN ({placeholders}) ORDER BY alignment", source_ids, ) out: dict[int, list[str]] = {sid: [] for sid in source_ids} for row in await cursor.fetchall(): out.setdefault(row["source_id"], []).append(row["alignment"]) return out async def _replace_alignments(db: aiosqlite.Connection, source_id: int, alignments: list[str]): """Ersetzt die alignments-Liste einer Quelle (DELETE + INSERT) — Aufrufer muss commit() machen.""" await db.execute("DELETE FROM source_alignments WHERE source_id = ?", (source_id,)) seen: set[str] = set() for raw in alignments: a = (raw or "").strip().lower() if not a or a in seen: continue if a not in ALIGNMENT_VALUES: raise HTTPException( status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=f"Ungueltiger alignment-Wert: '{a}'", ) seen.add(a) await db.execute( "INSERT INTO source_alignments (source_id, alignment) VALUES (?, ?)", (source_id, a), ) async def _clear_proposed(db: aiosqlite.Connection, source_id: int): await db.execute( """UPDATE sources SET proposed_political_orientation = NULL, proposed_media_type = NULL, proposed_reliability = NULL, proposed_state_affiliated = NULL, proposed_country_code = NULL, proposed_alignments_json = NULL, proposed_confidence = NULL, proposed_reasoning = NULL, proposed_at = NULL WHERE id = ?""", (source_id,), ) @router.get("/meta") async def get_sources_meta(admin: dict = Depends(get_current_admin)): """Liefert Kategorien und Typen als Single Source of Truth. Frontend lädt das beim Init und befüllt damit Filter-Dropdowns + Label-Lookups. Damit gibt es keine hardcoded Listen mehr im JS/HTML. """ return get_meta() @router.get("/tasks/summary") async def tasks_summary( admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Billiger Zähler für den Aufgaben-Badge im Quellen-Reiter. classification_pending nutzt DIESELBE Bedingung wie /classification/queue, damit Badge und Review-Karten nie auseinanderlaufen. """ suggestions_pending = 0 cur = await db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='source_suggestions'" ) if await cur.fetchone(): cur = await db.execute( "SELECT COUNT(*) AS cnt FROM source_suggestions WHERE status = 'pending'" ) suggestions_pending = (await cur.fetchone())["cnt"] cur = await db.execute( """SELECT COUNT(*) AS cnt FROM sources WHERE status = 'active' AND proposed_political_orientation IS NOT NULL""" ) classification_pending = (await cur.fetchone())["cnt"] return { "suggestions_pending": suggestions_pending, "classification_pending": classification_pending, "total": suggestions_pending + classification_pending, } class GlobalSourceCreate(BaseModel): name: str = Field(min_length=1, max_length=200) url: Optional[str] = None domain: Optional[str] = None source_type: str = Field(default="rss_feed", pattern="^(rss_feed|web_source|telegram_channel|podcast_feed|pdf_document|x_account)$") category: str = Field(default="sonstige") status: str = Field(default="active", pattern="^(active|inactive)$") notes: Optional[str] = None language: Optional[str] = Field(default=None, max_length=100) bias: Optional[str] = Field(default=None, max_length=500) fetch_strategy: Optional[str] = Field(default="default", pattern="^(default|googlebot|paywall|skip)$") class GlobalSourceUpdate(BaseModel): name: Optional[str] = Field(default=None, max_length=200) url: Optional[str] = None domain: Optional[str] = None source_type: Optional[str] = Field(default=None, pattern="^(rss_feed|web_source|telegram_channel|podcast_feed|pdf_document|x_account)$") category: Optional[str] = None status: Optional[str] = Field(default=None, pattern="^(active|inactive)$") notes: Optional[str] = None language: Optional[str] = Field(default=None, max_length=100) bias: Optional[str] = Field(default=None, max_length=500) political_orientation: Optional[str] = None media_type: Optional[str] = None reliability: Optional[str] = None state_affiliated: Optional[bool] = None country_code: Optional[str] = Field(default=None, max_length=8) alignments: Optional[list[str]] = None @router.get("") async def list_sources( scope: str = "all", admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Vereinte Quellenliste (Grund- und Kundenquellen). scope = all | global | tenant. Liefert pro Quelle den worst-case Health-Status aus source_health_checks (error > warning > ok > unknown) und den Organisationsnamen bei Kundenquellen (org_name, NULL bei global). """ scope_where = { "all": "1=1", "global": "s.tenant_id IS NULL", "tenant": "s.tenant_id IS NOT NULL", }.get(scope) if scope_where is None: raise HTTPException(status_code=422, detail="scope muss all, global oder tenant sein") cursor = await db.execute(f""" WITH article_stats AS ( -- Match per source-Name (case-insensitive). source_url im articles ist die -- Artikel-URL, nicht die Feed-URL - daher matcht das nicht mit sources.url. SELECT LOWER(source) AS s_lower, SUM(CASE WHEN collected_at > datetime('now', '-7 days') THEN 1 ELSE 0 END) AS a7d, SUM(CASE WHEN collected_at > datetime('now', '-30 days') THEN 1 ELSE 0 END) AS a30d FROM articles WHERE collected_at > datetime('now', '-30 days') AND source IS NOT NULL GROUP BY LOWER(source) ), health_agg AS ( SELECT source_id, MAX(CASE WHEN status = 'error' THEN 3 WHEN status = 'warning' THEN 2 WHEN status = 'ok' THEN 1 ELSE 0 END) AS rank FROM source_health_checks GROUP BY source_id ) SELECT s.*, o.name AS org_name, CASE ha.rank WHEN 3 THEN 'error' WHEN 2 THEN 'warning' WHEN 1 THEN 'ok' ELSE NULL END AS health_status, COALESCE(ast.a7d, 0) AS articles_7d, COALESCE(ast.a30d, 0) AS articles_30d FROM sources s LEFT JOIN organizations o ON o.id = s.tenant_id LEFT JOIN article_stats ast ON ast.s_lower = LOWER(s.name) LEFT JOIN health_agg ha ON ha.source_id = s.id WHERE {scope_where} ORDER BY s.category, s.source_type, s.name """) return [dict(row) for row in await cursor.fetchall()] @router.post("/global", status_code=201) async def create_global_source( data: GlobalSourceCreate, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Neue Grundquelle anlegen.""" if data.url: cursor = await db.execute( "SELECT id, name FROM sources WHERE url = ? AND tenant_id IS NULL", (data.url,), ) existing = await cursor.fetchone() if existing: raise HTTPException( status_code=409, detail=f"URL bereits vorhanden: {existing['name']}", ) cursor = await db.execute( """INSERT INTO sources (name, url, domain, source_type, category, status, notes, language, bias, fetch_strategy, added_by, tenant_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'system', NULL)""", (data.name, data.url, data.domain, data.source_type, data.category, data.status, data.notes, data.language, data.bias, data.fetch_strategy or "default"), ) src_id = cursor.lastrowid await db.commit() cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (src_id,)) new_src = dict(await cursor.fetchone()) await log_action( db, admin, get_client_ip(request), action="create", resource_type="source", resource_id=src_id, after=new_src, ) return new_src @router.put("/global/{source_id}") async def update_global_source( source_id: int, data: GlobalSourceUpdate, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Quelle bearbeiten (Grund- und Kundenquellen) — inkl. Klassifikation + alignments. Kundenquellen sind absichtlich mit erlaubt, damit der Deaktivieren-Knopf der vereinten Liste (status=inactive) fuer beide Herkuenfte funktioniert. """ cursor = await db.execute( "SELECT * FROM sources WHERE id = ?", (source_id,) ) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") before = dict(row) before_alignments = sorted((await _load_alignments_for(db, [source_id])).get(source_id, [])) payload = data.model_dump(exclude_none=True) alignments = payload.pop("alignments", None) if "political_orientation" in payload and payload["political_orientation"] not in POLITICAL_VALUES: raise HTTPException(status_code=422, detail=f"Ungueltige political_orientation: {payload['political_orientation']}") if "media_type" in payload and payload["media_type"] not in MEDIA_TYPE_VALUES: raise HTTPException(status_code=422, detail=f"Ungueltiger media_type: {payload['media_type']}") if "reliability" in payload and payload["reliability"] not in RELIABILITY_VALUES: raise HTTPException(status_code=422, detail=f"Ungueltige reliability: {payload['reliability']}") updates = {k: v for k, v in payload.items() if k in SOURCE_UPDATE_COLUMNS} if "state_affiliated" in updates: updates["state_affiliated"] = 1 if updates["state_affiliated"] else 0 classification_touched = any(k in updates for k in SOURCE_CLASSIFICATION_FIELDS) or alignments is not None if classification_touched: updates["classification_source"] = "manual" updates["classified_at"] = None # CURRENT_TIMESTAMP via SQL — siehe unten if updates: sets = [] vals = [] for k, v in updates.items(): if k == "classified_at": sets.append("classified_at = CURRENT_TIMESTAMP") else: sets.append(f"{k} = ?") vals.append(v) vals.append(source_id) await db.execute(f"UPDATE sources SET {', '.join(sets)} WHERE id = ?", vals) if alignments is not None: await _replace_alignments(db, source_id, alignments) await db.commit() cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) after = dict(await cursor.fetchone()) after_alignments = sorted((await _load_alignments_for(db, [source_id])).get(source_id, [])) if before_alignments != after_alignments: before["alignments"] = before_alignments after["alignments"] = after_alignments await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=source_id, before=before, after=after, ) return after @router.delete("/global/{source_id}", status_code=204) async def delete_global_source( source_id: int, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Grundquelle loeschen.""" cursor = await db.execute( "SELECT * FROM sources WHERE id = ? AND tenant_id IS NULL", (source_id,) ) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Grundquelle nicht gefunden") before = dict(row) await db.execute("DELETE FROM sources WHERE id = ?", (source_id,)) await db.commit() await log_action( db, admin, get_client_ip(request), action="delete", resource_type="source", resource_id=source_id, before=before, ) @router.get("/languages") async def get_languages( admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Distinct language-Werte aller Quellen - für Frontend-Filter-Dropdown.""" cur = await db.execute(""" SELECT DISTINCT language FROM sources WHERE language IS NOT NULL AND language != '' ORDER BY language """) return [r["language"] for r in await cur.fetchall()] @router.get("/stats") async def get_sources_stats( admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Aggregierte Stats für die Stats-Bar der vereinten Quellenliste. Zählt Grund- UND Kundenquellen, aktive wie inaktive (die Liste zeigt ebenfalls alle). by_origin liefert die Aufteilung, last_check den Zeitpunkt des jüngsten Health-Checks. """ cur = await db.execute(""" SELECT source_type, COUNT(*) AS count, COALESCE(SUM(article_count), 0) AS articles FROM sources GROUP BY source_type """) by_type = {} total = 0 total_articles = 0 for r in await cur.fetchall(): d = dict(r) by_type[d["source_type"]] = {"count": d["count"], "articles": d["articles"]} total += d["count"] total_articles += d["articles"] cur = await db.execute(""" SELECT CASE WHEN tenant_id IS NULL THEN 'global' ELSE 'tenant' END AS origin, COUNT(*) AS cnt FROM sources GROUP BY origin """) by_origin = {dict(r)["origin"]: dict(r)["cnt"] for r in await cur.fetchall()} # Health-Counter + Zeitpunkt des letzten Checks health = {"errors": 0, "warnings": 0, "ok": 0} last_check = None cur = await db.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name='source_health_checks' """) if await cur.fetchone(): cur = await db.execute(""" SELECT h.status AS hs, COUNT(DISTINCT h.source_id) AS cnt FROM source_health_checks h JOIN sources s ON s.id = h.source_id GROUP BY h.status """) for r in await cur.fetchall(): d = dict(r) if d["hs"] == "error": health["errors"] = d["cnt"] elif d["hs"] == "warning": health["warnings"] = d["cnt"] elif d["hs"] == "ok": health["ok"] = d["cnt"] cur = await db.execute("SELECT MAX(checked_at) AS lc FROM source_health_checks") row = await cur.fetchone() if row: last_check = dict(row)["lc"] return { "by_type": by_type, "by_origin": by_origin, "total": total, "total_articles": total_articles, "health": health, "last_check": last_check, } @router.post("/tenant/{source_id}/promote") async def promote_to_global( source_id: int, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Tenant-Quelle zur Grundquelle befoerdern.""" cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") if row["tenant_id"] is None: raise HTTPException(status_code=400, detail="Bereits eine Grundquelle") before = dict(row) if row["url"]: cursor = await db.execute( "SELECT id FROM sources WHERE url = ? AND tenant_id IS NULL", (row["url"],), ) if await cursor.fetchone(): raise HTTPException(status_code=409, detail="URL bereits als Grundquelle vorhanden") await db.execute( "UPDATE sources SET tenant_id = NULL, added_by = 'system' WHERE id = ?", (source_id,), ) await db.commit() cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) after = dict(await cursor.fetchone()) await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=source_id, before=before, after=after, ) return after class BulkPromoteRequest(BaseModel): source_ids: list[int] @router.post("/tenant/bulk-promote") async def bulk_promote_tenant_sources( data: BulkPromoteRequest, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Mehrere Tenant-Quellen auf einen Schlag zur Grundquelle befoerdern. Returns: promoted: int - Anzahl erfolgreich promoter Quellen skipped: list - {id, name, reason} fuer uebersprungene failed: list - {id, error} fuer Fehler """ promoted = 0 skipped = [] failed = [] for sid in data.source_ids: try: cur = await db.execute("SELECT * FROM sources WHERE id = ?", (sid,)) row = await cur.fetchone() if not row: failed.append({"id": sid, "error": "nicht gefunden"}) continue if row["tenant_id"] is None: skipped.append({"id": sid, "name": row["name"], "reason": "bereits Grundquelle"}) continue before = dict(row) if row["url"]: cur = await db.execute( "SELECT id FROM sources WHERE url = ? AND tenant_id IS NULL", (row["url"],), ) if await cur.fetchone(): skipped.append({"id": sid, "name": row["name"], "reason": "URL bereits als Grundquelle vorhanden"}) continue await db.execute( "UPDATE sources SET tenant_id = NULL, added_by = 'system' WHERE id = ?", (sid,), ) cur = await db.execute("SELECT * FROM sources WHERE id = ?", (sid,)) after = dict(await cur.fetchone()) await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=sid, before=before, after=after, ) promoted += 1 except Exception as e: failed.append({"id": sid, "error": str(e)}) await db.commit() return {"promoted": promoted, "skipped": skipped, "failed": failed} @router.post("/discover") async def discover_source_endpoint( url: str, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """URL analysieren: Domain, Kategorie und RSS-Feeds automatisch erkennen. Findet alle Feeds einer Domain, bewertet sie mit Claude und gibt die relevanten zurueck. Prueft auf bereits vorhandene Grundquellen. """ try: multi = await discover_all_feeds(url) except Exception as e: logger.error(f"Discovery fehlgeschlagen: {e}", exc_info=True) raise HTTPException(status_code=500, detail="Discovery fehlgeschlagen") domain = multi["domain"] category = multi["category"] feeds = multi.get("feeds", []) # Fallback auf Einzel-Discovery wenn keine Feeds gefunden if not feeds: try: single = await discover_source(url) if single.get("rss_url"): feeds = [{"name": single["name"], "url": single["rss_url"]}] domain = single.get("domain", domain) category = single.get("category", category) except Exception: pass if not feeds: return { "domain": domain, "category": category, "feeds": [], "existing": [], "message": "Keine RSS-Feeds gefunden", } # Mit Claude bewerten try: relevant_feeds = await evaluate_feeds_with_claude(domain, feeds) except Exception: relevant_feeds = feeds[:3] # Bereits vorhandene Grundquellen pruefen cursor = await db.execute( "SELECT url FROM sources WHERE tenant_id IS NULL AND url IS NOT NULL" ) existing_urls = {row["url"] for row in await cursor.fetchall()} result_feeds = [] existing = [] for feed in relevant_feeds: info = { "name": feed.get("name", domain_to_display_name(domain)), "url": feed["url"], "domain": domain, "category": category, } if feed["url"] in existing_urls: existing.append(info) else: result_feeds.append(info) return { "domain": domain, "category": category, "feeds": result_feeds, "existing": existing, } @router.post("/discover/add") async def add_discovered_sources( feeds: list[dict], request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Erkannte Feeds als Grundquellen anlegen. Erwartet eine Liste von {name, url, domain, category}. Ueberspringt bereits vorhandene URLs. """ cursor = await db.execute( "SELECT url FROM sources WHERE tenant_id IS NULL AND url IS NOT NULL" ) existing_urls = {row["url"] for row in await cursor.fetchall()} added = 0 skipped = 0 added_ids = [] for feed in feeds: if not feed.get("url"): continue if feed["url"] in existing_urls: skipped += 1 continue domain = feed.get("domain", "") cur = await db.execute( """INSERT INTO sources (name, url, domain, source_type, category, status, added_by, tenant_id) VALUES (?, ?, ?, 'rss_feed', ?, 'active', 'system', NULL)""", (feed["name"], feed["url"], domain, feed.get("category", "sonstige")), ) added_ids.append(cur.lastrowid) existing_urls.add(feed["url"]) added += 1 # Web-Source für die Domain anlegen wenn noch nicht vorhanden if feeds and feeds[0].get("domain"): domain = feeds[0]["domain"] cursor = await db.execute( "SELECT id FROM sources WHERE LOWER(domain) = ? AND source_type = 'web_source' AND tenant_id IS NULL", (domain.lower(),), ) if not await cursor.fetchone(): await db.execute( """INSERT INTO sources (name, url, domain, source_type, category, status, added_by, tenant_id) VALUES (?, ?, ?, 'web_source', ?, 'active', 'system', NULL)""", (domain_to_display_name(domain), f"https://{domain}", domain, feeds[0].get("category", "sonstige")), ) added += 1 await db.commit() if added_ids: await log_action( db, admin, get_client_ip(request), action="create", resource_type="source", after={"discovered_add": {"count": added, "ids": added_ids, "domain": feeds[0].get("domain") if feeds else None}}, ) return {"added": added, "skipped": skipped} # --- Health-Check & Vorschläge --- @router.get("/health/history") async def get_health_history( limit: int = 20, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Liefert die letzten N Health-Check-Runs aus source_health_history. Pro Run: run_id, archived_at (Run-Zeitpunkt), Counts pro Status. """ cursor = await db.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name='source_health_history' """) if not await cursor.fetchone(): return [] cursor = await db.execute(""" SELECT run_id, MIN(archived_at) AS archived_at, COUNT(*) AS total, SUM(CASE WHEN status = 'error' THEN 1 ELSE 0 END) AS errors, SUM(CASE WHEN status = 'warning' THEN 1 ELSE 0 END) AS warnings, SUM(CASE WHEN status = 'ok' THEN 1 ELSE 0 END) AS ok FROM source_health_history GROUP BY run_id ORDER BY archived_at DESC LIMIT ? """, (max(1, min(limit, 100)),)) return [dict(row) for row in await cursor.fetchall()] # Status des manuellen Health-Runs. In-Memory: nach einem Neustart wieder idle, # ein dabei abgebrochener Lauf hinterlässt höchstens einen partiellen # source_health_checks-Stand, den der nächste Lauf archiviert und heilt. _HEALTH_RUN_STATE = new_state() async def _health_run_background(): """Manueller Health-Run, exakt nach dem Vorbild des Monitor-Nachtjobs daily_source_health_check (eigene Connection, Checks + Vorschläge).""" from shared.services.source_health import run_health_checks from shared.services.source_suggester import generate_suggestions db = await get_db() try: result = await run_health_checks(db) suggestions = await generate_suggestions(db) mark_finished(_HEALTH_RUN_STATE, {**result, "suggestions": suggestions}) logger.info( "Manueller Health-Run: %s geprüft, %s Probleme, %s neue Vorschläge", result.get("checked"), result.get("issues"), suggestions, ) except Exception as e: logger.error("Manueller Health-Run fehlgeschlagen: %s", e, exc_info=True) mark_failed(_HEALTH_RUN_STATE, str(e)) finally: await db.close() @router.post("/health/run", status_code=202) async def start_health_run( background_tasks: BackgroundTasks, admin: dict = Depends(get_current_admin), ): """Health-Check + Vorschläge manuell anstoßen (läuft im Hintergrund weiter).""" if not try_start(_HEALTH_RUN_STATE): raise HTTPException(status_code=409, detail="Es läuft bereits eine Prüfung") background_tasks.add_task(_health_run_background) return {"status": "started", "started_at": _HEALTH_RUN_STATE["started_at"]} @router.get("/health/run-status") async def health_run_status(admin: dict = Depends(get_current_admin)): """Aktueller Zustand des manuellen Health-Runs (Polling-Ziel des Frontends).""" return _HEALTH_RUN_STATE @router.get("/{source_id}/health") async def get_source_health( source_id: int, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Alle Health-Checks einer einzelnen Quelle (für die Ausklapp-Zeile der Liste). Sortiert error > warning > ok, damit das Wichtigste oben steht. """ cursor = await db.execute("SELECT id FROM sources WHERE id = ?", (source_id,)) if not await cursor.fetchone(): raise HTTPException(status_code=404, detail="Quelle nicht gefunden") cursor = await db.execute(""" SELECT name FROM sqlite_master WHERE type='table' AND name='source_health_checks' """) if not await cursor.fetchone(): return {"source_id": source_id, "checks": []} cursor = await db.execute(""" SELECT check_type, status, message, checked_at FROM source_health_checks WHERE source_id = ? ORDER BY CASE status WHEN 'error' THEN 0 WHEN 'warning' THEN 1 ELSE 2 END, check_type """, (source_id,)) return {"source_id": source_id, "checks": [dict(row) for row in await cursor.fetchall()]} @router.get("/suggestions") async def get_suggestions( admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Alle OFFENEN Vorschläge plus die letzten 50 bearbeiteten. Kein Deckel auf pending, sonst liefe der Aufgaben-Badge (tasks/summary zählt ungedeckelt) der Tabelle davon. """ cursor = await db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='source_suggestions'" ) if not await cursor.fetchone(): return [] cursor = await db.execute(""" SELECT * FROM source_suggestions WHERE status = 'pending' ORDER BY created_at DESC """) pending = [dict(row) for row in await cursor.fetchall()] cursor = await db.execute(""" SELECT * FROM source_suggestions WHERE status != 'pending' ORDER BY created_at DESC LIMIT 50 """) return pending + [dict(row) for row in await cursor.fetchall()] class SuggestionAction(BaseModel): accept: bool async def _apply_suggestion_decision(db: aiosqlite.Connection, suggestion: dict, accept: bool): """Führt die Annahme/Ablehnung EINES pending-Vorschlags aus (ohne commit/Audit). Liefert (new_status, result_action). Wird vom Einzel- und vom Sammel-Endpoint identisch genutzt. """ import json as _json suggestion_id = suggestion["id"] new_status = "accepted" if accept else "rejected" result_action = None if accept: stype = suggestion["suggestion_type"] data = _json.loads(suggestion["suggested_data"]) if suggestion["suggested_data"] else {} if stype == "add_source": name = data.get("name", "Unbenannt") url = data.get("url") domain = data.get("domain", "") category = data.get("category", "sonstige") source_type = "rss_feed" if url and any( x in (url or "").lower() for x in ("rss", "feed", "xml", "atom") ) else "web_source" if url: cursor = await db.execute( "SELECT id FROM sources WHERE url = ? AND tenant_id IS NULL", (url,) ) if await cursor.fetchone(): result_action = "übersprungen (URL bereits vorhanden)" new_status = "rejected" else: await db.execute( "INSERT INTO sources (name, url, domain, source_type, category, status, added_by, tenant_id) " "VALUES (?, ?, ?, ?, ?, 'active', 'haiku-vorschlag', NULL)", (name, url, domain, source_type, category), ) result_action = f"Quelle '{name}' angelegt" else: result_action = "übersprungen (keine URL)" new_status = "rejected" elif stype == "deactivate_source": source_id = suggestion["source_id"] if source_id: await db.execute("UPDATE sources SET status = 'inactive' WHERE id = ?", (source_id,)) result_action = "Quelle deaktiviert" elif stype == "remove_source": source_id = suggestion["source_id"] if source_id: await db.execute("DELETE FROM sources WHERE id = ?", (source_id,)) result_action = "Quelle gelöscht" elif stype == "fix_url": source_id = suggestion["source_id"] new_url = data.get("url") if source_id and new_url: await db.execute("UPDATE sources SET url = ? WHERE id = ?", (new_url, source_id)) result_action = "URL aktualisiert" # Auto-Reject: Wenn fix_url oder add_source akzeptiert wird, # zugehörige deactivate_source-Vorschläge automatisch ablehnen if stype in ("fix_url", "add_source") and suggestion.get("source_id"): await db.execute( "UPDATE source_suggestions SET status = 'rejected', reviewed_at = CURRENT_TIMESTAMP " "WHERE source_id = ? AND suggestion_type = 'deactivate_source' AND status = 'pending' AND id != ?", (suggestion["source_id"], suggestion_id), ) await db.execute( "UPDATE source_suggestions SET status = ?, reviewed_at = CURRENT_TIMESTAMP WHERE id = ?", (new_status, suggestion_id), ) return new_status, result_action class BulkSuggestionAction(BaseModel): suggestion_ids: list[int] = Field(min_length=1, max_length=1000) accept: bool @router.put("/suggestions/bulk") async def bulk_update_suggestions( action: BulkSuggestionAction, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Mehrere Vorschläge auf einmal annehmen oder ablehnen. Nicht mehr pendende IDs (schon bearbeitet, per Auto-Reject erledigt oder unbekannt) werden übersprungen und gezählt, nicht als Fehler gewertet. """ accepted = 0 rejected = 0 skipped = 0 for sid in action.suggestion_ids: cursor = await db.execute("SELECT * FROM source_suggestions WHERE id = ?", (sid,)) row = await cursor.fetchone() if not row: skipped += 1 continue suggestion = dict(row) if suggestion["status"] != "pending": skipped += 1 continue new_status, _ = await _apply_suggestion_decision(db, suggestion, action.accept) if new_status == "accepted": accepted += 1 else: rejected += 1 await db.commit() await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", before={"suggestion_bulk": len(action.suggestion_ids), "accept": action.accept}, after={"accepted": accepted, "rejected": rejected, "skipped": skipped}, ) return {"accepted": accepted, "rejected": rejected, "skipped": skipped} @router.put("/suggestions/{suggestion_id}") async def update_suggestion( suggestion_id: int, action: SuggestionAction, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Vorschlag annehmen oder ablehnen.""" cursor = await db.execute( "SELECT * FROM source_suggestions WHERE id = ?", (suggestion_id,) ) suggestion = await cursor.fetchone() if not suggestion: raise HTTPException(status_code=404, detail="Vorschlag nicht gefunden") suggestion = dict(suggestion) if suggestion["status"] != "pending": raise HTTPException(status_code=400, detail=f"Vorschlag bereits {suggestion['status']}") new_status, result_action = await _apply_suggestion_decision(db, suggestion, action.accept) await db.commit() await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=suggestion.get("source_id"), before={"suggestion_id": suggestion_id, "status": "pending"}, after={"suggestion_id": suggestion_id, "status": new_status, "result_action": result_action}, ) return {"status": new_status, "action": result_action} @router.post("/health/search-fix/{source_id}") async def search_fix_for_source( source_id: int, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Sonnet mit WebSearch nach Lösung für eine kaputte Quelle suchen lassen.""" import json as _json cursor = await db.execute( "SELECT id, name, url, domain, source_type, category FROM sources WHERE id = ?", (source_id,), ) source = await cursor.fetchone() if not source: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") source = dict(source) # Health-Check-Probleme für diese Quelle laden cursor = await db.execute( "SELECT check_type, status, message FROM source_health_checks WHERE source_id = ?", (source_id,), ) issues = [dict(row) for row in await cursor.fetchall()] issues_text = "\n".join(f"- {i['check_type']}: {i['status']} - {i['message']}" for i in issues) prompt = f"""Du bist ein OSINT-Analyst. Folgende Quelle ist nicht mehr erreichbar: Name: {source['name']} URL: {source['url'] or 'keine'} Domain: {source['domain'] or 'unbekannt'} Typ: {source['source_type']} Kategorie: {source['category']} Probleme: {issues_text} Aufgabe: Suche im Internet nach funktionierenden Alternativen für diese Quelle. - Finde konkrete RSS-Feed-URLs die tatsächlich funktionieren - Prüfe ob es alternative Zugangswege gibt (andere Subdomains, Feed-Aggregatoren, alternative URLs) - Gibt es eine Lösung oder ist die Quelle nur noch per WebSearch erreichbar? Regeln: - Maximal 3 Lösungen vorschlagen (die besten) - Verwende echte deutsche Umlaute (ü, ä, ö, ß), keine Umschreibungen (ue, ae, oe, ss) Antworte NUR mit einem JSON-Objekt: {{ "fixable": true/false, "solutions": [ {{ "type": "replace_url|add_feed|deactivate", "name": "Anzeigename", "url": "https://...", "description": "Kurze Begründung" }} ], "summary": "Zusammenfassung in 1-2 Sätzen" }} Nur das JSON, kein anderer Text.""" from shared.agents.claude_client import call_claude try: response, usage = await call_claude(prompt, tools="WebSearch,WebFetch") import re json_match = re.search(r'\{.*\}', response, re.DOTALL) if json_match: result = _json.loads(json_match.group(0)) else: result = {"fixable": False, "solutions": [], "summary": response[:500]} # Lösungen als Vorschläge speichern for sol in result.get("solutions", []): sol_type = sol.get("type", "add_feed") suggestion_type = { "replace_url": "fix_url", "add_feed": "add_source", "deactivate": "deactivate_source", }.get(sol_type, "add_source") title = f"{source['name']}: {sol.get('description', sol_type)[:120]}" # Duplikat-Check: gleicher Typ + gleiche Quelle bereits pending? cursor = await db.execute( "SELECT id FROM source_suggestions WHERE suggestion_type = ? AND source_id = ? AND status = 'pending'", (suggestion_type, source_id), ) if await cursor.fetchone(): continue data = _json.dumps({ "name": sol.get("name", source["name"]), "url": sol.get("url", ""), "domain": source["domain"] or "", "category": source["category"], }, ensure_ascii=False) await db.execute( "INSERT INTO source_suggestions " "(suggestion_type, title, description, source_id, suggested_data, priority, status) " "VALUES (?, ?, ?, ?, ?, 'high', 'pending')", (suggestion_type, title, sol.get("description", ""), source_id, data), ) await db.commit() result["cost_usd"] = usage.cost_usd result["tokens"] = {"input": usage.input_tokens, "output": usage.output_tokens} return result except Exception as e: raise HTTPException(status_code=500, detail=f"Recherche fehlgeschlagen: {e}") # === Klassifikations-Review (LLM-Vorschlaege approve/reject/reclassify) === @router.get("/classification/queue") async def classification_queue( limit: int = 50, min_confidence: float = 0.0, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Liefert Quellen mit nicht-leeren proposed_*-Spalten (Review-Queue).""" cursor = await db.execute( """SELECT s.* FROM sources s WHERE s.proposed_political_orientation IS NOT NULL AND COALESCE(s.proposed_confidence, 0) >= ? ORDER BY s.proposed_confidence DESC, s.proposed_at DESC LIMIT ?""", (min_confidence, limit), ) rows = [dict(r) for r in await cursor.fetchall()] alignments_map = await _load_alignments_for(db, [r["id"] for r in rows]) out = [] for d in rows: try: proposed_aligns = json.loads(d.get("proposed_alignments_json") or "[]") except (json.JSONDecodeError, TypeError): proposed_aligns = [] out.append({ "id": d["id"], "name": d["name"], "url": d.get("url"), "domain": d.get("domain"), "source_type": d.get("source_type"), "category": d.get("category"), "is_global": d.get("tenant_id") is None, "current": { "political_orientation": d.get("political_orientation"), "media_type": d.get("media_type"), "reliability": d.get("reliability"), "state_affiliated": bool(d.get("state_affiliated")), "country_code": d.get("country_code"), "alignments": alignments_map.get(d["id"], []), "classification_source": d.get("classification_source"), }, "proposed": { "political_orientation": d.get("proposed_political_orientation"), "media_type": d.get("proposed_media_type"), "reliability": d.get("proposed_reliability"), "state_affiliated": bool(d.get("proposed_state_affiliated")), "country_code": d.get("proposed_country_code"), "alignments": proposed_aligns, "confidence": d.get("proposed_confidence"), "reasoning": d.get("proposed_reasoning"), "proposed_at": d.get("proposed_at"), }, }) return out @router.post("/{source_id}/classification/approve") async def approve_classification( source_id: int, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Uebernimmt proposed_* in echte Felder, setzt classification_source='llm_approved'.""" cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") src = dict(row) before_alignments = sorted((await _load_alignments_for(db, [source_id])).get(source_id, [])) before = {**src, "alignments": before_alignments} if src.get("proposed_political_orientation") is None: raise HTTPException(status_code=400, detail="Keine LLM-Vorschlaege fuer diese Quelle vorhanden") try: proposed_aligns = json.loads(src.get("proposed_alignments_json") or "[]") except (json.JSONDecodeError, TypeError): proposed_aligns = [] await db.execute( """UPDATE sources SET political_orientation = ?, media_type = ?, reliability = ?, state_affiliated = ?, country_code = ?, classification_source = 'llm_approved', classified_at = CURRENT_TIMESTAMP WHERE id = ?""", ( src["proposed_political_orientation"], src["proposed_media_type"], src["proposed_reliability"], 1 if src.get("proposed_state_affiliated") else 0, src.get("proposed_country_code"), source_id, ), ) await _replace_alignments(db, source_id, [a for a in proposed_aligns if a in ALIGNMENT_VALUES]) await _clear_proposed(db, source_id) await db.commit() try: await apply_reputation_overrides(db, source_id) except Exception as e: logger.warning("Reputation-Override fuer source_id=%s fehlgeschlagen: %s", source_id, e) cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) after_row = dict(await cursor.fetchone()) after_alignments = sorted((await _load_alignments_for(db, [source_id])).get(source_id, [])) after = {**after_row, "alignments": after_alignments} await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=source_id, before=before, after=after, ) return {"source_id": source_id, "status": "approved"} @router.post("/{source_id}/classification/reject") async def reject_classification( source_id: int, request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Verwirft die LLM-Vorschlaege ohne Uebernahme. classification_source: 'llm_pending' -> 'legacy'.""" cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") src = dict(row) before = dict(src) await _clear_proposed(db, source_id) if src.get("classification_source") == "llm_pending": await db.execute( "UPDATE sources SET classification_source = 'legacy' WHERE id = ?", (source_id,), ) await db.commit() cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (source_id,)) after = dict(await cursor.fetchone()) await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=source_id, before=before, after=after, ) return {"source_id": source_id, "status": "rejected"} @router.post("/{source_id}/classification/reclassify") async def reclassify_source( source_id: int, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Triggert eine LLM-Klassifikation einer einzelnen Quelle (synchron, ~3-5s).""" cursor = await db.execute("SELECT id FROM sources WHERE id = ?", (source_id,)) row = await cursor.fetchone() if not row: raise HTTPException(status_code=404, detail="Quelle nicht gefunden") try: result = await classify_source(db, source_id) except Exception as e: logger.error("Reclassify source_id=%s fehlgeschlagen: %s", source_id, e, exc_info=True) raise HTTPException(status_code=500, detail=f"Klassifikation fehlgeschlagen: {e}") return result async def _bulk_classify_background(limit: int, only_unclassified: bool): """Hintergrund-Task: oeffnet eigene DB-Connection.""" db = await get_db() try: await bulk_classify(db, limit=limit, only_unclassified=only_unclassified) finally: await db.close() @router.post("/classification/bulk-classify") async def trigger_bulk_classify( background_tasks: BackgroundTasks, limit: int = 50, only_unclassified: bool = True, admin: dict = Depends(get_current_admin), ): """Startet eine Bulk-Klassifikation im Hintergrund.""" if limit < 1 or limit > 500: raise HTTPException(status_code=400, detail="limit muss zwischen 1 und 500 liegen") background_tasks.add_task(_bulk_classify_background, limit, only_unclassified) return {"status": "started", "limit": limit, "only_unclassified": only_unclassified} @router.post("/external-reputation/sync") async def trigger_external_reputation_sync( background_tasks: BackgroundTasks, admin: dict = Depends(get_current_admin), ): """Startet Sync von IFCN- und EUvsDisinfo-Daten (Hintergrund).""" async def _bg(): db = await get_db() try: await sync_external_reputation(db) finally: await db.close() background_tasks.add_task(_bg) return {"status": "started"} @router.post("/classification/bulk-approve") async def bulk_approve_classifications( request: Request, min_confidence: float = 0.85, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), ): """Genehmigt alle Pending-Vorschlaege ueber dem confidence-Schwellwert.""" cursor = await db.execute( """SELECT id, proposed_political_orientation, proposed_media_type, proposed_reliability, proposed_state_affiliated, proposed_country_code, proposed_alignments_json FROM sources WHERE proposed_political_orientation IS NOT NULL AND COALESCE(proposed_confidence, 0) >= ?""", (min_confidence,), ) rows = [dict(r) for r in await cursor.fetchall()] approved_ids: list[int] = [] for src in rows: try: proposed_aligns = json.loads(src.get("proposed_alignments_json") or "[]") except (json.JSONDecodeError, TypeError): proposed_aligns = [] await db.execute( """UPDATE sources SET political_orientation = ?, media_type = ?, reliability = ?, state_affiliated = ?, country_code = ?, classification_source = 'llm_approved', classified_at = CURRENT_TIMESTAMP WHERE id = ?""", ( src["proposed_political_orientation"], src["proposed_media_type"], src["proposed_reliability"], 1 if src.get("proposed_state_affiliated") else 0, src.get("proposed_country_code"), src["id"], ), ) await _replace_alignments( db, src["id"], [a for a in proposed_aligns if a in ALIGNMENT_VALUES] ) await _clear_proposed(db, src["id"]) approved_ids.append(src["id"]) await db.commit() try: for sid in approved_ids: await apply_reputation_overrides(db, sid) except Exception as e: logger.warning("Bulk Reputation-Override fehlgeschlagen: %s", e) await log_action( db, admin, get_client_ip(request), action="update", resource_type="source", resource_id=None, after={"bulk_approved_ids": approved_ids, "min_confidence": min_confidence}, ) return {"approved": len(approved_ids), "ids": approved_ids} # --- PDF-Upload (Quelle vom Typ pdf_document) --- # Speicherort relativ zur DB: /pdfs/{sha256}.pdf # Der Monitor pollt pdf_document-Quellen mit processed_at IS NULL und # extrahiert Text + Uebersetzungen (DE/EN). Dieser Endpoint legt nur die # Datei + den Source-Eintrag an (kein LLM-Call hier). MAX_PDF_SIZE_BYTES = 50 * 1024 * 1024 # 50 MB PDF_DIR = os.path.join(os.path.dirname(os.path.abspath(DB_PATH)), "pdfs") def _pdf_dir() -> str: os.makedirs(PDF_DIR, exist_ok=True) return PDF_DIR @router.post("/global/upload-pdf", status_code=201) async def upload_pdf_source( request: Request, admin: dict = Depends(get_current_admin), db: aiosqlite.Connection = Depends(db_dependency), file: UploadFile = File(...), name: Optional[str] = Form(None), category: str = Form("sonstige"), language: Optional[str] = Form(None), notes: Optional[str] = Form(None), ): """PDF hochladen + als Grundquelle (source_type=pdf_document) registrieren. Idempotent ueber SHA256: bestehender Eintrag wird zurueckgegeben (409 mit Detail), die Datei wird nicht erneut gespeichert. """ # Magic-Bytes-Check (PDF beginnt mit %PDF-) head = await file.read(8) if not head.startswith(b"%PDF-"): raise HTTPException(status_code=415, detail="Datei ist kein gueltiges PDF (Magic-Bytes fehlen)") # Datei streaming in Temp lesen + sha256 berechnen + Groesse pruefen sha = hashlib.sha256() sha.update(head) total = len(head) tmp_path = os.path.join(_pdf_dir(), f".upload-{uuid.uuid4().hex}.tmp") try: with open(tmp_path, "wb") as out: out.write(head) while True: chunk = await file.read(1024 * 1024) if not chunk: break total += len(chunk) if total > MAX_PDF_SIZE_BYTES: raise HTTPException(status_code=413, detail=f"PDF ueberschreitet Maximum von {MAX_PDF_SIZE_BYTES // 1024 // 1024} MB") sha.update(chunk) out.write(chunk) sha_hex = sha.hexdigest() final_path = os.path.join(_pdf_dir(), f"{sha_hex}.pdf") rel_path = os.path.join("pdfs", f"{sha_hex}.pdf") # Duplikat-Check ueber sha256 cursor = await db.execute( "SELECT id, name FROM sources WHERE pdf_sha256 = ? AND tenant_id IS NULL", (sha_hex,), ) existing = await cursor.fetchone() if existing: # Datei wegwerfen, bestehende Quelle zurueckgeben os.unlink(tmp_path) raise HTTPException( status_code=409, detail=f"PDF bereits hochgeladen als Quelle '{existing['name']}' (id={existing['id']})", ) # Atomar umbenennen if not os.path.exists(final_path): os.replace(tmp_path, final_path) else: # Datei mit gleichem sha existiert physisch, aber keine Source -> wiederverwenden os.unlink(tmp_path) except HTTPException: if os.path.exists(tmp_path): try: os.unlink(tmp_path) except OSError: pass raise except Exception as e: if os.path.exists(tmp_path): try: os.unlink(tmp_path) except OSError: pass logger.exception("PDF-Upload fehlgeschlagen") raise HTTPException(status_code=500, detail=f"PDF-Upload fehlgeschlagen: {e}") # Name herleiten falls nicht angegeben display_name = (name or "").strip() or re.sub(r"\.pdf$", "", file.filename or "PDF", flags=re.I) display_name = display_name[:200] cursor = await db.execute( """INSERT INTO sources (name, url, domain, source_type, category, status, notes, language, pdf_path, pdf_sha256, added_by, tenant_id) VALUES (?, NULL, NULL, 'pdf_document', ?, 'active', ?, ?, ?, ?, ?, NULL)""", (display_name, category, notes, language, rel_path, sha_hex, admin.get("email") or "system"), ) src_id = cursor.lastrowid await db.commit() cursor = await db.execute("SELECT * FROM sources WHERE id = ?", (src_id,)) new_src = dict(await cursor.fetchone()) await log_action( db, admin, get_client_ip(request), action="upload_pdf", resource_type="source", resource_id=src_id, after={"name": display_name, "pdf_sha256": sha_hex, "size_bytes": total}, ) return new_src