From 5eb1c8ff005c72ba12733a5ea81ae5a5e292ec01 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 3 Nov 2026 19:34:28 +0800 Subject: [PATCH] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E5=8F=A3=E5=BE=84=E5=8F=AF?= =?UTF-8?q?=E8=B0=83=E7=94=A8=20descriptor=EF=BC=9A=E5=85=A5=E5=8F=A3=20Sc?= =?UTF-8?q?hema=E3=80=81=E4=BE=9D=E8=B5=96=E4=B8=8E=E5=87=BA=E5=85=B7?= =?UTF-8?q?=E7=AD=96=E7=95=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/ledger.py | 27 +++++ backend/app/ledger/descriptor.py | 128 ++++++++++++++++++++++++ backend/tests/test_ledger_descriptor.py | 107 ++++++++++++++++++++ 3 files changed, 262 insertions(+) create mode 100644 backend/app/ledger/descriptor.py create mode 100644 backend/tests/test_ledger_descriptor.py diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index c938d611..7cba10fe 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -16,6 +16,8 @@ （app/ledger/certify.py）。 - GET /api/ledger/recall/{workflow_id}/{version}、POST …/continue（router，搭建者，第 4 波）：口径升到这一版之后的召回批次—— 每组的结论、「另有 M 组未重算」、手动接着算（app/ledger/recall.py）。版本用量（workflow-versions）另加问数回答、导出、召回。 +- GET /api/ledger/calibers、GET /api/ledger/calibers/{workflow_id}/{version}/{node_id}（router，搭建者）：口径可调用描述 + （app/ledger/descriptor.py）：入口字段的 JSON Schema、依赖、能不能组合和出具；不返回 SQL 和参数值。 这几个新接口都不放给访问密钥（auth/keys.KEY_ROUTES 不登记）。 """ from __future__ import annotations @@ -256,3 +258,28 @@ async def certify_dismiss(payload: DismissIn, session: AsyncSession = Depends(ge await certify.dismiss(session, payload.item_id, kind=payload.kind, note=payload.note, principal=current_principal()) return {"ok": True, "item_id": payload.item_id} + + +@router.get("/calibers") +async def calibers(workflow_id: str | None = None, session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """口径可调用描述（app/ledger/descriptor.py）：每个工作流最新「已发布且未退役」版本里的源头口径卡一张一条，含入口字段的 + JSON Schema、指标、依赖的表和发布时钉住的数据目录/结构版本、能不能组合、出具策略。workflow_id 可只列一个工作流。 + + 不返回取数 SQL、不返回参数值（SQL 里可能写着遮罩值）。issuance 只描述这一版的策略（级别、有没有人工节点或写外部系统）， + 不代表这次运行一定能出具：全局自动出具开关、每日上限这类运行时条件不在这里判。""" + from app.ledger.descriptor import list_calibers as list_descriptors + + return {"calibers": await list_descriptors(session, workflow_id=workflow_id)} + + +@router.get("/calibers/{workflow_id}/{version}/{node_id}") +async def caliber_descriptor(workflow_id: str, version: int, node_id: str, + session: AsyncSession = Depends(get_session)) -> dict[str, Any]: + """一张源头口径卡的描述（形状同列表里的一条）。工作流不存在、这一版没发布过或已退役、节点不是源头口径卡（转抄卡也算） + 回 404。同样不返回 SQL、参数值；issuance 只描述版本策略，不代表这次运行一定能出具。""" + from app.ledger.descriptor import caliber_descriptor as descriptor_of + + made = await descriptor_of(session, workflow_id, version, node_id) + if made is None: + raise HTTPException(404, "没有这份口径：工作流不存在、这一版没发布过或已退役，或者节点不是源头口径卡") + return made diff --git a/backend/app/ledger/descriptor.py b/backend/app/ledger/descriptor.py new file mode 100644 index 00000000..e627b541 --- /dev/null +++ b/backend/app/ledger/descriptor.py @@ -0,0 +1,128 @@ +"""口径可调用描述（descriptor）：给目录页、应用表单、外部集成一份机器可读的稳定描述（next-round 02 的后续）。 + +口径已经能按固定版本实例化（look/calibers 的查看结果、ledger/compose 的 use_caliber），但入口字段、依赖、能不能出具这些 +契约只写在给模型看的文字里。这里把它们写成结构化的数据： + +- 只列每个工作流最新「已发布且未退役」版本里的**源头口径卡**（metrics 节点、caliber_from 为空），和 look/calibers、 + ledger/derive 的认法一致：转抄卡不列，源头才能复用； +- 入口字段用 ledger/derive.entry_fields，按 apps/forms.input_schema 转成 JSON Schema（和表单、use_caliber 校验同一个）； +- 依赖只从图里的取数节点认数据源和表名，**不返回 SQL、不返回参数值**（SQL 里可能写着遮罩值，组合时由服务端照抄）； +- 出具只描述这一版的策略：级别、图里有没有人工节点或写外部系统的节点。全局的自动出具开关、每日上限是运行时条件，这里 + 不查——descriptor 说的是「这一版允许什么」，不代表这次运行一定能出具。 +""" +from __future__ import annotations + +from typing import Any + +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.apps import versions as app_versions +from app.apps.backfill import graph_side_effect +from app.apps.forms import input_schema +from app.apps.tree import node_map +from app.db.models import Workflow +from app.engine.evidence import sql_tables +from app.ledger.derive import entry_fields +from app.look.calibers import card_inputs +from app.tools.datasource import QUERY_PREFIX + +#: 指标只写口径卡配置里真有的这几个键 +METRIC_KEYS = ("id", "name", "unit", "decimals", "format") + + +async def list_calibers(session: AsyncSession, *, workflow_id: str | None = None) -> list[dict[str, Any]]: + """每个工作流最新「已发布且未退役」版本里的源头口径卡，一张卡一条描述。给了 workflow_id 只列那个工作流。 + + 同一版图里的 graph_side_effect 按 (workflow_id, version) 只算一次：它要读子图、跑发布检查，比其余字段慢得多。""" + query = select(Workflow).order_by(Workflow.name, Workflow.id) + if workflow_id: + query = query.where(Workflow.id == workflow_id) + side_cache: dict[tuple[str, int], str | None] = {} + out: list[dict[str, Any]] = [] + for workflow in (await session.execute(query)).scalars().all(): + row = await app_versions.latest_published(session, workflow.id) + if row is None: + continue + for node_id, node in node_map(row.graph or {}).items(): + if node["type"] != "metrics" or node["config"].get("caliber_from") not in (None, "", {}): + continue + made = await _describe(session, workflow, row, node_id, side_cache) + if made is not None: + out.append(made) + return out + + +async def caliber_descriptor(session: AsyncSession, workflow_id: str, version: int, + node_id: str) -> dict[str, Any] | None: + """一个工作流版本里一张源头口径卡的描述；工作流不存在、版本没发布过或已退役、节点不是源头口径卡时返回 None。""" + workflow = await session.get(Workflow, str(workflow_id or "")) + if workflow is None: + return None + row = await app_versions.version_row(session, workflow.id, int(version)) + if row is None or not app_versions.is_published(row, current=workflow.published_version): + return None + return await _describe(session, workflow, row, str(node_id), {}) + + +async def _describe(session: AsyncSession, workflow: Workflow, row: Any, node_id: str, + side_cache: dict[tuple[str, int], str | None]) -> dict[str, Any] | None: + """一张卡的描述；节点不是源头口径卡时返回 None。""" + graph = row.graph or {} + nodes = node_map(graph) + card = nodes.get(node_id) + if card is None or card["type"] != "metrics" or card["config"].get("caliber_from") not in (None, "", {}): + return None + cfg = card["config"] + level = str(row.level or "") + _, reason = card_inputs(graph, node_id) + key = (workflow.id, int(row.version)) + if key not in side_cache: + side_cache[key] = await graph_side_effect(session, graph, level=level or None) + side_effect = side_cache[key] + sources, tables = _fetches(nodes) + reasons: list[str] = [] + if level != "governed": + reasons.append("这一版是已发布级别，只支持一键出具") + if side_effect is not None: + reasons.append(f"这一版{side_effect}，不能替人重跑") + return { + "workflow_id": workflow.id, + "workflow_name": workflow.name, + "version": int(row.version), + "level": level, + "graph_hash": row.graph_hash, + "card": {"node_id": node_id, "title": card["title"], "caliber": cfg.get("caliber"), + "caliber_version": cfg.get("caliber_version")}, + "metrics": [{k: m[k] for k in METRIC_KEYS if k in m} + for m in cfg.get("metrics") or [] + if isinstance(m, dict) and m.get("id")], + "entry_schema": input_schema(list(entry_fields(graph))), + "composable": reason is None, + "compose_reason": reason, + "dependencies": {"sources": sorted(sources), "tables": sorted(tables), + "catalog_versions": row.catalog_versions, "schema_versions": row.schema_versions}, + "issuance": {"level": level, "auto_issue": level == "governed" and side_effect is None, + "requires_independent_proof": True, "side_effects": side_effect, "reasons": reasons}, + } + + +def _fetches(nodes: dict[str, dict[str, Any]]) -> tuple[set[str], set[str]]: + """图里调用数据库查询工具、而且写了 SQL 的节点：数据源名、SQL 里 FROM / JOIN 的表名（去库名前缀）。只认表名，不读 SQL。""" + sources: set[str] = set() + tables: set[str] = set() + for node in nodes.values(): + if node["type"] != "tool": + continue + tool = str(node["config"].get("tool") or "") + args = node["config"].get("args") + sql = args.get("sql") if isinstance(args, dict) else None + # 工具名带模板的不是固定的数据源，认不出「哪一版库」 + if not tool.startswith(QUERY_PREFIX) or "{{" in tool or not isinstance(sql, str) or not sql.strip(): + continue + sources.add(tool[len(QUERY_PREFIX):]) + tables.update(t.rsplit(".", 1)[-1] for t in sql_tables(sql)) + return sources, tables + + +__all__ = ["caliber_descriptor", "list_calibers"] diff --git a/backend/tests/test_ledger_descriptor.py b/backend/tests/test_ledger_descriptor.py new file mode 100644 index 00000000..8f988b60 --- /dev/null +++ b/backend/tests/test_ledger_descriptor.py @@ -0,0 +1,107 @@ +"""口径可调用描述（app/ledger/descriptor.py、GET /api/ledger/calibers）：目录页、应用表单、外部集成用的机器可读契约。 + +- 只列最新「已发布且未退役」版本里的源头口径卡：入口字段成 JSON Schema、指标照配置、依赖只报数据源和表名； +- 能不能组合按 look.calibers.card_inputs（和对账、use_caliber 同一个认法）； +- 出具只描述版本策略：受管级别、图里没有人工节点或写外部系统的节点才自动出具，原因照写； +- 转抄别处口径卡的卡不列；响应里没有 SQL、没有参数值。 +数据是虚构店铺库。 +""" +from __future__ import annotations + +import json + +import pytest +from httpx import ASGITransport, AsyncClient + +from app.main import app +from tests import ledger_samples as S + + +@pytest.fixture(autouse=True) +async def shop(tmp_path): + await S.fresh_source(tmp_path / "shop.db") + yield + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def descriptors(client: AsyncClient, wf: str) -> list[dict]: + body = (await client.get("/api/ledger/calibers", params={"workflow_id": wf})).json() + assert set(body) == {"calibers"}, body + return body["calibers"] + + +def with_human() -> dict: + graph = S.caliber_graph() + graph["nodes"].append(S.node("approve", "human", assignee="运营")) + return graph + + +def with_copy() -> dict: + graph = S.caliber_graph() + graph["nodes"].append(S.node( + "copy", "metrics", caliber="抄来的周销售", caliber_version="v1", + caliber_from={"workflow_id": "elsewhere", "workflow_version": 1, "node_id": "card"}, + metrics=[{"id": "gmv", "name": "销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"}])) + return graph + + +async def test_a_source_caliber_is_described_as_a_callable_contract(client): + wf = await S.publish() + [made] = await descriptors(client, wf) + assert (made["workflow_id"], made["workflow_name"], made["version"]) == (wf, "周销售", 1) + assert made["level"] == "governed" and made["graph_hash"] + assert made["card"] == {"node_id": "card", "title": "card", "caliber": "周销售", "caliber_version": "v1"} + assert made["metrics"] == [{"id": "gmv", "name": "销售额", "unit": "元", "decimals": 2, "format": "thousands"}] + schema = made["entry_schema"] + assert schema["format"] == "agentlab.form/1" and schema["type"] == "object" + assert schema["required"] == ["start", "end"] + assert schema["properties"]["store"]["default"] == "all" + assert made["composable"] is True and made["compose_reason"] is None + assert made["dependencies"]["sources"] == [S.CURRENT["source"]] + assert made["dependencies"]["tables"] == ["orders"] + assert made["dependencies"]["catalog_versions"] is None and made["dependencies"]["schema_versions"] is None + issuance = made["issuance"] + assert issuance["level"] == "governed" and issuance["auto_issue"] is True + assert issuance["requires_independent_proof"] is True + assert issuance["side_effects"] is None and issuance["reasons"] == [] + + # 单条接口交同一份描述；转抄卡、没发布过的版本、不存在的节点都是 404 + single = (await client.get(f"/api/ledger/calibers/{wf}/1/card")).json() + assert single == made + for path in (f"/api/ledger/calibers/{wf}/1/nope", f"/api/ledger/calibers/{wf}/9/card", + "/api/ledger/calibers/nope/1/card"): + assert (await client.get(path)).status_code == 404, path + + # 契约里没有取数 SQL、没有参数值 + text = json.dumps((await client.get("/api/ledger/calibers", params={"workflow_id": wf})).json()) + assert "SELECT" not in text and '"sql"' not in text + assert "SUM(" not in text and "paid" not in text + + +async def test_a_human_node_turns_auto_issue_off_with_a_reason(client): + wf = await S.publish(name="审批周销售", graph=with_human(), level="governed") + [made] = await descriptors(client, wf) + assert made["level"] == "governed" + assert made["issuance"]["auto_issue"] is False + assert "approve" in made["issuance"]["side_effects"] + assert any("人工节点" in r for r in made["issuance"]["reasons"]) + assert made["issuance"]["requires_independent_proof"] is True + + +async def test_a_published_level_version_only_supports_one_click_issuance(client): + wf = await S.publish(name="已发布周销售", graph=S.caliber_graph(), level="published") + [made] = await descriptors(client, wf) + assert made["level"] == "published" and made["issuance"]["auto_issue"] is False + assert any("已发布级别" in r and "一键出具" in r for r in made["issuance"]["reasons"]) + + +async def test_a_card_that_copies_another_caliber_is_not_listed(client): + wf = await S.publish(graph=with_copy()) + listed = await descriptors(client, wf) + assert [d["card"]["node_id"] for d in listed] == ["card"] + assert (await client.get(f"/api/ledger/calibers/{wf}/1/copy")).status_code == 404