shira-hermes/api/routes/admin_kb.py

"""Admin endpoints for managing the Israeli National Insurance KB."""

from __future__ import annotations

import asyncio
import logging
import os
from typing import Literal

from fastapi import APIRouter, File, Form, HTTPException, Request, UploadFile
from pydantic import BaseModel, Field

from api.services.kb import admin_sources as kb_admin_sources
from api.services.kb import ingest as kb_ingest
from api.services.kb import ingest_jobs as kb_jobs
from api.services.kb import labels as kb_labels
from api.services.kb import s3 as kb_s3
from api.services.kb import topics as kb_topics
from api.services.kb.db import get_pool

# Arbitrary non-zero constant used as a pg_try_advisory_lock key. Any int fits;
# this one is a stable hash of "kb-scan-inbox" so it can't collide with any
# other advisory lock chosen the same way for a different purpose.
_SCAN_LOCK_KEY = 0x4B42_5343  # "KBSC" in ASCII

logger = logging.getLogger("shira.admin.kb")

router = APIRouter(prefix="/admin/kb", tags=["admin-kb"])


def _verify_admin(request: Request) -> None:
    expected = os.environ.get("ADMIN_API_KEY") or os.environ.get("API_KEY", "")
    if not expected:
        raise HTTPException(status_code=503, detail="Admin API key not configured")
    provided = request.headers.get("X-Admin-Key") or request.headers.get("X-Api-Key") or ""
    if provided != expected:
        raise HTTPException(status_code=401, detail="Unauthorized")


@router.post("/ingest")
async def ingest(
    request: Request,
    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
    title: str = Form(...),
    identifier: str | None = Form(None),
    published_at: str | None = Form(None),
    effective_at: str | None = Form(None),
    source_url: str | None = Form(None),
    original_path: str | None = Form(None),
    file: UploadFile = File(...),
):
    _verify_admin(request)
    data = await file.read()
    if not data:
        raise HTTPException(status_code=400, detail="Empty file")
    try:
        result = await kb_ingest.ingest_source(
            kind=kind,
            title=title,
            identifier=identifier,
            content=data,
            filename=file.filename or "upload",
            published_at=published_at,
            effective_at=effective_at,
            source_url=source_url,
            original_path=original_path,
        )
    except kb_ingest.IngestError as e:
        raise HTTPException(status_code=422, detail=str(e))
    except Exception as e:
        logger.exception("[admin.kb] ingest failed")
        raise HTTPException(status_code=500, detail=str(e))
    return result


@router.get("/stats")
async def stats(request: Request):
    _verify_admin(request)
    return await kb_ingest.stats()


@router.get("/inbox")
async def list_inbox(request: Request):
    _verify_admin(request)
    return {"items": kb_s3.list_inbox()}


def _derive_metadata(filename: str, kind: str) -> dict:
    """Best-effort metadata from the filename stem. User can override later."""
    stem = filename.rsplit("/", 1)[-1]
    for ext in (".pdf", ".docx", ".txt", ".md"):
        if stem.lower().endswith(ext):
            stem = stem[: -len(ext)]
            break
    return {
        "title": stem,
        "identifier": None,
        "published_at": None,
        "effective_at": None,
        "source_url": None,
    }


def _sidecar_metadata(item_key: str) -> dict:
    """Fetch the optional `<key>.meta.json` sidecar and return parsed fields.

    The sidecar is expected to be a JSON object with any of:
      {title, identifier, published_at, effective_at, source_url}

    Missing or unreadable sidecars return {}.
    """
    import json
    sidecar_key = item_key + ".meta.json"
    try:
        data = kb_s3.fetch(sidecar_key)
    except Exception:
        return {}
    try:
        parsed = json.loads(data.decode("utf-8"))
    except Exception as e:
        logger.warning("[admin.kb] bad sidecar %s: %s", sidecar_key, e)
        return {}
    # Accept only whitelisted keys — ignore typos / extra fields.
    allowed = {"title", "identifier", "published_at", "effective_at", "source_url"}
    return {k: v for k, v in parsed.items() if k in allowed and v}


@router.post("/requeue-failed")
async def requeue_failed(request: Request):
    """Move everything under failed/<kind>/ back to inbox/<kind>/ for retry.

    The sidecar `<name>.meta.json` (if present) is moved alongside.
    """
    _verify_admin(request)
    items = kb_s3.list_failed()
    moved = []
    for item in items:
        src = item["key"]
        dst = f"inbox/{item['kind']}/{item['filename']}"
        try:
            kb_s3.move(src, dst)
            # Best-effort sidecar move.
            try:
                kb_s3.move(src + ".meta.json", dst + ".meta.json")
            except Exception:
                pass
            moved.append({"from": src, "to": dst})
        except Exception as e:
            moved.append({"from": src, "error": str(e)})
    return {"requeued": len([m for m in moved if "to" in m]), "items": moved}


@router.post("/scan-inbox")
async def scan_inbox(request: Request):
    """Scan the inbox and ingest each pending file.

    Protected by a session-level Postgres advisory lock so two concurrent
    callers (e.g. overlapping n8n cron runs after a worker outage) can't both
    process the same file and create duplicate sources. A caller that fails
    to acquire the lock gets `skipped: lock held`.
    """
    _verify_admin(request)
    pool = await get_pool()
    async with pool.acquire() as conn:
        got_lock = await conn.fetchval(
            "SELECT pg_try_advisory_lock($1)", _SCAN_LOCK_KEY
        )
        if not got_lock:
            logger.info("[scan-inbox] skipped: another scan is in progress")
            return {"processed": 0, "failed": 0, "skipped": True, "items": []}
        try:
            return await _do_scan()
        finally:
            await conn.execute(
                "SELECT pg_advisory_unlock($1)", _SCAN_LOCK_KEY
            )


async def _do_scan() -> dict:
    items = kb_s3.list_inbox()
    results = []
    for item in items:
        src_key = item["key"]
        kind = item["kind"]
        filename = item["filename"]
        # Merge: filename-derived defaults < sidecar overrides.
        meta = _derive_metadata(filename, kind)
        meta.update(_sidecar_metadata(src_key))
        try:
            data = kb_s3.fetch(src_key)
            result = await kb_ingest.ingest_source(
                kind=kind,
                title=meta["title"],
                identifier=meta.get("identifier"),
                content=data,
                filename=filename,
                published_at=meta.get("published_at"),
                effective_at=meta.get("effective_at"),
                source_url=meta.get("source_url"),
                original_path=f"s3://{kb_s3._bucket()}/processed/{kind}/{filename}",
            )
            dst = kb_s3.mark_processed(src_key, filename, kind)
            results.append({
                "key": src_key,
                "status": "processed",
                "destination": dst,
                **result,
            })
        except Exception as e:
            logger.exception("[scan-inbox] failed for %s", src_key)
            dst = kb_s3.mark_failed(src_key, filename, kind, str(e))
            results.append({
                "key": src_key,
                "status": "failed",
                "destination": dst,
                "error": str(e),
            })
    return {"processed": sum(1 for r in results if r["status"] == "processed"),
            "failed": sum(1 for r in results if r["status"] == "failed"),
            "items": results}


# ── Phase 2: async upload + job tracking (Task #13) ────────────────────────

# Bytes; cap upload size so a stray 500MB scan doesn't OOM the API container.
_MAX_UPLOAD_BYTES = 50 * 1024 * 1024


@router.post("/upload")
async def upload(
    request: Request,
    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
    topic_id: int = Form(...),
    title: str | None = Form(None),
    identifier: str | None = Form(None),
    published_at: str | None = Form(None),
    effective_at: str | None = Form(None),
    source_url: str | None = Form(None),
    file: UploadFile = File(...),
):
    """Async upload: queue a kb_ingest_job, return immediately with job_id.

    The browser polls `/admin/kb/jobs/{id}` until it sees a terminal state.
    Errors during parse/embed are recorded on the job row, not bubbled here.
    """
    _verify_admin(request)

    data = await file.read()
    if not data:
        raise HTTPException(status_code=400, detail="Empty file")
    if len(data) > _MAX_UPLOAD_BYTES:
        raise HTTPException(
            status_code=413,
            detail=f"File too large ({len(data)} bytes; max {_MAX_UPLOAD_BYTES})",
        )

    # Validate the topic exists and is active. Using resolve_topic_id
    # mirrors what /kb/ask and /kb/search already do.
    try:
        topic = await kb_topics.get_topic(topic_id)
    except Exception:
        topic = None
    if not topic or not topic.get("is_active"):
        raise HTTPException(status_code=400, detail=f"Unknown or inactive topic_id={topic_id}")

    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )

    metadata = {
        "title": (title or "").strip() or None,
        "identifier": (identifier or "").strip() or None,
        "published_at": published_at or None,
        "effective_at": effective_at or None,
        "source_url": source_url or None,
    }

    try:
        job_id = await kb_jobs.create_job(
            topic_id=topic_id,
            topic_slug=topic["slug"],
            kind=kind,
            original_filename=file.filename or "upload",
            file_bytes=data,
            metadata=metadata,
            requested_by_user=requested_by,
        )
    except Exception as e:
        logger.exception("[admin.kb.upload] failed to create job")
        raise HTTPException(status_code=500, detail=f"Failed to queue job: {e}")

    # Fire-and-forget: don't await. The HTTP response returns now; the
    # background task transitions the job through processing → done|failed.
    asyncio.create_task(kb_jobs.process_job(job_id))

    return {"job_id": job_id, "status": "queued"}


@router.get("/jobs")
async def list_ingest_jobs(
    request: Request,
    topic_id: int | None = None,
    status: str | None = None,
    limit: int = 50,
):
    _verify_admin(request)
    if status and status not in ("queued", "processing", "done", "failed"):
        raise HTTPException(status_code=400, detail=f"Invalid status: {status}")
    items = await kb_jobs.list_jobs(topic_id=topic_id, status=status, limit=limit)
    return {"items": items}


@router.get("/jobs/{job_id}")
async def get_ingest_job(request: Request, job_id: int):
    _verify_admin(request)
    job = await kb_jobs.get_job(job_id)
    if not job:
        raise HTTPException(status_code=404, detail="Job not found")
    return job


# ── Phase 3: source management (Task #14) ──────────────────────────────────

class SourceUpdate(BaseModel):
    """Whitelist of editable fields. Anything outside this is rejected."""
    title: str | None = Field(None, max_length=500)
    identifier: str | None = Field(None, max_length=200)
    source_url: str | None = Field(None, max_length=1000)
    published_at: str | None = None
    effective_at: str | None = None
    description: str | None = Field(None, max_length=500)
    # Optional. None = leave labels unchanged; [] = clear all; [3,7] = exact set.
    label_ids: list[int] | None = None


@router.get("/sources")
async def list_admin_sources(
    request: Request,
    topic_id: int | None = None,
    kind: str | None = None,
    label_id: int | None = None,
    limit: int = 200,
):
    _verify_admin(request)
    if kind and kind not in ("law", "regulation", "circular", "caselaw", "tool"):
        raise HTTPException(status_code=400, detail=f"Invalid kind: {kind}")
    items = await kb_admin_sources.list_sources(
        topic_id=topic_id, kind=kind, label_id=label_id, limit=limit,
    )
    return {"items": items, "count": len(items)}


@router.put("/sources/{source_id}")
async def update_admin_source(request: Request, source_id: int, body: SourceUpdate):
    _verify_admin(request)
    raw = body.model_dump(exclude_unset=True)
    label_ids = raw.pop("label_ids", None)
    try:
        updated = await kb_admin_sources.update_source(
            source_id, raw, label_ids=label_ids,
        )
    except LookupError:
        raise HTTPException(status_code=404, detail="Source not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return updated


@router.delete("/sources/{source_id}")
async def delete_admin_source(request: Request, source_id: int):
    _verify_admin(request)
    try:
        result = await kb_admin_sources.delete_source(source_id)
    except LookupError:
        raise HTTPException(status_code=404, detail="Source not found")
    return result


@router.post("/sources/{source_id}/reingest")
async def reingest_admin_source(request: Request, source_id: int):
    _verify_admin(request)
    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )
    try:
        job_id = await kb_admin_sources.start_reingest(source_id, requested_by)
    except LookupError:
        raise HTTPException(status_code=404, detail="Source not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))

    asyncio.create_task(kb_admin_sources.process_reingest_job(job_id))
    return {"job_id": job_id, "status": "queued", "source_id": source_id}


# ── Phase 4: topic CRUD (Task #15) ─────────────────────────────────────────

class TopicCreate(BaseModel):
    slug: str = Field(..., min_length=2, max_length=63)
    name: str = Field(..., min_length=1, max_length=200)
    description: str | None = Field(None, max_length=1000)
    system_prompt_addendum: str | None = Field(None, max_length=4000)
    is_active: bool = True


class TopicUpdate(BaseModel):
    name: str | None = Field(None, max_length=200)
    description: str | None = Field(None, max_length=1000)
    system_prompt_addendum: str | None = Field(None, max_length=4000)
    is_active: bool | None = None


@router.get("/topics")
async def list_admin_topics(request: Request):
    _verify_admin(request)
    items = await kb_topics.list_topics_admin()
    return {"items": items, "count": len(items)}


@router.post("/topics")
async def create_admin_topic(request: Request, body: TopicCreate):
    _verify_admin(request)
    try:
        return await kb_topics.create_topic(
            slug=body.slug,
            name=body.name,
            description=body.description,
            system_prompt_addendum=body.system_prompt_addendum,
            is_active=body.is_active,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.put("/topics/{topic_id}")
async def update_admin_topic(request: Request, topic_id: int, body: TopicUpdate):
    _verify_admin(request)
    fields = body.model_dump(exclude_unset=True)
    try:
        return await kb_topics.update_topic(topic_id, fields)
    except LookupError:
        raise HTTPException(status_code=404, detail="Topic not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.delete("/topics/{topic_id}")
async def delete_admin_topic(request: Request, topic_id: int):
    _verify_admin(request)
    try:
        return await kb_topics.delete_topic(topic_id)
    except LookupError:
        raise HTTPException(status_code=404, detail="Topic not found")
    except kb_topics.ConflictError as e:
        raise HTTPException(status_code=409, detail=str(e))


# ── Phase 6 (Task #20): bulk upload + AI classification + labels ───────────

@router.post("/upload-batch")
async def upload_batch(
    request: Request,
    kind: Literal["law", "regulation", "circular", "caselaw", "tool"] = Form(...),
    topic_id: int = Form(...),
    # FastAPI accepts a list when the form field name is repeated; the
    # browser sends `files=...&files=...` for a multi-file <input multiple>
    # or for FormData.append('files', f) called N times.
    files: list[UploadFile] = File(...),
):
    """Multi-file upload. Each file gets its own kb_ingest_job and a shared
    batch_id (uuid). After insert, asyncio.create_task fires the
    classify stage in the background. The browser opens the review screen
    on /admin/kb/batch/{batch_id} and polls until all jobs are
    'awaiting_review' or 'failed'.
    """
    _verify_admin(request)

    if not files:
        raise HTTPException(status_code=400, detail="No files in batch")
    if len(files) > 50:
        raise HTTPException(status_code=400, detail="Max 50 files per batch")

    try:
        topic = await kb_topics.get_topic(topic_id)
    except Exception:
        topic = None
    if not topic or not topic.get("is_active"):
        raise HTTPException(status_code=400, detail=f"Unknown or inactive topic_id={topic_id}")

    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )

    import uuid as _uuid
    batch_id = _uuid.uuid4().hex

    out_jobs: list[dict] = []
    for f in files:
        data = await f.read()
        if not data:
            out_jobs.append({"original_filename": f.filename or "upload",
                             "error": "empty file"})
            continue
        if len(data) > _MAX_UPLOAD_BYTES:
            out_jobs.append({"original_filename": f.filename or "upload",
                             "error": f"too large ({len(data)} bytes)"})
            continue
        try:
            job_id = await kb_jobs.create_job(
                topic_id=topic_id,
                topic_slug=topic["slug"],
                kind=kind,
                original_filename=f.filename or "upload",
                file_bytes=data,
                # Empty user metadata at this stage — AI will populate
                # ai_suggestions, then user confirms via /commit.
                metadata={},
                requested_by_user=requested_by,
                batch_id=batch_id,
            )
        except Exception as e:
            logger.exception("[upload-batch] create_job failed")
            out_jobs.append({"original_filename": f.filename or "upload",
                             "error": f"queue failed: {e}"})
            continue
        # Fire-and-forget classify stage. The classifier is fail-soft so
        # the user always reaches the review screen.
        asyncio.create_task(kb_jobs.process_classify_stage(job_id))
        out_jobs.append({
            "job_id": job_id,
            "original_filename": f.filename or "upload",
            "status": "queued",
        })

    return {"batch_id": batch_id, "jobs": out_jobs}


@router.get("/batch/{batch_id}")
async def get_batch(request: Request, batch_id: str):
    _verify_admin(request)
    items = await kb_jobs.list_jobs_by_batch(batch_id)
    return {"batch_id": batch_id, "items": items, "count": len(items)}


@router.get("/batches")
async def list_batches(request: Request, limit: int = 50):
    """Batches that still have awaiting_review (or in-progress) jobs.

    Powers the 'ממתינים לאישור' panel in the management UI: a user can
    resume a review session even after a hard refresh wiped JS state.
    """
    _verify_admin(request)
    items = await kb_jobs.list_pending_batches(limit=limit)
    return {"items": items, "count": len(items)}


class JobCommit(BaseModel):
    """User-confirmed metadata from the review screen."""
    kind: Literal["law", "regulation", "circular", "caselaw", "tool"]
    title: str = Field(..., min_length=1, max_length=500)
    identifier: str | None = Field(None, max_length=200)
    source_url: str | None = Field(None, max_length=1000)
    published_at: str | None = None
    effective_at: str | None = None
    summary: str | None = Field(None, max_length=500)
    # Existing labels — by slug. Lookup is exact match on kb_label.slug.
    label_slugs: list[str] = Field(default_factory=list, max_length=10)
    # New labels — display names that the user accepted as "create on commit".
    # The route slugifies them and creates rows.
    new_labels: list[str] = Field(default_factory=list, max_length=10)


@router.post("/jobs/{job_id}/commit")
async def commit_job(request: Request, job_id: int, body: JobCommit):
    _verify_admin(request)
    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )

    # Resolve label_specs. Existing labels: pass slug + name (looked up
    # by slug, name only used if INSERT happens via the lookup_or_create
    # helper). New labels: slugify display names; reject empties.
    label_specs: list[dict] = []
    seen_slugs: set[str] = set()

    # Existing slugs: look up names from DB so the helper can re-insert
    # them safely (idempotent on conflict).
    if body.label_slugs:
        for slug in body.label_slugs:
            slug = (slug or "").strip()
            if not slug or slug in seen_slugs:
                continue
            existing = await kb_labels.list_labels(query=slug, limit=1)
            existing = [l for l in existing if l["slug"] == slug]
            if existing:
                label_specs.append({
                    "slug": slug,
                    "name": existing[0]["name"],
                    "topic_id": existing[0].get("topic_id"),
                })
                seen_slugs.add(slug)

    # New labels: derive slug from name, dedupe.
    if body.new_labels:
        # Find topic for the job so new labels are scoped to it.
        async with (await get_pool()).acquire() as conn:
            row = await conn.fetchrow(
                "SELECT topic_id FROM kb_ingest_job WHERE id = $1", job_id,
            )
        topic_id = (row or {}).get("topic_id") if row else None
        for name in body.new_labels:
            name = (name or "").strip()
            if not name:
                continue
            try:
                slug = kb_labels.slugify(name)
            except ValueError as e:
                logger.warning("[commit] bad new label name=%r: %s", name, e)
                continue
            if slug in seen_slugs:
                continue
            label_specs.append({"slug": slug, "name": name, "topic_id": topic_id})
            seen_slugs.add(slug)

    fields = {
        "kind": body.kind,
        "title": body.title.strip(),
        "identifier": (body.identifier or "").strip() or None,
        "source_url": (body.source_url or "").strip() or None,
        "published_at": body.published_at or None,
        "effective_at": body.effective_at or None,
        "summary": (body.summary or "").strip() or None,
    }

    try:
        result = await kb_jobs.commit_job(
            job_id=job_id,
            fields=fields,
            label_specs=label_specs,
            requested_by=requested_by,
        )
    except LookupError:
        raise HTTPException(status_code=404, detail="Job not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))
    return result


@router.post("/jobs/{job_id}/discard")
async def discard_job(request: Request, job_id: int):
    _verify_admin(request)
    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )
    try:
        return await kb_jobs.discard_job(job_id, requested_by)
    except LookupError:
        raise HTTPException(status_code=404, detail="Job not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


# ── Labels CRUD ────────────────────────────────────────────────────────────

class LabelCreate(BaseModel):
    slug: str = Field(..., min_length=1, max_length=200)
    name: str = Field(..., min_length=1, max_length=300)
    topic_id: int | None = None


class LabelMerge(BaseModel):
    into_label_id: int


@router.get("/labels")
async def list_admin_labels(
    request: Request,
    topic_id: int | None = None,
    q: str | None = None,
    limit: int = 50,
):
    _verify_admin(request)
    items = await kb_labels.list_labels(topic_id=topic_id, query=q, limit=limit)
    return {"items": items, "count": len(items)}


@router.post("/labels")
async def create_admin_label(request: Request, body: LabelCreate):
    _verify_admin(request)
    requested_by = (
        request.headers.get("X-User-Name")
        or request.headers.get("X-Forwarded-User")
        or None
    )
    try:
        return await kb_labels.create_label(
            slug=body.slug,
            name=body.name,
            topic_id=body.topic_id,
            created_by=requested_by,
        )
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.post("/labels/{label_id}/merge")
async def merge_admin_labels(request: Request, label_id: int, body: LabelMerge):
    _verify_admin(request)
    try:
        return await kb_labels.merge_labels(label_id, body.into_label_id)
    except LookupError as e:
        raise HTTPException(status_code=404, detail=str(e))
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))


@router.delete("/labels/{label_id}")
async def delete_admin_label(request: Request, label_id: int):
    _verify_admin(request)
    try:
        return await kb_labels.delete_label(label_id)
    except LookupError:
        raise HTTPException(status_code=404, detail="Label not found")
    except ValueError as e:
        raise HTTPException(status_code=400, detail=str(e))