From a43224fd9aacc5ae4d18dee2f394c0a691b2e7bf Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Fri, 16 Oct 2026 19:46:03 +0800 Subject: [PATCH] =?UTF-8?q?=E8=B4=A6=E6=9C=AC=E6=8E=A5=E5=8F=A3=E8=A1=A5?= =?UTF-8?q?=E5=87=A0=E6=A0=B7=E7=95=8C=E9=9D=A2=E8=A6=81=E7=94=BB=E7=9A=84?= =?UTF-8?q?=E4=B8=9C=E8=A5=BF=EF=BC=9A=E5=AF=B9=E8=B4=A6=E7=9A=84=E5=85=A5?= =?UTF-8?q?=E5=8F=A3=E5=AD=97=E6=AE=B5=E3=80=81=E9=85=8D=E5=88=97=E9=80=89?= =?UTF-8?q?=E9=A1=B9=E3=80=81=E5=AF=B9=E8=B4=A6=E5=8D=A1=E7=9A=84=E5=88=86?= =?UTF-8?q?=E7=BB=84=EF=BC=8C=E5=88=9A=E8=B7=91=E5=AE=8C=E6=9C=AA=E5=B0=81?= =?UTF-8?q?=E5=AD=98=E6=97=B6=E7=AD=94=20pending?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 账本前端（第 4 波 W4-c）照「界面层不放逻辑」只照服务端给的画，这几样以前接口不给，界面只能自己猜： - 「看差多少」那一项（reconcile.actions_of）带上口径的入口字段 fields（名字、必填、有没有缺省值，比对工件里记的那一份）： 核对参数的表单照它画，缺的参数也写得出是哪一项 - 认不出列怎么配（409 ledger_reconcile_needs_pairs）另带 options：这次结果里是数、没遮罩的列，口径卡上的指标（id、名字、 单位）；ReconcileRefused 加 extra，api/ledger 的对账接口把它并进返回体（CodedHTTPException 的 extra，改一行） - 对过的账（reconciled_view）带 card 和 groups：每一组对照的这次 / 口径 / 差值 / 相对差各是哪个指标、这一组怎么称呼 （从对账运行的图里读，不按命名规则拼），界面按 /evidence/metrics/{card}/{指标} 取值、三个数各自点得开 - GET /api/runs/{id}/ledger：runner 先发 run.finished、落库，再封存、再排比对，页面收到 run.finished 立刻来取时 还没封存。成功结束、要比对的运行这时答 pending（不另排按需比对），不再说成「运行还没有封存，不比对」 - 测试：对账的三样（单行、多行），刚跑完未封存答 pending、不排按需比对，失败的照旧 ineligible --- backend/app/api/ledger.py | 2 +- backend/app/ledger/reconcile.py | 62 +++++++++++++++++++++++--- backend/app/ledger/view.py | 5 +++ backend/tests/test_ledger_api.py | 24 ++++++++++ backend/tests/test_ledger_reconcile.py | 32 +++++++++++++ 5 files changed, 117 insertions(+), 8 deletions(-) diff --git a/backend/app/api/ledger.py b/backend/app/api/ledger.py index 5c8baa41..05ce37a3 100644 --- a/backend/app/api/ledger.py +++ b/backend/app/api/ledger.py @@ -159,7 +159,7 @@ async def reconcile_answer(run_id: str, payload: ReconcileIn) -> dict[str, Any]: query=payload.query, workflow_id=payload.workflow_id, version=payload.version, inputs=payload.inputs, confirm=payload.confirm, pairs=payload.pairs), principal=current_principal()) except ledger_reconcile.ReconcileRefused as e: - raise CodedHTTPException(e.status_code, e.message, e.code) from e + raise CodedHTTPException(e.status_code, e.message, e.code, extra=e.extra) from e return {"run_id": run.id, "reconciled": ledger_reconcile.reconciled_view(run)} diff --git a/backend/app/ledger/reconcile.py b/backend/app/ledger/reconcile.py index 89cf56de..dc997743 100644 --- a/backend/app/ledger/reconcile.py +++ b/backend/app/ledger/reconcile.py @@ -77,11 +77,16 @@ ROWS_CALIBER_COLUMNS = "认不出口径取数的结果列，按位置对不上" class ReconcileRefused(Exception): - def __init__(self, status_code: int, message: str, code: str = REFUSED) -> None: + """不能对账。extra 是给界面的几样（并进 409 的返回体）：认不出列怎么配时的 options（可选的列、口径卡上的指标），界面 + 照它画「选要对照的指标」那一步，不自己去猜有哪些列。""" + + def __init__(self, status_code: int, message: str, code: str = REFUSED, *, + extra: dict[str, Any] | None = None) -> None: super().__init__(message) self.status_code = status_code self.message = message self.code = code + self.extra = dict(extra or {}) # ========================================================================== @@ -106,7 +111,8 @@ def _needs_check(cand: dict[str, Any], fields: list[dict[str, Any]], hide: Any) def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None) -> list[dict[str, Any]]: """比对工件里每条有相关口径的查询、每份相关口径一项「看差多少」：{kind: reconcile, query, workflow_id, version, node_id, - needs_check, check_reason?, binding（遮过的）}。界面据此摆按钮、要不要先核对参数。""" + needs_check, check_reason?, binding（遮过的）, fields}。界面据此摆按钮、要不要先核对参数；fields 是那一版口径的入口字段 + （名字、必填、有没有缺省值，比对工件里记的那一份），核对参数的表单照它画（缺的参数也写得出是哪一项）。""" from app.ledger.labels import LEVEL_TEXT, binding_view hide = hide or (lambda _value: False) @@ -124,7 +130,10 @@ def actions_of(content: dict[str, Any], names: dict[str, str], hide: Any = None) "node_id": cand.get("node_id"), "workflow_name": names.get(str(cand.get("workflow_id") or "")), "level_text": LEVEL_TEXT.get(str(cand.get("level") or ""), ""), "needs_check": why is not None, "binding": binding_view(cand.get("binding"), hide), - "fallback": bool((cand.get("diff") or {}).get("fallback"))} + "fallback": bool((cand.get("diff") or {}).get("fallback")), + "fields": [{"name": str(f.get("name")), "required": bool(f.get("required")), + "has_default": bool(f.get("has_default"))} + for f in fields.get(cand.get("caliber")) or [] if f.get("name")]} if why is not None: item["check_reason"] = why out.append(item) @@ -141,11 +150,45 @@ async def reconciled_runs(session: Any, run_id: str) -> list[Run]: return list(rows) +#: 对账卡上一组对照的四个指标：id 的前缀（tail_nodes 的 _four）和名字的后缀 +_FOUR = (("ours", "这次"), ("caliber", "口径"), ("diff", "差值"), ("rel", "相对差")) +_PREFIX = {"ours": "ours_", "caliber": "cal_", "diff": "diff_", "rel": "rel_"} + + +def reconcile_groups(graph: Any) -> list[dict[str, str]]: + """对账运行的图里，对账卡上的每一组对照：{label, ours, caliber, diff, rel}（后四个是指标 id）。按卡上的先后；名字 + 是「<这一组的说法> · 这次」那样写的，label 去掉后缀。认不出的（不是派生的对账图）为空。""" + from app.ledger.derive import RECONCILE_CARD + + nodes = (graph or {}).get("nodes") if isinstance(graph, dict) else None + card = next((n for n in nodes or [] if isinstance(n, dict) and n.get("id") == RECONCILE_CARD + and n.get("type") == "metrics"), None) + metrics = (((card or {}).get("data") or {}).get("config") or {}).get("metrics") or [] + by_id = {str(m.get("id")): m for m in metrics if isinstance(m, dict) and m.get("id")} + out: list[dict[str, str]] = [] + for mid, metric in by_id.items(): + if not mid.startswith(_PREFIX["ours"]): + continue + n = mid[len(_PREFIX["ours"]):] + ids = {role: f"{_PREFIX[role]}{n}" for role, _ in _FOUR} + if not all(i in by_id for i in ids.values()): + continue + name = str(metric.get("name") or "") + label = name[:-len(" · 这次")] if name.endswith(" · 这次") else name + out.append({"label": label, **ids}) + return out + + def reconciled_view(run: Run) -> dict[str, Any]: + """GET 账本 reconciled 里的一项：哪一次对账运行、状态、对的哪条查询和哪份口径、参数从哪来；card 和 groups 是对账卡和 + 每一组对照的四个指标（界面按 /evidence/metrics/{card}/{指标} 取值、点开出处，不按命名规则自己拼）。""" + from app.ledger.derive import RECONCILE_CARD + origin = run.origin if isinstance(run.origin, dict) else {} return {"run_id": run.id, "status": run.status, "query": origin.get("query"), "caliber": origin.get("caliber"), "workflow_id": origin.get("workflow_id"), "version": origin.get("version"), "inputs_from": origin.get("inputs_from"), "rows": origin.get("rows"), + "card": RECONCILE_CARD, "groups": reconcile_groups(run.graph), "created_at": run.created_at.isoformat() if run.created_at else None} @@ -294,11 +337,16 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str by_id = {str(m.get("id")): (i, m) for i, m in enumerate(metrics)} multi = len(ours_rows) > 1 + # 要人选对照时给界面的选项：这次结果里是数、没遮罩的列，口径卡上的指标（id、名字、单位） + numeric = [c for i, c in enumerate(ours_columns) if c.lower() not in hidden and _numeric(ours_rows, i)] + options = {"options": {"columns": numeric, + "metrics": [{"id": str(m.get("id")), "name": str(m.get("name") or m.get("id")), + "unit": str(m.get("unit") or "")} for m in metrics if m.get("id")]}} if multi: return _plan_rows(ours_columns, ours_rows, hidden, caliber_columns, referenced, check_column) if chosen is not None: if not isinstance(chosen, list) or not chosen or not all(isinstance(p, dict) for p in chosen): - raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS) + raise ReconcileRefused(422, PAIRS_SHAPE, NEEDS_PAIRS, extra=options) pairs = [] for item in chosen: ours, metric = str(item.get("ours") or ""), str(item.get("metric") or "") @@ -310,7 +358,7 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str unit=str(m.get("unit") or ""), decimals=m.get("decimals"))) return Plan(pairs=pairs) if fallback: - raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS) + raise ReconcileRefused(409, PAIRS_FALLBACK, NEEDS_PAIRS, extra=options) pairs = [] if caliber_columns is not None and len(caliber_columns) == len(ours_columns): for index, m in enumerate(metrics): @@ -323,7 +371,7 @@ def plan_pairs(*, ours_columns: list[str], ours_rows: list[Any], hidden: set[str pairs.append(Pair(ours=ours, metric=str(m.get("id")), index=index, name=str(m.get("name") or m.get("id")), unit=str(m.get("unit") or ""), decimals=m.get("decimals"))) if not pairs: - raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS) + raise ReconcileRefused(409, PAIRS_NONE, NEEDS_PAIRS, extra=options) return Plan(pairs=pairs) @@ -521,4 +569,4 @@ def build(pinned: Any, *, sql: str, tool: str, plan: Plan, inputs: dict[str, Any __all__ = ["CHECK_INPUTS", "MAX_COLUMNS", "MAX_ROWS", "NEEDS_PAIRS", "ORIGIN", "REFUSED", "Pair", "Plan", "ReconcileRefused", "Request", "actions_of", "align_sql", "build", "plan_pairs", "reconcile", - "reconciled_runs", "reconciled_view", "tail_nodes"] + "reconcile_groups", "reconciled_runs", "reconciled_view", "tail_nodes"] diff --git a/backend/app/ledger/view.py b/backend/app/ledger/view.py index d1c8f356..81492c2b 100644 --- a/backend/app/ledger/view.py +++ b/backend/app/ledger/view.py @@ -116,6 +116,11 @@ async def run_view(run_id: str) -> dict[str, Any]: recorded = await store.latest(run_id) if recorded is None: why = ledger_match.ineligible(run) + if why == ledger_match.NOT_SEALED and run.status == "succeeded" \ + and ledger_match.wanted(run.run_class, run.origin_kind): + # runner 先发 run.finished、落库，再封存、再排上比对（engine/runner._finalize）：页面收到 run.finished 立刻来取， + # 这一下还没封存。这不是「不比对」，是封存完就比——照实答 pending，页面按退避接着取；这里不另排按需比对 + return {**base, "state": "pending"} sealed = await _sealed(run_id) if why is None else None if sealed is not None and not sealed.queries: why = ledger_match.NO_QUERIES diff --git a/backend/tests/test_ledger_api.py b/backend/tests/test_ledger_api.py index 24cc969a..e70a6561 100644 --- a/backend/tests/test_ledger_api.py +++ b/backend/tests/test_ledger_api.py @@ -93,6 +93,30 @@ async def test_an_unmatched_run_is_matched_on_demand_and_says_it_looks_back(clie assert body["retro_note"] == "按现在的口径回看" +async def test_a_run_that_just_finished_and_is_about_to_be_sealed_is_pending(client, monkeypatch): + """runner 先发 run.finished、落库，再封存、再排上比对：页面收到 run.finished 立刻来取账本，这一下运行已经成功、还没封存。 + 这不是「不比对」，是「封存完就比」——照实答 pending（不在这里另排一次按需比对：runner 封存之后自己会排），页面按退避 + 接着取；真不比的（正式运行、失败的）照旧 ineligible。""" + await S.publish() + row = await S.explore(monkeypatch, S.EQUIVALENT, text="本周销售额 [[v:Q1.r0.销售额]]。") + await ledger_match.drain() + from app.db.models import Run + + async with SessionLocal() as session: + fresh = Run(workflow_name="刚跑完", status="succeeded", run_class="exploratory", origin_kind="chat", graph={}, + input={}, output={}) + failed = Run(workflow_name="失败了", status="failed", run_class="exploratory", origin_kind="chat", graph={}, + input={}, output={}) + session.add_all([fresh, failed]) + await session.commit() + fresh_id, failed_id = fresh.id, failed.id + scheduled: list[str] = [] + monkeypatch.setattr(ledger_match, "schedule", lambda run_id, **kw: scheduled.append(run_id) or True) + body = await ledger_of(client, fresh_id) + assert body["state"] == "pending" and scheduled == [], body + assert (await ledger_of(client, failed_id))["state"] == "ineligible" + + async def test_runs_that_are_not_compared_say_why(client, monkeypatch): row = await S.explore(monkeypatch, S.EQUIVALENT, origin="api", text="本周销售额 [[v:Q1.r0.销售额]]。") body = await ledger_of(client, row.id) diff --git a/backend/tests/test_ledger_reconcile.py b/backend/tests/test_ledger_reconcile.py index a14b5175..68beded9 100644 --- a/backend/tests/test_ledger_reconcile.py +++ b/backend/tests/test_ledger_reconcile.py @@ -248,3 +248,35 @@ async def test_users_read_their_reconcile_runs_only_through_the_evidence(client, assert (await carol.get(f"/api/runs/{done.id}/evidence")).status_code == 200 finally: await auth_env.reset() + + +async def test_the_reconcile_view_gives_the_panel_what_it_draws(client, monkeypatch): + """界面层不放逻辑（W4-c 账本前端）：「看差多少」那一项带上口径的入口字段（核对参数的表单照它画：哪几项、必填、有没有 + 缺省值）；认不出列怎么配时 409 另带可选的列和口径卡上的指标（选对照的那一步照它画）；对过的账列出对账卡和每一组的 + 这次 / 口径 / 差值 / 相对差是哪几个指标（三个数各自点得开，界面不按命名规则去拼指标 id）。""" + wf = await S.publish(graph=S.caliber_graph(expression="cell(nodes.fetch, 0, 'gmv') * 1")) + row = await S.explore(monkeypatch, S.WITH_REFUNDS) + [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"] + assert action["fields"] == [{"name": "start", "required": True, "has_default": False}, + {"name": "end", "required": True, "has_default": False}, + {"name": "store", "required": False, "has_default": True}] + body = {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, "confirm": True} + res = await client.post(f"/api/ledger/answers/{row.id}/reconcile", json=body) + assert res.status_code == 409 and res.json()["code"] == ledger_reconcile.NEEDS_PAIRS + assert res.json()["options"] == {"columns": ["gmv"], "metrics": [{"id": "gmv", "name": "销售额", "unit": "元"}]} + done = await reconcile(client, row.id, {**body, "pairs": [{"ours": "gmv", "metric": "gmv"}]}) + [listed] = (await ledger_of(client, row.id))["reconciled"] + assert listed["run_id"] == done.id and listed["card"] == "reconcile" + assert listed["groups"] == [{"label": "销售额", "ours": "ours_0", "caliber": "cal_0", "diff": "diff_0", "rel": "rel_0"}] + + +async def test_multi_row_reconcile_groups_carry_the_row_labels(client, monkeypatch): + wf = await S.publish(graph=S.caliber_graph(BY_REGION)) + row = await S.explore(monkeypatch, BY_REGION_ALL, text="东区 [[v:Q1.r0.gmv]]，西区 [[v:Q1.r1.gmv]]。") + [action] = [a for a in (await ledger_of(client, row.id))["actions"] if a["kind"] == "reconcile"] + await reconcile(client, row.id, {"query": action["query"], "workflow_id": wf, "version": 1, "inputs": CHECKED, + "confirm": True}) + [listed] = (await ledger_of(client, row.id))["reconciled"] + assert [g["label"] for g in listed["groups"]] == ["区域=east · gmv", "区域=west · gmv"] + assert listed["groups"][1] == {"label": "区域=west · gmv", "ours": "ours_1_0", "caliber": "cal_1_0", "diff": "diff_1_0", + "rel": "rel_1_0"}