diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py new file mode 100644 index 00000000..3bde3acd --- /dev/null +++ b/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) diff --git a/backend/app/main.py b/backend/app/main.py index 0a17c1f8..0faed210 100644 --- a/backend/app/main.py +++ b/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) diff --git a/backend/tests/fixtures/catalog/fake_model.py b/backend/tests/fixtures/catalog/fake_model.py new file mode 100644 index 00000000..e9d26f3b --- /dev/null +++ b/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]} diff --git a/backend/tests/test_catalog_api.py b/backend/tests/test_catalog_api.py new file mode 100644 index 00000000..f0f1080b --- /dev/null +++ b/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"] == [] diff --git a/backend/tests/test_catalog_draft.py b/backend/tests/test_catalog_draft.py index 34d4f7bc..2de32020 100644 --- a/backend/tests/test_catalog_draft.py +++ b/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"},