From 958e096e56a326538e05b29cc94e7cbf3aa12d91 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Sat, 3 Oct 2026 11:52:49 +0800 Subject: [PATCH] =?UTF-8?q?=E6=95=B0=E6=8D=AE=E7=9B=AE=E5=BD=95=E9=98=B6?= =?UTF-8?q?=E6=AE=B5=204A=20=E5=BD=B1=E5=93=8D=E9=9D=A2=EF=BC=9A=E6=96=B0?= =?UTF-8?q?=E5=A2=9E=E6=8E=A5=E5=8F=A3=E5=88=97=E5=87=BA=E5=BC=95=E7=94=A8?= =?UTF-8?q?=E6=9F=90=E5=BC=A0=E8=A1=A8=E7=9A=84=E5=B7=B2=E5=8F=91=E5=B8=83?= =?UTF-8?q?=E5=92=8C=E5=8F=97=E7=AE=A1=E6=A8=A1=E6=9D=BF=EF=BC=8C=E5=8C=BA?= =?UTF-8?q?=E5=88=86=E7=9B=B4=E6=8E=A5=E5=BC=95=E7=94=A8=E5=92=8C=E5=8F=AF?= =?UTF-8?q?=E8=83=BD=E6=B6=89=E5=8F=8A=E3=80=81=E5=90=88=E5=B9=B6=E6=9F=A5?= =?UTF-8?q?=E8=AF=A2=E6=8C=89=E8=BE=93=E5=85=A5=E8=BF=BD=E6=BA=AF=EF=BC=8C?= =?UTF-8?q?=E8=A1=A8=E8=AF=A6=E6=83=85=E5=92=8C=E4=BF=9D=E5=AD=98=E7=BA=A0?= =?UTF-8?q?=E6=AD=A3=E5=90=8E=E9=83=BD=E6=98=BE=E7=A4=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/catalog.py | 15 +- backend/app/data/catalog_impact.py | 289 +++++++++++++++++++++ backend/tests/test_catalog_impact.py | 185 +++++++++++++ frontend/src/api/client.ts | 5 +- frontend/src/components/CatalogImpact.tsx | 122 +++++++++ frontend/src/lib/terms.ts | 24 ++ frontend/src/pages/catalog/TableDetail.tsx | 13 +- frontend/src/run/CatalogPatchCard.tsx | 4 +- frontend/src/types.ts | 31 +++ scripts/check-catalog.mjs | 48 ++++ scripts/check-stream.mjs | 11 +- 11 files changed, 741 insertions(+), 6 deletions(-) create mode 100644 backend/app/data/catalog_impact.py create mode 100644 backend/tests/test_catalog_impact.py create mode 100644 frontend/src/components/CatalogImpact.tsx diff --git a/backend/app/api/catalog.py b/backend/app/api/catalog.py index e1c96af0..10a26f87 100644 --- a/backend/app/api/catalog.py +++ b/backend/app/api/catalog.py @@ -6,6 +6,7 @@ - POST /catalog/{table}/review 单项审阅：确认、驳回、撤销审阅 - POST /catalog/{table}/patch/preview 目录修改提案的预览（只读）：对着当前目录算改前、改后 - POST /catalog/{table}/patch 保存目录修改提案（乐观锁），记为人工确认 +- GET /catalog/{table}/impact 影响面：引用这张表的已发布、受管模板（data/catalog_impact.py） - POST /catalog/draft 同步起草（注释、外键、命名推断，可选模型） 规则都在 data/catalog.py，这里只做取源、转换形状和把异常翻成状态码：版本不符 409，结构不合规、审阅路径 @@ -22,7 +23,7 @@ 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.data import catalog, catalog_impact, introspect, table_versions from app.db.base import get_session from app.db.models import DataSource @@ -280,3 +281,15 @@ async def save_catalog_patch(source_id: str, table: str, payload: CatalogPatchIn 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} diff --git a/backend/app/data/catalog_impact.py b/backend/app/data/catalog_impact.py new file mode 100644 index 00000000..7fd60f17 --- /dev/null +++ b/backend/app/data/catalog_impact.py @@ -0,0 +1,289 @@ +"""数据目录的影响面与发布时的目录版本（阶段 4A：维护闭环）。 + +目录改了一项（「在园人数是存量」），用到这张表的模板算出来的数可能就该变了。这里回答两件事： + +- **影响面**：哪些已发布、受管的模板用到了这张表（table_impact）。按每个模板当前的已发布版本算，不看画布上的 + 草稿——正式运行跑的是已发布的那一版。调用工具节点写死的 SQL 用到了这张表算「直接引用」；Agent 绑定了这个源 + 的查询工具算「可能涉及」（SQL 是运行时写的，静态看不出）；合并查询按它的输入一路追溯。 +- **发布时的目录版本**：发布时记下模板 SQL 用到的每张表的目录版本 {源名: {表名: 版本}}（catalog_versions_for）， + 从这一版发起正式运行时和当前的比（catalog_drift），有变化只提醒、不拦运行——目录是给人和助手看的说明， + 不是运行的输入；拦下来等于让一次文字修订卡住正式出具。 + +表名的取法和证据台账同一个口径：evidence.sql_tables 取 FROM / JOIN 后面的表（去掉注释、字符串、CTE）， +去掉 schema 前缀后按 name_key 对到表结构上（和 catalog.table_usage 一致）。 +""" +from __future__ import annotations + +import logging +from collections.abc import Iterable, Mapping +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.data.names import name_key + +logger = logging.getLogger(__name__) + +#: 影响程度：direct 直接引用（SQL 里写着这张表）；possible 可能涉及（Agent 运行时自己写 SQL） +IMPACTS = ("direct", "possible") + + +def _lookup(tables: Iterable[str]) -> dict[str, str]: + out: dict[str, str] = {} + for name in tables: + out.setdefault(name_key(name), name) + return out + + +def sql_table_keys(sql: str, tables: Iterable[str] | None) -> set[str]: + """SQL 用到的表，对到表结构的写法上。tables 不给（不知道表结构）时用 SQL 里去掉 schema 前缀的原名。""" + from app.engine.evidence import sql_tables + + try: + names = sql_tables(sql or "") + except Exception: # noqa: BLE001 - 解析不了就当没用到表：影响面和版本记录都是提示，不能因此报错 + return set() + shorts = [str(n).rsplit(".", 1)[-1] for n in names] + if tables is None: + return set(shorts) + lookup = _lookup(tables) + return {hit for s in shorts if (hit := lookup.get(name_key(s))) is not None} + + +def graph_sql_tables(spec: Any, tables_by_source: Mapping[str, Iterable[str]] | None = None) -> dict[str, set[str]]: + """一张图里调用工具节点写死的 SQL 用到的表：{源名: {表名}}。Agent 的 SQL 运行时才有，不在这里。 + + tables_by_source 是各源表结构里的表名（schema_cache 的 tables 的键）：给了的源按它规范表名、对不上的不算； + 没给的源用 SQL 里的原名。 + """ + from app.data.sqlcheck import tool_query + + out: dict[str, set[str]] = {} + for node in getattr(spec, "nodes", []) or []: + found = tool_query(node) + if not found: + continue + source, sql = found + known = (tables_by_source or {}).get(source) + hits = sql_table_keys(sql, list(known) if known is not None else None) + if hits: + out.setdefault(source, set()).update(hits) + return out + + +def _kind(node: Any) -> str: + return str(getattr(node.type, "value", node.type)) + + +def _setting(spec: Any, node: Any, key: str) -> Any: + """节点配置的实际取值：节点上没写就取图级 defaults（和运行时 NodeContext.cfg 同一个规则）。""" + value = (node.config or {}).get(key) + return (getattr(spec, "defaults", None) or {}).get(key) if value in (None, "") else value + + +def template_refs(spec: Any, source: str, table: str, tables: Iterable[str] | None) -> list[dict[str, Any]]: + """一张图里哪些节点用到了某个源的某张表，按图里的顺序。每项 {node_id, label, type, impact, via?, member?}。 + + - 调用工具节点：调用 db_query__<源>、SQL 里有这张表 → direct。 + - Agent 节点、协作成员：绑定了 db_query__<源> → possible（每个成员各一项，带 member）。 + - 合并查询：输入（别名 → 节点）里有用到这张表的 → 取输入里最强的那一级，via 写经由哪些输入。合并的合并 + 一路追到底；成环（校验会报错的图）时停下。 + """ + from app.data.sqlcheck import tool_query + from app.tools.datasource import QUERY_PREFIX + + tool = f"{QUERY_PREFIX}{source}" + wanted = name_key(table) + known = list(tables) if tables is not None else None + nodes = list(getattr(spec, "nodes", []) or []) + by_id = {n.id: n for n in nodes} + own: dict[str, dict[str, Any]] = {} + out: list[dict[str, Any]] = [] + + def base(node: Any, impact: str, **extra: Any) -> dict[str, Any]: + return {"node_id": node.id, "label": node.title, "type": _kind(node), "impact": impact, **extra} + + for node in nodes: + kind = _kind(node) + if kind == "tool": + found = tool_query(node) + if found and found[0] == source and wanted in {name_key(t) for t in sql_table_keys(found[1], known)}: + own[node.id] = base(node, "direct") + elif kind == "agent": + if tool in [str(t).strip() for t in _setting(spec, node, "tools") or [] if isinstance(t, str)]: + own[node.id] = base(node, "possible") + elif kind == "supervisor": + for member in _setting(spec, node, "agents") or []: + if isinstance(member, Mapping) and tool in [str(t).strip() for t in member.get("tools") or [] + if isinstance(t, str)]: + out.append(base(node, "possible", member=str(member.get("name") or ""))) + + resolved: dict[str, dict[str, Any] | None] = {} + + def through(node_id: str, trail: frozenset[str]) -> dict[str, Any] | None: + """合并查询节点经由输入用到这张表的样子；用不到返回 None。""" + if node_id in resolved: + return resolved[node_id] + node = by_id.get(node_id) + if node is None or _kind(node) != "merge" or node_id in trail: + return own.get(node_id) + via: list[dict[str, Any]] = [] + best: str | None = None + inputs = (node.config or {}).get("inputs") + for alias, target in (inputs.items() if isinstance(inputs, Mapping) else []): + if not isinstance(target, str): + continue + hit = own.get(target) or through(target, trail | {node_id}) + if hit is None: + continue + via.append({"node_id": target, "label": by_id[target].title if target in by_id else target, "alias": alias}) + best = "direct" if hit["impact"] == "direct" or best == "direct" else "possible" + resolved[node_id] = base(node, best, via=via) if best else None + return resolved[node_id] + + for node in nodes: + if node.id in own: + out.append(own[node.id]) + elif _kind(node) == "merge" and (hit := through(node.id, frozenset())): + out.append(hit) + order = {n.id: i for i, n in enumerate(nodes)} + out.sort(key=lambda r: order.get(r["node_id"], 0)) + return out + + +async def _published(session: AsyncSession) -> list[tuple[Any, Any]]: + """每个发布过的模板和它当前的已发布版本：[(Workflow, WorkflowVersion)]。""" + from app.db.models import Workflow, WorkflowVersion + + rows = (await session.execute( + select(Workflow, WorkflowVersion) + .join(WorkflowVersion, (WorkflowVersion.workflow_id == Workflow.id) + & (WorkflowVersion.version == Workflow.published_version)) + .where(Workflow.published_version.is_not(None)) + )).all() + return [(w, v) for w, v in rows] + + +async def table_impact(session: AsyncSession, source: str, table: str, + tables: Iterable[str] | None) -> list[dict[str, Any]]: + """引用某个源某张表的已发布、受管模板，直接引用的在前，同级按名字。 + + 每项 {workflow_id, name, version, level, impact, nodes}：version 是当前的已发布版本，level 是它发布时的级别 + （published / governed；记级别之前发布的老版本按工作流现在的状态）。图解析不了的版本跳过。 + """ + from app.engine.schema import GraphSpec + + known = list(tables) if tables is not None else None + out: list[dict[str, Any]] = [] + for workflow, version in await _published(session): + try: + spec = GraphSpec.model_validate(version.graph or {}) + except Exception: # noqa: BLE001 - 解析不了的版本发起不了正式运行，也就谈不上影响 + continue + refs = template_refs(spec, source, table, known) + if not refs: + continue + level = version.level or (workflow.status if workflow.status in ("published", "governed") else "published") + out.append({ + "workflow_id": workflow.id, "name": workflow.name, "version": version.version, "level": level, + "impact": "direct" if any(r["impact"] == "direct" for r in refs) else "possible", "nodes": refs, + }) + out.sort(key=lambda t: (t["impact"] != "direct", t["name"])) + return out + + +# ========================================================================== +# 发布时的目录版本 +# ========================================================================== + + +async def _sources_by_name(session: AsyncSession, names: Iterable[str]) -> dict[str, Any]: + """启用着的数据源：{源名: (DataSource 行, 用来读表结构的源)}。上传源解析成当前快照的视图，解析不了用原行。""" + from app.data import table_versions + from app.db.models import DataSource + + wanted = sorted({n for n in names if n}) + if not wanted: + return {} + rows = (await session.execute( + select(DataSource).where(DataSource.name.in_(wanted), DataSource.enabled.is_(True)) + )).scalars() + out: dict[str, Any] = {} + for row in rows: + try: + view = await table_versions.resolve_source(session, row) + except table_versions.SnapshotMissing: + view = row + out[row.name] = (row, view) + return out + + +def _schema_tables(view: Any) -> list[str] | None: + tables = (getattr(view, "schema_cache", None) or {}).get("tables") + return list(tables) if isinstance(tables, Mapping) and tables else None + + +async def catalog_versions_for(session: AsyncSession, spec: Any) -> dict[str, dict[str, int]]: + """发布时要记下的目录版本：{源名: {表名: 版本}}，只收模板 SQL 用到的表；还没有目录的表记 0。 + + 记 0 是有意的：发布之后有人给这张表补了目录，同样算「自发布以来有变化」。找不到的源（停用、删了）不记： + 没有目录可比，正式运行时工具自己会报源不存在。读库出错时返回空 dict——记不下版本不该挡住发布。 + """ + from app.data import catalog + + try: + names = graph_sql_tables(spec).keys() + sources = await _sources_by_name(session, names) + tables_by_source = {name: _schema_tables(view) for name, (_, view) in sources.items()} + used = graph_sql_tables(spec, {k: v for k, v in tables_by_source.items() if v is not None}) + out: dict[str, dict[str, int]] = {} + for name, tables in sorted(used.items()): + if name not in sources: + continue + entries = await catalog.read_catalog(session, sources[name][0].id) + out[name] = {t: (entries[t].version if t in entries else 0) for t in sorted(tables)} + return out + except Exception: # noqa: BLE001 + logger.exception("记录发布时的目录版本失败，本次不记") + return {} + + +async def catalog_drift(session: AsyncSession, recorded: Mapping[str, Any] | None) -> list[dict[str, Any]]: + """发布时记下的目录版本和现在比：变了的表 [{source, source_id, table, label, published, current}]。 + + 没记过（这个功能之前发布的版本）返回空列表——不知道当时是哪一版，不能说它变了。现在找不到的源跳过（理由同 + catalog_versions_for）。label 是表现在的中文名（没有或被驳回时为 None）。读库出错返回空列表：提醒是附加的。 + """ + from app.data import catalog + + if not isinstance(recorded, Mapping) or not recorded: + return [] + try: + sources = await _sources_by_name(session, recorded.keys()) + out: list[dict[str, Any]] = [] + for name in sorted(recorded): + pinned = recorded[name] + if name not in sources or not isinstance(pinned, Mapping): + continue + row = sources[name][0] + entries = await catalog.read_catalog(session, row.id) + for table in sorted(pinned): + then = pinned[table] + entry = entries.get(table) + now = entry.version if entry else 0 + if isinstance(then, int) and then == now: + continue + label = (entry.notes.get("label") if entry else None) or {} + out.append({"source": name, "source_id": row.id, "table": table, + "label": label.get("value") if isinstance(label, dict) and label.get("status") != "rejected" + else None, + "published": then if isinstance(then, int) else None, "current": now}) + return out + except Exception: # noqa: BLE001 + logger.exception("比对发布时的目录版本失败，本次不提醒") + return [] + + +__all__ = ["IMPACTS", "catalog_drift", "catalog_versions_for", "graph_sql_tables", "sql_table_keys", "table_impact", + "template_refs"] diff --git a/backend/tests/test_catalog_impact.py b/backend/tests/test_catalog_impact.py new file mode 100644 index 00000000..7e07e2af --- /dev/null +++ b/backend/tests/test_catalog_impact.py @@ -0,0 +1,185 @@ +"""目录的影响面：改了一张表的目录，哪些已发布、受管的模板会受影响（阶段 4A）。 + +GET /api/datasources/{id}/catalog/{table}/impact，每个模板按它当前的已发布版本算（不是画布上的草稿）： +- 调用工具节点调用这个源的 db_query__，SQL 里用到了这张表：直接引用。表名的取法和证据台账同一个口径 + （evidence.sql_tables，按 name_key 对到表结构上）。 +- Agent 节点（协作成员也算）绑定了这个源的查询工具：可能涉及——SQL 是运行时写的，静态看不出用不用这张表。 +- 合并查询节点按它的输入追溯：输入里有直接引用的，它也算直接引用，并写明经由哪个输入。 +- 没发布过的草稿、别的源的同名表、只在未发布草稿里用到的，都不算。 +""" +from __future__ import annotations + +import uuid +from types import SimpleNamespace + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.data.engine import engines +from app.data.introspect import introspect +# 模型要在建表之前注册进 metadata：conftest 的建表只导入了 app.db.base，这个文件单独跑时没有别的模块先导入它们 +from app.db import models # noqa: F401 + +_CACHE: dict[str, dict] = {} + + +async def _schema(path: str) -> dict: + if path not in _CACHE: + probe = SimpleNamespace(id=f"probe-{uuid.uuid4().hex[:6]}", name="scenic", kind="sqlite", database=path, + host=None, port=None, username=None, password=None, options={}, readonly=True, + description="", schema_cache={}, origin="manual") + try: + _CACHE[path] = await introspect(probe) + finally: + await engines.invalidate(probe.id) + return _CACHE[path] + + +def _node(nid, ntype, label=None, **config): + return {"id": nid, "type": ntype, "data": {"label": label or nid, "config": config}} + + +def _graph(*nodes): + nodes = [_node("start", "input"), *nodes, _node("out", "output", fields=[])] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def _query(nid, source, sql, label=None): + return _node(nid, "tool", label, tool=f"db_query__{source}", args={"sql": sql}) + + +@pytest.fixture +async def world(scenic_db): + """一个景区数据源、另一个同结构的源，和几份模板；用完全部删掉。""" + from app.db.base import SessionLocal + from app.db.models import DataSource, Workflow, WorkflowVersion + + tag = uuid.uuid4().hex[:8] + src, other = f"scenic_{tag}", f"mirror_{tag}" + created: list[str] = [] + async with SessionLocal() as session: + rows = [] + for name in (src, other): + row = DataSource(name=name, kind="sqlite", database=scenic_db, readonly=True) + row.schema_cache = await _schema(scenic_db) + session.add(row) + rows.append(row) + await session.commit() + + async def template(name, versions, *, published=None, level="published"): + wf = Workflow(name=f"{name}-{tag}", graph=versions[-1], version=len(versions), + status=level if published else "draft", published_version=published) + session.add(wf) + await session.flush() + for i, g in enumerate(versions, start=1): + session.add(WorkflowVersion(workflow_id=wf.id, version=i, graph=g, + level=level if published == i else None)) + created.append(wf.id) + return wf + + direct = await template("入园日报", [_graph(_query("q", src, "SELECT COUNT(*) FROM main.VISITS v WHERE v.status = 1", + "查询入园人数"))], published=1) + agent = await template("入园分析", [_graph(_node("ask", "agent", "分析入园", prompt="分析", + tools=[f"db_query__{src}", f"db_schema__{src}"]))], + published=1, level="governed") + team = await template("协作分析", [_graph(_node("team", "supervisor", "协作", agents=[ + {"name": "analyst", "tools": [f"db_query__{src}"]}]))], published=1) + merged = await template("渠道合并", [_graph( + _query("q1", src, "SELECT id, park_id FROM visits", "入园"), + _query("q2", src, "SELECT id, channel_id FROM orders", "订单"), + _node("m", "merge", "合并入园和订单", inputs={"a": "q1", "b": "q2"}, sql="SELECT * FROM a JOIN b ON a.id = b.id"), + _node("m2", "merge", "再合并", inputs={"x": "m"}, sql="SELECT * FROM x"), + )], published=1) + # 不算的：草稿、别的源的同名表、只在未发布的新版本里用到 + await template("草稿", [_graph(_query("q", src, "SELECT * FROM visits"))]) + await template("镜像", [_graph(_query("q", other, "SELECT * FROM visits"))], published=1) + await template("后来才用", [_graph(_query("q", src, "SELECT * FROM orders")), + _graph(_query("q", src, "SELECT * FROM visits"))], published=1) + # 已发布的是旧版本：画布上后来删掉了这张表，影响面仍按已发布的那一版算 + stale = await template("改过画布", [_graph(_query("q", src, "SELECT * FROM visits")), + _graph(_query("q", src, "SELECT * FROM orders"))], published=1) + await session.commit() + ids = SimpleNamespace(source=rows[0].id, other=rows[1].id, src=src, direct=direct.id, agent=agent.id, + team=team.id, merged=merged.id, stale=stale.id) + yield ids + async with SessionLocal() as session: + for wid in created: + if (wf := await session.get(Workflow, wid)) is not None: + await session.delete(wf) + for sid in (ids.source, ids.other): + if (row := await session.get(DataSource, sid)) is not None: + await session.delete(row) + await session.commit() + + +async def _impact(source_id: str, table: str) -> dict: + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + r = await client.get(f"/api/datasources/{source_id}/catalog/{table}/impact") + assert r.status_code == 200, r.text + return r.json() + + +async def test_impact_classifies_direct_possible_and_merge(world): + out = await _impact(world.source, "visits") + assert out["table"] == "visits" + by_id = {t["workflow_id"]: t for t in out["templates"]} + assert set(by_id) == {world.direct, world.agent, world.team, world.merged, world.stale} + + direct = by_id[world.direct] + assert direct["impact"] == "direct" and direct["version"] == 1 and direct["level"] == "published" + assert [(n["node_id"], n["label"], n["impact"]) for n in direct["nodes"]] == [("q", "查询入园人数", "direct")] + + agent = by_id[world.agent] + assert agent["impact"] == "possible" and agent["level"] == "governed" + assert [(n["node_id"], n["impact"]) for n in agent["nodes"]] == [("ask", "possible")] + assert by_id[world.team]["impact"] == "possible" + assert by_id[world.team]["nodes"][0]["member"] == "analyst" + + merged = by_id[world.merged] + assert merged["impact"] == "direct" + nodes = {n["node_id"]: n for n in merged["nodes"]} + assert set(nodes) == {"q1", "m", "m2"} # 查订单的 q2 不涉及这张表 + assert nodes["m"]["impact"] == "direct" and nodes["m"]["via"] == [{"node_id": "q1", "label": "入园", "alias": "a"}] + # 合并的合并：一路追到底 + assert nodes["m2"]["impact"] == "direct" and nodes["m2"]["via"] == [{"node_id": "m", "label": "合并入园和订单", + "alias": "x"}] + # 直接引用的排在前面 + impacts = [t["impact"] for t in out["templates"]] + assert impacts == sorted(impacts, key=lambda i: i != "direct") + + +async def test_impact_of_another_table(world): + out = await _impact(world.source, "orders") + by_id = {t["workflow_id"]: t for t in out["templates"]} + # 已发布的旧版本只查了入园记录：画布上后来改成查订单，不算 + assert world.stale not in by_id + assert by_id[world.merged]["impact"] == "direct" + assert {n["node_id"] for n in by_id[world.merged]["nodes"]} == {"q2", "m", "m2"} + # 只绑了查询工具的 Agent 对每张表都是「可能涉及」 + assert by_id[world.agent]["impact"] == "possible" + assert world.direct not in by_id + + +async def test_impact_unknown_table_is_404(world): + from app.main import app + + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as client: + r = await client.get(f"/api/datasources/{world.source}/catalog/ghost_table/impact") + assert r.status_code == 404 + + +def test_graph_sql_tables_uses_the_evidence_rules(): + """发布时记目录版本和影响面共用这一份：只看调用工具节点写死的 SQL，表名按 name_key 对到表结构上。""" + from app.data.catalog_impact import graph_sql_tables + from app.engine.schema import GraphSpec + + spec = GraphSpec.model_validate(_graph( + _query("q1", "shop", "SELECT * FROM main.Orders o JOIN order_items i ON i.order_id = o.id -- FROM ghost"), + _query("q2", "shop", "WITH t AS (SELECT 1) SELECT * FROM t, customers"), + _query("q3", "other", "SELECT * FROM orders"), + _node("ask", "agent", tools=["db_query__shop"]), + )) + tables = {"shop": {"orders": {}, "order_items": {}, "customers": {}}, "other": {"orders": {}}} + assert graph_sql_tables(spec, tables) == {"shop": {"orders", "order_items", "customers"}, "other": {"orders"}} diff --git a/frontend/src/api/client.ts b/frontend/src/api/client.ts index 237d5778..572ed88e 100644 --- a/frontend/src/api/client.ts +++ b/frontend/src/api/client.ts @@ -1,5 +1,5 @@ import type { - ActivateSnapshotBody, ActivateSnapshotOut, AiPreview, CatalogDetail, CatalogDraftOut, CatalogList, CatalogNotes, + ActivateSnapshotBody, ActivateSnapshotOut, AiPreview, CatalogDetail, CatalogDraftOut, CatalogImpact, CatalogList, CatalogNotes, CatalogPatchPreview, CatalogPatchSubmit, CatalogReviewAction, CommitOut, CurrentRecipe, EditPreview, EditRequest, ImportRecord, ManifestOut, PurgeRawBody, PurgeRawOut, QuestionAnswer, Recipe, RedraftRulesOut, RemovePeriodBody, RemovePeriodOut, RevokeAcceptanceBody, RevokeAcceptanceOut, SnapshotOut, Staging, @@ -669,6 +669,9 @@ export const api = { /** 单项审阅。path：表级项写字段名（grain）；列级项 columns.<列名>.<字段>；关系 relations.<编号> */ review: (sourceId: string, table: string, body: { path: string; action: CatalogReviewAction; if_version: number }) => post(`/datasources/${encodeURIComponent(sourceId)}/catalog/${encodeURIComponent(table)}/review`, body), + /** 影响面：引用这张表的已发布、受管模板（直接引用 / 可能涉及，合并查询按输入追溯） */ + impact: (sourceId: string, table: string, opts?: RequestOptions) => + get(`/datasources/${encodeURIComponent(sourceId)}/catalog/${encodeURIComponent(table)}/impact`, opts), /** 目录修改提案的预览：对着当前目录重算改前、改后，不写库（保存遇到 409 后重新载入用） */ patchPreview: (sourceId: string, table: string, changes: CatalogPatchSubmit[]) => post( diff --git a/frontend/src/components/CatalogImpact.tsx b/frontend/src/components/CatalogImpact.tsx new file mode 100644 index 00000000..97542951 --- /dev/null +++ b/frontend/src/components/CatalogImpact.tsx @@ -0,0 +1,122 @@ +import { useCallback, useEffect, useRef, useState } from 'react' +import { Link } from 'react-router-dom' +import { GitBranch } from 'lucide-react' +import clsx from 'clsx' +import { api } from '../api/client' +import { humanizeError } from '../lib/errors' +import { CATALOG_IMPACT_TEXT as T, WORKFLOW_STATUS_LABEL } from '../lib/terms' +import type { CatalogImpact, CatalogImpactNode, CatalogImpactTemplate } from '../types' +import { Skeleton } from './ui' + +// =========================================================================== +// 影响面：哪些已发布、受管的模板引用这张表（GET /catalog/{table}/impact）。数据目录的表详情里一栏，助手的 +// 「建议更新数据目录」卡片保存之后也列一遍——改了目录，人要知道哪些正式出具的数可能跟着变。 +// +// 「直接引用」和「可能涉及」分开写：前者 SQL 里写着这张表，后者是 Agent 运行时自己写 SQL，静态看不出。 +// 合并查询写明经由哪个输入。模板名点进去是画布（定位到那个节点）。 +// =========================================================================== + +const SHOWN = 5 + +export function CatalogImpactList({ sourceId, table, refreshKey, compact = false }: { + sourceId: string + table: string + /** 变了就重新统计（表详情传目录版本：保存之后刷新） */ + refreshKey?: number | string + /** 卡片里的紧凑版：只列前几个模板，节点收成一行 */ + compact?: boolean +}) { + const [data, setData] = useState(null) + const [error, setError] = useState(null) + const [all, setAll] = useState(false) + const seq = useRef(0) + + const load = useCallback(() => { + const mine = ++seq.current + setError(null) + api.dataCatalog.impact(sourceId, table) + .then((d) => { if (mine === seq.current) setData(d) }) + .catch((e) => { if (mine === seq.current) setError(e) }) + }, [sourceId, table]) + + useEffect(() => { + setData(null) + setAll(false) + load() + }, [load, refreshKey]) + + if (error && !data) { + return ( +

+ {T.error}：{humanizeError(error).title} + +

+ ) + } + if (!data) return
+ + const list = data.templates + if (!list.length) { + return ( +

+ {compact ? T.afterSaveNone : T.none} +

+ ) + } + const shown = all || !compact ? list : list.slice(0, SHOWN) + return ( +
+ {compact &&

{T.afterSave(list.length)}

} +
    + {shown.map((t) => )} +
+ {compact && list.length > shown.length && ( + + )} +
+ ) +} + +function ImpactBadge({ impact }: { impact: 'direct' | 'possible' }) { + return ( + + {T.impact[impact]} + + ) +} + +function nodeText(n: CatalogImpactNode): string { + const who = n.member ? `「${n.label}」${T.member(n.member)}` : `「${n.label}」` + const via = n.via?.length ? `（${T.via(n.via.map((v) => `「${v.label}」`).join('、'))}）` : '' + return `${who}${via}` +} + +function TemplateRow({ t, compact }: { t: CatalogImpactTemplate; compact: boolean }) { + const first = t.nodes.find((n) => n.impact === t.impact) ?? t.nodes[0] + const href = `/studio/${encodeURIComponent(t.workflow_id)}${first ? `?focus=${encodeURIComponent(first.node_id)}` : ''}` + return ( +
  • +
    + + {t.name} + {T.version(t.version)} + {WORKFLOW_STATUS_LABEL[t.level] ?? t.level} + +
    + {!compact && ( +
      + {t.nodes.map((n) => ( +
    • + {nodeText(n)} + {T.impact[n.impact]} +
      • + ))} +
    + )} +
    • + ) +} diff --git a/frontend/src/lib/terms.ts b/frontend/src/lib/terms.ts index 40bee922..d3e14c11 100644 --- a/frontend/src/lib/terms.ts +++ b/frontend/src/lib/terms.ts @@ -2187,3 +2187,27 @@ export const CATALOG_PATCH_TEXT = { whereRelationNew: (to: string) => `新增指向 ${to} 的关联关系`, sourceOf: (source: string) => `数据源「${source}」`, } + +/** + * 影响面：哪些已发布、受管的模板引用这张表。按每个模板当前的已发布版本统计（正式运行用的是它）。 + * 「直接引用」：查询的 SQL 里写着这张表；「可能涉及」：Agent 绑定了这个数据源的查询工具，SQL 运行时才生成。 + */ +export const CATALOG_IMPACT_TEXT = { + title: '引用这张表的模板', + hint: '按每个模板当前的已发布版本统计', + none: '没有已发布或受管的模板引用这张表', + count: (n: number) => `${formatNumber(n)} 个模板`, + impact: { direct: '直接引用', possible: '可能涉及' } as Record<'direct' | 'possible', string>, + impactHint: { + direct: '查询的 SQL 中写着这张表', + possible: 'Agent 绑定了这个数据源的查询工具，SQL 在运行时生成，可能用到这张表', + } as Record<'direct' | 'possible', string>, + version: (v: number) => `v${v}`, + via: (labels: string) => `经由输入${labels}`, + member: (name: string) => `成员「${name}」`, + afterSave: (n: number) => `${formatNumber(n)} 个已发布模板引用这张表，下次正式运行时会提示数据目录有变化：`, + afterSaveNone: '没有已发布或受管的模板引用这张表', + more: (n: number) => `另有 ${formatNumber(n)} 个`, + error: '无法统计引用这张表的模板', + retry: '重试', +} diff --git a/frontend/src/pages/catalog/TableDetail.tsx b/frontend/src/pages/catalog/TableDetail.tsx index 614d5e68..0016dd7c 100644 --- a/frontend/src/pages/catalog/TableDetail.tsx +++ b/frontend/src/pages/catalog/TableDetail.tsx @@ -11,9 +11,10 @@ import { DeleteButton, EmptyState, ErrorState, Notice, Skeleton, Spinner, confir import { useLeaveGuard } from '../../lib/leave' import { formatDateTime, formatNumber } from '../../lib/format' import { - CATALOG_CARDINALITY_LABEL, CATALOG_KIND_HINT, CATALOG_KIND_LABEL, CATALOG_TABLE_FIELD_HINT, CATALOG_TABLE_FIELD_LABEL, - CATALOG_TEXT as CT, + CATALOG_CARDINALITY_LABEL, CATALOG_IMPACT_TEXT, CATALOG_KIND_HINT, CATALOG_KIND_LABEL, CATALOG_TABLE_FIELD_HINT, + CATALOG_TABLE_FIELD_LABEL, CATALOG_TEXT as CT, } from '../../lib/terms' +import { CatalogImpactList } from '../../components/CatalogImpact' import { CountBadges, ItemMark, StatusChip, StatusLegend, SystemNotesNotice } from './parts' import type { ReviewTarget } from './parts' import { ColumnTable, Value } from './ColumnTable' @@ -382,6 +383,14 @@ export function TableDetail({ sourceId, table, rows, drafted, prev, next, onOpen onChange={setRelations} /> : } + +
    +
    +

    {CATALOG_IMPACT_TEXT.title}

    + {CATALOG_IMPACT_TEXT.hint} +
    + +
    diff --git a/frontend/src/run/CatalogPatchCard.tsx b/frontend/src/run/CatalogPatchCard.tsx index 867592a7..8fac3074 100644 --- a/frontend/src/run/CatalogPatchCard.tsx +++ b/frontend/src/run/CatalogPatchCard.tsx @@ -4,6 +4,7 @@ import { BookMarked, Check, ExternalLink, RotateCw } from 'lucide-react' import clsx from 'clsx' import { ApiError, api } from '../api/client' import { Notice, Spinner, toast } from '../components/ui' +import { CatalogImpactList } from '../components/CatalogImpact' import { CATALOG_PATCH_TEXT as T, CATALOG_STATUS_LABEL } from '../lib/terms' import type { CatalogPatchChange } from '../types' import { @@ -15,7 +16,7 @@ import { // 改后和理由，人点「保存到数据目录」才写入——服务端从不自动写。 // // 保存带着提案对照的版本：别人在这之后改过这张表就 409，卡片写明「这张表刚被修改过」，「重新载入」按最新的 -// 目录重算改前和改后（预览接口，只读），人看过再保存。保存成功后给到数据目录那张表的入口。 +// 目录重算改前和改后（预览接口，只读），人看过再保存。保存成功后列出受影响的模板（影响面），给到数据目录那张表的入口。 // // 卡片的结局（已保存、已忽略、重新载入后的内容）记在模块级的表里：画布右栏在对话层和运行层之间切换会卸载 // 卡片，回来时不能又变回「未保存」，再点一次只会撞上 409。 @@ -151,6 +152,7 @@ export function CatalogPatchCard({ patch, turnId, dense = false }: { {T.open}

    + {saved && } ) : ( diff --git a/frontend/src/types.ts b/frontend/src/types.ts index d41cc963..f1eb560f 100644 --- a/frontend/src/types.ts +++ b/frontend/src/types.ts @@ -2739,3 +2739,34 @@ export interface CatalogPatchPreview { /** 不合法的项的说明（目录刚被改过、列已删除……） */ problems: string[] } + +/** 影响面里模板的一个节点（服务端 catalog_impact.template_refs） */ +export interface CatalogImpactNode { + node_id: string + label: string + type: string + /** direct：SQL 里写着这张表；possible：Agent 运行时自己写 SQL，可能用到 */ + impact: 'direct' | 'possible' + /** 合并查询：经由哪些输入（别名 → 节点） */ + via?: { node_id: string; label: string; alias: string }[] + /** 协作节点：绑定了查询工具的成员 */ + member?: string +} + +/** 引用这张表的一个已发布或受管模板（按它当前的已发布版本算） */ +export interface CatalogImpactTemplate { + workflow_id: string + name: string + /** 当前的已发布版本 */ + version: number + level: 'published' | 'governed' + impact: 'direct' | 'possible' + nodes: CatalogImpactNode[] +} + +/** GET /datasources/{id}/catalog/{table}/impact */ +export interface CatalogImpact { + table: string + /** 直接引用的在前 */ + templates: CatalogImpactTemplate[] +} diff --git a/scripts/check-catalog.mjs b/scripts/check-catalog.mjs index 32431142..697eec9b 100644 --- a/scripts/check-catalog.mjs +++ b/scripts/check-catalog.mjs @@ -395,6 +395,11 @@ function catalogHandlers(state, replies = {}) { json({ tables: state.tables[parts(key)[2]].filter((t) => !t.missing).map((t) => t.name), summary: '', synced_at: T0 })(route)], [/^GET \/datasources\/check-cat-[a-z]+\/catalog$/, (route, { key }) => json(listOf(state, parts(key)[2]))(route)], [/^POST \/datasources\/check-cat-[a-z]+\/catalog\/draft$/, reply('draft', draftReply(state))], + // 影响面：引用这张表的已发布、受管模板（只读）。没配的表答空 + [/^GET \/datasources\/check-cat-[a-z]+\/catalog\/[^/]+\/impact$/, reply('impact', (route, { key }) => { + const name = parts(key)[4] + return json({ table: name, templates: state.impact?.[name] ?? [] })(route) + })], [/^GET \/datasources\/check-cat-[a-z]+\/catalog\/[^/]+$/, reply('get', (route, { key }) => { const { src, t } = findTable(state, key) return t ? json(detailOf(src, t))(route) : json({ detail: `数据源中没有表 ${parts(key)[4]}，可能已被删除或尚未探查结构` }, 404)(route) @@ -775,6 +780,49 @@ await section('数据目录 · 表详情与单项审阅', async () => { await close() }) +await section('数据目录 · 引用这张表的模板（影响面）', async () => { + const state = freshState() + state.impact = { + visits: [ + { workflow_id: 'wf-daily', name: '入园日报', version: 4, level: 'governed', impact: 'direct', nodes: [ + { node_id: 'q', label: '查询入园人数', type: 'tool', impact: 'direct' }, + { node_id: 'm', label: '合并入园和订单', type: 'merge', impact: 'direct', via: [{ node_id: 'q', label: '查询入园人数', alias: 'v' }] }, + ] }, + { workflow_id: 'wf-ask', name: '客流分析', version: 2, level: 'published', impact: 'possible', nodes: [ + { node_id: 'ask', label: '分析客流', type: 'agent', impact: 'possible' }, + ] }, + ], + } + const { page, sent, errors, close } = await open(`/data/catalog/${S1}`, { handlers: catalogHandlers(state) }) + await openDetail(page, S1, 'visits') + const sec = page.locator('[data-catalog-section="impact"]') + await sec.locator('[data-catalog-impact="2"]').waitFor({ timeout: 3000 }) + const text = await sec.innerText() + check('表详情有「引用这张表的模板」一栏，写明按已发布版本统计', text.includes('引用这张表的模板') && text.includes('按每个模板当前的已发布版本统计')) + const daily = sec.locator('[data-impact-template="wf-daily"]') + check('直接引用：模板名、版本、受管、「直接引用」', ['入园日报', 'v4', '受管', '直接引用'].every((x) => text.includes(x)) + && await daily.getAttribute('data-impact-level') === 'direct') + check('合并查询写明经由哪个输入', (await daily.locator('[data-impact-node="m"]').innerText()).includes('经由输入「查询入园人数」')) + check('Agent 写「可能涉及」，悬停说明原因', (await sec.locator('[data-impact-template="wf-ask"] [data-impact="possible"]').getAttribute('title')).includes('运行时生成')) + check('模板名点进去是画布，定位到那个节点', (await daily.locator('a').first().getAttribute('href')) === '/studio/wf-daily?focus=q') + // 开发模式下 StrictMode 会把挂载时的取数做两遍，只查有没有取、没有写 + check('统计影响面只读', writes(sent, /impact/).length === 0 && count(sent, /\/visits\/impact$/) >= 1, + `${count(sent, /\/visits\/impact$/)} 次`) + // 没有模板引用的表：直说没有 + await openDetail(page, S1, 'orders') + check('没有模板引用：直说没有', (await page.locator('[data-catalog-section="impact"] [data-catalog-impact="0"]').innerText()) + .includes('没有已发布或受管的模板引用这张表')) + // 表详情保存之后（版本变了）重新统计 + const before = count(sent, /\/orders\/impact$/) + await page.locator('[data-catalog-edit]').click() + await page.locator('[data-input="grain"]').fill('一笔订单一行（检查）') + await page.locator('[data-catalog-save]').click() + await until(async () => count(sent, /\/orders\/impact$/) > before, 3000) + check('保存之后重新统计影响面', count(sent, /\/orders\/impact$/) > before, `${before} → ${count(sent, /\/orders\/impact$/)}`) + check('没有运行时报错', errors.length === 0, errors.join(' | ')) + await close() +}) + await section('数据目录 · 批量确认', async () => { const state = freshState() const { page, sent, errors, natives, close } = await open(`/data/catalog/${S1}`, { handlers: catalogHandlers(state) }) diff --git a/scripts/check-stream.mjs b/scripts/check-stream.mjs index 530dbee9..4af14dc9 100644 --- a/scripts/check-stream.mjs +++ b/scripts/check-stream.mjs @@ -795,7 +795,12 @@ await section('建议更新数据目录：卡片的保存、忽略和 409（cata return route.fulfill({ json: detail(5) }) }) await c.page.route('**/api/datasources/src-scenic/catalog/visits/impact', (route) => - route.fulfill({ json: { table: 'visits', templates: [] } })) + route.fulfill({ json: { table: 'visits', templates: [ + { workflow_id: 'wf-daily', name: '入园日报', version: 4, level: 'governed', impact: 'direct', + nodes: [{ node_id: 'q', label: '查询入园人数', type: 'tool', impact: 'direct' }] }, + { workflow_id: 'wf-ask', name: '客流分析', version: 2, level: 'published', impact: 'possible', + nodes: [{ node_id: 'ask', label: '分析客流', type: 'agent', impact: 'possible' }] }, + ] } })) const card = c.page.locator('[data-catalog-patch="visits"]') check('卡片在：标题写「建议更新数据目录」', await card.count() === 1 && (await card.innerText()).includes('建议更新数据目录')) const codes = card.locator('[data-patch-change="columns.status.codes"]') @@ -830,6 +835,10 @@ await section('建议更新数据目录：卡片的保存、忽略和 409（cata check('保存成功：写明第几版，给到数据目录的入口', (await card.locator('[data-patch-done]').innerText()).includes('已保存到数据目录（第 5 版）') && (await card.locator('[data-patch-open]').getAttribute('href')) === '/data/catalog/src-scenic/visits') check('保存后不再有保存按钮', await card.locator('[data-patch-save]').count() === 0) + await card.locator('[data-catalog-impact="2"]').waitFor({ timeout: 3000 }) + const impact = await card.locator('[data-catalog-impact]').innerText() + check('保存后列出受影响的模板：直接引用、可能涉及', impact.includes('2 个已发布模板引用这张表') && impact.includes('入园日报') + && impact.includes('直接引用') && impact.includes('可能涉及'), impact.replace(/\s+/g, ' ')) check('没有运行时报错', c.errors.length === 0, c.errors.join(' | ')) await c.page.close()