AegisSight-Monitor-Verwaltung/src/routers/sources.py

"""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


@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."""
    import json as _json

    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 = "accepted" if action.accept else "rejected"
    result_action = None

    if action.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),
    )
    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: <dirname(DB_PATH)>/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