From 86e52192a9c231b9e4e4fa045ca5ba7e82cb22d9 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Mon, 5 Oct 2026 08:49:52 +0800 Subject: [PATCH] =?UTF-8?q?=E8=BF=90=E8=A1=8C=E5=AF=B9=E6=AF=94=E5=90=8E?= =?UTF-8?q?=E7=AB=AF=EF=BC=88=E5=8F=98=E5=8C=96=E4=B8=89=E5=88=86=E6=B3=95?= =?UTF-8?q?=EF=BC=89=EF=BC=9A=E4=B8=A4=E6=AC=A1=E8=BF=90=E8=A1=8C=E6=8C=89?= =?UTF-8?q?=E6=8C=87=E6=A0=87=E6=8B=86=E6=B8=85=E6=98=AF=E6=95=B0=E6=8D=AE?= =?UTF-8?q?=E5=8F=98=E4=BA=86=E3=80=81=E5=8F=A3=E5=BE=84=E5=8F=98=E4=BA=86?= =?UTF-8?q?=E8=BF=98=E6=98=AF=E6=A8=A1=E6=9D=BF=20/=20SQL=20=E5=8F=98?= =?UTF-8?q?=E4=BA=86=EF=BC=8C=E6=95=B0=E6=8D=AE=E5=8F=98=E5=8C=96=E6=8C=89?= =?UTF-8?q?=E7=BB=B4=E5=BA=A6=E5=BD=92=E5=9B=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit GET /api/runs/{id}/compare/base 找默认基准：定时任务发起的先找同一定时任务更早结束的一次，否则同一工作流、 同一运行类别；只挑封存过的终态运行，找不到 404。GET /api/runs/{id}/compare?base= 对比两次运行：两次都要 已结束、封存且核对一致（照搬证据接口的封存判定），否则 409；不是同一个工作流、同一次运行 422。只读，不追加 事件、不落工件。 engine/compare.py 是纯函数： - 查询按「节点 + 第几次 + 工具」配对，配不上再按规范化 SQL 配；normalize_sql 判 same / params_only / changed，结构变了的用 diff_facets 写出差在哪（时间范围排在最后）；核查类查询不参与 - 口径卡同一节点按指标 id 配对，表达式、口径版本、小数位、单位、显示格式不同算口径变了 - 每个指标按 added/removed → 口径 → SQL 或上游配置 → 说不清 → 只参数 → 数据 的次序分类；输入落在快照上 的直接比两边快照的 SQL，不落在快照上的看写它的节点和画布上游 - 口径变了时严格拆分：旧表达式配这次的输入用推演重算的还原和求值算 f_old(x_new)，十进制精确相减，两效应之和 逐位等于总变化；SQL 也变了、用到遮罩列、单位不同、旧叶子没记录的不拆并说明 - 维度归因按维度键配对两边的行、取变化最大的 10 个；度量之和等于输入的值（含合计行那一格）才给占比；截断、 度量列遮罩照实写；遮罩列、日期列、每期都不同的列不当维度 --- backend/app/api/compare.py | 167 +++- backend/app/engine/compare.py | 1152 ++++++++++++++++++++++++ backend/tests/fixtures/compare_runs.py | 211 +++++ backend/tests/test_compare_api.py | 387 ++++++++ backend/tests/test_compare_engine.py | 558 ++++++++++++ 5 files changed, 2472 insertions(+), 3 deletions(-) create mode 100644 backend/app/engine/compare.py create mode 100644 backend/tests/fixtures/compare_runs.py create mode 100644 backend/tests/test_compare_api.py create mode 100644 backend/tests/test_compare_engine.py diff --git a/backend/app/api/compare.py b/backend/app/api/compare.py index 582d926c..5b648c45 100644 --- a/backend/app/api/compare.py +++ b/backend/app/api/compare.py @@ -1,10 +1,171 @@ -"""运行对比：两次运行按「变化三分法」对照。 +"""运行对比：同一个工作流的两次运行按「变化三分法」对照——数据变了、口径变了、还是模板 / SQL 变了（engine/compare.py）。 -目前只登记了路由前缀，接口由运行对比的工作包补上。路径挂在 /api/runs/{run_id}/… 下面； -这个路由登记在 runs.router 之前，以后加 /api/runs/compare 这类固定路径也不会被 /{run_id} 吃掉。 +GET /api/runs/{run_id}/compare/base：默认拿哪一次来比，{base_run_id, reason, base}。这次运行是定时任务发起的，取同一 +定时任务、结束得比它早的最近一次终态运行（reason: same_schedule）；不是、或者同一任务还没有更早的，取同一工作流、 +同一运行类别、结束得比它早的最近一次终态运行（reason: same_workflow）。只挑封存过的：没封存的拿来也比不了。 +找不到 404。 + +GET /api/runs/{run_id}/compare?base={base_id}：{target, base, summary, template, metrics, queries, notes}，每个字段 +的形状见 engine/compare.py。两次运行都必须已结束、封存且封存核对一致，封存判定照搬证据接口（api/evidence._sealed）： +没封存的运行事件还会变，封存核对不一致的事件被改过，拿它们比出来的结论都不作数，回 409。不是同一个工作流、或者 +是同一次运行，回 422。 + +只读：两次运行都不追加事件、不落工件、不改运行记录；结果只交回给发请求的人。只认各自封存范围内的工件（证据接口的 +白名单，加上封存的节点产出和指标集里写着的完整值，同推演），遮罩列（查询当时记下的加数据源现在设的）的值不出现在 +结果里。读工件、比 SQL、重算都在线程里做：快照大的时候不挡住同一进程里别的请求。 + +路径挂在 /api/runs/{run_id}/… 下面；这个路由登记在 runs.router 之前，以后加 /api/runs/compare 这类固定路径也不会 +被 /{run_id} 吃掉。定时任务结束后自动对比（期 4）直接调 default_base、compare_runs。 """ from __future__ import annotations +import asyncio +from collections.abc import Callable +from typing import Any + from fastapi import APIRouter +from sqlalchemy import select + +from app.api.coded import CodedHTTPException +from app.api.evidence import _TERMINAL, RUN_NOT_FOUND, _Masks, _Sealed, _sealed, _sealed_cards +from app.api.whatif import _finished_before, _loader, _mask_of, _outputs_before, _returned +from app.db.base import SessionLocal +from app.db.models import DataSource, Run +from app.engine import compare as engine +from app.engine import recompute router = APIRouter(prefix="/api/runs", tags=["compare"]) + +BAD_REQUEST = "compare_bad_request" +MISMATCH = "compare_mismatch" +UNSEALED = "compare_unsealed" +SEAL_BROKEN = "compare_seal_broken" +NO_BASE = "compare_no_base" + +BAD_REQUEST_TEXT = "请求格式有误：base 应为作为对比基准的运行 ID" +BASE_NOT_FOUND = "作为对比基准的运行记录不存在" +SAME_RUN = "不能和同一次运行自己对比，请选择另一次运行作为基准" +UNSAVED = "{who}不属于任何已保存的工作流，无法确认两次运行是同一个工作流，不能对比" +OTHER_WORKFLOW = "两次运行不属于同一个工作流（「{target}」与「{base}」），不能对比" +UNSEALED_TEXT = "{who}尚未结束并封存：对比只用封存的运行记录，运行结束后才能对比" +SEAL_BROKEN_TEXT = "{who}的封存核对不一致：封存后有事件被修改、删除或插入，不能用于对比。可在记录页点「核对」查看核对结果" +NO_BASE_TEXT = ("没有找到可以对比的上一次运行：需要同一工作流、同一运行类别（定时任务发起的优先找同一定时任务）、" + "更早结束并已封存的运行") +TARGET = "这次运行" +BASE = "作为基准的运行" + + +@router.get("/{run_id}/compare/base") +async def compare_base(run_id: str) -> dict[str, Any]: + """默认拿来对比的那一次运行：{base_run_id, reason: same_schedule | same_workflow, base}。""" + found = await default_base(run_id) + if found is None: + raise CodedHTTPException(404, NO_BASE_TEXT, NO_BASE) + run, reason = found + return {"base_run_id": run.id, "reason": reason, "base": run_ref(run)} + + +@router.get("/{run_id}/compare") +async def compare_with(run_id: str, base: str | None = None) -> dict[str, Any]: + """这次运行和 base 那一次的对比，形状见 engine/compare.py。""" + if base is None or not base.strip(): + raise CodedHTTPException(422, BAD_REQUEST_TEXT, BAD_REQUEST) + return await compare_runs(run_id, base.strip()) + + +async def default_base(run_id: str) -> tuple[Run, str] | None: + """(默认的基准运行, 怎么选的)；找不到返回 None，这次运行本身不存在 404。 + + 「早」按结束时间比：这次运行还没结束（还在跑、停在审批）的按开始时间。只挑终态、封存过的运行——没封存的拿来 + 对比只会得到 409。定时任务发起的先找同一定时任务（同一个工作流）的；它是这个任务的第一次时，退回同一工作流、 + 同一运行类别。没挂在已保存的工作流上的运行（画布上没保存就运行的图）没有可比的。""" + async with SessionLocal() as session: + run = await session.get(Run, run_id) + if run is None: + raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND) + if not run.workflow_id: + return None + before = run.finished_at or run.started_at or run.created_at + query = (select(Run) + .where(Run.id != run.id, Run.workflow_id == run.workflow_id, Run.status.in_(_TERMINAL), + Run.manifest_hash.is_not(None), Run.finished_at.is_not(None), Run.finished_at < before) + .order_by(Run.finished_at.desc(), Run.created_at.desc()).limit(1)) + if run.schedule_id: + found = (await session.execute(query.where(Run.schedule_id == run.schedule_id))).scalars().first() + if found is not None: + return found, "same_schedule" + found = (await session.execute(query.where(Run.run_class == run.run_class))).scalars().first() + return (found, "same_workflow") if found is not None else None + + +async def compare_runs(run_id: str, base_id: str) -> dict[str, Any]: + """对比两次运行（run_id 是这次，base_id 是基准）。拒绝时抛 CodedHTTPException：运行不存在 404，同一次运行、 + 不是同一个工作流 422，没结束封存、封存核对不一致 409。""" + async with SessionLocal() as session: + target = await session.get(Run, run_id) + base = await session.get(Run, base_id) + kinds = {str(name): kind for name, kind in (await session.execute(select(DataSource.name, DataSource.kind)))} + if target is None: + raise CodedHTTPException(404, "运行记录不存在", RUN_NOT_FOUND) + if base is None: + raise CodedHTTPException(404, BASE_NOT_FOUND, RUN_NOT_FOUND) + if base.id == target.id: + raise CodedHTTPException(422, SAME_RUN, MISMATCH) + for run, who in ((target, TARGET), (base, BASE)): + if not run.workflow_id: + raise CodedHTTPException(422, UNSAVED.format(who=who), MISMATCH) + if target.workflow_id != base.workflow_id: + raise CodedHTTPException(422, OTHER_WORKFLOW.format(target=target.workflow_name, base=base.workflow_name), + MISMATCH) + sealed = [] + for run, who in ((target, TARGET), (base, BASE)): + found = await _sealed(run.id) + if not found.sealed or found.status not in _TERMINAL: + raise CodedHTTPException(409, UNSEALED_TEXT.format(who=who), UNSEALED) + if not found.trusted: + raise CodedHTTPException(409, SEAL_BROKEN_TEXT.format(who=who), SEAL_BROKEN) + sealed.append(found) + masks = _Masks() + target_side = await _side(target, sealed[0], masks, kinds) + base_side = await _side(base, sealed[1], masks, kinds) + return await asyncio.to_thread(engine.compare, base_side, target_side) + + +def run_ref(run: Run) -> dict[str, Any]: + """一次运行的概要（RunRef）：对比结果和默认基准里都用这一个形状。时间带时区（UTC）。""" + return {"id": run.id, "workflow_name": run.workflow_name, "version": run.version, "run_class": run.run_class, + "started_at": run.started_at.isoformat() if run.started_at else None, + "finished_at": run.finished_at.isoformat() if run.finished_at else None, + "schedule_id": run.schedule_id} + + +async def _side(run: Run, sealed: _Sealed, masks: _Masks, kinds: dict[str, Any]) -> engine.Side: + """一次运行交给对比的东西（engine.compare.Side）：封存范围内的事件、复验过哈希的口径卡、只认封存范围内工件的 + 读法、每份查询快照要遮的列，以及按口径卡那次执行还原输入用的 Sources。""" + cards = await asyncio.to_thread(_sealed_cards, sealed) + loader = _loader(sealed, cards) + # 遮罩要查库（数据源现在设的），在这里每份快照算一遍交进去，对比本身在线程里不碰库。快照先在线程里读好， + # 同一个 loader 记着读过的，对比时不再读盘 + snaps = await asyncio.to_thread(_read_all, loader, list(sealed.queries)) + masked = {artifact: frozenset(await _mask_of(snaps.get(artifact), info, masks)) + for artifact, info in sealed.queries.items()} + returned = _returned(sealed) + + def sources(started: int) -> recompute.Sources: + return recompute.Sources(load=loader, queries=tuple(sealed.queries), outputs=_outputs_before(sealed, started), + masked=masked, returned=returned, finished=_finished_before(sealed, started)) + + return engine.Side(run=run_ref(run), graph=sealed.graph, events=sealed.events, cards=cards, load=loader, + masked=masked, sources=sources, dialects=kinds, inputs=dict(run.input or {})) + + +def _read_all(loader: Callable[[str], Any], artifacts: list[str]) -> dict[str, Any]: + """读出这几份查询快照：读不出来、被改过的记 None（遮罩照样按台账记的数据源算）。""" + out: dict[str, Any] = {} + for artifact in artifacts: + try: + content = loader(artifact) + except ValueError: + content = None + out[artifact] = content if isinstance(content, dict) else None + return out diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py new file mode 100644 index 00000000..32e022f0 --- /dev/null +++ b/backend/app/engine/compare.py @@ -0,0 +1,1152 @@ +"""运行对比：同一个工作流的两次运行（本周和上周的周报），把每个指标的变化拆清楚——是数据变了、口径变了， +还是模板 / SQL 变了；数据变化再按维度归因。每一项都追得到两边的查询快照。 + +只读：这里只算，不读库、不写工件、不追加事件。接口（api/compare.py）把两次运行封存范围内的事件、工件的读法和 +遮罩准备好交进来（Side），结果原样交回给发请求的人。 + +怎么对： + +- 查询：两次运行各自封存范围内交回过的查询快照，按「节点 + 该节点里第几次查询 + 工具」配对，配不上的再按规范化 + SQL（data/sql_semantics.normalize_sql）配。结构：规范文本和字面量都相同是 same；规范文本相同、只有字面量不同是 + params_only（定时运行每次的日期不同，属于参数变化不是模板变化）；否则 changed（数据源换了也算）。核查类的查询 + （双盲复算、反方审稿发的 role: verify）不进报告，不参与对比。 +- 口径卡：同一个口径卡节点按指标 id 配对；循环里执行了几次的取最后一次。表达式、口径版本、小数位、单位、显示格式 + 有一样不同就是口径变了：小数位改变取整，单位、比率与否改变数值的含义，都不只是换个写法。 +- 每个指标一个分类，按这个次序判：一边有一边没有（added / removed）；口径变了（caliber_and_data，可能连同数据）； + 喂它的查询结构变了、上游节点改了配置（sql_changed，标出、不拆）；输入对不上具体的查询，上游又有变化，说不清变化 + 从哪来（incomparable）；只有参数变了（params_only，含运行输入）；都没变时数值相等是 unchanged，不等是 data_only。 + 「喂它的查询」按输入的出处认：cell() 那一格、Agent 按出处交来的字段落在哪份快照上就是哪条查询（两边各自的快照 + 直接比 SQL）；不落在快照上的，看写它的节点自己查过的库，节点不查库（代码、数据整形……）就看它的配置和画布上的 + 全部上游。 +- 严格拆分（只对口径变了的）：旧表达式的每个叶子路径都能在这次的口径卡输入里取得值时， + 口径效应 = f_new(x_new) − f_old(x_new)，数据效应 = f_old(x_new) − f_old(x_old)。f_old(x_new) 用推演重算 + （engine/recompute.py）的还原和求值算，取整按旧口径卡的小数位；三个数按十进制精确相减，两效应之和逐位等于 + 总变化。取数的 SQL 也变了、用到了遮罩列、单位不同的不拆，说明为什么。 +- 维度归因：指标的输入落在查询快照的某一格（或 Agent 数组字段的一段行）、两边快照都没截断时，取快照里不是数值的 + 列作维度（遮罩的、日期类型的不算；两边都只有一个值的、两边没有一个值相同的——日期、期次这类每期都不同的列—— + 也不拿来配对），按维度键配对两边的行，算每个键的 base / target / delta，按 |delta| 取前 10。只有度量列之和 + 等于这个输入的值（两边都等）时才给占比；取的是合计行那一格时，其余各行之和等于它也算，合计行不列进去。 +- 遮罩列的值绝不出现在结果里：取自遮罩列的输入不给值，遮罩列不当维度、不当度量，用到遮罩列的指标不拆分（拆出来的 + f_old(x_new) 配上公开的数就能反推出遮住的那一格）。 +""" +from __future__ import annotations + +import ast +from collections.abc import Callable, Iterable, Mapping +from dataclasses import dataclass, field +from decimal import Decimal, InvalidOperation +from typing import Any + +from app.core.events import EventType +from app.data.sql_semantics import NormalizedSql, diff_facets, facets, normalize_sql +from app.engine import recompute +from app.engine.evidence import DEFAULT_FORMAT, RenderError, is_number, render_number +from app.engine.expressions import ExpressionError, cell_value, leaf_refs, parse_expression +from app.engine.labels import field_label +from app.engine.merge_query import MERGE_SOURCE +from app.engine.nodes.metrics import _REF_HEAD + +UNCHANGED = "unchanged" +DATA_ONLY = "data_only" +CALIBER_AND_DATA = "caliber_and_data" +SQL_CHANGED = "sql_changed" +PARAMS_ONLY = "params_only" +ADDED = "added" +REMOVED = "removed" +INCOMPARABLE = "incomparable" +#: summary 里的各类，次序同接口约定 +CLASSES = (UNCHANGED, DATA_ONLY, CALIBER_AND_DATA, SQL_CHANGED, PARAMS_ONLY, ADDED, REMOVED, INCOMPARABLE) +#: 这几类才做维度归因：数据（连同参数、口径）变了，两边的行说的是同一回事 +_ATTRIBUTED = frozenset({DATA_ONLY, PARAMS_ONLY, CALIBER_AND_DATA}) + +#: 维度归因最多列几个维度值（按变化的绝对值从大到小） +TOP_KEYS = 10 +#: 一条查询的结构差异最多写几条（diff_facets 已按对结果的影响从大到小排） +MAX_DIFFERENCES = 5 +#: 表达式、取值写进说明时最多几个字 +_SHOWN_MAX = 60 +#: 比较 SQL 时两边的叫法（diff_facets 的 labels） +_LABELS = ("基准", "这次") +#: 合并查询的 SQL 在 SQLite 里执行（engine/merge_query.py） +_MERGE_DIALECT = "sqlite" +#: 日期类的列不当维度：每期都不同，按它配对两边的行一个都配不上 +_TEMPORAL = frozenset({"date", "datetime", "time"}) +_FORMAT_LABEL = {"thousands": "千分位", "plain": "不分组", "percent_of_ratio": "比率按百分数显示"} + +NO_CARDS = "两次运行都没有口径卡，没有可以对比的指标" +EDGES_CHANGED = "节点之间的连线有变化" +DEFAULTS_CHANGED = "工作流默认设置有变化" +MASKED_NOTE = "部分列在数据源中设置了遮罩，对比中不显示这些列的值，也不用它们做拆分和归因" +NO_LINEAGE = "口径卡没有记录输入（生成于系统升级前），无法判断变化来源" +DATA_FROM_QUERIES = "取数的 SQL 没有变化，查询结果有变化" +DATA_FROM_UPSTREAM = "模板和取数的 SQL 都没有变化，变化来自上游数据" +SQL_NOT_SPLIT = "取数的 SQL 或上游配置有变化，变化无法在口径与数据之间拆分" +NOT_NUMBERS = "有一边的值不是数值，无法拆分" +SPLIT_MASKED = "用到了设置了遮罩的列，不做拆分" +NO_DIFFERENCE = "查询的写法有变化（选取的列、计算方式等）" +TRUNCATED = "查询结果被截断（{side}只取回了前 {rows} 行），不做维度归因" +MASKED_MEASURE = "度量列「{column}」设置了遮罩，不做维度归因" +UNREADABLE_SNAPSHOT = "查询快照无法读取，不做维度归因" +NOT_A_MEASURE = "度量列「{column}」有不是数值的格，不做维度归因" +NO_SHARE = "各维度值的度量之和与输入的值不相等，变化仅供参考，不给占比" +ZERO_TOTAL = "总变化为 0，不计算占比" +TOTAL_ROW = "输入取自合计行（第 {row} 行），按其余各行归因" +MORE_KEYS = "共 {count} 个维度值，列出变化最大的 {top} 个" +NO_OVERLAP = "列「{columns}」两边没有相同的值（可能是日期、期次这类每期都不同的列），不按它配对" + + +@dataclass +class Side: + """一次运行在对比里要用到的东西，接口从这次运行的封存范围里准备好。 + + - run：运行的概要（id、工作流名、版本、类别、起止时间、定时任务），原样放进结果 + - graph：运行时的图快照；events：封存范围内的事件 (seq, type, node_id, data) + - cards：封存范围内复验过哈希的口径卡 (节点, 工件, 指标集)（api/evidence._sealed_cards）。事件里记着、这里 + 没有的，就是读不出来或者被改过 + - load：工件 id → 内容，只认封存范围内的；文件不在返回 None，内容和哈希对不上抛 ValueError + - masked：查询快照 → 要遮的列（小写）：查询当时记下的加数据源现在设的 + - sources：口径卡那次执行开始的事件序号 → 推演重算还原输入要用的 Sources（节点产出、跑完的时刻都按那一刻）。 + 不给就按 load、masked 拼一个最小的：nodes.* 这类容器输入还原不出来 + - dialects：数据源名 → 类型（sqlite / postgres …），规范化 SQL 时按它的方言解析 + - inputs：运行输入（Run.input） + """ + + run: dict[str, Any] + graph: dict[str, Any] + events: list[tuple[int, str, str | None, Any]] + cards: list[tuple[str, str, dict[str, Any]]] = field(default_factory=list) + load: Callable[[str], Any] = lambda _: None + masked: Mapping[str, frozenset[str]] = field(default_factory=dict) + sources: Callable[[int], recompute.Sources] | None = None + dialects: Mapping[str, str | None] = field(default_factory=dict) + inputs: dict[str, Any] = field(default_factory=dict) + + +def compare(base: Side, target: Side) -> dict[str, Any]: + """对比两次运行（base 是基准，target 是这次）：{target, base, summary, template, metrics, queries, notes}， + 形状见接口约定（api/compare.py）。只读，不碰两次运行的任何东西。""" + return _Compare(base, target).result() + + +# -------------------------------------------------------------------------- +# 查询：收集、配对、比结构 +# -------------------------------------------------------------------------- + + +@dataclass +class _Query: + """一次运行里交回过的一份查询快照（核查类的不算）。""" + + artifact: str + node_id: str + tool: str | None + #: 这个节点里第几次查询，从 1 数（循环里执行几次的接着数） + ordinal: int + call_id: str | None = None + #: 快照里的 SQL；快照读不出来时取 tool.start 的参数（合并查询取 merge.end 的） + sql: str | None = None + snap: dict[str, Any] | None = None + #: 快照读不出来的原因：missing / tampered + problem: str | None = None + source: str | None = None + + +def _snapshot(load: Callable[[str], Any], artifact: str) -> tuple[dict[str, Any] | None, str | None]: + try: + content = load(artifact) + except ValueError: + return None, "tampered" + if isinstance(content, dict) and isinstance(content.get("rows"), list): + return content, None + return None, "missing" + + +def _queries(side: Side) -> list[_Query]: + """封存范围内交回过的查询快照，按交回的先后。 + + 调用工具节点、Agent 的查询在 tool.end 里（query_artifact）；合并查询只记在 node.finished 的台账里。同一件快照 + 两处都有的按 tool.end 算一次。核查类的查询（role: verify）不进写作目录，这里也不收。""" + out: list[_Query] = [] + seen: set[tuple[str, str]] = set() + asked: dict[tuple[str, Any], str] = {} # (节点, call_id) → tool.start 里的 SQL + merged: dict[str, str] = {} # 合并节点 → merge.end 里的 SQL + count: dict[str, int] = {} + + def add(node_id: str, artifact: str, tool: Any, call_id: Any, sql: Any, source: Any) -> None: + count[node_id] = count.get(node_id, 0) + 1 + snap, problem = _snapshot(side.load, artifact) + own = snap.get("sql") if snap is not None and isinstance(snap.get("sql"), str) else None + out.append(_Query(artifact=artifact, node_id=node_id, tool=str(tool) if tool else None, + ordinal=count[node_id], call_id=str(call_id) if call_id else None, + sql=own or (sql if isinstance(sql, str) and sql.strip() else None), snap=snap, + problem=problem, source=(snap or {}).get("source") or source)) + + for _, etype, node_id, data in side.events: + if not node_id or not isinstance(data, dict) or data.get("role") == "verify": + continue + if etype == EventType.TOOL_START: + args = data.get("args") + if isinstance(args, dict) and isinstance(args.get("sql"), str): + asked[(node_id, data.get("call_id"))] = args["sql"] + elif etype == EventType.TOOL_END and data.get("query_artifact"): + artifact = str(data["query_artifact"]) + seen.add((node_id, artifact)) + add(node_id, artifact, data.get("tool"), data.get("call_id"), asked.get((node_id, data.get("call_id"))), + None) + elif etype == EventType.MERGE_END and isinstance(data.get("sql"), str): + merged[node_id] = data["sql"] + elif etype == EventType.NODE_FINISHED: + for entry in data.get("evidence") or []: + if not (isinstance(entry, dict) and entry.get("kind") == "query" and entry.get("artifact")): + continue + artifact = str(entry["artifact"]) + if (node_id, artifact) in seen: + # tool.end 已经收过：台账里多记着数据源，读不出快照时用得上 + for q in out: + if q.node_id == node_id and q.artifact == artifact and not q.source: + q.source = entry.get("source") + continue + seen.add((node_id, artifact)) + add(node_id, artifact, entry.get("tool"), entry.get("call_id"), merged.get(node_id), + entry.get("source")) + return out + + +def _pair(base: list[_Query], target: list[_Query], + text: Callable[[_Query], str | None]) -> list[tuple[_Query | None, _Query | None]]: + """先按「节点 + 第几次 + 工具」配，配不上的再按规范化 SQL 配（节点换了 ID、同一个模板只有日期不同的也配得上）。 + 结果按这次运行的先后排，基准运行多出来的接在后面。""" + slots = {(q.node_id, q.ordinal, q.tool): i for i, q in enumerate(base)} + matched: dict[int, int] = {} + for j, q in enumerate(target): + i = slots.get((q.node_id, q.ordinal, q.tool)) + if i is not None: + matched[j] = i + taken = set(matched.values()) + for j, q in enumerate(target): + if j in matched or (key := text(q)) is None: + continue + i = next((i for i, b in enumerate(base) if i not in taken and text(b) == key), None) + if i is not None: + matched[j] = i + taken.add(i) + pairs: list[tuple[_Query | None, _Query | None]] = [ + (base[matched[j]] if j in matched else None, q) for j, q in enumerate(target)] + return pairs + [(b, None) for i, b in enumerate(base) if i not in taken] + + +@dataclass +class _Shape: + """一对查询的结构：same / params_only / changed / added / removed，不同的字面量，结构上差在哪。""" + + structure: str + literals: list[dict[str, str]] = field(default_factory=list) + differences: list[str] = field(default_factory=list) + #: 两边的 SQL 都拿得到（拿不到的只能说不知道，不能说变了） + known: bool = True + + +# -------------------------------------------------------------------------- +# 口径卡 +# -------------------------------------------------------------------------- + + +@dataclass +class _Card: + """一个口径卡节点在一次运行里最后一次执行的指标集。""" + + node_id: str + artifact: str + content: dict[str, Any] | None + #: 指标集读不出来、被改过的说法 + problem: str | None + #: 那次执行的 node.started 序号：输入按那一刻的节点产出还原 + started: int + #: 执行了几次 + runs: int + plans: dict[str, Any] = field(default_factory=dict) + #: 这张卡上全部指标的输入，按路径（同一次执行里同一条路径的值是同一个） + inputs: dict[str, Any] = field(default_factory=dict) + sources: recompute.Sources | None = None + + def metrics(self) -> list[dict[str, Any]]: + return [m for m in (self.content or {}).get("metrics") or [] if isinstance(m, dict) and m.get("id") is not None] + + +def _cards(side: Side) -> dict[str, _Card]: + """每个口径卡节点最后一次执行的指标集，按第一次执行的先后。哪些 node.finished 算口径卡的执行，和 + api/evidence._sealed_cards 一样认：台账里的 metric_set 条目；一期以前没有台账的，画布上是口径卡的节点的产出。""" + verified = {(node_id, artifact): card for node_id, artifact, card in side.cards} + calibers = {str(n.get("id")) for n in side.graph.get("nodes") or [] + if isinstance(n, dict) and n.get("type") == "metrics"} + runs: dict[str, list[tuple[int, str]]] = {} + started: dict[str, list[int]] = {} + for seq, etype, node_id, data in side.events: + if not node_id or not isinstance(data, dict): + continue + if etype == EventType.NODE_STARTED: + started.setdefault(node_id, []).append(seq) + if etype != EventType.NODE_FINISHED: + continue + arts = [str(e["artifact"]) for e in data.get("evidence") or [] + if isinstance(e, dict) and e.get("kind") == "metric_set" and e.get("artifact")] + if not arts and node_id in calibers and data.get("artifact"): + arts = [str(data["artifact"])] + runs.setdefault(node_id, []).extend((seq, art) for art in arts) + out: dict[str, _Card] = {} + for node_id, done in runs.items(): + if not done: + continue + seq, artifact = done[-1] + content = verified.get((node_id, artifact)) + out[node_id] = _Card(node_id=node_id, artifact=artifact, content=content, + problem=None if content is not None else _card_problem(side.load, artifact), + started=max((s for s in started.get(node_id, []) if s < seq), default=seq), + runs=len(done)) + return out + + +def _card_problem(load: Callable[[str], Any], artifact: str) -> str: + try: + content = load(artifact) + except ValueError: + return "指标集与哈希不一致，疑似被修改" + return "无法读取指标集" if content is None else "指标集的内容不完整" + + +def _plan_card(side: Side, card: _Card, queries: list[_Query]) -> None: + """按推演重算的规则还原卡上每个指标的输入：落在哪份快照上、是不是遮罩列、容器的完整值。""" + if card.content is None: + return + sources = side.sources(card.started) if side.sources is not None else recompute.Sources( + load=side.load, queries=tuple(dict.fromkeys(q.artifact for q in queries)), masked=side.masked) + store = recompute._Store(sources) + card.sources = sources + for metric in card.metrics(): + plan = recompute._plan(metric, store) + card.plans[str(metric["id"])] = plan + for path, inp in plan.inputs.items(): + card.inputs.setdefault(path, inp) + + +# -------------------------------------------------------------------------- +# 数：精确相减、渲染 +# -------------------------------------------------------------------------- + + +def _exact(value: Any) -> Decimal | None: + """有限的数换成十进制：浮点按最短的写法（repr），相减不带二进制的误差。""" + if not recompute._number(value): + return None + try: + return Decimal(repr(value)) if isinstance(value, float) else Decimal(value) + except (InvalidOperation, ValueError): + return None + + +def _plain(d: Decimal, *, whole: bool) -> int | float: + """十进制换回 JSON 里的数：参与的都是整数时给整数，否则给浮点。""" + if whole and d == d.to_integral_value(): + return int(d) + return float(d) + + +def _places(*metrics: dict[str, Any]) -> int | None: + """几个指标里最多的小数位；有一个没设的就不定小数位（按值本来的样子写）。""" + found = [m.get("decimals") for m in metrics] + if any(not isinstance(d, int) or isinstance(d, bool) for d in found): + return None + return max(found) + + +def _scale(metric: dict[str, Any]) -> tuple[str, bool]: + """数值的含义：单位，以及是不是比率（percent_of_ratio 的 0.087 和单位为 % 的 8.7 不能直接相减）。""" + return str(metric.get("unit") or ""), (metric.get("format") or DEFAULT_FORMAT) == "percent_of_ratio" + + +def _signed(meta: dict[str, Any], value: Any) -> str | None: + """变化量的写法：+300.0元、-2单、+10.3 个百分点（比率、百分数的变化写百分点，和推演同一个写法）。为 0 写不带 + 符号的 0。渲染不出来的给 None。""" + if not recompute._number(value): + return None + if value != 0: + return recompute._delta(meta, 0, value)[1] + fmt, unit, places = meta.get("format") or DEFAULT_FORMAT, str(meta.get("unit") or ""), meta.get("decimals") + try: + if fmt == "percent_of_ratio": + return render_number(0, decimals=places, fmt=fmt).rstrip("%") + " 个百分点" + if unit in ("%", "％"): + return render_number(0, decimals=places, fmt=fmt) + " 个百分点" + return render_number(0, unit=unit, decimals=places, fmt=fmt) + except RenderError: + return None + + +def _shown(value: Any) -> str: + """一个值写进说明：文本、数照写（太长的截断），空值写「空值」，容器写大概。""" + if value is None: + return "空值" + if isinstance(value, (list, dict)): + return f"[{len(value)} 项]" if isinstance(value, list) else f"{{{len(value)} 个字段}}" + text = str(value) + return text if len(text) <= _SHOWN_MAX else text[:_SHOWN_MAX] + "…" + + +def _expression(metric: dict[str, Any]) -> str: + """表达式的规范写法：空白、括号写法不同的算同一个。解析不了的照原文。""" + raw = str(metric.get("expression") or "").strip() + try: + return ast.unparse(parse_expression(raw)[0]) + except (ExpressionError, ValueError): + return raw + + +# -------------------------------------------------------------------------- +# 画布 +# -------------------------------------------------------------------------- + + +def _nodes(graph: dict[str, Any]) -> dict[str, dict[str, Any]]: + return {str(n.get("id")): n for n in graph.get("nodes") or [] if isinstance(n, dict) and n.get("id")} + + +def _data(node: dict[str, Any]) -> dict[str, Any]: + return node.get("data") if isinstance(node.get("data"), dict) else {} + + +def _config(node: dict[str, Any]) -> dict[str, Any]: + config = _data(node).get("config") + return config if isinstance(config, dict) else {} + + +def _edges(graph: dict[str, Any]) -> set[tuple[str, str, str]]: + """连线：(起点, 终点, 出口)。默认出口的几种写法算同一个（和 autofix 认连线的规则一致）。""" + out = set() + for edge in graph.get("edges") or []: + if isinstance(edge, dict): + handle = edge.get("sourceHandle") + out.add((str(edge.get("source")), str(edge.get("target")), + "" if handle in (None, "out", "source") else str(handle))) + return out + + +def _config_keys(was: dict[str, Any], now: dict[str, Any]) -> tuple[list[str], bool]: + """一个节点改了哪些配置项：(键名列表, 改的是不是只有名称)。值是字典的往下看一层（args.sql），不展示值本身。""" + keys: list[str] = [] + if was.get("type") != now.get("type"): + keys.append("type") + renamed = _data(was).get("label") != _data(now).get("label") + a, b = _config(was), _config(now) + for key in dict.fromkeys([*a, *b]): + x, y = a.get(key), b.get(key) + if x == y: + continue + if isinstance(x, dict) and isinstance(y, dict): + keys += [f"{key}.{sub}" for sub in dict.fromkeys([*x, *y]) if x.get(sub) != y.get(sub)] + else: + keys.append(key) + return (keys + ["label"] if renamed else keys), renamed and not keys + + +def _key_label(key: str, node_type: Any) -> str: + return "节点类型" if key == "type" else field_label(key, node_type) + + +# -------------------------------------------------------------------------- +# 输入的出处有没有变 +# -------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class _Change: + """一个输入的出处在两次运行之间变没变：same / params / sql / template / unknown，加一句说明。""" + + kind: str + reason: str | None = None + + +#: 合并几处出处变化时的轻重：模板、SQL 变了 > 说不清 > 只有参数变了 > 没变 +_WEIGHT = {"sql": 3, "template": 3, "unknown": 2, "params": 1, "same": 0} + + +def _worst(changes: Iterable[_Change]) -> _Change: + found = list(changes) + return max(found, key=lambda c: _WEIGHT[c.kind]) if found else _Change("same") + + +def _from_input(path: str, item: dict[str, Any]) -> bool: + head = _REF_HEAD.match(path) + return item.get("via") == "input" or (head is not None and head.group(1) == "input") + + +# -------------------------------------------------------------------------- +# 对比 +# -------------------------------------------------------------------------- + + +class _Compare: + def __init__(self, base: Side, target: Side) -> None: + self.base, self.target = base, target + self.notes: list[str] = [] + self.masked_seen = False + self._norm: dict[tuple[str, str | None], NormalizedSql] = {} + self._shapes: dict[tuple[int, int], _Shape] = {} + self._ancestors: dict[str, set[str]] = {} + self.base_queries, self.target_queries = _queries(base), _queries(target) + self.pairs = _pair(self.base_queries, self.target_queries, self._text) + self.template, self.changed = self._template() + self.titles = {**{k: self._title_of(n) for k, n in _nodes(base.graph).items()}, + **{k: self._title_of(n) for k, n in _nodes(target.graph).items()}} + self.parents: dict[str, set[str]] = {} + for source, target_id, _ in _edges(base.graph) | _edges(target.graph): + self.parents.setdefault(target_id, set()).add(source) + + # ---------------------------------------------------------------- 结果 + + def result(self) -> dict[str, Any]: + metrics = self._metrics() + queries = [self._query_view(b, t) for b, t in self.pairs] + self._run_notes() + if self.masked_seen: + self.notes.append(MASKED_NOTE) + summary = {name: 0 for name in CLASSES} + for m in metrics: + summary[m["classification"]] += 1 + return {"target": self.target.run, "base": self.base.run, "summary": summary, "template": self.template, + "metrics": metrics, "queries": queries, "notes": list(dict.fromkeys(self.notes))} + + def _title_of(self, node: dict[str, Any]) -> str: + return str(_data(node).get("label") or node.get("id")) + + def title(self, node_id: Any) -> str: + return self.titles.get(str(node_id), str(node_id)) + + def _run_notes(self) -> None: + """运行输入的变化：键名，值短的连值一起写。值可能是一大段文字，太长的只说有变化。""" + before, after = self.base.inputs or {}, self.target.inputs or {} + for key in dict.fromkeys([*before, *after]): + a, b = before.get(key), after.get(key) + if a == b: + continue + short = all(v is None or isinstance(v, (int, float, str, bool)) and len(str(v)) <= 40 for v in (a, b)) + self.notes.append(f"运行输入「{key}」由「{_shown(a)}」改为「{_shown(b)}」" if short + else f"运行输入「{key}」有变化") + + # ---------------------------------------------------------------- 查询 + + def _dialect(self, source: Any) -> str | None: + if source == MERGE_SOURCE: + return _MERGE_DIALECT + return self.target.dialects.get(str(source)) or self.base.dialects.get(str(source)) if source else None + + def _normalize(self, sql: str, dialect: str | None) -> NormalizedSql: + key = (sql, dialect) + if key not in self._norm: + self._norm[key] = normalize_sql(sql, dialect) + return self._norm[key] + + def _text(self, q: _Query) -> str | None: + return self._normalize(q.sql, self._dialect(q.source)).text if q.sql else None + + def shape(self, b: _Query | None, t: _Query | None) -> _Shape: + if b is None: + return _Shape("added") + if t is None: + return _Shape("removed") + key = (id(b), id(t)) + if key not in self._shapes: + self._shapes[key] = self._compare_sql(b, t) + return self._shapes[key] + + def _compare_sql(self, b: _Query, t: _Query) -> _Shape: + if b.sql is None or t.sql is None: + which = "基准运行" if b.sql is None else "这次运行" + return _Shape("changed", known=False, differences=[f"{which}的查询快照无法读取，无法比较 SQL"]) + differences = [] + if b.source and t.source and b.source != t.source: + differences.append(f"数据源由「{b.source}」改为「{t.source}」") + dialect = self._dialect(t.source or b.source) + nb, nt = self._normalize(b.sql, dialect), self._normalize(t.sql, dialect) + if nb.text != nt.text: + # 时间范围的不同排到最后：定时运行每期的日期本来就不同，结构上真正改了的（多了筛选、换了时间列……） + # 才是要先看的 + found = sorted(diff_facets(facets(b.sql, dialect), facets(t.sql, dialect), _LABELS), + key=lambda d: d.aspect == "time_range") + texts = [d.text for d in found] or [NO_DIFFERENCE] + return _Shape("changed", differences=(differences + texts)[:MAX_DIFFERENCES]) + if differences: + return _Shape("changed", differences=differences) + literals = [{"base": x, "target": y} for x, y in zip(nb.literals, nt.literals) if x != y] + return _Shape("params_only", literals) if literals else _Shape("same") + + def _query_view(self, b: _Query | None, t: _Query | None) -> dict[str, Any]: + shape = self.shape(b, t) + q = t or b + assert q is not None + view = { + "node_id": q.node_id, "node_title": self.title(q.node_id), "ordinal": q.ordinal, + "structure": shape.structure, + "base_sql": b.sql if b else None, "target_sql": t.sql if t else None, "literals": shape.literals, + "base_rows": len(b.snap["rows"]) if b and b.snap else None, + "target_rows": len(t.snap["rows"]) if t and t.snap else None, + "base_truncated": bool(b.snap.get("truncated")) if b and b.snap else None, + "target_truncated": bool(t.snap.get("truncated")) if t and t.snap else None, + "base_artifact": b.artifact if b else None, "target_artifact": t.artifact if t else None, + "differences": shape.differences, + } + problems = [f"{side}的查询快照{'与哈希不一致，疑似被修改' if one.problem == 'tampered' else '无法读取'}" + for side, one in (("基准运行", b), ("这次运行", t)) if one is not None and one.problem] + if problems: + view["problem"] = "；".join(problems) + return view + + def _side_query(self, queries: list[_Query], inp: Any) -> _Query | None: + """一个输入落在的那份快照，是这次运行里哪一次查询交回的（同一件快照出现几次时先找写它的节点）。""" + found = [q for q in queries if q.artifact == inp.snapshot] + node_id = inp.item.get("node_id") + return next((q for q in found if q.node_id == node_id), found[0] if found else None) + + def _query_change(self, b: _Query | None, t: _Query | None) -> _Change: + q = t or b + assert q is not None + name = f"查询「{self.title(q.node_id)}」" + if b is None: + return _Change("sql", f"节点「{self.title(q.node_id)}」这次多了一次查询") + if t is None: + return _Change("sql", f"节点「{self.title(q.node_id)}」这次少了一次查询") + shape = self.shape(b, t) + if not shape.known: + return _Change("unknown", f"{name}的快照无法读取，无法判断 SQL 是否有变化") + if shape.structure == "changed": + return _Change("sql", f"{name}的 SQL 结构有变化" + (f"：{shape.differences[0]}" if shape.differences else "")) + if shape.structure == "params_only": + shown = "，".join(f"{x['base']} → {x['target']}" for x in shape.literals[:2]) + more = f" 等 {len(shape.literals)} 处" if len(shape.literals) > 2 else "" + return _Change("params", f"{name}的 SQL 只有参数不同（{shown}{more}）") + return _Change("same") + + def _node_change(self, node_id: str) -> _Change | None: + """一个节点自己查过的库在两次运行之间变没变；它两边都没查过库返回 None。""" + own = [(b, t) for b, t in self.pairs if (b and b.node_id == node_id) or (t and t.node_id == node_id)] + return _worst(self._query_change(b, t) for b, t in own) if own else None + + def _upstream(self, node_id: str) -> set[str]: + """画布上 node_id 的全部上游（两次运行的图合起来，宁可多算）。""" + if node_id not in self._ancestors: + seen: set[str] = set() + todo = [node_id] + while todo: + for parent in self.parents.get(todo.pop(), ()): + if parent not in seen: + seen.add(parent) + todo.append(parent) + self._ancestors[node_id] = seen + return self._ancestors[node_id] + + def provenance(self, path: str, b: Any, t: Any) -> _Change: + """一个输入的出处在两次运行之间变没变（b、t 是推演重算还原出来的输入）。""" + if b is None or t is None: + return _Change("unknown", f"输入「{path}」只在一边的口径卡里有记录") + if b.snapshot or t.snapshot: + # 落在查询快照上：两边各自的那份快照直接比 SQL，不管全局配对把它们配给了谁 + if not (b.snapshot and t.snapshot): + return _Change("unknown", f"输入「{path}」只有一边对得上查询快照，无法比较取数") + qb, qt = self._side_query(self.base_queries, b), self._side_query(self.target_queries, t) + if qb is None or qt is None: + return _Change("unknown", f"输入「{path}」落在的查询快照不在封存范围内的查询里，无法比较取数") + return self._query_change(qb, qt) + nb, nt = b.item.get("node_id"), t.item.get("node_id") + if nb != nt: + return _Change("template", f"输入「{path}」的来源节点由「{self.title(nb)}」换成了「{self.title(nt)}」") + if _from_input(path, b.item): + if recompute._same(b.item.get("value"), t.item.get("value")): + return _Change("same") + return _Change("params", f"运行输入「{path}」由「{_shown(b.item.get('value'))}」" + f"改为「{_shown(t.item.get('value'))}」") + if nb and (own := self._node_change(str(nb))) is not None: + # 写它的节点自己查过库（Agent 没按出处交、调用工具节点的结果整份用）：它查的那几条就是喂它的查询 + if self.changed.get(str(nb)) and own.kind in ("same", "params"): + return _Change("template", f"上游节点「{self.title(nb)}」的配置有变化") + return own + if nb and self.changed.get(str(nb)): + return _Change("template", f"上游节点「{self.title(nb)}」的配置有变化") + # 写它的节点不查库（代码、数据整形、模型调用……）：看画布上它的全部上游；写它的节点不知道就看整张图 + scope = self._upstream(str(nb)) if nb else set(self.titles) + moved = [n for n in scope if self.changed.get(n)] + changes = [self._query_change(b2, t2) for b2, t2 in self.pairs if (t2 or b2).node_id in scope] + worst = _worst(changes) + where = f"来自「{self.title(nb)}」" if nb else "不知道由哪个节点写入" + if moved or worst.kind in ("sql", "template", "unknown"): + return _Change("unknown", f"输入「{path}」{where}，对不上具体的查询；它的上游有变化，无法判断变化来源") + return worst + + # ---------------------------------------------------------------- 模板 + + def _template(self) -> tuple[dict[str, Any], dict[str, str]]: + """两次运行的图快照比较：新增、删除的节点，配置有变化的节点和配置项名（不展示值）。changed 是影响取值的 + 变化（新增、删除、类型或配置变了；只改名称的不算），判断输入的出处时用。""" + before, after = _nodes(self.base.graph), _nodes(self.target.graph) + nodes: list[dict[str, Any]] = [] + changed: dict[str, str] = {} + + def entry(node_id: str, node: dict[str, Any], change: str, keys: list[str]) -> dict[str, Any]: + return {"node_id": node_id, "title": self._title_of(node), "change": change, "keys": keys, + "labels": [_key_label(k, node.get("type")) for k in keys]} + + for node_id, node in after.items(): + was = before.get(node_id) + if was is None: + nodes.append(entry(node_id, node, "added", [])) + changed[node_id] = "added" + continue + keys, renamed_only = _config_keys(was, node) + if keys: + nodes.append(entry(node_id, node, "config", keys)) + if not renamed_only: + changed[node_id] = "config" + for node_id, node in before.items(): + if node_id not in after: + nodes.append(entry(node_id, node, "removed", [])) + changed[node_id] = "removed" + same_edges = _edges(self.base.graph) == _edges(self.target.graph) + same_defaults = (self.base.graph.get("defaults") or {}) == (self.target.graph.get("defaults") or {}) + if not same_edges: + self.notes.append(EDGES_CHANGED) + if not same_defaults: + self.notes.append(DEFAULTS_CHANGED) + return {"base_version": self.base.run.get("version"), "target_version": self.target.run.get("version"), + "same_graph": not nodes and same_edges and same_defaults, "nodes": nodes}, changed + + # ---------------------------------------------------------------- 指标 + + def _metrics(self) -> list[dict[str, Any]]: + before, after = _cards(self.base), _cards(self.target) + if not before and not after: + self.notes.append(NO_CARDS) + return [] + out: list[dict[str, Any]] = [] + for node_id in [*after, *(n for n in before if n not in after)]: + b, t = before.get(node_id), after.get(node_id) + for side, card in (("基准运行", b), ("这次运行", t)): + if card is not None and card.runs > 1: + self.notes.append(f"口径卡「{self.title(node_id)}」在{side}中执行了 {card.runs} 次，按最后一次对比") + if b is not None: + _plan_card(self.base, b, self.base_queries) + if t is not None: + _plan_card(self.target, t, self.target_queries) + for side, card in (("基准运行", b), ("这次运行", t)): + if card is not None and card.problem: + self.notes.append(f"{side}的口径卡「{self.title(node_id)}」{card.problem}，其中的指标无法对比") + self._card_notes(node_id, b, t) + ids = [str(m["id"]) for m in (t.metrics() if t else [])] + ids += [mid for mid in (str(m["id"]) for m in (b.metrics() if b else [])) if mid not in ids] + for mid in ids: + bm = next((m for m in (b.metrics() if b else []) if str(m["id"]) == mid), None) + tm = next((m for m in (t.metrics() if t else []) if str(m["id"]) == mid), None) + out.append(self._metric(node_id, b, t, bm, tm)) + return out + + def _card_notes(self, node_id: str, b: _Card | None, t: _Card | None) -> None: + """口径卡本身的变化（口径名称、固定的来源版本）：不改变指标的分类，写进说明。""" + if b is None or t is None or b.content is None or t.content is None: + return + before, after = b.content.get("caliber") or "", t.content.get("caliber") or "" + if before != after: + self.notes.append(f"口径卡「{self.title(node_id)}」的口径名称由「{before}」改为「{after}」") + src_b, src_t = b.content.get("source"), t.content.get("source") + if src_b != src_t: + def version(src: Any) -> str: + return f"v{src.get('workflow_version')}" if isinstance(src, dict) else "本地定义" + self.notes.append(f"口径卡「{self.title(node_id)}」固定的口径来源由 {version(src_b)} 改为 {version(src_t)}") + + def _metric(self, card_id: str, b: _Card | None, t: _Card | None, bm: dict[str, Any] | None, + tm: dict[str, Any] | None) -> dict[str, Any]: + meta = tm or bm or {} + mid = str(meta.get("id")) + caliber = next((c.content.get("caliber") for c in (t, b) if c is not None and c.content), None) + out: dict[str, Any] = { + # 画布上的节点名；图里找不到这个节点（老运行的图读不出来）就写口径名 + "card_id": card_id, "card_title": self.title(card_id) if card_id in self.titles else caliber or card_id, + "metric_id": mid, "name": meta.get("name") or mid, "unit": meta.get("unit") or "", + "decimals": meta.get("decimals"), + "base": _value(bm), "target": _value(tm), + "base_rendered": _rendered(bm), "target_rendered": _rendered(tm), + "delta": None, "delta_rendered": None, "delta_pct": None, + "classification": INCOMPARABLE, "reasons": [], "decomposition": None, "decomposition_note": None, + "inputs": [], "attribution": None, + } + bp = b.plans.get(mid) if b is not None and bm is not None else None + tp = t.plans.get(mid) if t is not None and tm is not None else None + out["inputs"] = self._inputs(bp, tp) + # 一边有一边没有：整张卡都没有、卡里没有这个指标；卡读不出来的说不清有没有 + for card, metric, cls, side in ((b, bm, ADDED, "基准运行"), (t, tm, REMOVED, "这次运行")): + if metric is not None: + continue + if card is None: + out.update(classification=cls, reasons=[f"{side}没有口径卡「{self.title(card_id)}」"]) + elif card.content is None: + out["reasons"] = [f"{side}的口径卡「{self.title(card_id)}」{card.problem}，无法对比"] + else: + out.update(classification=cls, reasons=[f"{side}的口径卡「{self.title(card_id)}」里没有这个指标"]) + return out + assert bm is not None and tm is not None and b is not None and t is not None + caliber = _caliber_reasons(bm, tm, b.content or {}, t.content or {}) + self._delta(out, bm, tm) + lineage = isinstance(bm.get("inputs"), list) and isinstance(tm.get("inputs"), list) + if caliber: + out.update(classification=CALIBER_AND_DATA, reasons=caliber) + out["decomposition"], out["decomposition_note"] = self._decompose(bm, tm, b, t, bp, tp) + elif not lineage: + quiet = self.template["same_graph"] and all(self.shape(x, y).structure == "same" for x, y in self.pairs) + same = recompute._same(bm.get("value"), tm.get("value")) + out.update(classification=UNCHANGED if quiet and same else INCOMPARABLE, + reasons=[] if quiet and same else [NO_LINEAGE]) + else: + paths = list(dict.fromkeys([*tp.inputs, *bp.inputs])) + changes = [self.provenance(p, bp.inputs.get(p), tp.inputs.get(p)) for p in paths] + worst = _worst(changes) + reasons = list(dict.fromkeys(c.reason for c in changes if c.kind == worst.kind and c.reason)) + if worst.kind in ("sql", "template"): + out.update(classification=SQL_CHANGED, reasons=reasons, decomposition_note=SQL_NOT_SPLIT) + elif worst.kind == "unknown": + out.update(classification=INCOMPARABLE, reasons=reasons) + elif worst.kind == "params": + out.update(classification=PARAMS_ONLY, reasons=reasons) + elif recompute._same(bm.get("value"), tm.get("value")): + out.update(classification=UNCHANGED) + else: + linked = all(inp.snapshot for inp in tp.inputs.values()) and bool(tp.inputs) + out.update(classification=DATA_ONLY, reasons=[DATA_FROM_QUERIES if linked else DATA_FROM_UPSTREAM]) + for side, metric in (("基准运行", bm), ("这次运行", tm)): + if metric.get("status") == "missing_input": + out["reasons"].append(f"{side}这个指标缺少输入，没有值") + if out["classification"] in _ATTRIBUTED and bp is not None and tp is not None: + out["attribution"] = self._attribution(bp, tp) + if any(inp.masked for plan in (bp, tp) if plan is not None for inp in plan.inputs.values()): + self.masked_seen = True + return out + + def _delta(self, out: dict[str, Any], bm: dict[str, Any], tm: dict[str, Any]) -> None: + """总变化：两边都是数、数值的含义相同（单位一样、比率与否一样）才相减，按十进制精确算。""" + b, t = _exact(bm.get("value")), _exact(tm.get("value")) + if b is None or t is None or _scale(bm) != _scale(tm): + return + whole = all(isinstance(m.get("value"), int) for m in (bm, tm)) + delta = _plain(t - b, whole=whole) + meta = {**tm, "decimals": _places(bm, tm)} + out.update(delta=delta, delta_rendered=_signed(meta, delta) if delta != 0 else None, + delta_pct=float(round((t - b) / abs(b) * 100, 2)) if b != 0 else None) + + def _decompose(self, bm: dict[str, Any], tm: dict[str, Any], b: _Card, t: _Card, bp: Any, + tp: Any) -> tuple[dict[str, Any] | None, str | None]: + """严格拆分：(拆分, None) 或 (None, 为什么没法拆)。""" + old, new = _exact(bm.get("value")), _exact(tm.get("value")) + if old is None or new is None: + return None, NOT_NUMBERS + if _scale(bm) != _scale(tm): + return None, "单位或比率的写法不同，两边的数值不能直接相减，不做拆分" + try: + leaves = leaf_refs(parse_expression(str(bm.get("expression") or ""))[0]) + except ExpressionError as e: + return None, f"旧口径的表达式无法解析（{e}），无法严格拆分" + if bp is None or not isinstance(bm.get("inputs"), list): + return None, "基准运行的口径卡没有记录输入（生成于系统升级前），无法确认取数是否有变化，无法严格拆分" + missing = [p for p in leaves if p not in t.inputs] + if missing: + more = f"等 {len(missing)} 个输入" if len(missing) > 1 else "" + return None, f"旧口径用到的「{missing[0]}」{more}在这次运行的口径卡里没有记录，无法严格拆分" + used = [t.inputs[p] for p in leaves] + plans = [p for p in (bp, tp) if p is not None] + if any(inp.masked for inp in used) or any(inp.masked for plan in plans for inp in plan.inputs.values()): + return None, SPLIT_MASKED + for path in leaves: + change = self.provenance(path, bp.inputs.get(path), t.inputs[path]) + if change.kind in ("sql", "template", "unknown"): + return None, f"旧口径用到的输入取数也有变化，无法严格拆分：{change.reason}" + # f_old(x_new)：旧表达式配这次的输入，按推演重算的规则还原、求值，取整按旧口径卡的小数位 + synthetic = {"id": bm.get("id"), "expression": bm.get("expression"), "decimals": bm.get("decimals"), + "inputs": [inp.item for inp in used], "substituted": None, "value": None} + assert t.sources is not None + plan = recompute._plan(synthetic, recompute._Store(t.sources)) + if plan.problem: + return None, f"无法严格拆分：{plan.problem}" + for path in plan.tops: + if problem := plan.inputs[path].problem: + return None, f"无法严格拆分：这次运行的输入「{path}」无法还原（{problem}）" + value, reason = recompute._evaluate(plan, {p: inp.value for p, inp in plan.inputs.items()}) + if reason is not None: + return None, f"按旧口径在这次的数据上无法计算（{reason}），无法严格拆分" + middle = _exact(value) + if middle is None: + return None, "按旧口径在这次的数据上算出的不是有限的数值，无法严格拆分" + whole = all(isinstance(v, int) and not isinstance(v, bool) for v in (bm.get("value"), tm.get("value"), value)) + caliber, data = _plain(new - middle, whole=whole), _plain(middle - old, whole=whole) + meta = {**tm, "decimals": _places(bm, tm)} + return {"caliber_effect": caliber, "data_effect": data, "caliber_rendered": _signed(meta, caliber), + "data_rendered": _signed(meta, data)}, None + + def _inputs(self, bp: Any, tp: Any) -> list[dict[str, Any]]: + """两边的输入按路径对齐：值（遮罩列、核对不了遮没遮的不给），各自落在的查询快照。""" + b_in = bp.inputs if bp is not None else {} + t_in = tp.inputs if tp is not None else {} + out = [] + for path in dict.fromkeys([*t_in, *b_in]): + b, t = b_in.get(path), t_in.get(path) + entry: dict[str, Any] = {"path": path, "base": _input_value(b), "target": _input_value(t)} + if any(inp is not None and inp.masked for inp in (b, t)): + entry["masked"] = True + if b is not None and b.snapshot: + entry["base_artifact"] = b.snapshot + if t is not None and t.snapshot: + entry["target_artifact"] = t.snapshot + out.append(entry) + return out + + # ---------------------------------------------------------------- 维度归因 + + def _attribution(self, bp: Any, tp: Any) -> dict[str, Any] | None: + """挑一个落在查询快照上、两边都变了的输入做维度归因：先找给得出占比的，没有就取第一个列得出行的，再没有 + 就取第一个说明（截断、度量列遮罩……）。一个都挑不出来返回 None。""" + found: list[dict[str, Any]] = [] + for path, t in tp.inputs.items(): + b = bp.inputs.get(path) + if b is None or not (b.snapshot and t.snapshot) or recompute._same(b.value, t.value): + continue + if self.provenance(path, b, t).kind not in ("same", "params"): + continue + result = self._attribute(b, t) + if result is None: + continue + if result["rows"] and all(r["share"] is not None for r in result["rows"]): + return result + found.append(result) + return next((r for r in found if r["rows"]), found[0] if found else None) + + def _attribute(self, b: Any, t: Any) -> dict[str, Any] | None: + """一个输入的维度归因：{columns, measure, rows, total_delta, note}；归不了因的照实写在 note 里（rows 为空）。 + 快照里没有可以当维度的列（只有一行汇总这种）返回 None：没什么可归的。""" + measure = _locator(b).get("column") + if not isinstance(measure, str) or _locator(t).get("column") != measure: + return None + sb, st = self._snap(self.base_queries, b), self._snap(self.target_queries, t) + empty = {"columns": [], "measure": measure, "rows": [], "total_delta": None} + if sb is None or st is None: + return {**empty, "note": UNREADABLE_SNAPSHOT} + for side, snap in (("基准运行", sb), ("这次运行", st)): + if snap.get("truncated"): + # 截断的结果只是前几行：各键的和不是完整的和，按它归因会把没取回的行的变化算到别的键头上 + return {**empty, "note": TRUNCATED.format(side=side, rows=len(snap["rows"]))} + hidden = (self.base.masked.get(b.snapshot) or frozenset()) | (self.target.masked.get(t.snapshot) or frozenset()) + if any(str(c).lower() in hidden for c in [*(sb.get("columns") or []), *(st.get("columns") or [])]): + self.masked_seen = True + if measure.lower() in hidden: + return {**empty, "note": MASKED_MEASURE.format(column=measure)} + rb, rt = _records(sb, b), _records(st, t) + if rb is None or rt is None: + return None + values = (_input_total(b), _input_total(t)) + notes: list[str] = [] + if b.row is not None and t.row is not None and not _adds_up(rb, rt, measure, values): + # 取的是合计行那一格：其余各行之和等于它（两边都等）时按其余各行归因，合计行不列进去 + kb, kt = [x for x in rb if x[0] != b.row], [x for x in rt if x[0] != t.row] + if kb and kt and len(kb) < len(rb) and len(kt) < len(rt) and _adds_up(kb, kt, measure, values): + rb, rt = kb, kt + notes.append(TOTAL_ROW.format(row=t.row + 1)) + dims, apart = _dimensions(sb, st, rb, rt, measure, hidden) + if not dims: + return {**empty, "note": "；".join(apart)} if apart else None + notes += apart + groups_b, groups_t = _group(rb, dims, measure), _group(rt, dims, measure) + if groups_b is None or groups_t is None: + return {**empty, "note": NOT_A_MEASURE.format(column=measure)} + total_b, total_t = sum(groups_b.values(), Decimal(0)), sum(groups_t.values(), Decimal(0)) + total = total_t - total_b + # 占比只在这份结果正好是这个输入的拆分时才有意义：各键之和等于输入的值，两边都等 + share = _equal(values[0], total_b) and _equal(values[1], total_t) + if not share: + notes.append(NO_SHARE) + elif total == 0: + notes.append(ZERO_TOTAL) + rows = [] + for key in dict.fromkeys([*groups_b, *groups_t]): + vb, vt = groups_b.get(key), groups_t.get(key) + delta = (vt or Decimal(0)) - (vb or Decimal(0)) + rows.append((abs(delta), {"key": list(key), "base": None if vb is None else _plain(vb, whole=False), + "target": None if vt is None else _plain(vt, whole=False), + "delta": _plain(delta, whole=False), + "share": float(delta / total) if share and total != 0 else None})) + rows.sort(key=lambda r: -r[0]) # 稳定排序：变化一样大的按先出现的 + if len(rows) > TOP_KEYS: + notes.append(MORE_KEYS.format(count=len(rows), top=TOP_KEYS)) + return {"columns": dims, "measure": measure, "rows": [r for _, r in rows[:TOP_KEYS]], + "total_delta": _plain(total, whole=False), "note": "；".join(notes) or None} + + def _snap(self, queries: list[_Query], inp: Any) -> dict[str, Any] | None: + q = self._side_query(queries, inp) + return q.snap if q is not None else None + + +# -------------------------------------------------------------------------- +# 指标、输入的几样小事 +# -------------------------------------------------------------------------- + + +def _value(metric: dict[str, Any] | None) -> Any: + """指标的值：有限的数才给，别的（空值、文本、溢出成无穷大的）给 None，写法看 rendered。""" + value = (metric or {}).get("value") + return value if recompute._number(value) else None + + +def _rendered(metric: dict[str, Any] | None) -> str | None: + """卡上记的写法（报告里就是这几个字）；没记的按卡上的格式渲染。""" + if metric is None: + return None + rendered = metric.get("rendered") + return rendered if isinstance(rendered, str) else recompute._render(metric, metric.get("value")) + + +def _caliber_reasons(bm: dict[str, Any], tm: dict[str, Any], bc: dict[str, Any], tc: dict[str, Any]) -> list[str]: + """口径变了的地方，一处一句；没变返回空列表。""" + out = [] + if _expression(bm) != _expression(tm): + out.append(f"表达式由「{_shown(bm.get('expression'))}」改为「{_shown(tm.get('expression'))}」") + before, after = str(bc.get("caliber_version") or ""), str(tc.get("caliber_version") or "") + if before != after: + out.append(f"口径版本由 {before or '未设置'} 改为 {after or '未设置'}") + if bm.get("decimals") != tm.get("decimals"): + def places(d: Any) -> str: + return "未设置" if d is None else str(d) + out.append(f"小数位由 {places(bm.get('decimals'))} 改为 {places(tm.get('decimals'))}") + if (bm.get("unit") or "") != (tm.get("unit") or ""): + out.append(f"单位由「{bm.get('unit') or '无'}」改为「{tm.get('unit') or '无'}」") + fb, ft = bm.get("format") or DEFAULT_FORMAT, tm.get("format") or DEFAULT_FORMAT + if fb != ft: + out.append(f"显示格式由「{_FORMAT_LABEL.get(fb, fb)}」改为「{_FORMAT_LABEL.get(ft, ft)}」") + return out + + +def _input_value(inp: Any) -> Any: + """输入在对比里给人看的值：遮罩列、落在读不出来的快照上的不给；容器写卡上记的大概。""" + if inp is None or inp.masked or (inp.unreadable and inp.shape == "scalar"): + return None + if inp.shape == "container": + return inp.item.get("value") + return recompute._finite(inp.item.get("value")) + + +def _locator(inp: Any) -> dict[str, Any]: + locator = inp.item.get("locator") + return locator if isinstance(locator, dict) else {} + + +def _records(snap: dict[str, Any], inp: Any) -> list[tuple[int, dict[str, Any]]] | None: + """归因用到的行：(行号, {列: 值})。数组字段只用它那一段行，单格用整份结果。""" + columns = [str(c) for c in snap.get("columns") or []] + rows = snap.get("rows") or [] + first, last = (inp.rows if inp.rows is not None else (0, len(rows) - 1)) + out = [] + for i in range(first, min(last, len(rows) - 1) + 1): + row = rows[i] + if isinstance(row, dict): + out.append((i, {str(k): v for k, v in row.items()})) + elif isinstance(row, (list, tuple)) and len(row) == len(columns): + out.append((i, dict(zip(columns, row)))) + else: + return None + return out + + +def _kind(snap: dict[str, Any], column: str) -> str | None: + types = snap.get("column_types") + kind = types.get(column) if isinstance(types, dict) else None + return kind if isinstance(kind, str) else None + + +def _numeric(values: Iterable[Any], kind: str | None) -> bool: + """这一列是数值列：查询时记的类型是 number，或者没记类型、值（数字文本也算）全是数。""" + if kind is not None: + return kind == "number" + present = [v for v in values if v is not None] + return bool(present) and all(is_number(cell_value(v)) for v in present) + + +def _dimensions(sb: dict[str, Any], st: dict[str, Any], rb: list[tuple[int, dict[str, Any]]], + rt: list[tuple[int, dict[str, Any]]], measure: str, + hidden: frozenset[str] | set[str]) -> tuple[list[str] | None, list[str]]: + """维度列：(列名列表, 说明)。两边都有、不是数值、不是日期类型、没遮罩的列；两边各自都只有一个值的不算（分不出 + 维度）；两边没有一个值相同的不拿来配对（日期、期次这类每期都不同的列），写进说明。一个候选都没有返回 None。""" + common = [c for c in (str(c) for c in sb.get("columns") or []) if c in {str(x) for x in st.get("columns") or []}] + candidates = [] + for column in common: + if column == measure or column.lower() in hidden: + continue + kinds = (_kind(sb, column), _kind(st, column)) + if any(k in _TEMPORAL for k in kinds): + continue + vb, vt = [r.get(column) for _, r in rb], [r.get(column) for _, r in rt] + if _numeric(vb, kinds[0]) or _numeric(vt, kinds[1]): + continue + candidates.append((column, vb, vt)) + if not candidates: + return None, [] + dims, apart = [], [] + for column, vb, vt in candidates: + seen_b, seen_t = {_hashable(v) for v in vb}, {_hashable(v) for v in vt} + if len(seen_b) <= 1 and len(seen_t) <= 1: + continue + if seen_b and seen_t and not seen_b & seen_t: + apart.append(column) + continue + dims.append(column) + return dims, ([NO_OVERLAP.format(columns="、".join(apart))] if apart else []) + + +def _hashable(value: Any) -> Any: + return value if isinstance(value, (str, int, float, bool)) or value is None else repr(value) + + +def _measure(raw: Any) -> Decimal | None: + """度量列的一格，十进制：空值按 0（SQL 的 SUM 也不计空值），数字文本按数，别的不是数返回 None。""" + return Decimal(0) if raw is None else _exact(cell_value(raw)) + + +def _group(rows: list[tuple[int, dict[str, Any]]], dims: list[str], + measure: str) -> dict[tuple[Any, ...], Decimal] | None: + """按维度键汇总度量：{键: 和}（去掉了期次列之后同一个键可能有几行）。有不是数值的格返回 None。""" + out: dict[tuple[Any, ...], Decimal] = {} + for _, row in rows: + number = _measure(row.get(measure)) + if number is None: + return None + key = tuple(_hashable(row.get(d)) for d in dims) + out[key] = out.get(key, Decimal(0)) + number + return out + + +def _input_total(inp: Any) -> Decimal | None: + """输入的值（数组字段是各项之和），十进制。不是数的给 None：这时不给占比。""" + value = inp.value + if isinstance(value, list): + parts = [_exact(cell_value(v)) for v in value if v is not None] + return sum(parts, Decimal(0)) if all(p is not None for p in parts) else None # type: ignore[arg-type] + return _exact(cell_value(value)) + + +def _adds_up(rb: list[tuple[int, dict[str, Any]]], rt: list[tuple[int, dict[str, Any]]], measure: str, + values: tuple[Decimal | None, Decimal | None]) -> bool: + """度量列之和等于输入的值（两边都等）。""" + for rows, value in ((rb, values[0]), (rt, values[1])): + parts = [_measure(row.get(measure)) for _, row in rows] + if any(p is None for p in parts) or not _equal(value, sum(parts, Decimal(0))): # type: ignore[arg-type] + return False + return True + + +def _equal(value: Decimal | None, total: Decimal) -> bool: + """输入的值和度量之和相等：十进制逐位比；浮点算出来的值差在最后几位的也算（相对误差 1e-9 以内）。""" + if value is None: + return False + if value == total: + return True + scale = max(abs(value), abs(total)) + return scale != 0 and abs(value - total) <= scale * Decimal("1e-9") diff --git a/backend/tests/fixtures/compare_runs.py b/backend/tests/fixtures/compare_runs.py new file mode 100644 index 00000000..1d8bb6e2 --- /dev/null +++ b/backend/tests/fixtures/compare_runs.py @@ -0,0 +1,211 @@ +"""运行对比的虚构运行（tests/test_compare_engine.py、tests/test_compare_api.py 共用）。 + +两次周报运行照真实运行的事件形状手写（区域、金额、日期都是编的）：取数节点 fetch 查一条按区域汇总的 SQL，Agent 节点 +bot 自己再查一次、把各区域的金额作为数组字段 amounts 交给口径卡，口径卡算几个指标。工件按内容寻址放在字典里， +接口测试把它们原样落进工件库、事件落进运行记录。 +""" +from __future__ import annotations + +import json + +from app.core.artifact_store import canonical_json, content_hash +from app.engine import compare, recompute +from app.engine.evidence import render_number +from app.engine.expressions import parse_expression, substitute + +TOOL = "db_query__shop" +WEEK36 = ("2026-08-31", "2026-09-06") +WEEK37 = ("2026-09-07", "2026-09-13") +COLUMNS = ["region", "gmv", "refund"] +TYPES = {"region": "text", "gmv": "number", "refund": "number"} +BASE_ROWS = [["东区", 1200.5, 100.0], ["南区", 800.0, 50.0], ["西区", 600.0, 0.0]] +NEXT_ROWS = [["东区", 1500.5, 120.0], ["南区", 700.0, 40.0], ["西区", 600.0, 10.0]] +EAST = "cell(nodes.fetch, 0, 'gmv')" +REFUND = "cell(nodes.fetch, 0, 'refund')" + + +def region_sql(days=WEEK36, extra=""): + return ("SELECT region, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders " + f"WHERE day BETWEEN '{days[0]}' AND '{days[1]}'{extra} GROUP BY region ORDER BY region") + + +def node(nid, ntype, label, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": label, "config": config}} + + +def chain(*nodes): + return {"nodes": list(nodes), "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +def weekly_graph(sql=None, *, between=(), card_label="周报口径卡"): + return chain( + node("start", "input", "开始"), + node("fetch", "tool", "区域取数", tool=TOOL, args={"sql": sql or region_sql()}), + node("bot", "agent", "区域分析", prompt="按区域列出金额", tools=[TOOL], cite_fields=True), + *between, + node("card", "metrics", card_label, caliber="周报口径", caliber_version="v1"), + node("write", "report", "写周报"), + node("out", "output", "成果"), + ) + + +class Fake: + """一次虚构的运行：工件按内容寻址放在字典里，事件照真实运行的形状写（tool.start / tool.end / node.finished + 的证据台账、口径卡的 metric_set 条目、节点产出工件）。""" + + def __init__(self, run_id, *, graph=None, version=3, inputs=None, masked=None, source="shop"): + self.id = run_id + self.source = source + self.graph = graph or weekly_graph() + self.version = version + self.inputs = inputs or {} + self.masked = masked or {} + self.store: dict[str, object] = {} + self.tampered: set[str] = set() + self.events: list[tuple[int, str, str | None, dict]] = [] + self.cards: list[tuple[str, str, dict]] = [] + self.queries: list[str] = [] + self.elapsed = 0 + self.add("run.started", None, {"nodes": len(self.graph["nodes"])}) + + def put(self, obj): + text = canonical_json(obj) + artifact = content_hash(text) + self.store[artifact] = json.loads(text) + return artifact + + def load(self, artifact): + if artifact in self.tampered: + raise ValueError("工件内容与哈希不符，疑似被篡改") + return self.store.get(artifact) + + def add(self, etype, node_id, data): + self.events.append((len(self.events) + 1, etype, node_id, data)) + + def snapshot(self, sql, columns, rows, *, truncated=False, types=None, masks=None): + self.elapsed += 1 # 真实快照带着耗时：同一条 SQL 查两次也是两件工件 + snap = {"columns": columns, "rows": rows, "row_count": len(rows), "truncated": truncated, + "elapsed_ms": self.elapsed, "sql": sql, "source": self.source, + **({"column_types": types} if types else {}), **({"mask_columns": masks} if masks else {})} + artifact = self.put(snap) + self.queries.append(artifact) + return artifact + + def entry(self, node_id, artifact, columns, rows, *, tool=TOOL, call_id=None, truncated=False): + return {"kind": "query", "node_id": node_id, "exec": 1, "artifact": artifact, "tool": tool, + "source": self.source, + "columns": columns, "rows": len(rows), "truncated": truncated, + **({"call_id": call_id} if call_id else {})} + + def tool(self, node_id, sql, columns=COLUMNS, rows=BASE_ROWS, *, tool=TOOL, truncated=False, types=TYPES, + masks=None): + """调用工具节点查一次库。masks 是查询当时数据源设的遮罩（记进快照的 mask_columns）。""" + artifact = self.snapshot(sql, columns, rows, truncated=truncated, types=types, masks=masks) + output = self.put({"columns": columns, "rows": rows, "sql": sql, "source": self.source, "artifact": artifact}) + self.add("node.started", node_id, {"node_type": "tool"}) + self.add("tool.start", node_id, {"tool": tool, "args": {"sql": sql}}) + self.add("tool.end", node_id, {"tool": tool, "duration_ms": 3, "query_artifact": artifact}) + self.add("node.finished", node_id, {"duration_ms": 4, "attempt": 1, "artifact": output, "evidence": [ + self.entry(node_id, artifact, columns, rows, tool=tool, truncated=truncated)]}) + return artifact + + def agent(self, node_id, sql, columns=COLUMNS, rows=BASE_ROWS, *, types=TYPES, truncated=False, masks=None, + verify=()): + """Agent 节点自己查一次库，按出处交回数组字段 amounts（快照 gmv 列的全部行）。verify 是同一节点里核查类的 + 查询（role: verify，不进台账）。""" + artifact = self.snapshot(sql, columns, rows, truncated=truncated, types=types, masks=masks) + last = len(rows) - 1 + gmv = columns.index("gmv") + data = {"amounts": [r[gmv] for r in rows]} + evidence = {"amounts": {"ref": f"Q1.r0-{last}.gmv", "call": "Q1", "artifact": artifact, + "locator": {"rows": [0, last], "column": "gmv"}, "status": "verified"}} + output = self.put({"text": "按区域列出了金额", "data": data, "data_evidence": evidence}) + self.add("node.started", node_id, {"node_type": "agent"}) + self.add("tool.start", node_id, {"tool": TOOL, "args": {"sql": sql}, "call_id": "c1"}) + self.add("tool.end", node_id, {"tool": TOOL, "call_id": "c1", "duration_ms": 5, "query_artifact": artifact}) + for i, extra in enumerate(verify): + check = self.snapshot(extra, ["n"], [[1]]) + self.add("tool.end", node_id, {"tool": TOOL, "call_id": f"v{i}", "role": "verify", "query_artifact": check}) + self.add("node.finished", node_id, {"duration_ms": 9, "attempt": 1, "artifact": output, "evidence": [ + self.entry(node_id, artifact, columns, rows, call_id="c1", truncated=truncated)]}) + return artifact + + def code(self, node_id, value, var): + """沙箱代码节点算出一个数写进变量：它不查库，对不上具体的查询。""" + output = self.put({var: value}) + self.add("node.started", node_id, {"node_type": "code"}) + self.add("node.finished", node_id, {"duration_ms": 2, "attempt": 1, "artifact": output}) + + def card(self, metrics, *, node_id="card", caliber="周报口径", version="v1", runs=1): + card = {"kind": "metric_set", "caliber": caliber, "caliber_version": version, "metrics": metrics} + artifact = self.put(card) + for _ in range(runs): + self.add("node.started", node_id, {"node_type": "metrics"}) + self.add("node.finished", node_id, {"duration_ms": 1, "attempt": 1, "artifact": self.put({**card, "text": "…"}), + "evidence": [{"kind": "metric_set", "node_id": node_id, "exec": 1, + "artifact": artifact, "caliber": caliber, + "version": version}]}) + self.cards.append((node_id, artifact, card)) + return artifact + + def side(self): + def sources(started): + outputs = {} + for seq, etype, nid, data in self.events: + if seq < started and etype == "node.finished" and data.get("artifact"): + outputs[nid] = data["artifact"] + return recompute.Sources(load=self.load, queries=tuple(self.queries), outputs=outputs, masked=self.masked) + + run = {"id": self.id, "workflow_name": "区域周报", "version": self.version, "run_class": "formal", + "started_at": "2026-09-07T01:00:00+00:00", "finished_at": "2026-09-07T01:02:00+00:00", + "schedule_id": None} + return compare.Side(run=run, graph=self.graph, events=self.events, cards=self.cards, load=self.load, + masked=self.masked, sources=sources, dialects={self.source: "sqlite"}, inputs=self.inputs) + + +def cell_in(row, column, value, artifact, node_id="fetch"): + return {"path": f"cell(nodes.{node_id}, {row}, '{column}')", "value": value, "node_id": node_id, "via": "tool_cell", + "locator": {"row": row, "column": column}, "artifact": artifact, + "eid": f"ev:cell:{artifact[:8]}:{row}:{column}", "status": "verified"} + + +def amounts_in(artifact, rows): + last = len(rows) - 1 + return {"path": "vars.kpi.amounts", "value": f"[{len(rows)} 项]", "node_id": "bot", "via": "agent_field", + "field": "amounts", "ref": f"Q1.r0-{last}.gmv", "artifact": artifact, + "locator": {"rows": [0, last], "column": "gmv"}, "eid": f"ev:rows:{artifact[:8]}", "status": "verified"} + + +def scalar_in(path, value, node_id, via): + return {"path": path, "value": value, "node_id": node_id, "via": via, "field": path.split(".", 2)[-1], + "status": "ok" if value is not None else "missing"} + + +def metric(mid, expression, value, inputs, *, name=None, unit="元", decimals=1, fmt="thousands"): + tree = parse_expression(expression)[0] + scalars = {i["path"]: i["value"] for i in inputs if not str(i["value"]).endswith(" 项]")} + return {"id": mid, "name": name or mid, "unit": unit, "value": value, "expression": expression, + "decimals": decimals, "format": fmt, + "rendered": render_number(value, unit=unit, decimals=decimals, fmt=fmt), + "inputs": inputs, "substituted": substitute(tree, scalars), "recompute_ok": True, + "status": "ok" if value is not None else "missing_input"} + + +def week(run_id, rows=BASE_ROWS, *, sql=None, metrics=None, graph=None, version="v1", bot_rows=None, **kw): + """一次周报运行：fetch 和 bot 各查一次，口径卡算 east（东区金额）和 total（各区域合计）。 + metrics(fetch, bot, rows) 返回口径卡的指标，不给就是这两个。""" + run = Fake(run_id, graph=graph or weekly_graph(sql), **kw) + sql = sql or region_sql() + fetch = run.tool("fetch", sql, rows=rows) + bot_rows = bot_rows if bot_rows is not None else rows + bot = run.agent("bot", sql, rows=bot_rows) + if metrics is None: + metrics = default_metrics + run.card(metrics(fetch, bot, rows, bot_rows), version=version) + return run + + +def default_metrics(fetch, bot, rows, bot_rows): + total = sum(r[1] for r in bot_rows) + return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], name="东区金额"), + metric("total", "sum(vars.kpi.amounts)", total, [amounts_in(bot, bot_rows)], name="各区域合计")] diff --git a/backend/tests/test_compare_api.py b/backend/tests/test_compare_api.py new file mode 100644 index 00000000..7e92273a --- /dev/null +++ b/backend/tests/test_compare_api.py @@ -0,0 +1,387 @@ +"""运行对比的接口：GET /api/runs/{run_id}/compare/base 和 GET /api/runs/{run_id}/compare?base=。 + +两次周报运行用 tests/fixtures/compare_runs.py 的虚构运行：工件落进工件库，事件落进运行记录，按 runner 的算法封存 +（不连库、不调模型）。另有一组真跑：虚构的 SQLite 库在两次运行之间改了数，再一次代进去的日期参数不同。守的是： +- 两次都结束、封存且核对一致才比：没封存、还在运行、封存后事件被改过都回 409，说清是哪一次；不是同一个工作流、 + 同一次运行回 422；运行不存在 404 +- 对比只读：两次运行的 last_seq、事件、工件引用、工件文件一样不多 +- 遮罩列（查询当时记下的、数据源现在设的）的值不出现在结果里 +- 默认基准：同一定时任务优先，否则同一工作流、同一运行类别；只挑更早结束、封存过的终态运行 +- 真跑：同一条 SQL 数据变了是 data_only；只是代进去的日期不同是 params_only，图一样 +""" +from __future__ import annotations + +import asyncio +import json +import sqlite3 +from datetime import datetime, timedelta, timezone + +import pytest +from httpx import ASGITransport, AsyncClient +from sqlalchemy import func, select + +from app.core.artifact_store import manifest_hash, put_json +from app.core.config import settings +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import Artifact, DataSource, Run, RunEvent, Workflow +from app.engine.runner import run_manager, verify_manifest +from app.main import app +from tests.fixtures.compare_runs import ( + BASE_ROWS, + EAST, + NEXT_ROWS, + Fake, + amounts_in, + cell_in, + chain, + metric, + node, + week, +) +from tests.fixtures.sources import drop_source + +SOURCE = "cmp_shop" +T0 = datetime(2026, 9, 7, 1, 0, tzinfo=timezone.utc) + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +async def new_workflow(name="区域周报") -> str: + async with SessionLocal() as session: + workflow = Workflow(name=name) + session.add(workflow) + await session.commit() + return workflow.id + + +async def persist(fake: Fake, workflow_id: str | None, *, finished=T0, status="succeeded", run_class="formal", + schedule_id=None, seal=True) -> str: + """把一次虚构运行落库：工件进工件库（内容寻址，id 和虚构时一样），事件进运行记录，按 runner 的算法封存。""" + for content in fake.store.values(): + await put_json(content, kind="compare_fixture") + async with SessionLocal() as session: + run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, graph=fake.graph, + input=dict(fake.inputs), run_class=run_class, version=fake.version, schedule_id=schedule_id, + started_at=finished - timedelta(minutes=2), finished_at=finished if status != "running" else None, + agent_limits={}) + session.add(run) + await session.flush() + rows = [] + for seq, etype, node_id, data in fake.events: + session.add(RunEvent(run_id=run.id, seq=seq, type=etype, node_id=node_id, ts=float(seq), data=data)) + rows.append((seq, etype, node_id, data)) + run.last_seq = len(rows) + if seal: + run.manifest_hash, run.manifest_seq = manifest_hash(rows), rows[-1][0] + await session.commit() + return run.id + + +async def pair(base=None, target=None, **kw) -> tuple[str, str, str]: + """(工作流, 基准运行, 这次运行)：同一个工作流的两次周报，这次的数变了。""" + workflow = await new_workflow() + base_id = await persist(base or week("r1", source=SOURCE), workflow, finished=T0) + target_id = await persist(target or week("r2", NEXT_ROWS, source=SOURCE), workflow, + finished=T0 + timedelta(days=7), **kw) + return workflow, base_id, target_id + + +async def footprint(*run_ids: str) -> tuple: + """几次运行留在库里、盘上的痕迹：各自的 last_seq、封存、事件数，加上全部工件引用行数和工件文件数。""" + async with SessionLocal() as session: + runs = [] + for run_id in run_ids: + run = await session.get(Run, run_id) + count = await session.scalar(select(func.count()).select_from(RunEvent).where(RunEvent.run_id == run_id)) + runs.append((run.last_seq, run.manifest_hash, run.manifest_seq, run.status, count)) + artifacts = await session.scalar(select(func.count()).select_from(Artifact)) + files = sum(1 for _ in (settings.data_dir / "artifacts").rglob("*.json")) + return runs, artifacts, files + + +async def get(client, path: str, status: int = 200) -> dict: + resp = await client.get(path) + assert resp.status_code == status, resp.text + return resp.json() + + +# -------------------------------------------------------------------------- +# 对比 +# -------------------------------------------------------------------------- + + +async def test_compares_two_sealed_runs(client): + _, base_id, target_id = await pair() + body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}") + assert body["target"]["id"] == target_id and body["base"]["id"] == base_id + assert set(body["target"]) == {"id", "workflow_name", "version", "run_class", "started_at", "finished_at", + "schedule_id"} + assert body["target"]["finished_at"].endswith("+00:00") and body["base"]["run_class"] == "formal" + assert body["summary"]["data_only"] == 2 and sum(body["summary"].values()) == 2 + east = next(m for m in body["metrics"] if m["metric_id"] == "east") + assert (east["classification"], east["base"], east["target"], east["delta_rendered"]) == ( + "data_only", 1200.5, 1500.5, "+300.0元") + total = next(m for m in body["metrics"] if m["metric_id"] == "total") + assert [r["share"] for r in total["attribution"]["rows"]] == [1.5, -0.5, 0.0] + assert [(q["node_id"], q["structure"]) for q in body["queries"]] == [("fetch", "same"), ("bot", "same")] + assert body["template"] == {"base_version": 3, "target_version": 3, "same_graph": True, "nodes": []} + assert body["notes"] == [] + + +async def test_comparing_writes_nothing(client): + _, base_id, target_id = await pair() + before = await footprint(base_id, target_id) + await get(client, f"/api/runs/{target_id}/compare?base={base_id}") + await get(client, f"/api/runs/{target_id}/compare/base") + assert await footprint(base_id, target_id) == before + for run_id in (base_id, target_id): + assert (await verify_manifest(run_id))["ok"] is True + + +async def test_unsealed_runs_are_refused(client): + workflow = await new_workflow() + base_id = await persist(week("r1", source=SOURCE), workflow) + target_id = await persist(week("r2", NEXT_ROWS, source=SOURCE), workflow, finished=T0 + timedelta(days=7), + seal=False) + body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409) + assert body["code"] == "compare_unsealed" and body["detail"].startswith("这次运行") + body = await get(client, f"/api/runs/{base_id}/compare?base={target_id}", 409) + assert body["code"] == "compare_unsealed" and body["detail"].startswith("作为基准的运行") + # 封存过、又被接着运行（还在运行）的：封存范围之后还会变，同样不比 + running = await persist(week("r3", NEXT_ROWS, source=SOURCE), workflow, status="running") + body = await get(client, f"/api/runs/{running}/compare?base={base_id}", 409) + assert body["code"] == "compare_unsealed" + + +async def test_a_broken_seal_is_refused(client): + _, base_id, target_id = await pair() + async with SessionLocal() as session: + event = (await session.execute(select(RunEvent).where( + RunEvent.run_id == base_id, RunEvent.type == "tool.end"))).scalars().first() + event.data = {**event.data, "duration_ms": 999} + await session.commit() + body = await get(client, f"/api/runs/{target_id}/compare?base={base_id}", 409) + assert body["code"] == "compare_seal_broken" and body["detail"].startswith("作为基准的运行") + assert "封存核对不一致" in body["detail"] + + +async def test_mismatched_runs_are_refused(client): + _, base_id, target_id = await pair() + other = await persist(week("r9", source=SOURCE), await new_workflow("另一个周报")) + body = await get(client, f"/api/runs/{target_id}/compare?base={other}", 422) + assert body["code"] == "compare_mismatch" and "同一个工作流" in body["detail"] + body = await get(client, f"/api/runs/{target_id}/compare?base={target_id}", 422) + assert body["code"] == "compare_mismatch" and "同一次运行" in body["detail"] + unsaved = await persist(week("r8", source=SOURCE), None) + body = await get(client, f"/api/runs/{unsaved}/compare?base={base_id}", 422) + assert body["code"] == "compare_mismatch" + body = await get(client, f"/api/runs/{target_id}/compare", 422) + assert body["code"] == "compare_bad_request" + await get(client, f"/api/runs/no-such-run/compare?base={base_id}", 404) + body = await get(client, f"/api/runs/{target_id}/compare?base=no-such-run", 404) + assert "基准" in body["detail"] + + +async def test_masked_columns_never_leave_the_server(client): + """快照里记着查询当时遮了 refund，数据源现在又遮了 manager：两样的值都不出现在结果里。""" + columns = ["region", "manager", "gmv", "refund"] + types = {"region": "text", "manager": "text", "gmv": "number", "refund": "number"} + sql = "SELECT region, manager, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders GROUP BY region, manager" + + def make(run_id, rows, expression, version): + run = Fake(run_id, source=SOURCE) + fetch = run.tool("fetch", sql, columns, rows, types=types, masks=["refund"]) + bot = run.agent("bot", sql, columns, rows, types=types, masks=["refund"]) + inputs = [cell_in(0, "gmv", rows[0][2], fetch)] + value = rows[0][2] + if "refund" in expression: + inputs.append(cell_in(0, "refund", rows[0][3], fetch)) + value = rows[0][2] - rows[0][3] + run.card([metric("net", expression, value, inputs), + metric("total", "sum(vars.kpi.amounts)", sum(r[2] for r in rows), [amounts_in(bot, rows)])], + version=version) + return run + + async with SessionLocal() as session: + session.add(DataSource(name=SOURCE, kind="sqlite", database=":memory:", options={"mask_columns": ["manager"]})) + await session.commit() + try: + base = make("r1", [["东区", "经理甲", 1200.5, 101.25], ["南区", "经理乙", 800.0, 52.75]], EAST, "v1") + target = make("r2", [["东区", "经理甲", 1500.5, 133.5], ["南区", "经理乙", 700.0, 47.25]], + f"{EAST} - cell(nodes.fetch, 0, 'refund')", "v2") + _, base_id, target_id = await pair(base, target) + resp = await client.get(f"/api/runs/{target_id}/compare?base={base_id}") + assert resp.status_code == 200, resp.text + for secret in ("经理甲", "经理乙", "101.25", "52.75", "133.5", "47.25"): + assert secret not in resp.text, secret + body = resp.json() + net = next(m for m in body["metrics"] if m["metric_id"] == "net") + refund = next(i for i in net["inputs"] if "refund" in i["path"]) + assert refund["masked"] is True and net["decomposition"] is None and "遮罩" in net["decomposition_note"] + total = next(m for m in body["metrics"] if m["metric_id"] == "total") + assert total["attribution"]["columns"] == ["region"] + finally: + await drop_source(name=SOURCE) + + +# -------------------------------------------------------------------------- +# 默认基准 +# -------------------------------------------------------------------------- + + +async def bare(workflow_id, finished, *, status="succeeded", run_class="formal", schedule_id=None, sealed=True) -> str: + """只为挑基准用的运行记录：挑基准只看运行记录上的几列。""" + async with SessionLocal() as session: + run = Run(workflow_id=workflow_id, workflow_name="区域周报", status=status, run_class=run_class, + schedule_id=schedule_id, started_at=finished - timedelta(minutes=2), + finished_at=finished if status in ("succeeded", "failed", "cancelled") else None, + manifest_hash="ab" * 32 if sealed else None, manifest_seq=1 if sealed else None) + session.add(run) + await session.commit() + return run.id + + +async def test_the_default_base_prefers_the_same_schedule(client): + workflow = await new_workflow() + day = timedelta(days=1) + scheduled = await bare(workflow, T0, schedule_id="sched-weekly") + manual = await bare(workflow, T0 + day) # 更近，但不是同一个定时任务 + current = await bare(workflow, T0 + 2 * day, schedule_id="sched-weekly") + body = await get(client, f"/api/runs/{current}/compare/base") + assert (body["base_run_id"], body["reason"]) == (scheduled, "same_schedule") + assert body["base"]["id"] == scheduled and body["base"]["schedule_id"] == "sched-weekly" + # 不是定时任务发起的：同一工作流、同一运行类别里最近的一次（定时的也算） + later = await bare(workflow, T0 + 3 * day) + body = await get(client, f"/api/runs/{later}/compare/base") + assert (body["base_run_id"], body["reason"]) == (current, "same_workflow") + # 定时任务的第一次：退回同一工作流、同一运行类别 + first = await bare(workflow, T0 + 4 * day, schedule_id="sched-daily") + body = await get(client, f"/api/runs/{first}/compare/base") + assert (body["base_run_id"], body["reason"]) == (later, "same_workflow") + assert manual != body["base_run_id"] + + +async def test_the_default_base_skips_what_cannot_be_compared(client): + workflow = await new_workflow() + day = timedelta(days=1) + good = await bare(workflow, T0) + await bare(workflow, T0 + day, sealed=False) # 没封存 + await bare(workflow, T0 + 2 * day, status="running") # 还在运行 + await bare(workflow, T0 + 3 * day, run_class="exploratory") # 别的运行类别 + await bare(await new_workflow("另一个周报"), T0 + 4 * day) # 别的工作流 + current = await bare(workflow, T0 + 5 * day) + await bare(workflow, T0 + 6 * day) # 比它晚结束 + body = await get(client, f"/api/runs/{current}/compare/base") + assert (body["base_run_id"], body["reason"]) == (good, "same_workflow") + # 失败、取消的终态运行也算（封存过） + failed = await bare(workflow, T0 + 5 * day - timedelta(hours=1), status="failed") + assert (await get(client, f"/api/runs/{current}/compare/base"))["base_run_id"] == failed + # 第一次运行、没保存的工作流：没有可比的 + body = await get(client, f"/api/runs/{good}/compare/base", 404) + assert body["code"] == "compare_no_base" and "上一次运行" in body["detail"] + await get(client, f"/api/runs/{await bare(None, T0)}/compare/base", 404) + await get(client, "/api/runs/no-such-run/compare/base", 404) + + +# -------------------------------------------------------------------------- +# 真跑：虚构的 SQLite 库，两次运行之间改了数、换了日期参数 +# -------------------------------------------------------------------------- + +E2E_SOURCE = "cmp_e2e" +E2E_SQL = ("SELECT region, SUM(amount) AS gmv FROM orders WHERE day BETWEEN '{{ input.start }}' AND '{{ input.end }}' " + "GROUP BY region ORDER BY region") +WEEK36_INPUT = {"start": "2026-08-31", "end": "2026-09-06"} +WEEK37_INPUT = {"start": "2026-09-07", "end": "2026-09-13"} + + +@pytest.fixture +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def orders(tmp_path): + path = tmp_path / "orders.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL);") + db.executemany("INSERT INTO orders (day, region, amount) VALUES (?, ?, ?)", [ + ("2026-09-01", "东区", 1200.5), ("2026-09-02", "南区", 800.0), ("2026-09-03", "西区", 600.0), + ("2026-09-08", "东区", 900.0), ("2026-09-09", "南区", 650.0), ("2026-09-10", "西区", 640.0)]) + db.commit() + db.close() + async with SessionLocal() as session: + row = DataSource(name=E2E_SOURCE, kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield path, source_id + await drop_source(source_id) + + +def e2e_graph(): + return chain( + node("start", "input", "开始"), + node("fetch", "tool", "区域取数", tool=f"db_query__{E2E_SOURCE}", args={"sql": E2E_SQL}), + node("card", "metrics", "周报口径卡", caliber="周报口径", caliber_version="v1", metrics=[ + {"id": "east", "name": "东区金额", "unit": "元", "decimals": 1, "format": "thousands", + "expression": "cell(nodes.fetch, 0, 'gmv')"}]), + node("out", "output", "成果", fields=[{"name": "指标", "value": "{{ nodes.card.text }}"}]), + ) + + +async def finished_run(workflow_id: str, inputs: dict) -> str: + run = await run_manager.start(graph=e2e_graph(), input_payload=inputs, workflow_id=workflow_id, + workflow_name="区域周报") + for _ in range(400): + await asyncio.sleep(0.05) + async with SessionLocal() as session: + row = await session.get(Run, run.id) + if row.status in ("succeeded", "failed") and row.manifest_seq is not None: + assert row.status == "succeeded", row.error + return run.id + raise AssertionError(f"运行 {run.id} 没有结束：{row.status}") + + +async def test_real_runs_split_data_and_parameter_changes(client, engine_up, orders): + path, source_id = orders + workflow = await new_workflow() + first = await finished_run(workflow, WEEK36_INPUT) + db = sqlite3.connect(path) + db.execute("UPDATE orders SET amount = 1500.5 WHERE day = '2026-09-01'") + db.commit() + db.close() + await engines.invalidate(source_id) + second = await finished_run(workflow, WEEK36_INPUT) + third = await finished_run(workflow, WEEK37_INPUT) + + body = await get(client, f"/api/runs/{second}/compare?base={first}") + [east] = body["metrics"] + assert (east["classification"], east["base"], east["target"], east["delta"]) == ("data_only", 1200.5, 1500.5, 300.0) + [fetch] = body["queries"] + assert (fetch["node_id"], fetch["structure"], fetch["base_rows"], fetch["target_rows"]) == ("fetch", "same", 3, 3) + assert east["inputs"][0]["base_artifact"] == fetch["base_artifact"] + # 东区金额只是三个区域里的一格：各区域的变化照列，占比为空 + assert [(r["key"], r["delta"]) for r in east["attribution"]["rows"]] == [ + (["东区"], 300.0), (["南区"], 0), (["西区"], 0)] + + body = await get(client, f"/api/runs/{third}/compare?base={second}") + [east] = body["metrics"] + assert east["classification"] == "params_only" and east["target"] == 900.0 + [fetch] = body["queries"] + assert fetch["structure"] == "params_only" + assert fetch["literals"] == [{"base": "'2026-08-31'", "target": "'2026-09-07'"}, + {"base": "'2026-09-06'", "target": "'2026-09-13'"}] + assert body["template"]["same_graph"] is True + assert any("start" in n for n in body["notes"]), body["notes"] + + body = await get(client, f"/api/runs/{third}/compare/base") + assert (body["base_run_id"], body["reason"]) == (second, "same_workflow") + json.dumps(body, allow_nan=False) diff --git a/backend/tests/test_compare_engine.py b/backend/tests/test_compare_engine.py new file mode 100644 index 00000000..2b7a49c7 --- /dev/null +++ b/backend/tests/test_compare_engine.py @@ -0,0 +1,558 @@ +"""运行对比的纯函数（engine/compare.py）：同一个工作流的两次运行，每个指标的变化拆成数据、口径、模板 / SQL。 + +两次运行都是照真实运行的事件形状手写的虚构数据（区域、金额、日期都是编的）：取数节点 fetch 查一条按区域汇总的 +SQL，Agent 节点 bot 自己再查一次、把各区域的金额作为数组字段交给口径卡，口径卡算几个指标。工件放在字典里。守的是： +- 只数据变（SQL 一字不差、结果变了）是 data_only；只改日期字面量是 params_only；SQL 结构变了是 sql_changed、不拆 +- 口径变了（表达式、口径版本）是 caliber_and_data：旧表达式的叶子路径在这次的输入里都取得到时严格拆分， + 口径效应 + 数据效应 = 总变化（逐位相等）；取不到就不拆，说明缺的是哪一个 +- 维度归因：按维度键配对两边的行，按变化的绝对值取前 10；度量列之和等于输入的值（两边都等）才给占比，占比之和 + 为 1；对不上时占比为空并说明；截断的快照不归因，照实写 +- 遮罩列的值绝不出现在结果里；用到遮罩列的指标不拆分 +- 模板对比：新增、删除的节点，配置有变化的节点和配置项名 +""" +from __future__ import annotations + +import json +from decimal import Decimal + +from app.engine import compare +from tests.fixtures.compare_runs import ( + BASE_ROWS, + EAST, + NEXT_ROWS, + REFUND, + WEEK37, + Fake, + amounts_in, + cell_in, + default_metrics, + metric, + node, + region_sql, + scalar_in, + week, + weekly_graph, +) + + +def run_compare(base, target): + result = compare.compare(base.side(), target.side()) + json.dumps(result, allow_nan=False) # 结果要能原样交给接口 + return result + + +def by_id(result): + return {m["metric_id"]: m for m in result["metrics"]} + + +def exact(value): + return Decimal(repr(value)) if isinstance(value, float) else Decimal(value) + + +# -------------------------------------------------------------------------- +# 分类：只数据变、只参数变、SQL 结构变 +# -------------------------------------------------------------------------- + + +def test_same_sql_new_rows_is_data_only(): + result = run_compare(week("r1"), week("r2", NEXT_ROWS)) + metrics = by_id(result) + east, total = metrics["east"], metrics["total"] + assert (east["classification"], total["classification"]) == ("data_only", "data_only") + assert (east["base"], east["target"], east["delta"]) == (1200.5, 1500.5, 300.0) + assert (east["base_rendered"], east["target_rendered"], east["delta_rendered"]) == ("1,200.5元", "1,500.5元", "+300.0元") + assert east["delta_pct"] == round(300 / 1200.5 * 100, 2) + assert east["card_id"] == "card" and east["card_title"] == "周报口径卡" and east["name"] == "东区金额" + assert east["decomposition"] is None and east["decomposition_note"] is None and east["reasons"] + assert total["delta"] == 200.0 and total["delta_rendered"] == "+200.0元" + assert result["summary"] == {"unchanged": 0, "data_only": 2, "caliber_and_data": 0, "sql_changed": 0, + "params_only": 0, "added": 0, "removed": 0, "incomparable": 0} + # 输入两边的值都列出来，带着各自的查询快照 + [cell] = east["inputs"] + assert (cell["path"], cell["base"], cell["target"]) == (EAST, 1200.5, 1500.5) + assert cell["base_artifact"] != cell["target_artifact"] + # 查询：节点 + 第几次 + 工具配上对，SQL 一字不差 + assert [(q["node_id"], q["ordinal"], q["structure"]) for q in result["queries"]] == [ + ("fetch", 1, "same"), ("bot", 1, "same")] + fetch = result["queries"][0] + assert fetch["node_title"] == "区域取数" and fetch["base_sql"] == fetch["target_sql"] == region_sql() + assert fetch["literals"] == [] and (fetch["base_rows"], fetch["target_rows"]) == (3, 3) + assert (fetch["base_truncated"], fetch["target_truncated"]) == (False, False) + assert result["template"]["same_graph"] is True and result["template"]["nodes"] == [] + assert (result["base"]["id"], result["target"]["id"]) == ("r1", "r2") + + +def test_identical_runs_are_unchanged(): + result = run_compare(week("r1"), week("r2")) + assert {m["classification"] for m in result["metrics"]} == {"unchanged"} + east = by_id(result)["east"] + assert east["delta"] == 0 and east["delta_rendered"] is None and east["attribution"] is None + assert result["summary"]["unchanged"] == 2 + + +def test_only_date_literals_changed_is_params_only(): + # 图里的 SQL 是模板（两次一样），执行时代进去的日期不同 + graph = weekly_graph(region_sql(("{{ input.start }}", "{{ input.end }}"))) + result = run_compare(week("r1", graph=graph), week("r2", NEXT_ROWS, sql=region_sql(WEEK37), graph=graph)) + metrics = by_id(result) + assert {m["classification"] for m in metrics.values()} == {"params_only"} + assert any("参数" in r for r in metrics["east"]["reasons"]) + fetch = result["queries"][0] + assert fetch["structure"] == "params_only" + assert fetch["literals"] == [{"base": "'2026-08-31'", "target": "'2026-09-07'"}, + {"base": "'2026-09-06'", "target": "'2026-09-13'"}] + # 定时运行换了日期属于参数变化：模板（图）一样 + assert result["summary"]["params_only"] == 2 and result["template"]["same_graph"] is True + + +def test_a_changed_sql_structure_is_flagged_not_split(): + paid = region_sql(extra=" AND status = 'paid'") + result = run_compare(week("r1"), week("r2", NEXT_ROWS, sql=paid)) + metrics = by_id(result) + assert {m["classification"] for m in metrics.values()} == {"sql_changed"} + east = metrics["east"] + assert east["decomposition"] is None and east["attribution"] is None + assert any("区域取数" in r and "SQL" in r for r in east["reasons"]) + fetch = result["queries"][0] + assert fetch["structure"] == "changed" and fetch["literals"] == [] + assert any("status = 'paid'" in d for d in fetch["differences"]) + # 图里 fetch 的 SQL 也改了：模板对比列出它和改了的配置项 + [changed] = result["template"]["nodes"] + assert (changed["node_id"], changed["change"], changed["keys"]) == ("fetch", "config", ["args.sql"]) + assert result["template"]["same_graph"] is False + + +# -------------------------------------------------------------------------- +# 口径变了：严格拆分 +# -------------------------------------------------------------------------- + + +def net_metrics(expression): + """净额指标 net：表达式由调用方定（v1 只取金额，v2 减去退款）。""" + def build(fetch, bot, rows, bot_rows): + gmv, refund = rows[0][1], rows[0][2] + inputs = [cell_in(0, "gmv", gmv, fetch)] + value = gmv + if REFUND in expression: + inputs.append(cell_in(0, "refund", refund, fetch)) + value = gmv - refund + return [metric("net", expression, value, inputs, name="东区净额")] + return build + + +def test_a_caliber_change_is_split_into_caliber_and_data_effects(): + base = week("r1", metrics=net_metrics(EAST), version="v1") + target = week("r2", NEXT_ROWS, metrics=net_metrics(f"{EAST} - {REFUND}"), version="v2") + net = by_id(run_compare(base, target))["net"] + assert net["classification"] == "caliber_and_data" + assert any("表达式" in r for r in net["reasons"]) and any("v1" in r and "v2" in r for r in net["reasons"]) + assert (net["base"], net["target"], net["delta"]) == (1200.5, 1380.5, 180.0) + split = net["decomposition"] + assert net["decomposition_note"] is None + # 口径效应 = f_new(x_new) − f_old(x_new) = 1380.5 − 1500.5；数据效应 = f_old(x_new) − f_old(x_old) = 1500.5 − 1200.5 + assert (split["caliber_effect"], split["data_effect"]) == (-120.0, 300.0) + assert (split["caliber_rendered"], split["data_rendered"]) == ("-120.0元", "+300.0元") + assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(net["delta"]) + + +def test_the_split_adds_up_exactly_with_awkward_decimals(): + """小数位不同、浮点相减有误差的数：两效应之和逐位等于总变化。""" + def build(expression, decimals): + def make(fetch, bot, rows, bot_rows): + gmv, refund = rows[0][1], rows[0][2] + inputs = [cell_in(0, "gmv", gmv, fetch), cell_in(0, "refund", refund, fetch)] + value = round(eval(expression.replace(EAST, repr(gmv)).replace(REFUND, repr(refund))), decimals) + return [metric("ratio", expression, value, inputs, decimals=decimals, unit="")] + return make + + base_rows = [["东区", 1000.1, 0.3], *BASE_ROWS[1:]] + next_rows = [["东区", 1000.7, 0.2], *NEXT_ROWS[1:]] + base = week("r1", base_rows, metrics=build(f"{EAST} * 0.1 - {REFUND}", 2)) + target = week("r2", next_rows, metrics=build(f"{EAST} * 0.3 + {REFUND}", 3), version="v2") + ratio = by_id(run_compare(base, target))["ratio"] + split = ratio["decomposition"] + assert split is not None, ratio["decomposition_note"] + # f_old(x_new) 按旧口径的小数位取整：round(1000.7 * 0.1 - 0.2, 2) = 99.87 + assert exact(split["data_effect"]) == exact(99.87) - exact(ratio["base"]) + assert exact(split["caliber_effect"]) + exact(split["data_effect"]) == exact(ratio["delta"]) + assert exact(ratio["delta"]) == exact(ratio["target"]) - exact(ratio["base"]) + + +def test_a_version_bump_with_the_same_formula_is_all_data(): + base = week("r1", metrics=net_metrics(EAST), version="v1") + target = week("r2", NEXT_ROWS, metrics=net_metrics(EAST), version="v2") + net = by_id(run_compare(base, target))["net"] + assert net["classification"] == "caliber_and_data" + assert (net["decomposition"]["caliber_effect"], net["decomposition"]["data_effect"]) == (0, 300.0) + assert net["decomposition"]["caliber_rendered"] == "0.0元" + + +def test_no_split_when_an_old_leaf_is_not_recorded_this_time(): + base = week("r1", metrics=net_metrics(f"{EAST} - {REFUND}"), version="v1") + target = week("r2", NEXT_ROWS, metrics=net_metrics(EAST), version="v2") + net = by_id(run_compare(base, target))["net"] + assert net["classification"] == "caliber_and_data" and net["decomposition"] is None + assert REFUND in net["decomposition_note"] and "无法严格拆分" in net["decomposition_note"] + # 总变化照算 + assert net["delta"] == 1500.5 - (1200.5 - 100.0) + + +def test_no_split_when_the_sql_changed_too(): + base = week("r1", metrics=net_metrics(EAST), version="v1") + target = week("r2", NEXT_ROWS, sql=region_sql(extra=" AND status = 'paid'"), metrics=net_metrics(EAST), version="v2") + net = by_id(run_compare(base, target))["net"] + assert net["classification"] == "caliber_and_data" and net["decomposition"] is None + assert "SQL" in net["decomposition_note"] + + +def test_a_unit_change_is_not_subtracted(): + def build(unit): + def make(fetch, bot, rows, bot_rows): + return [metric("east", EAST, rows[0][1], [cell_in(0, "gmv", rows[0][1], fetch)], unit=unit)] + return make + east = by_id(run_compare(week("r1", metrics=build("元")), week("r2", NEXT_ROWS, metrics=build("万元"))))["east"] + assert east["classification"] == "caliber_and_data" and any("单位" in r for r in east["reasons"]) + assert east["delta"] is None and east["delta_rendered"] is None and east["decomposition"] is None + assert "单位" in east["decomposition_note"] + + +# -------------------------------------------------------------------------- +# 维度归因 +# -------------------------------------------------------------------------- + + +def test_attribution_pairs_keys_and_shares_add_up_to_one(): + total = by_id(run_compare(week("r1"), week("r2", NEXT_ROWS)))["total"] + attribution = total["attribution"] + assert attribution["columns"] == ["region"] and attribution["measure"] == "gmv" + assert attribution["total_delta"] == 200.0 + rows = attribution["rows"] + assert [(r["key"], r["base"], r["target"], r["delta"]) for r in rows] == [ + (["东区"], 1200.5, 1500.5, 300.0), (["南区"], 800.0, 700.0, -100.0), (["西区"], 600.0, 600.0, 0.0)] + assert [r["share"] for r in rows] == [1.5, -0.5, 0.0] + assert abs(sum(r["share"] for r in rows) - 1) < 1e-12 + + +def test_attribution_pairs_by_key_not_by_row_position(): + """这次多了一个区域、顺序也变了：按区域配对，只有一边有的那一边写空。""" + next_rows = [["北区", 90.0, 0.0], ["东区", 1500.5, 120.0], ["西区", 600.0, 10.0], ["南区", 700.0, 40.0]] + total = by_id(run_compare(week("r1"), week("r2", next_rows)))["total"] + rows = {tuple(r["key"]): r for r in total["attribution"]["rows"]} + assert rows[("北区",)]["base"] is None and rows[("北区",)]["target"] == 90.0 and rows[("北区",)]["delta"] == 90.0 + assert rows[("东区",)]["delta"] == 300.0 and rows[("南区",)]["delta"] == -100.0 + assert abs(sum(r["share"] for r in rows.values()) - 1) < 1e-12 + + +def test_attribution_keeps_the_ten_largest_changes(): + names = [f"{n}区" for n in "一二三四五六七八九十甲乙"] + base_rows = [[name, 100.0, 0.0] for name in names] + next_rows = [[name, 100.0 + (i + 1) * 10 * (-1) ** i, 0.0] for i, name in enumerate(names)] + total = by_id(run_compare(week("r1", base_rows), week("r2", next_rows)))["total"] + rows = total["attribution"]["rows"] + assert len(rows) == 10 + assert [abs(r["delta"]) for r in rows] == [120.0, 110.0, 100.0, 90.0, 80.0, 70.0, 60.0, 50.0, 40.0, 30.0] + assert "12" in total["attribution"]["note"] + # 占比按全部维度值的总变化算：前 10 个之和不必为 1，全部之和为 1 + assert total["attribution"]["total_delta"] == sum((i + 1) * 10 * (-1) ** i for i in range(12)) + + +def test_no_share_when_the_measure_does_not_add_up_to_the_input(): + """东区金额只是快照里的一格：各区域之和不等于它，各键的变化照列，占比为空并说明。""" + east = by_id(run_compare(week("r1"), week("r2", NEXT_ROWS)))["east"] + attribution = east["attribution"] + assert attribution["measure"] == "gmv" and [r["key"] for r in attribution["rows"]] == [["东区"], ["南区"], ["西区"]] + assert all(r["share"] is None for r in attribution["rows"]) and attribution["note"] + + +def test_a_total_row_is_left_out_of_its_own_breakdown(): + """合计行和各区域在同一份结果里：指标取合计那一格，按其余各行归因，占比之和为 1。""" + base_rows = [*BASE_ROWS, ["合计", 2600.5, 150.0]] + next_rows = [*NEXT_ROWS, ["合计", 2800.5, 170.0]] + + def build(fetch, bot, rows, bot_rows): + return [metric("all", "cell(nodes.fetch, 3, 'gmv')", rows[3][1], [cell_in(3, "gmv", rows[3][1], fetch)])] + total = by_id(run_compare(week("r1", base_rows, metrics=build), week("r2", next_rows, metrics=build)))["all"] + rows = total["attribution"]["rows"] + assert ["合计"] not in [r["key"] for r in rows] and len(rows) == 3 + assert abs(sum(r["share"] for r in rows) - 1) < 1e-12 and total["attribution"]["total_delta"] == 200.0 + + +def test_a_truncated_snapshot_is_not_attributed(): + base, target = week("r1"), Fake("r2") + sql = region_sql() + fetch = target.tool("fetch", sql, rows=NEXT_ROWS) + bot = target.agent("bot", sql, rows=NEXT_ROWS, truncated=True) + target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS)) + total = by_id(run_compare(base, target))["total"] + attribution = total["attribution"] + assert attribution["rows"] == [] and "截断" in attribution["note"] + assert result_queries(run_compare(base, target), "bot")["target_truncated"] is True + + +def result_queries(result, node_id): + return next(q for q in result["queries"] if q["node_id"] == node_id) + + +def test_a_period_column_that_never_matches_is_not_a_dimension(): + """每期都不同的列（这里是日期写成的文本）不拿来配对：按其余的维度列归因。""" + columns = ["day", "region", "gmv"] + types = {"day": "text", "region": "text", "gmv": "number"} + + def make(run_id, days, values): + run = Fake(run_id) + rows = [[day, region, value] for day in days for region, value in zip(("东区", "南区"), values)] + sql = "SELECT day, region, SUM(amount) AS gmv FROM orders GROUP BY day, region" + bot = run.agent("bot", sql, columns, rows, types=types) + total = sum(r[2] for r in rows) + run.card([metric("total", "sum(vars.kpi.amounts)", total, [amounts_in(bot, rows)])]) + return run + + total = by_id(run_compare(make("r1", ("2026-09-05", "2026-09-06"), (10.0, 20.0)), + make("r2", ("2026-09-12", "2026-09-13"), (15.0, 18.0))))["total"] + attribution = total["attribution"] + assert attribution["columns"] == ["region"] + assert [(r["key"], r["delta"]) for r in attribution["rows"]] == [(["东区"], 10.0), (["南区"], -4.0)] + assert "day" in attribution["note"] + + +# -------------------------------------------------------------------------- +# 遮罩列 +# -------------------------------------------------------------------------- + + +def test_masked_columns_never_show_up(): + """数据源遮了 manager（经理名）和 refund：经理名不当维度、不出现在结果里；用到 refund 的指标输入不给值、不拆分。""" + columns = ["region", "manager", "gmv", "refund"] + types = {"region": "text", "manager": "text", "gmv": "number", "refund": "number"} + + def make(run_id, rows, expression, version): + run = Fake(run_id) + sql = "SELECT region, manager, SUM(amount) AS gmv, SUM(refund) AS refund FROM orders GROUP BY region, manager" + fetch = run.tool("fetch", sql, columns, rows, types=types) + bot = run.agent("bot", sql, columns, rows, types=types) + run.masked = {fetch: frozenset({"manager", "refund"}), bot: frozenset({"manager", "refund"})} + inputs = [cell_in(0, "gmv", rows[0][2], fetch)] + value = rows[0][2] + if "refund" in expression: + inputs.append(cell_in(0, "refund", rows[0][3], fetch)) + value = rows[0][2] - rows[0][3] + run.card([metric("net", expression, value, inputs), + metric("total", "sum(vars.kpi.amounts)", sum(r[2] for r in rows), [amounts_in(bot, rows)])], + version=version) + return run + + base_rows = [["东区", "经理甲", 1200.5, 101.25], ["南区", "经理乙", 800.0, 52.75]] + next_rows = [["东区", "经理甲", 1500.5, 133.5], ["南区", "经理乙", 700.0, 47.25]] + result = run_compare(make("r1", base_rows, EAST, "v1"), + make("r2", next_rows, "cell(nodes.fetch, 0, 'gmv') - cell(nodes.fetch, 0, 'refund')", "v2")) + text = json.dumps(result, ensure_ascii=False) + for secret in ("经理甲", "经理乙", "101.25", "52.75", "133.5", "47.25"): + assert secret not in text, secret + metrics = by_id(result) + net = metrics["net"] + refund = next(i for i in net["inputs"] if "refund" in i["path"]) + assert refund["masked"] is True and refund["base"] is None and refund["target"] is None + assert net["decomposition"] is None and "遮罩" in net["decomposition_note"] + # 维度只用没遮的 region；占比照给 + attribution = metrics["total"]["attribution"] + assert attribution["columns"] == ["region"] and abs(sum(r["share"] for r in attribution["rows"]) - 1) < 1e-12 + assert any("遮罩" in n for n in result["notes"]) + + +def test_a_masked_measure_is_not_attributed(): + def masked(run): + run.masked = {a: frozenset({"gmv"}) for a in run.queries} + return run + result = run_compare(masked(week("r1")), masked(week("r2", NEXT_ROWS))) + total = by_id(result)["total"] + assert total["attribution"]["rows"] == [] and "遮罩" in total["attribution"]["note"] + east = by_id(result)["east"] + assert east["inputs"][0]["base"] is None and east["inputs"][0]["masked"] is True + assert "1200.5" not in json.dumps(east["inputs"]) and "800.0" not in json.dumps(result) + + +# -------------------------------------------------------------------------- +# 对不上具体查询的输入、运行输入、一边才有的指标 +# -------------------------------------------------------------------------- + + +def code_metrics(value): + def build(fetch, bot, rows, bot_rows): + return [metric("calc", "vars.adj * 2", value * 2, [scalar_in("vars.adj", value, "calc", "code")])] + return build + + +def test_an_input_from_a_code_node_follows_its_upstream(): + def make(run_id, rows, value, sql=None): + run = Fake(run_id, graph=weekly_graph(sql, between=(node("calc", "code", "调整", code="print(1)", + assign_to="adj"),))) + sql = sql or region_sql() + run.tool("fetch", sql, rows=rows) + run.agent("bot", sql, rows=rows) + run.code("calc", value, "adj") + run.card(code_metrics(value)(None, None, rows, rows)) + return run + + # 上游什么都没变：变化来自数据 + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), make("r2", NEXT_ROWS, 12)))["calc"] + assert calc["classification"] == "data_only" and calc["delta"] == 4 + # 上游的 SQL 结构变了：说不清它变没变，不能报成数据变化 + calc = by_id(run_compare(make("r1", BASE_ROWS, 10), + make("r2", NEXT_ROWS, 12, sql=region_sql(extra=" AND status = 'paid'"))))["calc"] + assert calc["classification"] == "incomparable" and any("调整" in r for r in calc["reasons"]) + + +def test_the_code_node_itself_changed_is_a_template_change(): + def make(run_id, code, value): + run = Fake(run_id, graph=weekly_graph(between=(node("calc", "code", "调整", code=code, assign_to="adj"),))) + run.tool("fetch", region_sql()) + run.agent("bot", region_sql()) + run.code("calc", value, "adj") + run.card(code_metrics(value)(None, None, BASE_ROWS, BASE_ROWS)) + return run + calc = by_id(run_compare(make("r1", "x = 1", 10), make("r2", "x = 2", 11)))["calc"] + assert calc["classification"] == "sql_changed" and any("调整" in r for r in calc["reasons"]) + + +def test_a_run_input_change_is_a_parameter_change(): + def make(run_id, goal): + run = Fake(run_id, inputs={"goal": goal}) + run.add("node.started", "start", {"node_type": "input"}) + run.add("node.finished", "start", {"artifact": run.put({"goal": goal})}) + fetch = run.tool("fetch", region_sql()) + run.agent("bot", region_sql()) + run.card([metric("reach", f"round({EAST} / input.goal * 100, 1)", round(1200.5 / goal * 100, 1), + [cell_in(0, "gmv", 1200.5, fetch), scalar_in("input.goal", goal, "start", "input")], + unit="%", fmt="plain")]) + return run + result = run_compare(make("r1", 1000), make("r2", 1500)) + reach = by_id(result)["reach"] + assert reach["classification"] == "params_only" and any("input.goal" in r for r in reach["reasons"]) + assert reach["delta_rendered"] == "-40.0 个百分点" + assert any("goal" in n for n in result["notes"]) + + +def test_metrics_and_cards_on_one_side_only(): + def extra(fetch, bot, rows, bot_rows): + return [*default_metrics(fetch, bot, rows, bot_rows), + metric("west", "cell(nodes.fetch, 2, 'gmv')", rows[2][1], [cell_in(2, "gmv", rows[2][1], fetch)])] + base = week("r1") + target = week("r2", metrics=extra) + target.card([metric("cnt", "len(vars.kpi.amounts)", 3, [amounts_in(target.queries[1], BASE_ROWS)], unit="个", + decimals=None)], node_id="card2", caliber="补充口径") + result = run_compare(base, target) + metrics = by_id(result) + assert metrics["west"]["classification"] == "added" and metrics["west"]["base"] is None + assert metrics["cnt"]["classification"] == "added" and metrics["cnt"]["card_id"] == "card2" + assert result["summary"]["added"] == 2 + swapped = by_id(run_compare(target, base)) + assert swapped["west"]["classification"] == "removed" and swapped["west"]["target"] is None + + +def test_an_unreadable_card_makes_its_metrics_incomparable(): + base, target = week("r1"), week("r2", NEXT_ROWS) + base.tampered.add(base.cards[0][1]) + base.cards = [] # 接口只交回复验过哈希的口径卡 + result = run_compare(base, target) + assert {m["classification"] for m in result["metrics"]} == {"incomparable"} + assert any("哈希" in r for r in by_id(result)["east"]["reasons"]) + + +def test_a_card_executed_several_times_uses_the_last_one(): + base = week("r1") + target = Fake("r2") + fetch = target.tool("fetch", region_sql(), rows=NEXT_ROWS) + bot = target.agent("bot", region_sql(), rows=NEXT_ROWS) + target.card(default_metrics(fetch, bot, BASE_ROWS, BASE_ROWS)) + target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS), runs=2) + result = run_compare(base, target) + assert by_id(result)["east"]["target"] == 1500.5 + assert any("执行了 3 次" in n for n in result["notes"]) + + +def test_verify_queries_are_left_out(): + base = week("r1") + target = Fake("r2") + fetch = target.tool("fetch", region_sql(), rows=NEXT_ROWS) + bot = target.agent("bot", region_sql(), rows=NEXT_ROWS, verify=["SELECT COUNT(*) AS n FROM orders"]) + target.card(default_metrics(fetch, bot, NEXT_ROWS, NEXT_ROWS)) + result = run_compare(base, target) + assert [(q["node_id"], q["structure"]) for q in result["queries"]] == [("fetch", "same"), ("bot", "same")] + + +# -------------------------------------------------------------------------- +# 查询配对、模板对比 +# -------------------------------------------------------------------------- + + +def test_queries_fall_back_to_normalized_sql_pairing(): + """节点换了 ID：按「节点 + 第几次 + 工具」配不上，再按规范化 SQL 配（大小写、空白不同也认）。""" + base = week("r1") + target = Fake("r2", graph=weekly_graph()) + target.tool("pull", region_sql(WEEK37).lower(), rows=NEXT_ROWS) + target.agent("bot", region_sql(WEEK37), rows=NEXT_ROWS) + result = run_compare(base, target) + pull = result_queries(result, "pull") + assert pull["structure"] == "params_only" and pull["base_sql"] == region_sql() + assert len(result["queries"]) == 2 + + +def test_unpaired_queries_are_added_or_removed(): + base = week("r1") + target = week("r2") + target.tool("extra", "SELECT COUNT(*) AS n FROM refunds", ["n"], [[7]], types={"n": "number"}) + result = run_compare(base, target) + extra = result_queries(result, "extra") + assert extra["structure"] == "added" and extra["base_sql"] is None and extra["target_rows"] == 1 + removed = result_queries(run_compare(target, base), "extra") + assert removed["structure"] == "removed" and removed["target_sql"] is None + + +def test_the_template_lists_added_removed_and_reconfigured_nodes(): + base_graph = weekly_graph(between=(node("calc", "code", "调整", code="x = 1"),)) + target_graph = weekly_graph(region_sql(WEEK37), between=(node("check", "dual_check", "双盲复算"),)) + target_graph["nodes"][-2]["data"]["label"] = "写周报（新）" + target_graph["nodes"][-1]["data"]["config"] = {"fields": [{"name": "周报", "value": "{{ nodes.write.text }}"}]} + base = week("r1", graph=base_graph) + target = week("r2", graph=target_graph, version="v1") + base.version, target.version = 3, 4 + template = run_compare(base, target)["template"] + assert (template["base_version"], template["target_version"], template["same_graph"]) == (3, 4, False) + changes = {n["node_id"]: (n["change"], n["keys"]) for n in template["nodes"]} + assert changes == {"fetch": ("config", ["args.sql"]), "check": ("added", []), "write": ("config", ["label"]), + "out": ("config", ["fields"]), "calc": ("removed", [])} + labels = {n["node_id"]: n["labels"] for n in template["nodes"]} + assert labels["write"] == ["节点名称"] and labels["out"] == ["成果字段"] + # 配置的值（这里是 SQL 和模板）一个字都不展示 + assert "2026-09" not in json.dumps(template, ensure_ascii=False) + + +def test_a_missing_snapshot_file_is_reported_and_withheld(): + """基准运行的快照文件不在了：SQL 从 tool.start 的参数里取，结构照比；输入的值核对不了遮没遮，不给；不归因。""" + base, target = week("r1"), week("r2", NEXT_ROWS) + fetch = base.queries[0] + del base.store[fetch] + result = run_compare(base, target) + query = result_queries(result, "fetch") + assert query["structure"] == "same" and query["base_sql"] == region_sql() and query["base_rows"] is None + assert "无法读取" in query["problem"] + east = by_id(result)["east"] + assert east["classification"] == "data_only" and east["inputs"][0]["base"] is None + assert east["attribution"]["rows"] == [] and "无法读取" in east["attribution"]["note"] + + +def test_runs_without_metric_cards_still_compare_queries(): + base, target = Fake("r1"), Fake("r2") + base.tool("fetch", region_sql()) + target.tool("fetch", region_sql(WEEK37), rows=NEXT_ROWS) + result = run_compare(base, target) + assert result["metrics"] == [] and sum(result["summary"].values()) == 0 + assert any("没有口径卡" in n for n in result["notes"]) + assert [q["structure"] for q in result["queries"]] == ["params_only"] + + +def test_the_comparison_does_not_touch_the_runs(): + base, target = week("r1"), week("r2", NEXT_ROWS) + before = (list(base.events), list(target.events), dict(base.store), dict(target.store)) + run_compare(base, target) + assert (base.events, target.events, base.store, target.store) == before