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          同步起草（注释、外键、命名推断，可选模型）。界面改用下面的后台起草任务，这个留着；
                                 和后台任务双向互斥：同一个源有排队或进行中的后台任务（或正在同步起草）时 409
                                 catalog_draft_busy，同步起草期间建任务、重试、导入也是 409
- PUT    /catalog/guidance       数据源级的起草要求（使用者补充的背景和要求，起草时交给模型）
- POST   /catalog/draft-jobs     建后台起草任务（202）；GET /draft-jobs/active、/latest、/{job_id} 读任务；
                                 POST /draft-jobs/{job_id}/cancel 停止，/retry 只重试没写好的表（data/catalog_jobs.py）
- POST   /catalog/profile        同步剖析：对业务库发少量只读查询，核对关系、取码值候选（data/catalog_profile.py）

规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。剖析没开启、正在剖析 409。
起草任务的 409 带机读码（api/coded.py）：同一个源已有排队或进行中的任务、起草模型用不了、要停止的任务已经结束。

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

from typing import Any, Literal

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

from app.api.coded import (
    CATALOG_DRAFT_BUSY, CATALOG_DRAFT_FINISHED, CATALOG_DRAFT_MODEL_UNAVAILABLE, PROFILE_BUSY, SCHEMA_MISSING,
    SNAPSHOT_TAMPERED, CodedHTTPException,
)
from app.api.runs import actor_of
from app.data import catalog, catalog_impact, catalog_jobs, 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 DraftJobIn(BaseModel):
    #: 要起草的表；不给或为空时按使用次数取前 20 张。最多 500 张（超了 422，说明是中文）
    tables: list[str] | None = None
    #: 是否请起草模型写中文名、粒度、列的含义等
    use_model: bool = False
    #: 这一次的补充说明，和数据源级的起草要求一起交给模型（最多 1000 字）
    note: str | None = None


#: PUT /catalog/guidance 的请求 {guidance: string}：数据源级的起草要求，空串（或 null）清除，最多 2000 字。必须写
#: 这一项：表名恰好叫 guidance 时，整份提交那张表目录的请求也落到这个路径上，不能把它当成「清除起草要求」。
#: 没写、不是文字都回 422，说明是中文的一句话（同超长）
GUIDANCE_MISSING = "请求里要写起草要求（guidance 一项，一段文字；要清除起草要求请传空串）"
GUIDANCE_NOT_TEXT = "起草要求要是一段文字（要清除起草要求请传空串）"


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, last_draft: dict[str, Any] | None = None) -> 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,
        # 这张表最近一次起草任务里的那一项：{status, job_id, finished_at, error, model_error, blank_columns}，没起草过
        # 是 null
        "last_draft": last_draft,
    }


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 张」。老缓存没记
    总数时按探查到的张数算。
    """
    row, 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)
    drafts = await catalog_jobs.last_drafts(session, source.id)
    order = {name: i for i, name in enumerate(tables)}
    rows = [_summary_row(name, meta, entries.get(name), usage.get(name, 0), drafts.get(name))
            for name, meta in tables.items()]
    rows.sort(key=lambda r: (-r["usage"], order[r["table_name"]]))
    rows += [_summary_row(name, None, entry, 0, drafts.get(name))
             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)),
        # 数据源级的起草要求（没有是 null）
        "guidance": catalog_jobs.guidance_of(row),
        # 排队或进行中的起草任务 {id, status, total, done}，界面据此显示进度；没有是 null
        "active_job": await catalog_jobs.job_brief(session, await catalog_jobs.active_job(session, source.id)),
    }


@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)


async def _source_row(session: AsyncSession, source_id: str) -> DataSource:
    row = await session.get(DataSource, source_id)
    if row is None:
        raise HTTPException(404, "数据源不存在，可能已被删除")
    return row


@router.put("/{source_id}/catalog/guidance")
async def put_catalog_guidance(source_id: str, body: dict[str, Any] = Body(...),
                               x_actor: str | None = Header(default=None),
                               session: AsyncSession = Depends(get_session)) -> Any:
    """存数据源级的起草要求：去首尾空白，空串清除，最多 2000 字（超了 422）。上传表格的源存在它自己的数据源记录上。

    要登记在 PUT /catalog/{table} 前面：不然「guidance」会被当成表名。反过来，表名恰好叫 guidance 时，整份提交那张表
    目录的请求（{notes, if_version}，没有 guidance 这一项）也落到这里：照旧当作那张表的整份提交。"""
    if "guidance" not in body and "notes" in body:
        try:
            table_payload = CatalogPutIn.model_validate(body)
        except ValidationError as e:
            raise RequestValidationError(e.errors()) from e
        return await put_catalog_table(source_id, "guidance", table_payload, x_actor=x_actor, session=session)
    if "guidance" not in body:
        raise HTTPException(422, GUIDANCE_MISSING)
    guidance = body["guidance"]
    if guidance is not None and not isinstance(guidance, str):
        raise HTTPException(422, GUIDANCE_NOT_TEXT)
    row = await _source_row(session, source_id)
    try:
        value = await catalog_jobs.save_guidance(session, row, guidance)
    except catalog_jobs.TextTooLong as e:
        raise HTTPException(422, str(e)) from e
    return {"guidance": value}


async def _ensure_model(session: AsyncSession) -> None:
    """起草模型用不了（没有接入、停用、只有演示接入……）：409，不建任务。说法同同步起草时的 model_error。"""
    try:
        await catalog_jobs.check_model(session)
    except catalog.CatalogModelUnavailable as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_MODEL_UNAVAILABLE) from e


def _no_schema(cache: dict[str, Any]) -> HTTPException:
    return HTTPException(409, f"{introspect.why_empty(cache)}，无法起草数据目录。请先在数据源卡片上点「探查结构」")


@router.post("/{source_id}/catalog/draft-jobs", status_code=202)
async def create_draft_job(source_id: str, payload: DraftJobIn, x_actor: str | None = Header(default=None),
                           session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """建后台起草任务，马上返回（202，任务是排队中）。同一个源已有排队或进行中的任务 409 catalog_draft_busy（界面改去
    GET /draft-jobs/active 显示那一个）；用模型而起草模型用不了 409 catalog_draft_model_unavailable；没有表结构 409。
    表结构里没有的表照样建进任务、状态 skipped。"""
    if payload.tables and len(payload.tables) > catalog_jobs.MAX_TABLES:
        raise HTTPException(422, f"一次最多起草 {catalog_jobs.MAX_TABLES} 张表，当前 {len(payload.tables)} 张，"
                                 "请分几次起草")
    try:
        note = catalog_jobs.clean_text(payload.note, catalog_jobs.NOTE_MAX, "本次说明")
    except catalog_jobs.TextTooLong as e:
        raise HTTPException(422, str(e)) from e
    row, source = await _resolved(session, source_id)
    try:
        await catalog_jobs.ensure_idle(session, row)
    except catalog_jobs.DraftBusy as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
    cache = source.schema_cache or {}
    if not cache.get("tables"):
        raise _no_schema(cache)
    if payload.use_model:
        await _ensure_model(session)
    try:
        job, tasks = await catalog_jobs.create_job(session, row, source, tables=payload.tables or None,
                                                   use_model=payload.use_model, note=note, actor=actor_of(x_actor))
    except catalog_jobs.DraftBusy as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
    catalog_jobs.manager.start(job.id)
    return {"job": catalog_jobs.job_dict(job, tasks)}


@router.get("/{source_id}/catalog/draft-jobs/active")
async def active_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """排队或进行中的那个起草任务，没有是 null。"""
    row = await _source_row(session, source_id)
    return {"job": await catalog_jobs.job_out(session, await catalog_jobs.active_job(session, row.id))}


@router.get("/{source_id}/catalog/draft-jobs/latest")
async def latest_draft_job(source_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """最近建的一个起草任务（任何状态），没有是 null。"""
    row = await _source_row(session, source_id)
    return {"job": await catalog_jobs.job_out(session, await catalog_jobs.latest_job(session, row.id))}


async def _job_or_404(session: AsyncSession, source_id: str, job_id: str) -> tuple[DataSource, Any]:
    row = await _source_row(session, source_id)
    job = await catalog_jobs.load(session, row.id, job_id)
    if job is None:
        raise HTTPException(404, "起草任务不存在，可能属于别的数据源，或者已随数据源删除")
    return row, job


@router.get("/{source_id}/catalog/draft-jobs/{job_id}")
async def get_draft_job(source_id: str, job_id: str, session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    _, job = await _job_or_404(session, source_id, job_id)
    return {"job": await catalog_jobs.job_out(session, job)}


@router.post("/{source_id}/catalog/draft-jobs/{job_id}/cancel")
async def cancel_draft_job(source_id: str, job_id: str,
                           session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """停止：排队的直接停止；进行中的停掉正在进行的模型调用，没开始和没写完的表记停止，已经写进目录的不回滚。
    已经结束的 409 catalog_draft_finished。返回停止之后的任务。"""
    row, job = await _job_or_404(session, source_id, job_id)
    owner = row.id
    try:
        await catalog_jobs.cancel(session, job)
    except catalog_jobs.DraftFinished as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_FINISHED) from e
    session.expire_all()                  # 后台在别的会话里收的尾：重新读
    return {"job": await catalog_jobs.job_out(session, await catalog_jobs.load(session, owner, job_id))}


@router.post("/{source_id}/catalog/draft-jobs/{job_id}/retry", status_code=202)
async def retry_draft_job(source_id: str, job_id: str, x_actor: str | None = Header(default=None),
                          session: AsyncSession = Depends(get_session)) -> dict[str, Any]:
    """重试：新任务，只含原任务里没写好（failed、partial）和停止了（cancelled）的表；用不用模型、本次说明沿用，起草要求
    取数据源当前的。有排队或进行中的任务 409 catalog_draft_busy；没有可重试的表 422；起草模型用不了 409。"""
    _, old = await _job_or_404(session, source_id, job_id)
    row, source = await _resolved(session, source_id)
    try:
        await catalog_jobs.ensure_idle(session, row)
    except catalog_jobs.DraftBusy as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
    try:
        tables = await catalog_jobs.retry_tables(session, old)
    except catalog_jobs.NothingToRetry as e:
        raise HTTPException(422, str(e)) from e
    cache = source.schema_cache or {}
    if not cache.get("tables"):
        raise _no_schema(cache)
    if old.use_model:
        await _ensure_model(session)
    try:
        job, tasks = await catalog_jobs.create_job(session, row, source, tables=tables, use_model=old.use_model,
                                                   note=old.note, actor=actor_of(x_actor), retry_of=old.id)
    except catalog_jobs.DraftBusy as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
    catalog_jobs.manager.start(job.id)
    return {"job": catalog_jobs.job_dict(job, tasks)}


@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 有值）。

    双向互斥（catalog_jobs.sync_draft，进程内登记）：同一个源有排队或进行中的后台起草任务、或者正在同步起草时 409
    catalog_draft_busy（和建任务同一句话）；反过来，同步起草进行期间同一个源建任务、重试、导入目录也是 409。两边同时写
    同一个源的目录、抢模型名额。
    """
    row, source = await _resolved(session, source_id)
    cache = source.schema_cache or {}
    try:
        async with catalog_jobs.sync_draft(session, row):
            if not cache.get("tables"):
                raise _no_schema(cache)
            report = await catalog.draft_catalog(session, source, tables=payload.tables or None,
                                                 use_model=payload.use_model, actor=actor_of(x_actor),
                                                 guidance=catalog_jobs.guidance_of(row))
    except catalog_jobs.DraftBusy as e:
        raise CodedHTTPException(409, str(e), CATALOG_DRAFT_BUSY) from e
    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}