AgentLab/backend/app/api/catalog.py

"""数据目录接口：/api/datasources/{source_id}/catalog。

- GET    /catalog                表清单（没有目录的表也列出来）
- GET    /catalog/{table}        单表目录 + 表结构
- PUT    /catalog/{table}        整份提交（乐观锁），改动过的项记为人工确认
- POST   /catalog/{table}/review 单项审阅：确认、驳回、撤销审阅
- POST   /catalog/{table}/patch/preview  目录修改提案的预览（只读）：对着当前目录算改前、改后
- POST   /catalog/{table}/patch  保存目录修改提案（乐观锁），记为人工确认
- GET    /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py）
- POST   /catalog/draft          同步起草（注释、外键、命名推断，可选模型）
- POST   /catalog/profile        同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py）

规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。

上传表格的源按当前快照的表结构回答（table_versions.resolve_source），和查询实际用的库是同一个版本。
"""
from __future__ import annotations

from typing import Any, Literal

from fastapi import APIRouter, Depends, Header, HTTPException
from pydantic import BaseModel, Field
from sqlalchemy.ext.asyncio import AsyncSession

from app.api.coded import PROFILE_BUSY, SCHEMA_MISSING, SNAPSHOT_TAMPERED, CodedHTTPException
from app.api.runs import actor_of
from app.data import catalog, catalog_impact, catalog_profile, introspect, table_versions
from app.data.engine import SnapshotTampered
from app.db.base import get_session
from app.db.models import DataSource

router = APIRouter(prefix="/api/datasources", tags=["catalog"])


class CatalogPutIn(BaseModel):
    notes: dict[str, Any]
    #: 读到的版本；这张表还没有目录时为 0
    if_version: int = Field(ge=0)


class CatalogReviewIn(BaseModel):
    #: grain、columns.amount.measure、relations.<编号>
    path: str = Field(min_length=1, max_length=600)
    action: Literal["confirm", "reject", "reset"]
    if_version: int = Field(ge=0)


class CatalogPatchChange(BaseModel):
    #: 审阅接口的路径写法；新增关联关系写 relations.new
    path: str = Field(min_length=1, max_length=600)
    value: Any
    reason: str | None = Field(default=None, max_length=2000)


class CatalogPatchPreviewIn(BaseModel):
    changes: list[CatalogPatchChange] = Field(min_length=1, max_length=catalog.PATCH_MAX_CHANGES)


class CatalogPatchIn(CatalogPatchPreviewIn):
    if_version: int = Field(ge=0)


class CatalogDraftIn(BaseModel):
    #: 要起草的表；不给或为空时按使用次数取前 20 张
    tables: list[str] | None = Field(default=None, max_length=200)
    #: 是否请助手的模型起草中文名、粒度、列的含义等
    use_model: bool = False


class CatalogProfileIn(BaseModel):
    #: 要剖析的表；不给或为空时按使用次数取前 10 张目录里有待核实关系的表
    tables: list[str] | None = Field(default=None, max_length=200)


async def _resolved(session: AsyncSession, source_id: str) -> tuple[DataSource, Any]:
    """(数据源记录, 用来读表结构的源)：上传源解析成绑定当前快照的视图。"""
    row = await session.get(DataSource, source_id)
    if row is None:
        raise HTTPException(404, "数据源不存在，可能已被删除")
    try:
        return row, await table_versions.resolve_source(session, row)
    except table_versions.SnapshotMissing as e:
        raise HTTPException(409, str(e)) from e


def _iso(value: Any) -> str | None:
    return value.isoformat() if value is not None else None


def _int(value: Any) -> int:
    return value if isinstance(value, int) and not isinstance(value, bool) and value > 0 else 0


def _item_value(notes: dict[str, Any], name: str) -> tuple[Any, str | None]:
    item = notes.get(name)
    if not isinstance(item, dict) or item.get("status") == "rejected":
        return None, None
    return item.get("value"), item.get("status")


def _summary_row(name: str, meta: dict[str, Any] | None, entry: catalog.CatalogEntry | None,
                 usage: int) -> dict[str, Any]:
    notes = entry.notes if entry else {}
    label, label_status = _item_value(notes, "label")
    kind, _ = _item_value(notes, "kind")
    relations = notes.get("relations") if isinstance(notes.get("relations"), list) else []
    return {
        "table_name": name,
        "qualified": (meta or {}).get("qualified", name),
        "is_view": bool((meta or {}).get("is_view")),
        # 表结构里已经没有这张表（被删了、改名了），目录还留着：界面据此提示清理
        "in_schema": meta is not None,
        "label": label,
        "label_status": label_status,
        "kind": kind,
        "counts": catalog.status_counts(notes),
        "relations": sum(1 for r in relations if isinstance(r, dict) and r.get("status") != "rejected"),
        "usage": usage,
        "version": entry.version if entry else 0,
        "updated_at": _iso(entry.updated_at) if entry else None,
        "updated_by": entry.updated_by if entry else None,
    }


def _structure(meta: dict[str, Any]) -> dict[str, Any]:
    return {
        "qualified": meta.get("qualified"),
        "is_view": bool(meta.get("is_view")),
        "comment": meta.get("comment"),
        "columns": introspect.table_columns(meta),
        "primary_key": list(meta.get("primary_key") or []),
        # 升级前探查的缓存里没有这两项：为 null 表示「不知道」，和空列表（读过、没有）分开
        "foreign_keys": meta.get("foreign_keys"),
        "unique": meta.get("unique"),
    }


async def _detail(session: AsyncSession, source: Any, table: str, entry: catalog.CatalogEntry | None,
                  usage: int | None = None) -> dict[str, Any]:
    meta = ((source.schema_cache or {}).get("tables") or {}).get(table)
    if usage is None:
        usage = (await catalog.table_usage(session, source)).get(table, 0)
    return {
        "table_name": table,
        "in_schema": meta is not None,
        "notes": entry.notes if entry else {},
        "version": entry.version if entry else 0,
        "updated_at": _iso(entry.updated_at) if entry else None,
        "updated_by": entry.updated_by if entry else None,
        "structure": _structure(meta) if meta is not None else None,
        "system_notes": catalog.system_notes_source(source),
        "usage": usage,
    }


async def _table_key(session: AsyncSession, source: Any, table: str) -> tuple[str, catalog.CatalogEntry | None]:
    """路径里的表名 → 目录的键。表结构里有的按表结构的写法；表结构里没有、但还留着目录的也认（便于清理）。"""
    key = catalog.resolve_table_name(source.schema_cache, table)
    entry = await catalog.read_entry(session, source.id, key or table)
    if key is None and entry is None:
        raise HTTPException(404, f"数据源中没有表 {table}，可能已被删除或尚未探查结构")
    return key or table, entry


def _conflict(e: catalog.CatalogConflict) -> HTTPException:
    return HTTPException(409, str(e))


def _invalid(e: catalog.CatalogInvalid) -> HTTPException:
    shown = "；".join(e.problems[:8]) + (f"；另有 {len(e.problems) - 8} 处" if len(e.problems) > 8 else "")
    return HTTPException(422, f"数据目录的格式不正确：{shown}")


@router.get("/{source_id}/catalog")
async def list_catalog(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """表清单：表结构里的每张表各一行（没有目录的也列），使用次数多的在前；目录还在、表已不在的排最后。

    探查结构每个源最多取 200 张表（introspect._MAX_TABLES），多出来的不在表结构里、这里也列不出：schema_truncated
    说明有没有截断，schema_total 是数据库里一共几张表（含视图），界面据此说明「只探查了前 200 张」。老缓存没记
    总数时按探查到的张数算。
    """
    _, source = await _resolved(session, source_id)
    cache = source.schema_cache or {}
    tables = cache.get("tables") or {}
    entries = await catalog.read_catalog(session, source.id)
    usage = await catalog.table_usage(session, source)
    order = {name: i for i, name in enumerate(tables)}
    rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0)) for name, meta in tables.items()]
    rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]]))
    rows += [_summary_row(name, None, entry, 0) for name, entry in sorted(entries.items()) if name not in tables]
    return {
        "tables": rows,
        "system_notes": catalog.system_notes_source(source),
        # 没有表结构时说明原因（尚未探查 / 探查失败），界面据此引导去探查
        "schema_note": None if tables else introspect.why_empty(cache),
        "schema_truncated": bool(cache.get("truncated")),
        "schema_total": max(_int(cache.get("total")), len(tables)),
    }


@router.get("/{source_id}/catalog/{table}")
async def get_catalog_table(source_id: str, table: str,
                            session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    _, source = await _resolved(session, source_id)
    key, entry = await _table_key(session, source, table)
    return await _detail(session, source, key, entry)


@router.put("/{source_id}/catalog/{table}")
async def put_catalog_table(source_id: str, table: str, payload: CatalogPutIn,
                            x_actor: str | None = Header(default=None),
                            session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """整份提交一张表的目录。改动过的项记为人工确认；版本不符 409。"""
    _, source = await _resolved(session, source_id)
    key, _ = await _table_key(session, source, table)
    try:
        entry = await catalog.save_human_edit(session, source.id, key, payload.notes, if_version=payload.if_version,
                                              actor=actor_of(x_actor))
    except catalog.CatalogConflict as e:
        raise _conflict(e) from e
    except catalog.CatalogInvalid as e:
        raise _invalid(e) from e
    return await _detail(session, source, key, entry)


@router.post("/{source_id}/catalog/draft")
async def draft_catalog(source_id: str, payload: CatalogDraftIn,
                        x_actor: str | None = Header(default=None),
                        session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """同步起草。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。

    model_items：模型给了这张表几项可用内容（没用模型、模型这部分失败时为 null）。界面据此区分「模型给了内容、
    只是和现有目录一致」（model_items > 0、三个数都是 0）和「模型没给内容」（model_error 有值）。
    """
    _, source = await _resolved(session, source_id)
    cache = source.schema_cache or {}
    if not cache.get("tables"):
        raise HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」")
    report = await catalog.draft_catalog(session, source, tables=payload.tables or None,
                                         use_model=payload.use_model, actor=actor_of(x_actor))
    rows = [{"table_name": r.table, "added": r.added, "updated": r.updated, "removed": r.removed,
             "version": r.version, "error": r.error, "model_error": r.model_error, "model_items": r.model_items}
            for r in report.tables]
    return {
        "tables": rows,
        "total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")},
        "model_used": report.model_used,
        "model": report.model or None,
        "model_error": report.model_error,
    }


def _profile_table_out(t: catalog_profile.TableProfile) -> dict[str, Any]:
    return {
        "table_name": t.table,
        "queries": t.queries,
        "row_estimate": t.row_estimate.to_dict() if t.row_estimate else None,
        "findings": t.findings,
        "skipped": [s.to_dict() for s in t.skipped],
        "date_ranges": t.date_ranges,
        "added": t.added,
        "updated": t.updated,
        "removed": t.removed,
        "version": t.version,
        "error": t.error,
    }


@router.post("/{source_id}/catalog/profile")
async def profile_catalog_tables(source_id: str, payload: CatalogProfileIn,
                                 x_actor: str | None = Header(default=None),
                                 session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """同步剖析。返回每张表用了几条查询、写进目录的发现、跳过了什么及原因。

    默认关闭：数据源设置里没开启数据剖析、数据源停用，409；同一个数据源已经在剖析，409。
    """
    row, source = await _resolved(session, source_id)
    # 409 都带机读码（api/coded.py），界面按它分五种情况给下一步，不从原话里认
    try:
        catalog_profile.ensure_enabled(row)
    except catalog_profile.ProfileDisabled as e:
        raise CodedHTTPException(409, str(e), e.code) from e
    cache = source.schema_cache or {}
    if not cache.get("tables"):
        raise CodedHTTPException(409, f"{introspect.why_empty(cache)}，无法进行数据剖析。请先在数据源卡片上点「探查结构」",
                                 SCHEMA_MISSING)
    actor = actor_of(x_actor)
    try:
        report = await catalog_profile.profile_catalog(session, row, source, tables=payload.tables or None,
                                                       actor=actor)
    except catalog_profile.ProfileDisabled as e:
        raise CodedHTTPException(409, str(e), e.code) from e
    except catalog_profile.ProfileBusy as e:
        raise CodedHTTPException(409, str(e), PROFILE_BUSY) from e
    except SnapshotTampered as e:
        raise CodedHTTPException(409, str(e), SNAPSHOT_TAMPERED) from e
    rows = [_profile_table_out(t) for t in report.tables]
    note = None
    if not rows and report.empty_tables:
        note = (f"有待核对关联关系的 {len(report.empty_tables)} 张表都是空表，没有可以核实的数据。"
                "请在数据有了之后再剖析，或指定要剖析的表")
    elif not rows:
        note = "目录里还没有待核实的关联关系，请先起草数据目录，或指定要剖析的表"
    return {
        "profiled_at": report.profiled_at,
        "actor": actor,
        "settings": report.settings.to_dict(),
        "queries_used": report.queries_used,
        "stopped": report.stopped,
        "tables": rows,
        "total": {k: sum(r[k] for r in rows) for k in ("added", "updated", "removed")},
        # 不指定表时挑表途中确知是空表、跳过了的表（指定了表时总是空列表）
        "empty_tables": report.empty_tables,
        "note": note,
    }


@router.post("/{source_id}/catalog/{table}/review")
async def review_catalog_item(source_id: str, table: str, payload: CatalogReviewIn,
                              x_actor: str | None = Header(default=None),
                              session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """单项审阅。版本不符 409；路径不对、项不存在 422。"""
    _, source = await _resolved(session, source_id)
    key, _ = await _table_key(session, source, table)
    try:
        entry = await catalog.review_entry(session, source.id, key, payload.path, payload.action,
                                           if_version=payload.if_version, actor=actor_of(x_actor))
    except catalog.CatalogConflict as e:
        raise _conflict(e) from e
    except catalog.CatalogPathError as e:
        raise HTTPException(422, str(e)) from e
    return await _detail(session, source, key, entry)


def _tables_of(source: Any) -> dict[str, Any] | None:
    """核对提案用的表结构；没有表结构（尚未探查）时不核对列名。"""
    return ((source.schema_cache or {}).get("tables") or None) if source is not None else None


@router.post("/{source_id}/catalog/{table}/patch/preview")
async def preview_catalog_patch(source_id: str, table: str, payload: CatalogPatchPreviewIn,
                                session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """目录修改提案的预览：对着当前目录逐项算改前、改后，不写库。

    助手转出的提案卡片在保存遇到 409（别人刚改过这张表）时用它重新载入：改前的值、码值补充后的样子都按
    最新的目录重算，人看过之后再带新版本保存。
    """
    _, source = await _resolved(session, source_id)
    key, entry = await _table_key(session, source, table)
    plan = catalog.plan_patch(entry.notes if entry else {}, [c.model_dump() for c in payload.changes], table=key,
                              tables=_tables_of(source))
    return {"table_name": key, "version": entry.version if entry else 0,
            "changes": [c.as_dict() for c in plan.changes], "problems": plan.problems}


@router.post("/{source_id}/catalog/{table}/patch")
async def save_catalog_patch(source_id: str, table: str, payload: CatalogPatchIn,
                             x_actor: str | None = Header(default=None),
                             session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """保存一条目录修改提案：改动过的项记为人工填写、已确认。版本不符 409；有一项不合法 422，整条不写。"""
    _, source = await _resolved(session, source_id)
    key, _ = await _table_key(session, source, table)
    try:
        entry = await catalog.save_patch(session, source.id, key, [c.model_dump() for c in payload.changes],
                                         if_version=payload.if_version, actor=actor_of(x_actor),
                                         tables=_tables_of(source))
    except catalog.CatalogConflict as e:
        raise _conflict(e) from e
    except catalog.CatalogInvalid as e:
        raise _invalid(e) from e
    return await _detail(session, source, key, entry)


@router.get("/{source_id}/catalog/{table}/impact")
async def catalog_table_impact(source_id: str, table: str,
                               session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """引用这张表的已发布、受管模板，每个模板按当前的已发布版本算：调用工具节点的 SQL 用到了这张表算直接引用，
    Agent 绑定了这个源的查询工具算可能涉及，合并查询按输入追溯。改目录之前、保存纠正之后都看它。"""
    row, source = await _resolved(session, source_id)
    key, _ = await _table_key(session, source, table)
    tables = _tables_of(source)
    templates = await catalog_impact.table_impact(session, row.name, key, list(tables) if tables else None)
    return {"table": key, "templates": templates}