Create-knowledge-find-decode/app/routes/query_generation.py

"""Formal query-generation preview API routes."""
from __future__ import annotations

from typing import Any
from uuid import UUID

from fastapi import APIRouter, Depends, Query

from acquisition.queries.builder import QueryBuildOptions, TREES, build_creation_query_batch
from app.dependencies import _env_file, get_acquisition_repository
from app.routes.acquisition import _model_dump
from app.schemas import QuerySchema
from core.config import Settings

router = APIRouter(prefix="/api/query-generation", tags=["query-generation"])


def _summary(generated: dict[str, Any]) -> dict[str, Any]:
    families = generated.get("families") or []
    return {
        "family_count": len(families),
        "family_keys": [family.get("key") for family in families],
        "query_count": sum(len(family.get("items") or []) for family in families),
        "kept_count": sum(
            1
            for family in families
            for item in family.get("items") or []
            if item.get("keep", True)
        ),
    }


LIGHT_ITEM_METADATA_KEYS = {
    "page_index",
    "page_rank",
    "source_cursor",
    "content_mode",
    "search_provider",
    "detail_provider",
    "acquisition_match_status",
    "matched_unique_key",
    "skip_reason",
    "unsupported_raw_type",
    "video_url_missing",
}


def _light_metadata(metadata: dict[str, Any] | None) -> dict[str, Any]:
    if not isinstance(metadata, dict):
        return {}
    return {key: metadata[key] for key in LIGHT_ITEM_METADATA_KEYS if key in metadata}


def _normalize_light_query_detail(detail: dict[str, Any]) -> dict[str, Any]:
    media_by_item = {media["item_id"]: media for media in detail.get("media_assets") or []}
    classification_by_item = {row["item_id"]: row for row in detail.get("classifications") or []}
    decode_by_item = {row["item_id"]: row for row in detail.get("decode_summaries") or []}
    platforms: dict[str, dict[str, Any]] = {}
    for raw_job in detail.get("jobs") or []:
        job = _model_dump(raw_job)
        platform = job.get("platform")
        if not platform:
            continue
        platforms.setdefault(
            platform,
            {
                "platform": platform,
                "status": job.get("status") or "pending",
                "attempt_count": job.get("attempt_count"),
                "display_limit": job.get("display_limit"),
                "search_limit": job.get("search_limit"),
                "error_message": job.get("error_message"),
                "item_ids": [],
                "item_count": 0,
            },
        )

    items: list[dict[str, Any]] = []
    for row in detail.get("items") or []:
        item = _model_dump(row)
        item_id = item["id"]
        classification = classification_by_item.get(item_id)
        payload = {
            "id": item_id,
            "platform": item.get("platform"),
            "title": item.get("title"),
            "raw_summary": item.get("raw_summary"),
            "status": item.get("status"),
            "content_mode": item.get("content_mode"),
            "metadata": _light_metadata(item.get("metadata")),
            "classification": _model_dump(classification) if classification else None,
            "decode_summary": _model_dump(decode_by_item[item_id]) if item_id in decode_by_item else None,
            "media_assets": [],
        }
        if item_id in media_by_item:
            payload["media_assets"] = [_model_dump(media_by_item[item_id])]
        items.append(payload)
        platform = payload["platform"]
        group = platforms.setdefault(
            platform,
            {"platform": platform, "status": "done", "item_ids": [], "item_count": 0},
        )
        if group.get("status") in {None, "pending"}:
            group["status"] = "done"
        group.setdefault("item_ids", []).append(item_id)
        group["item_count"] = len(group["item_ids"])

    return {
        "query": QuerySchema.model_validate(detail["query"]).model_dump(mode="json"),
        "run": _model_dump(detail["run"]) if detail.get("run") else None,
        "jobs": [_model_dump(job) for job in detail.get("jobs") or []],
        "items": items,
        "platforms": platforms,
    }


@router.get("/preview")
def query_generation_preview(
    per: int = Query(default=0, ge=0, le=10000),
    batch_n: int = Query(default=0, ge=0, le=1000),
) -> dict[str, Any]:
    """Preview the currently active formal query families without writing DB rows."""

    settings = Settings.from_env(_env_file())
    generated = build_creation_query_batch(
        settings,
        tree_path=TREES,
        options=QueryBuildOptions(
            per=per,
            batch_n=batch_n,
            active_family_keys=("f1", "f2"),
        ),
    )
    generated["summary"] = _summary(generated)
    return generated


@router.get("/latest-singleton")
def latest_singleton_overview(repo: Any = Depends(get_acquisition_repository)) -> dict[str, Any]:
    """Return links from the query preview board to the latest real singleton run."""

    getter = getattr(repo, "get_latest_singleton_overview", None)
    if getter is None:
        return {"batch": None, "run": None, "queries": [], "decoded_items": []}
    return getter()


@router.get("/latest/queries/{query_id}")
def latest_query_detail(
    query_id: UUID,
    repo: Any = Depends(get_acquisition_repository),
) -> dict[str, Any]:
    """Return search material for one query in the latest real query board batch."""

    detail_getter = getattr(repo, "get_latest_query_result_list", None)
    if detail_getter is None:
        return {"query": None, "jobs": [], "items": [], "platforms": {}}
    return _normalize_light_query_detail(detail_getter(query_id))