数据目录阶段 1A：新增 /api/datasources/{id}/catalog 接口——表清单、单表、整份提交与单项审阅（乐观锁 409）、同步起草

backend/app/api/catalog.py

@@ -0,0 +1,226 @@
+"""数据目录接口：/api/datasources/{source_id}/catalog。
+
+- GET    /catalog                表清单（没有目录的表也列出来）
+- GET    /catalog/{table}        单表目录 + 表结构
+- PUT    /catalog/{table}        整份提交（乐观锁），改动过的项记为人工确认
+- POST   /catalog/{table}/review 单项审阅：确认、驳回、撤销审阅
+- POST   /catalog/draft          同步起草（注释、外键、命名推断，可选模型）
+
+规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径
+不对 422，源或表不存在 404。所有写操作记录请求头 X-Actor 的署名（自报、未认证）。
+
+上传表格的源按当前快照的表结构回答（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.runs import actor_of
+from app.data import catalog, introspect, table_versions
+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 CatalogDraftIn(BaseModel):
+    #: 要起草的表；不给或为空时按使用次数取前 20 张
+    tables: list[str] | None = Field(default=None, max_length=200)
+    #: 是否请助手的模型起草中文名、粒度、列的含义等
+    use_model: bool = False
+
+
+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 _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]:
+    """表清单：表结构里的每张表各一行（没有目录的也列），使用次数多的在前；目录还在、表已不在的排最后。"""
+    _, 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),
+    }
+
+
+@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]:
+    """同步起草。返回每张表新增、更新、删除了几项；模型用不了或某张表的模型起草失败，照实写在结果里。"""
+    _, 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} 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,
+    }
+
+
+@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)

backend/app/main.py

@@ -9,7 +9,7 @@ from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 
 from app.api import (
-    artifacts, conversations, copilot, datasources, evidence, governance, knowledge, runs,
+    artifacts, catalog, conversations, copilot, datasources, evidence, governance, knowledge, runs,
     sandbox, settings as settings_api, source_versions, table_imports, tools, workflows,
 )
 from app.api.coded import CodedHTTPException, coded_handler
@@ -120,6 +120,7 @@ for router in (
     # 按配方导入的 /api/datasources/imports/… 要在 datasources 的通配 /{source_id} 之前登记
     table_imports.router,
     source_versions.router,
+    catalog.router,
     datasources.router,
 ):
     app.include_router(router)

backend/tests/fixtures/catalog/fake_model.py

@@ -0,0 +1,30 @@
+"""数据目录起草用的假模型：不发请求，按表名回固定的草稿。
+
+    model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}}, fail={"orders"})
+    await catalog.draft_with_model(model, source, ["visits"])
+
+- 结构化输出（with_structured_output）返回自己；ainvoke 从提示词里认出这一批有哪些表（「表 <表名>」），
+  返回 {"tables": [{"table": 表名, **replies[表名]}, …]}；
+- fail 里的表所在的那一批抛异常，用来验证「一批失败只影响这一批」；
+- calls 记下每次收到的提示词全文，用来核对交给模型的内容（只有结构，没有数据）。
+"""
+from __future__ import annotations
+
+from typing import Any
+
+
+class FakeModel:
+    def __init__(self, replies: dict[str, dict], fail: set[str] = frozenset()) -> None:
+        self.replies, self.fail, self.calls = replies, set(fail), []
+
+    def with_structured_output(self, schema: Any, **_: Any) -> "FakeModel":
+        assert schema["title"]
+        return self
+
+    async def ainvoke(self, messages: list[Any]) -> dict:
+        text = "\n".join(str(getattr(m, "content", m)) for m in messages)
+        self.calls.append(text)
+        names = [t for t in self.replies if f"表 {t}\n" in text or f"表 {t}（" in text]
+        if any(t in self.fail for t in names):
+            raise RuntimeError("网关超时")
+        return {"tables": [{"table": t, **self.replies[t]} for t in names]}

backend/tests/test_catalog_api.py

@@ -0,0 +1,189 @@
+"""数据目录接口：/api/datasources/{id}/catalog 下的表清单、单表、整份提交、单项审阅、起草。
+
+所有写操作记录请求头 X-Actor 的署名；版本不符一律 409，detail 说明这张表刚被修改过、请重新载入。
+"""
+from __future__ import annotations
+
+import uuid
+from urllib.parse import quote
+
+import pytest
+from httpx import ASGITransport, AsyncClient
+
+from app.data import catalog
+from app.data.engine import engines
+from app.db.base import SessionLocal
+from app.db.models import DataSource
+from app.main import app as _fastapi_app
+from tests.fixtures.catalog import scenic
+from tests.fixtures.catalog.fake_model import FakeModel
+
+ACTOR = {"X-Actor": quote("王敏")}
+
+
+@pytest.fixture
+async def client():
+    async with AsyncClient(transport=ASGITransport(app=_fastapi_app), base_url="http://test") as c:
+        yield c
+
+
+@pytest.fixture
+async def source_id(client, scenic_db):
+    """接上合成景区库并探查结构（走真实的探查接口）。"""
+    async with SessionLocal() as session:
+        row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=scenic_db)
+        session.add(row)
+        await session.commit()
+        sid = row.id
+    r = await client.post(f"/api/datasources/{sid}/introspect")
+    assert r.status_code == 200, r.text
+    yield sid
+    await engines.invalidate(sid)
+
+
+def _base(sid: str) -> str:
+    return f"/api/datasources/{sid}/catalog"
+
+
+async def test_list_includes_tables_without_notes(client, source_id):
+    r = await client.get(_base(source_id))
+    assert r.status_code == 200
+    body = r.json()
+    rows = {t["table_name"]: t for t in body["tables"]}
+    assert len(rows) == scenic.TABLE_COUNT + len(scenic.VIEWS)
+    visits = rows["visits"]
+    assert visits["version"] == 0 and visits["label"] is None and visits["usage"] == 0
+    assert visits["counts"] == {"proposed": 0, "verified": 0, "confirmed": 0, "rejected": 0}
+    assert visits["relations"] == 0 and visits["updated_at"] is None
+    assert body["system_notes"] is False
+
+
+async def test_draft_then_read_table(client, source_id):
+    r = await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits", "nope"], "use_model": False},
+                          headers=ACTOR)
+    assert r.status_code == 200, r.text
+    body = r.json()
+    by = {t["table_name"]: t for t in body["tables"]}
+    # visits：两条外键（景区、票种）+ 两条命名推断（闸口、会员）
+    assert by["visits"]["added"] == 4 and by["visits"]["version"] == 1 and by["visits"]["error"] is None
+    assert by["nope"]["error"] and body["model_used"] is False
+    assert body["total"] == {"added": 4, "updated": 0, "removed": 0}
+
+    r = await client.get(f"{_base(source_id)}/visits")
+    assert r.status_code == 200
+    detail = r.json()
+    assert detail["version"] == 1 and detail["updated_by"] == "王敏"
+    sources = sorted(rel["source"] for rel in detail["notes"]["relations"])
+    assert sources == ["fk", "fk", "name", "name"]
+    structure = detail["structure"]
+    assert structure["primary_key"] == ["id"]
+    assert {fk["to_table"] for fk in structure["foreign_keys"]} == {"parks", "ticket_types"}
+    assert ["ticket_no"] in structure["unique"]
+    assert any(c["name"] == "visit_time" for c in structure["columns"])
+
+    listed = {t["table_name"]: t for t in (await client.get(_base(source_id))).json()["tables"]}
+    assert listed["visits"]["counts"] == {"proposed": 2, "verified": 2, "confirmed": 0, "rejected": 0}
+    assert listed["visits"]["relations"] == 4 and listed["visits"]["version"] == 1
+
+
+async def test_put_marks_edits_human_confirmed_and_bumps_version(client, source_id):
+    await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
+    detail = (await client.get(f"{_base(source_id)}/visits")).json()
+    notes = detail["notes"]
+    notes["label"] = {"value": "入园记录"}
+    notes["grain"] = {"value": "每张门票每次检票一行", "source": "llm", "status": "proposed"}
+    r = await client.put(f"{_base(source_id)}/visits", json={"notes": notes, "if_version": detail["version"]},
+                         headers={"X-Actor": quote("李雷")})
+    assert r.status_code == 200, r.text
+    saved = r.json()
+    assert saved["version"] == 2 and saved["updated_by"] == "李雷"
+    assert saved["notes"]["label"]["source"] == "human" and saved["notes"]["label"]["status"] == "confirmed"
+    assert saved["notes"]["grain"]["source"] == "human"
+    assert {r["source"] for r in saved["notes"]["relations"]} == {"fk", "name"}     # 没动的保持原样
+
+
+async def test_put_with_stale_version_conflicts(client, source_id):
+    await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
+    r = await client.put(f"{_base(source_id)}/visits", json={"notes": {"label": {"value": "x"}}, "if_version": 0},
+                         headers=ACTOR)
+    assert r.status_code == 409
+    assert "刚被修改过" in r.json()["detail"] and "重新载入" in r.json()["detail"]
+
+
+async def test_put_rejects_malformed_notes(client, source_id):
+    r = await client.put(f"{_base(source_id)}/visits",
+                         json={"notes": {"formula": {"value": "sum(amount)"}}, "if_version": 0}, headers=ACTOR)
+    assert r.status_code == 422
+    assert "formula" in r.json()["detail"]
+
+
+async def test_review_confirm_reject_and_conflict(client, source_id):
+    await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"]}, headers=ACTOR)
+    detail = (await client.get(f"{_base(source_id)}/visits")).json()
+    gate = next(r for r in detail["notes"]["relations"] if r["columns"] == ["gate_id"])
+    r = await client.post(f"{_base(source_id)}/visits/review",
+                          json={"path": f"relations.{gate['id']}", "action": "confirm", "if_version": 1},
+                          headers={"X-Actor": quote("韩梅")})
+    assert r.status_code == 200, r.text
+    after = r.json()
+    assert after["version"] == 2 and after["updated_by"] == "韩梅"
+    assert next(x for x in after["notes"]["relations"] if x["id"] == gate["id"])["status"] == "confirmed"
+    stale = await client.post(f"{_base(source_id)}/visits/review",
+                              json={"path": f"relations.{gate['id']}", "action": "reject", "if_version": 1})
+    assert stale.status_code == 409 and "重新载入" in stale.json()["detail"]
+    bad = await client.post(f"{_base(source_id)}/visits/review",
+                            json={"path": "columns.visit_time.formula", "action": "confirm", "if_version": 2})
+    assert bad.status_code == 422
+    wrong = await client.post(f"{_base(source_id)}/visits/review",
+                              json={"path": "grain", "action": "approve", "if_version": 2})
+    assert wrong.status_code == 422
+
+
+async def test_unknown_table_and_source(client, source_id):
+    assert (await client.get(f"{_base(source_id)}/nope")).status_code == 404
+    assert (await client.get(_base("missing-source"))).status_code == 404
+    r = await client.put(f"{_base(source_id)}/nope", json={"notes": {}, "if_version": 0})
+    assert r.status_code == 404
+
+
+async def test_draft_defaults_to_top_tables_and_model_via_settings(client, source_id, monkeypatch):
+    model = FakeModel({"visits": {"label": "入园记录", "kind": "fact"}})
+
+    async def fake_resolve(session):
+        return model, "fake-model"
+
+    async def usage(session, source):
+        return {"visits": 7}
+
+    monkeypatch.setattr(catalog, "resolve_draft_model", fake_resolve)
+    monkeypatch.setattr(catalog, "table_usage", usage)
+    r = await client.post(f"{_base(source_id)}/draft", json={"use_model": True}, headers=ACTOR)
+    assert r.status_code == 200, r.text
+    body = r.json()
+    assert len(body["tables"]) == 20 and body["tables"][0]["table_name"] == "visits"
+    assert body["model_used"] is True and body["model"] == "fake-model"
+    detail = (await client.get(f"{_base(source_id)}/visits")).json()
+    assert detail["notes"]["label"]["source"] == "llm" and detail["usage"] == 7
+
+
+async def test_draft_reports_model_unavailable(client, source_id, monkeypatch):
+    async def unavailable(session):
+        raise catalog.CatalogModelUnavailable("未配置模型接入，无法由助手起草数据目录。请到「设置 → 模型接入」添加")
+
+    monkeypatch.setattr(catalog, "resolve_draft_model", unavailable)
+    r = await client.post(f"{_base(source_id)}/draft", json={"tables": ["visits"], "use_model": True})
+    assert r.status_code == 200
+    body = r.json()
+    assert body["model_used"] is False and "模型接入" in body["model_error"]
+    assert body["tables"][0]["added"] == 4                          # 外键和命名推断照常写入
+
+
+async def test_draft_without_schema_is_refused(client):
+    async with SessionLocal() as session:
+        row = DataSource(name=f"cat_{uuid.uuid4().hex[:10]}", kind="sqlite", database=":memory:")
+        session.add(row)
+        await session.commit()
+        sid = row.id
+    r = await client.post(f"{_base(sid)}/draft", json={})
+    assert r.status_code == 409 and "探查结构" in r.json()["detail"]
+    assert (await client.get(_base(sid))).json()["tables"] == []

backend/tests/test_catalog_draft.py

@@ -7,7 +7,6 @@ from __future__ import annotations
 import json
 import uuid
 from types import SimpleNamespace
-from typing import Any
 
 import pytest
 
@@ -15,6 +14,7 @@ from app.data import catalog
 from app.data.catalog import make_item
 from app.db.base import SessionLocal
 from app.db.models import DataSource
+from tests.fixtures.catalog.fake_model import FakeModel
 
 
 def _col(name: str, type_: str = "INTEGER", comment: str | None = None, nullable: bool = True) -> dict:
@@ -165,25 +165,6 @@ def test_name_inference_one_to_one_when_column_is_own_key():
 # ---------------------------------------------------------------- 模型起草
 
 
-class FakeModel:
-    """假模型：结构化输出返回 replies[表名] 拼成的 {"tables": [...]}；fail 里的表所在的那一批抛异常。"""
-
-    def __init__(self, replies: dict[str, dict], fail: set[str] = frozenset()) -> None:
-        self.replies, self.fail, self.calls = replies, set(fail), []
-
-    def with_structured_output(self, schema: Any, **_: Any) -> "FakeModel":
-        assert schema["title"]
-        return self
-
-    async def ainvoke(self, messages: list[Any]) -> dict:
-        text = "\n".join(str(getattr(m, "content", m)) for m in messages)
-        self.calls.append(text)
-        names = [t for t in self.replies if f"表 {t}" in text]
-        if any(t in self.fail for t in names):
-            raise RuntimeError("网关超时")
-        return {"tables": [{"table": t, **self.replies[t]} for t in names]}
-
-
 _VISITS_REPLY = {
     "label": "入园记录", "grain": "每张门票每次检票一行", "kind": "fact", "keys": ["ticket_no"],
     "business_date": {"column": "visit_time", "rule": "按检票时间计", "timezone": "+08:00"},