From b686685f4387c6edd36dfcca8aae1f71589bc827 Mon Sep 17 00:00:00 2001 From: claude-dev Date: Fri, 24 Jul 2026 23:18:15 +0200 Subject: [PATCH] feat(studio): Phase 2 - Studio-Bausteine, Backend-Kern (Sammeln/Analyse/Faktencheck) Backend fuer die modularen Studio-Pipeline-Bausteine, alles ueber Claude. DB (additiv, idempotent): - Tabellen incident_events + fact_check_runs (+ Indizes) - Funktion log_incident_event() - Spalten incidents.summary_at, incidents.executive_summary, articles.geoparsed_at agents/stage_runners.py (neu): isolierte Bausteine analyze + factcheck auf dem vorhandenen Bestand, komplett neu, Historie bleibt (Snapshot/fact_check_runs). orchestrator.py: collect_only in die Lane-Multi-Tenant-Queue eingebaut (enqueue_refresh 4-Tupel, _worker-Entpacken, Multi-Pass-Gate, _run_refresh-Signatur + Analyse/FC-Skip). Zusaetzlich summary_at beim Schreiben des Lagebilds gestempelt (Studio-Freshness). incidents.py: Endpunkte POST /{id}/run/{stage} (collect|analyze|factcheck), GET /{id}/run-status, GET /{id}/freshness, GET /{id}/factcheck-runs[/{run_id}]. Busy-Check auf Online-Orchestrator (_current_tasks), Tenant-Zugriffspruefung. Kein lokaler LLM-Unterbau, kein lokales Modell. /ask + /events folgen in Phase 3. Co-Authored-By: Claude Opus 4.8 (1M context) --- src/agents/orchestrator.py | 34 +++-- src/agents/stage_runners.py | 257 ++++++++++++++++++++++++++++++++++++ src/database.py | 78 +++++++++++ src/routers/incidents.py | 180 ++++++++++++++++++++++++- 4 files changed, 538 insertions(+), 11 deletions(-) create mode 100644 src/agents/stage_runners.py diff --git a/src/agents/orchestrator.py b/src/agents/orchestrator.py index 6cfd306..f5fe8cb 100644 --- a/src/agents/orchestrator.py +++ b/src/agents/orchestrator.py @@ -458,9 +458,13 @@ class AgentOrchestrator: task.cancel() logger.info("Agenten-Orchestrator gestoppt") - async def enqueue_refresh(self, incident_id: int, trigger_type: str = "manual", user_id: int = None) -> bool: + async def enqueue_refresh(self, incident_id: int, trigger_type: str = "manual", user_id: int = None, collect_only: bool = False) -> bool: """Refresh-Auftrag in die Lane der Organisation stellen. Gibt False zurueck - wenn die Lage dort bereits wartet oder gerade laeuft.""" + wenn die Lage dort bereits wartet oder gerade laeuft. + + collect_only=True (Studio-Baustein "Sammeln"): sammelt nur Artikel, ueberspringt + Analyse und Faktencheck. + """ if incident_id in self._queued_ids or incident_id in self._current_tasks: logger.info(f"Refresh fuer Lage {incident_id} uebersprungen: bereits aktiv/in Queue") return False @@ -476,7 +480,7 @@ class AgentOrchestrator: self._lane_workers[lane_key] = asyncio.create_task(self._worker(lane_key, queue)) logger.info(f"Neue Worker-Lane fuer Organisation {lane_key} gestartet") self._queued_ids.add(incident_id) - queue.put_nowait((incident_id, trigger_type, user_id)) + queue.put_nowait((incident_id, trigger_type, user_id, collect_only)) queue_size = queue.qsize() logger.info(f"Refresh fuer Lage {incident_id} eingereiht (Lane {lane_key}, Queue: {queue_size}, Trigger: {trigger_type})") @@ -582,11 +586,15 @@ class AgentOrchestrator: return continue - if len(item) == 3: + if len(item) == 4: + incident_id, trigger_type, user_id, collect_only = item + elif len(item) == 3: incident_id, trigger_type, user_id = item + collect_only = False else: incident_id, trigger_type = item user_id = None + collect_only = False self._queued_ids.discard(incident_id) # Session-Start EINMAL setzen — bleibt ueber Multi-Pass/Retry hinweg stabil self._current_tasks[incident_id] = datetime.utcnow().strftime('%Y-%m-%dT%H:%M:%SZ') @@ -610,14 +618,14 @@ class AgentOrchestrator: try: # Research-Lagen: Automatisch 3 Durchläufe nur beim ersten Refresh incident_type, has_summary = await self._get_incident_info(incident_id) - use_multi_pass = incident_type == "research" and not has_summary + use_multi_pass = incident_type == "research" and not has_summary and not collect_only for attempt in range(3): try: if use_multi_pass: await self._run_research_multi_pass(incident_id, trigger_type=trigger_type, user_id=user_id) else: - await self._run_refresh(incident_id, trigger_type=trigger_type, retry_count=attempt, user_id=user_id) + await self._run_refresh(incident_id, trigger_type=trigger_type, retry_count=attempt, user_id=user_id, collect_only=collect_only) last_error = None break # Erfolg except asyncio.CancelledError: @@ -783,7 +791,7 @@ class AgentOrchestrator: await db.close() return visibility, created_by, tenant_id - async def _run_refresh(self, incident_id: int, trigger_type: str = "manual", retry_count: int = 0, user_id: int = None, _suppress_complete: bool = False, _pass_info: dict = None): + async def _run_refresh(self, incident_id: int, trigger_type: str = "manual", retry_count: int = 0, user_id: int = None, _suppress_complete: bool = False, _pass_info: dict = None, collect_only: bool = False): """Führt einen kompletten Refresh-Zyklus durch.""" import aiosqlite from database import get_db @@ -1433,7 +1441,8 @@ class AgentOrchestrator: logger.warning(f"Quellen-Statistiken konnten nicht aktualisiert werden: {e}") # Schritt 3+4: Analyse und Faktencheck PARALLEL - if new_count > 0 or not previous_summary: + # collect_only (Studio-Baustein "Sammeln"): ueberspringt Analyse/Faktencheck. + if (new_count > 0 or not previous_summary) and not collect_only: is_first_summary = not previous_summary # Snapshot des alten Lagebilds sichern BEVOR parallele Verarbeitung startet @@ -1775,9 +1784,14 @@ class AgentOrchestrator: pass sources_json = json.dumps(sources, ensure_ascii=False) if sources else previous_sources_json + # summary_at haelt fest, WANN das Lagebild entstand (Studio-Freshness). + # JETZT stempeln, nicht 'now' vom Lauf-Beginn: sonst saehen die in diesem + # Lauf gesammelten Artikel neuer aus als der Bericht, der sie schon enthaelt. + summary_now = datetime.now(TIMEZONE).strftime('%Y-%m-%d %H:%M:%S') await db.execute( - "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, updated_at = ? WHERE id = ?", - (new_summary, sources_json, now, incident_id), + "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, " + "updated_at = ?, summary_at = ? WHERE id = ?", + (new_summary, sources_json, summary_now, summary_now, incident_id), ) # Beim ersten Refresh: Snapshot des neuen Lagebilds erstellen diff --git a/src/agents/stage_runners.py b/src/agents/stage_runners.py new file mode 100644 index 0000000..9e604e4 --- /dev/null +++ b/src/agents/stage_runners.py @@ -0,0 +1,257 @@ +""" +Modulare Pipeline-Bausteine fuers Studio. + +Fuehrt einzelne Pipeline-Stufen ISOLIERT auf dem vorhandenen DB-Bestand aus, +statt des durchlaufenden orchestrator._run_refresh. Das grosse _run_refresh +bleibt unangetastet (dient weiter als "kompletter Lauf"). + +Nutzt die bereits reinen Agenten (AnalyzerAgent/FactCheckerAgent) und bildet nur +die Lade-/Persistenz-Logik aus _run_refresh nach. + +Verhalten: KOMPLETT NEU (ersetzt das aktive Ergebnis), Historie bleibt einsehbar: + - Analyse -> altes Lagebild wird als incident_snapshots archiviert, dann neu gesetzt + - Faktencheck -> alter Faktenstand wird als fact_check_runs archiviert, dann ersetzt + +Phase 1: analyze, factcheck. (Spaeter: collect, geoparse, network.) +""" +from __future__ import annotations + +import asyncio +import json +import logging +from datetime import datetime + +from config import TIMEZONE +from database import get_db + +logger = logging.getLogger("osint.stages") + +# Im Studio einzeln startbare Bausteine (Phase 1) +STAGES = {"analyze", "factcheck"} + +# Status-Registry je Incident (Single-Flight) + Task-Referenzen (GC-Schutz) +_STATE: dict[int, dict] = {} +_TASKS: set = set() + + +def _now() -> str: + return datetime.now(TIMEZONE).strftime("%Y-%m-%d %H:%M:%S") + + +def get_state(incident_id: int) -> dict | None: + return _STATE.get(incident_id) + + +def is_running(incident_id: int) -> bool: + st = _STATE.get(incident_id) + return bool(st and st.get("status") == "running") + + +async def _load_incident(db, incident_id: int) -> dict | None: + cur = await db.execute("SELECT * FROM incidents WHERE id = ?", (incident_id,)) + row = await cur.fetchone() + return dict(row) if row else None + + +async def _output_language(db, tenant_id) -> str: + from services.org_settings import get_org_language, language_display + iso = await get_org_language(db, tenant_id) if tenant_id else "de" + return language_display(iso) + + +async def _count(db, sql: str, params) -> int: + row = await (await db.execute(sql, params)).fetchone() + return (row[0] if row else 0) or 0 + + +# --------------------------------------------------------------------------- +# Baustein: Analyse (Lagebild / Recherchebericht) — komplett neu +# --------------------------------------------------------------------------- +async def _run_analysis(db, inc: dict, user_id) -> dict: + from agents.analyzer import AnalyzerAgent, build_fact_context_block + from services.license_service import charge_usage_to_tenant + + incident_id = inc["id"] + tenant_id = inc.get("tenant_id") + incident_type = inc.get("type") or "adhoc" + title = inc.get("title") or "" + description = inc.get("description") or "" + prev_summary = inc.get("summary") or "" + prev_sources = inc.get("sources_json") + now = _now() + output_language = await _output_language(db, tenant_id) + + # 1) Altes Lagebild als Snapshot archivieren (Historie), bevor es ersetzt wird + if prev_summary: + acnt = await _count(db, "SELECT COUNT(*) FROM articles WHERE incident_id = ?", (incident_id,)) + fcnt = await _count(db, "SELECT COUNT(*) FROM fact_checks WHERE incident_id = ?", (incident_id,)) + await db.execute( + """INSERT INTO incident_snapshots + (incident_id, summary, sources_json, article_count, fact_check_count, + refresh_log_id, created_at, tenant_id) + VALUES (?,?,?,?,?,?,?,?)""", + (incident_id, prev_summary, prev_sources, acnt, fcnt, None, now, tenant_id), + ) + await db.commit() + + # 2) Alle Artikel + bestehende Fakten laden (Frische-Bias bei adhoc) + order = "published_at IS NULL, published_at DESC" if incident_type == "adhoc" else "collected_at DESC" + arts = [dict(r) for r in await (await db.execute( + f"SELECT * FROM articles WHERE incident_id = ? ORDER BY {order}", (incident_id,) + )).fetchall()] + if not arts: + raise ValueError("Keine Artikel vorhanden — bitte zuerst sammeln.") + + facts = [dict(r) for r in await (await db.execute( + "SELECT id, claim, status, sources_count, evidence FROM fact_checks WHERE incident_id = ?", + (incident_id,), + )).fetchall()] + fact_ctx = "" + try: + fact_ctx = build_fact_context_block(facts, [], incident_type) + except Exception as e: + logger.warning(f"Faktenkontext fuer Analyse fehlgeschlagen: {e}") + + # 3) Analyse komplett neu ueber ALLE Artikel + analyzer = AnalyzerAgent() + analysis, usage = await analyzer.analyze( + title, description, arts, incident_type, + fact_context_block=fact_ctx, output_language=output_language, + ) + if usage: + try: + await charge_usage_to_tenant(db, tenant_id, usage, source="analysis") + except Exception: + pass + if not analysis or not (analysis.get("summary") or "").strip(): + raise ValueError("Analyse lieferte kein Lagebild.") + + summary = analysis.get("summary") or "" + sources = analysis.get("sources") or [] + for s in sources: + if isinstance(s.get("nr"), str): + try: + s["nr"] = int(s["nr"]) + except ValueError: + pass + sources_json = json.dumps(sources, ensure_ascii=False) if sources else prev_sources + + # summary_at = Entstehungszeit des Lagebilds. JETZT stempeln, nicht 'now' vom Beginn + # des Bausteins: die Analyse laeuft Minuten, und der Stempel muss den Stand abdecken, + # der tatsaechlich verarbeitet wurde. (updated_at wird auch beim Sammeln gesetzt und + # taugt als Bericht-Zeitpunkt ohnehin nicht.) + summary_now = _now() + await db.execute( + "UPDATE incidents SET summary = ?, sources_json = ?, executive_summary = NULL, " + "updated_at = ?, summary_at = ? WHERE id = ?", + (summary, sources_json, summary_now, summary_now, incident_id), + ) + await db.commit() + return {"articles": len(arts), "summary_len": len(summary), "sources": len(sources)} + + +# --------------------------------------------------------------------------- +# Baustein: Faktencheck — komplett neu (alter Stand als Lauf archiviert) +# --------------------------------------------------------------------------- +async def _run_factcheck(db, inc: dict, user_id) -> dict: + from agents.factchecker import FactCheckerAgent, deduplicate_new_facts + from services.license_service import charge_usage_to_tenant + + incident_id = inc["id"] + tenant_id = inc.get("tenant_id") + incident_type = inc.get("type") or "adhoc" + title = inc.get("title") or "" + now = _now() + output_language = await _output_language(db, tenant_id) + + # 1) Aktuellen Faktenstand als Lauf archivieren (Historie) + cur_facts = [dict(r) for r in await (await db.execute( + "SELECT claim, status, sources_count, evidence, is_notification, checked_at, status_history " + "FROM fact_checks WHERE incident_id = ? ORDER BY id", (incident_id,) + )).fetchall()] + if cur_facts: + await db.execute( + "INSERT INTO fact_check_runs (incident_id, tenant_id, created_at, facts_json, fact_count) VALUES (?,?,?,?,?)", + (incident_id, tenant_id, now, json.dumps(cur_facts, ensure_ascii=False), len(cur_facts)), + ) + await db.commit() + + # 2) Alle Artikel laden + arts = [dict(r) for r in await (await db.execute( + "SELECT * FROM articles WHERE incident_id = ? ORDER BY collected_at DESC", (incident_id,) + )).fetchall()] + if not arts: + raise ValueError("Keine Artikel vorhanden — bitte zuerst sammeln.") + + # 3) Faktencheck komplett neu + fc = FactCheckerAgent() + facts, usage = await fc.check(title, arts, incident_type, output_language=output_language) + if usage: + try: + await charge_usage_to_tenant(db, tenant_id, usage, source="factcheck") + except Exception: + pass + facts = deduplicate_new_facts(facts or []) + + # 4) Bestehende Fakten ersetzen + await db.execute("DELETE FROM fact_checks WHERE incident_id = ?", (incident_id,)) + for f in facts: + init_hist = json.dumps([{"status": f.get("status", "developing"), "at": now}]) + await db.execute( + """INSERT INTO fact_checks + (incident_id, claim, status, sources_count, evidence, is_notification, tenant_id, status_history, checked_at) + VALUES (?,?,?,?,?,?,?,?,?)""", + (incident_id, f.get("claim", ""), f.get("status", "developing"), + f.get("sources_count", 0), f.get("evidence"), f.get("is_notification", 0), + tenant_id, init_hist, now), + ) + await db.commit() + return {"facts": len(facts), "archived": len(cur_facts), "articles": len(arts)} + + +# --------------------------------------------------------------------------- +# Orchestrierung: Start (Single-Flight) + Status +# --------------------------------------------------------------------------- +_RUNNERS = {"analyze": _run_analysis, "factcheck": _run_factcheck} + +_LABELS = {"analyze": "Analyse", "factcheck": "Faktencheck"} + + +def start_stage(incident_id: int, stage: str, user_id) -> bool: + """Startet einen Baustein im Hintergrund. False, wenn schon einer laeuft.""" + if stage not in STAGES: + raise ValueError(f"Unbekannter Baustein: {stage}") + if is_running(incident_id): + return False + _STATE[incident_id] = { + "stage": stage, "label": _LABELS.get(stage, stage), + "status": "running", "started_at": _now(), + "finished_at": None, "error": None, "result": None, + } + t = asyncio.create_task(_execute(incident_id, stage, user_id)) + _TASKS.add(t) + t.add_done_callback(_TASKS.discard) + return True + + +async def _execute(incident_id: int, stage: str, user_id): + db = await get_db() + started_at = _STATE.get(incident_id, {}).get("started_at") + try: + inc = await _load_incident(db, incident_id) + if not inc: + raise ValueError("Lage nicht gefunden") + res = await _RUNNERS[stage](db, inc, user_id) + _STATE[incident_id] = { + "stage": stage, "label": _LABELS.get(stage, stage), "status": "done", + "started_at": started_at, "finished_at": _now(), "error": None, "result": res, + } + logger.info(f"Baustein {stage} Lage {incident_id} fertig: {res}") + except Exception as e: + logger.warning(f"Baustein {stage} Lage {incident_id} Fehler: {e}", exc_info=True) + _STATE[incident_id] = { + "stage": stage, "label": _LABELS.get(stage, stage), "status": "error", + "started_at": started_at, "finished_at": _now(), "error": str(e)[:400], "result": None, + } + finally: + await db.close() diff --git a/src/database.py b/src/database.py index 9f3ae88..3ac64e9 100644 --- a/src/database.py +++ b/src/database.py @@ -1,5 +1,6 @@ """SQLite Datenbank-Setup und Zugriff.""" import aiosqlite +import json import logging import os from config import DB_PATH, DATA_DIR @@ -133,6 +134,32 @@ CREATE TABLE IF NOT EXISTS refresh_pipeline_steps ( CREATE INDEX IF NOT EXISTS idx_pipeline_steps_incident ON refresh_pipeline_steps(incident_id, started_at DESC); CREATE INDEX IF NOT EXISTS idx_pipeline_steps_log ON refresh_pipeline_steps(refresh_log_id); +-- Aktivitaets-/Ereignisprotokoll einer Lage (Studio-Ereignis-Timeline). +-- Erfasst nur Ereignisse, die sonst nirgends stehen: Chat-Q&A + Quellen-Aenderungen. +CREATE TABLE IF NOT EXISTS incident_events ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE, + event_type TEXT NOT NULL, -- 'chat_qa' | 'source_change' + title TEXT, + detail TEXT, + meta TEXT, -- optionales JSON + user_id INTEGER REFERENCES users(id) ON DELETE SET NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + tenant_id INTEGER REFERENCES organizations(id) +); +CREATE INDEX IF NOT EXISTS idx_incident_events ON incident_events(incident_id, created_at DESC); + +-- Archivierte Faktencheck-Laeufe (Studio-Faktencheck-Verlauf). +CREATE TABLE IF NOT EXISTS fact_check_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE, + tenant_id INTEGER REFERENCES organizations(id), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + facts_json TEXT, + fact_count INTEGER DEFAULT 0 +); +CREATE INDEX IF NOT EXISTS idx_fact_check_runs ON fact_check_runs(incident_id, created_at DESC); + CREATE TABLE IF NOT EXISTS incident_snapshots ( id INTEGER PRIMARY KEY AUTOINCREMENT, incident_id INTEGER REFERENCES incidents(id) ON DELETE CASCADE, @@ -369,6 +396,28 @@ async def get_db() -> aiosqlite.Connection: return db +async def log_incident_event(db, incident_id, event_type, title, detail=None, + meta=None, user_id=None, tenant_id=None, commit=True): + """Schreibt ein Ereignis ins Aktivitaetsprotokoll einer Lage (Studio-Timeline). + + Bewusst tolerant: Fehler beim Protokollieren duerfen die eigentliche Aktion + (Chat-Antwort, Quelle anlegen) niemals scheitern lassen. + """ + try: + await db.execute( + """INSERT INTO incident_events + (incident_id, event_type, title, detail, meta, user_id, tenant_id) + VALUES (?, ?, ?, ?, ?, ?, ?)""", + (incident_id, event_type, title, detail, + json.dumps(meta, ensure_ascii=False) if meta is not None else None, + user_id, tenant_id), + ) + if commit: + await db.commit() + except Exception as e: # pragma: no cover - Protokoll ist nie kritisch + logger.warning(f"incident_event nicht protokolliert (incident={incident_id}, typ={event_type}): {e}") + + async def init_db(): """Initialisiert die Datenbank mit dem Schema.""" db = await get_db() @@ -444,6 +493,24 @@ async def init_db(): await db.commit() logger.info("Migration: public_mood_updated_at zu incidents hinzugefuegt") + # Migration (Studio): summary_at = Entstehungszeit des Lagebilds. Grundlage der + # Studio-Anzeige "N neue Artikel seit dem letzten Bericht". updated_at taugt dafuer + # nicht (wird auch beim reinen Sammeln gesetzt). Backfill mit updated_at genuegt. + if "summary_at" not in columns: + await db.execute("ALTER TABLE incidents ADD COLUMN summary_at TEXT") + await db.execute( + "UPDATE incidents SET summary_at = updated_at " + "WHERE summary IS NOT NULL AND TRIM(summary) <> ''" + ) + await db.commit() + logger.info("Migration: summary_at zu incidents hinzugefuegt (Studio)") + + # Migration (Studio): executive_summary (stage_runners setzt es beim Analyse-Baustein) + if "executive_summary" not in columns: + await db.execute("ALTER TABLE incidents ADD COLUMN executive_summary TEXT") + await db.commit() + logger.info("Migration: executive_summary zu incidents hinzugefuegt (Studio)") + # Migration: Tabelle podcast_transcripts (URL-Cache fuer Transkripte) cursor = await db.execute( "SELECT name FROM sqlite_master WHERE type='table' AND name='podcast_transcripts'" @@ -606,6 +673,17 @@ async def init_db(): await db.execute("ALTER TABLE articles ADD COLUMN tenant_id INTEGER REFERENCES organizations(id)") await db.commit() + # Migration (Studio): geoparsed_at fuer articles (Merker "schon verortet", + # Grundlage der Studio-Freshness fuer Geoparsing). Backfill aus article_locations. + if "geoparsed_at" not in art_columns: + await db.execute("ALTER TABLE articles ADD COLUMN geoparsed_at TEXT") + await db.execute( + """UPDATE articles SET geoparsed_at = COALESCE(collected_at, CURRENT_TIMESTAMP) + WHERE id IN (SELECT DISTINCT article_id FROM article_locations)""" + ) + await db.commit() + logger.info("Migration: geoparsed_at zu articles hinzugefuegt (Studio)") + # Migration: tenant_id fuer fact_checks cursor = await db.execute("PRAGMA table_info(fact_checks)") fc_columns = [row[1] for row in await cursor.fetchall()] diff --git a/src/routers/incidents.py b/src/routers/incidents.py index f9c6144..a6aa6b7 100644 --- a/src/routers/incidents.py +++ b/src/routers/incidents.py @@ -5,7 +5,7 @@ from models import IncidentCreate, IncidentUpdate, IncidentResponse, IncidentLis from auth import get_current_user from middleware.license_check import require_writable_license from database import db_dependency, get_db -from datetime import datetime +from datetime import datetime, timezone from config import TIMEZONE import asyncio import aiosqlite @@ -1128,6 +1128,184 @@ async def cancel_refresh( return {"status": "cancelling" if cancelled else "not_running"} +# --- Modulare Pipeline-Bausteine (nur Studio): einzelne Stufen isoliert ------ +@router.post("/{incident_id}/run/{stage}") +async def run_stage( + incident_id: int, + stage: str, + current_user: dict = Depends(require_writable_license), + db: aiosqlite.Connection = Depends(db_dependency), +): + """Einzelnen Pipeline-Baustein auf dem vorhandenen Datenbestand starten + (collect | analyze | factcheck). Komplett neu; Historie bleibt erhalten.""" + from agents import stage_runners + from agents.orchestrator import orchestrator + user_id = current_user["id"] + tenant_id = current_user.get("tenant_id") + await _check_incident_access(db, incident_id, user_id, tenant_id) + + orch_busy = (incident_id in getattr(orchestrator, "_current_tasks", {})) or \ + (incident_id in getattr(orchestrator, "_queued_ids", set())) + busy_msg = "Es laeuft bereits eine Aktualisierung/ein Baustein fuer diese Lage." + + # "Sammeln" nutzt die bewaehrte Sammel-Pipeline (Orchestrator, collect_only) + if stage == "collect": + if stage_runners.is_running(incident_id): + raise HTTPException(status_code=409, detail=busy_msg) + ok = await orchestrator.enqueue_refresh( + incident_id, trigger_type="collect", user_id=user_id, collect_only=True) + if not ok: + raise HTTPException(status_code=409, detail=busy_msg) + return {"started": True, "stage": "collect", "via": "refresh"} + + # Analyse/Faktencheck: isolierte Bausteine (stage_runners) + if stage not in stage_runners.STAGES: + raise HTTPException(status_code=400, detail=f"Unbekannter Baustein: {stage}") + if orch_busy: + raise HTTPException(status_code=409, detail=busy_msg) + started = stage_runners.start_stage(incident_id, stage, user_id) + if not started: + raise HTTPException(status_code=409, detail=busy_msg) + return {"started": True, "stage": stage} + + +def _app_ts_to_utc(ts) -> str | None: + """App-Zeitstempel (lokale Zeitzone) -> UTC-String, fuer den Vergleich mit + articles.collected_at (das SQLite in UTC setzt). Ohne Umrechnung waere der + Vergleich im Sommer zwei Stunden falsch.""" + if not ts: + return None + s = str(ts).strip().replace("T", " ")[:19] + try: + dt = datetime.strptime(s, "%Y-%m-%d %H:%M:%S") + except ValueError: + return None + return dt.replace(tzinfo=TIMEZONE).astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%S") + + +@router.get("/{incident_id}/freshness") +async def get_freshness( + incident_id: int, + current_user: dict = Depends(get_current_user), + db: aiosqlite.Connection = Depends(db_dependency), +): + """Datenstand je Artefakt: erzeugt? wie alt? wie viele Artikel kamen seither dazu? + + Grundlage fuer die Veraltet-Anzeige der Studio-Karten ("12 neue Artikel seit dem + letzten Lagebild"). + """ + await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id")) + + async def one(sql: str, params=()): + row = await (await db.execute(sql, params)).fetchone() + return row[0] if row else None + + async def newer_than(app_ts) -> int: + utc = _app_ts_to_utc(app_ts) + if not utc: + return 0 + return await one( + "SELECT COUNT(*) FROM articles WHERE incident_id = ? AND collected_at > ?", + (incident_id, utc), + ) or 0 + + articles = await one("SELECT COUNT(*) FROM articles WHERE incident_id = ?", (incident_id,)) or 0 + + inc = await (await db.execute( + "SELECT summary, summary_at, updated_at FROM incidents WHERE id = ?", (incident_id,) + )).fetchone() + has_summary = bool(inc and (inc["summary"] or "").strip()) + # summary_at ist die Entstehungszeit des Lagebilds; updated_at nur der Notnagel. + summary_at = (inc["summary_at"] or inc["updated_at"]) if inc else None + + facts = await one("SELECT COUNT(*) FROM fact_checks WHERE incident_id = ?", (incident_id,)) or 0 + fc_at = await one("SELECT MAX(checked_at) FROM fact_checks WHERE incident_id = ?", (incident_id,)) + + geo_pending = await one( + "SELECT COUNT(*) FROM articles WHERE incident_id = ? AND geoparsed_at IS NULL", + (incident_id,), + ) or 0 + geo_at = await one( + "SELECT MAX(geoparsed_at) FROM articles WHERE incident_id = ?", (incident_id,) + ) + + return { + "articles": articles, + "summary": { + "exists": has_summary, + "last": summary_at if has_summary else None, + "pending": await newer_than(summary_at) if has_summary else articles, + }, + "factcheck": { + "exists": facts > 0, + "facts": facts, + "last": fc_at, + "pending": await newer_than(fc_at) if facts else articles, + }, + "geoparse": { + "exists": geo_at is not None, + "last": geo_at, + "pending": geo_pending, + }, + "snapshots": await one( + "SELECT COUNT(*) FROM incident_snapshots WHERE incident_id = ?", (incident_id,) + ) or 0, + "events": await one( + "SELECT COUNT(*) FROM incident_events WHERE incident_id = ?", (incident_id,) + ) or 0, + } + + +@router.get("/{incident_id}/run-status") +async def run_stage_status( + incident_id: int, + current_user: dict = Depends(get_current_user), + db: aiosqlite.Connection = Depends(db_dependency), +): + """Status des zuletzt/aktuell laufenden Bausteins.""" + from agents import stage_runners + await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id")) + return {"state": stage_runners.get_state(incident_id)} + + +@router.get("/{incident_id}/factcheck-runs") +async def list_factcheck_runs( + incident_id: int, + current_user: dict = Depends(get_current_user), + db: aiosqlite.Connection = Depends(db_dependency), +): + """Faktencheck-Historie (archivierte Laeufe mit Zeitstempel).""" + await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id")) + cur = await db.execute( + "SELECT id, created_at, fact_count FROM fact_check_runs WHERE incident_id = ? ORDER BY created_at DESC, id DESC LIMIT 50", + (incident_id,), + ) + return {"runs": [dict(r) for r in await cur.fetchall()]} + + +@router.get("/{incident_id}/factcheck-runs/{run_id}") +async def factcheck_run_detail( + incident_id: int, + run_id: int, + current_user: dict = Depends(get_current_user), + db: aiosqlite.Connection = Depends(db_dependency), +): + """Faktenstand eines archivierten Laufs.""" + await _check_incident_access(db, incident_id, current_user["id"], current_user.get("tenant_id")) + cur = await db.execute( + "SELECT facts_json, created_at FROM fact_check_runs WHERE id = ? AND incident_id = ?", + (run_id, incident_id), + ) + row = await cur.fetchone() + if not row: + raise HTTPException(status_code=404, detail="Lauf nicht gefunden.") + try: + facts = json.loads(row["facts_json"] or "[]") + except (ValueError, TypeError): + facts = [] + return {"created_at": row["created_at"], "facts": facts} + + def _slugify(text: str) -> str: """Dateinamen-sicherer Slug aus Titel."""