From 6a96083dcb03b299b383c72e25492a30be535330 Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Wed, 21 Oct 2026 13:03:38 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=86=E6=9E=90=E8=83=BD=E5=8A=9B=E5=85=A5?= =?UTF-8?q?=E5=8F=A3=EF=BC=88=E5=85=AD=EF=BC=89=EF=BC=9A=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=A1=E7=9A=84=E5=8F=AF=E6=AF=94=E6=80=A7=E6=8C=82=E9=92=A9?= =?UTF-8?q?=EF=BC=88engine/nodes/metrics.py=20=E7=9A=84=20=5Fcomparability?= =?UTF-8?q?=EF=BC=89=EF=BC=9A=E4=B8=80=E4=B8=AA=E6=8C=87=E6=A0=87=E7=9A=84?= =?UTF-8?q?=E8=BE=93=E5=85=A5=E9=87=8C=E5=8F=96=E8=87=AA=E6=9F=A5=E8=AF=A2?= =?UTF-8?q?=E5=BF=AB=E7=85=A7=E7=9A=84=E6=A0=BC=EF=BC=88cell()=20=E5=8F=96?= =?UTF-8?q?=E7=9A=84=E6=A0=BC=E3=80=81=E6=8C=89=E4=BD=8D=E7=BD=AE=E5=8F=96?= =?UTF-8?q?=E7=9A=84=E6=A0=BC=E3=80=81Agent=20=E6=8C=89=E5=87=BA=E5=A4=84?= =?UTF-8?q?=E4=BA=A4=E6=9D=A5=E7=9A=84=E5=AD=97=E6=AE=B5=EF=BC=89=E6=81=B0?= =?UTF-8?q?=E5=A5=BD=E8=90=BD=E5=9C=A8=E4=B8=A4=E4=BB=BD=E4=B8=8D=E5=90=8C?= =?UTF-8?q?=E7=9A=84=E5=BF=AB=E7=85=A7=E4=B8=8A=EF=BC=8C=E4=B8=A4=E4=BB=BD?= =?UTF-8?q?=20SQL=20=E6=8C=89=E8=A7=84=E8=8C=83=E5=8C=96=EF=BC=88data/sql?= =?UTF-8?q?=5Fsemantics.normalize=5Fsql=EF=BC=8C=E6=8C=89=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E6=BA=90=E7=9A=84=E6=96=B9=E8=A8=80=EF=BC=89=E5=90=8C?= =?UTF-8?q?=E5=BD=A2=E3=80=81=E5=8F=AA=E5=B7=AE=E5=AD=97=E9=9D=A2=E9=87=8F?= =?UTF-8?q?=E3=80=81=E6=9F=A5=E7=9A=84=E6=98=AF=E5=90=8C=E4=B8=80=E4=B8=AA?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E6=BA=90=E6=97=B6=EF=BC=88=E6=9C=AC=E5=91=A8?= =?UTF-8?q?=E5=AF=B9=E4=B8=8A=E5=91=A8=EF=BC=89=EF=BC=8C=E5=8F=A3=E5=BE=84?= =?UTF-8?q?=E5=8D=A1=E6=89=A7=E8=A1=8C=E6=97=B6=E8=B0=83=E5=88=86=E6=9E=90?= =?UTF-8?q?=E7=AE=97=E5=AD=90=20comparable=EF=BC=88=E6=97=A9=E7=9A=84?= =?UTF-8?q?=E9=82=A3=E6=9C=9F=E4=BD=9C=E5=9F=BA=E6=9C=9F=EF=BC=9A=E7=AC=AC?= =?UTF-8?q?=E4=B8=80=E5=A4=84=E4=B8=8D=E5=90=8C=E7=9A=84=E5=AD=97=E9=9D=A2?= =?UTF-8?q?=E9=87=8F=E6=98=AF=E6=97=A5=E6=9C=9F=E6=88=96=E5=90=8C=E6=A0=B7?= =?UTF-8?q?=E5=86=99=E6=B3=95=E7=9A=84=E6=9C=9F=E6=AC=A1=E7=9A=84=E6=8C=89?= =?UTF-8?q?=E5=85=88=E5=90=8E=EF=BC=8C=E8=AE=A4=E4=B8=8D=E5=87=BA=E7=9A=84?= =?UTF-8?q?=E8=A1=A8=E8=BE=BE=E5=BC=8F=E9=87=8C=E5=85=88=E5=87=BA=E7=8E=B0?= =?UTF-8?q?=E7=9A=84=E9=82=A3=E4=BB=BD=E7=AE=97=E6=9C=AC=E6=9C=9F=EF=BC=89?= =?UTF-8?q?=EF=BC=8C=E7=BB=93=E8=AE=BA=E8=90=BD=E6=88=90=E5=B7=A5=E4=BB=B6?= =?UTF-8?q?=EF=BC=8C=E6=8C=87=E6=A0=87=E4=B8=8A=E8=AE=B0=20comparability?= =?UTF-8?q?=EF=BC=88contract.MetricComparability=EF=BC=9A=E5=B7=A5?= =?UTF-8?q?=E4=BB=B6=E3=80=81=E5=88=A4=E5=AE=9A=E3=80=81=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E8=AF=AD=E3=80=81fail=20/=20warn=20=E7=9A=84=E5=8E=9F=E5=9B=A0?= =?UTF-8?q?=EF=BC=8C=E5=8E=9F=E5=9B=A0=E5=85=88=E6=8C=89=E7=BB=93=E8=AE=BA?= =?UTF-8?q?=E8=A1=A8=E7=9A=84=20status=20=E4=B8=80=E5=88=97=E5=8F=96?= =?UTF-8?q?=EF=BC=8C=E6=B2=A1=E6=9C=89=E5=86=8D=E6=8C=89=E5=88=A4=E5=AE=9A?= =?UTF-8?q?=E8=AF=AD=E6=8B=86=EF=BC=89=E3=80=82SQL=20=E7=BB=93=E6=9E=84?= =?UTF-8?q?=E4=B8=8D=E5=90=8C=EF=BC=88=E9=80=80=E6=AC=BE=E9=A2=9D=20/=20?= =?UTF-8?q?=E9=94=80=E5=94=AE=E9=A2=9D=EF=BC=89=E3=80=81=E4=B8=A4=E6=AC=A1?= =?UTF-8?q?=E6=98=AF=E5=90=8C=E4=B8=80=E6=9D=A1=20SQL=E3=80=81=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E6=BA=90=E4=B8=8D=E5=90=8C=E7=9A=84=E4=B8=8D=E6=98=AF?= =?UTF-8?q?=E5=90=8C=E4=B8=80=E4=B8=AA=E9=87=8F=E7=9A=84=E4=B8=A4=E6=9C=9F?= =?UTF-8?q?=EF=BC=8C=E4=B8=8D=E6=9F=A5=E4=B9=9F=E4=B8=8D=E8=AE=B0=E6=97=A5?= =?UTF-8?q?=E5=BF=97=EF=BC=9B=E5=90=8C=E4=B8=80=E5=AF=B9=E5=BF=AB=E7=85=A7?= =?UTF-8?q?=E4=B8=80=E5=BC=A0=E5=8D=A1=E9=87=8C=E5=8F=AA=E6=9F=A5=E4=B8=80?= =?UTF-8?q?=E6=AC=A1=E3=80=82=E5=BF=AB=E7=85=A7=E8=AF=BB=E4=B8=8D=E5=87=BA?= =?UTF-8?q?=E6=9D=A5=E3=80=81=E6=8E=A2=E6=9F=A5=E5=A4=B1=E8=B4=A5=E3=80=81?= =?UTF-8?q?=E7=AE=97=E5=AD=90=E5=87=BA=E9=94=99=E3=80=81=E7=AE=97=E4=B8=8D?= =?UTF-8?q?=E4=BA=86=E3=80=81=E8=BF=98=E6=B2=A1=E6=9C=89=E7=AE=97=E5=AD=90?= =?UTF-8?q?=EF=BC=88=E7=AC=AC=201=20=E6=B3=A2=E7=9A=84=20run=20=E6=8A=9B?= =?UTF-8?q?=20NotImplementedError=EF=BC=89=E9=83=BD=E4=B8=8D=E6=8C=A1?= =?UTF-8?q?=E5=8F=A3=E5=BE=84=E5=8D=A1=EF=BC=8C=E7=85=A7=E5=AE=9E=E8=AE=B0?= =?UTF-8?q?=E4=B8=80=E6=9D=A1=20metric=5Fcomparability=5Fskipped=EF=BC=8C?= =?UTF-8?q?=E8=BF=99=E4=B8=AA=E6=8C=87=E6=A0=87=E4=B8=8D=E5=86=99=20compar?= =?UTF-8?q?ability?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/engine/nodes/metrics.py | 181 +++++++++++++++++++++ backend/tests/test_metric_comparability.py | 168 +++++++++++++++++++ 2 files changed, 349 insertions(+) create mode 100644 backend/tests/test_metric_comparability.py diff --git a/backend/app/engine/nodes/metrics.py b/backend/app/engine/nodes/metrics.py index d040fa29..7e90092e 100644 --- a/backend/app/engine/nodes/metrics.py +++ b/backend/app/engine/nodes/metrics.py @@ -55,6 +55,9 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: 取数的那条 SQL 对照数据目录查出 error 级问题的（查询快照的 checks，见 data/sqlcheck.py：一对多关联后重复计算、 存量跨期加总），同样值照算，标 sql_check_failed 并写明原因，出具契约据此降档（issuance.sql_check_gaps）。 + + 一个指标的输入落在两份不同的查询快照上、两份 SQL 同形只差字面量（本周对上周）时，自动做一次可比性检查（分析算子 + comparable），结论落成工件、记在指标的 comparability 上（_comparability）。做不成不挡口径卡，照实记一条日志。 """ definitions = ctx.cfg("metrics", []) or [] caliber_raw, caliber_version = ctx.cfg("caliber", "") or ctx.node.title, str(ctx.cfg("caliber_version", "") or "v1") @@ -84,6 +87,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: failed_by: dict[str, list[QueryProblem]] = {} #: 查询快照工件 id → 快照本身（读输入的格落在哪一列「查询区间里这张表没有数据」，_no_data_inputs） snapshots: dict[str, Any] = {} + #: 可比性检查：同一对查询快照只查一次（几个指标都用本周、上周两份快照时），和做检查要用的几样（_Comparing） + comparing = _Comparing() for definition in definitions: metric_id = str(definition.get("id") or "").strip() expr = str(definition.get("expression") or "").strip() @@ -167,6 +172,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: no_data: dict[str, Any] = {} if lacking := _no_data_inputs(inputs, carried, snapshots): no_data = {"no_data": True, "no_data_reason": _no_data_reason(lacking), "no_data_sources": lacking} + comparability = await _comparability(metric_id, str(definition.get("name") or metric_id), inputs, ctx, + comparing, snapshots) metrics.append( { "id": metric_id, @@ -192,6 +199,8 @@ async def run_metrics(state: GraphState, ctx: NodeContext) -> dict[str, Any]: **sql_check, # 只在输入取自「查询区间里这张表没有数据」的列时出现，同上 **no_data, + # 只在输入落在两份同形、只差字面量的查询快照上、可比性检查做成了时出现（contract.MetricComparability） + **({"comparability": comparability} if comparability else {}), } ) @@ -462,6 +471,178 @@ def _rendered(value: Any, definition: dict[str, Any], display: dict[str, Any]) - return MISSING +# -------------------------------------------------------------------------- +# 可比性检查（docs/design/analysis-round/00-总方案.md §3.4 第 3 条） +# -------------------------------------------------------------------------- + +#: 可比性检查没做成（探查失败、算子出错、还没有算子……）时那条日志的 code +COMPARABILITY_SKIPPED = "metric_comparability_skipped" +#: 那条日志的说法 +COMPARABILITY_SKIPPED_TEXT = "指标「{name}」用到的两次查询没做成可比性检查：{why}。指标照常计算，两期能不能直接比请自行核对" +_DAY = re.compile(r"^(\d{4})-?(\d{2})-?(\d{2})(?:[ T].*)?$") + + +class _Comparing: + """一张口径卡执行时做可比性检查记下的：同一对快照的结论（没做成的记 None，不再试）、数据源的类型（规范化 SQL 按 + 它的方言解析）。阈值、运行期依赖到真要检查时才读。""" + + def __init__(self) -> None: + self.done: dict[tuple[str, str], dict[str, Any] | None] = {} + self.kinds: dict[str, str | None] = {} + + +async def _comparability(metric_id: str, name: str, inputs: list[dict[str, Any]], ctx: NodeContext, + comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any] | None: + """指标的可比性（contract.MetricComparability：结论工件、判定、判定语、原因）。只在这种时候做：输入里取自查询快照的格 + （cell() 取的格、按位置取的格、Agent 按出处交来的字段）恰好落在两份不同的快照上，两份 SQL 同形、只差字面量，而且查的是 + 同一个数据源——本周对上周。SQL 结构不同的（退款额 / 销售额）不是同一个量的两期，不查。 + + 哪份是基期：两边第一处不同的字面量是日期（或同样写法的期次）的，早的是基期；认不出的，表达式里先出现的那份算本期 + （「(本周 - 上周) / 上周」的写法）。 + + 做不成（快照读不出来、探查失败、算子出错、算不了、还没有算子）不挡口径卡：照实记一条日志（COMPARABILITY_SKIPPED）， + 这个指标不写 comparability。同一对快照只查一次。""" + from app.engine.lineage import SNAPSHOT_VIAS + + artifacts = list(dict.fromkeys(str(i["artifact"]) for i in inputs if i.get("via") in SNAPSHOT_VIAS + and isinstance(i.get("artifact"), str) and i["artifact"])) + if len(artifacts) != 2: + return None + pair = (artifacts[0], artifacts[1]) + key = (min(pair), max(pair)) # 本周减上周、上周减本周用的是同一对快照 + if key in comparing.done: + return comparing.done[key] + found: dict[str, Any] | None = None + try: + found = await _check_comparable(pair, inputs, ctx, comparing, snapshots) + except _NotPeriods: + found = None + except Exception as e: # noqa: BLE001 - 检查是附加的，不能挡住口径卡 + why = e.message if isinstance(e, _Skipped) else describe_exception(e) + ctx.emit(EventType.LOG, level="warn", code=COMPARABILITY_SKIPPED, metric=metric_id, metric_name=name, + reason=why, message=COMPARABILITY_SKIPPED_TEXT.format(name=name, why=why)) + comparing.done[key] = found + return found + + +class _NotPeriods(Exception): + """两份快照不是同一个量的两期（SQL 结构不同、数据源不同、字面量一样、没有 SQL）：不查，也不记日志。""" + + +class _Skipped(Exception): + """该查、却做不成（结论算不了、判定认不出）：记一条日志，message 是给人看的原因。""" + + def __init__(self, message: str) -> None: + super().__init__(message) + self.message = message + + +async def _check_comparable(pair: tuple[str, str], inputs: list[dict[str, Any]], ctx: NodeContext, + comparing: _Comparing, snapshots: dict[str, Any]) -> dict[str, Any]: + import app.analysis as analysis_ops # 按模块取 run：测试、合并前的替身都只换模块上的这一个函数 + + from app.analysis.contract import VERDICTS + from app.data.sql_semantics import normalize_sql + from app.tools.analysis import ( + context_of, input_snapshot, masked_of, run_snapshot, settings_of, store, tool_context, + ) + + contents = [] + for artifact in pair: + if artifact not in snapshots: + try: + snapshots[artifact] = artifact_store.load(artifact) + except (ValueError, OSError): + snapshots[artifact] = None + if not isinstance(snapshots[artifact], dict): + raise _Skipped("查询快照读不出来") + contents.append(snapshots[artifact]) + sqls = [c.get("sql") for c in contents] + sources = [str(c.get("source") or "") for c in contents] + if not all(isinstance(q, str) and q.strip() for q in sqls) or sources[0] != sources[1]: + raise _NotPeriods() + dialect = await _dialect_of(sources[0], comparing) + norms = [normalize_sql(q, dialect) for q in sqls] + if not norms[0].same_shape(norms[1]) or norms[0].literals == norms[1].literals: + raise _NotPeriods() + first, second = _period_order(norms[0].literals, norms[1].literals) + order = (pair[first], pair[second]) # (基期, 本期) + refs: dict[str, str] = {} + for role, artifact in zip(("base", "target"), order): + producer = next((str(i.get("node_id") or "") for i in inputs if i.get("artifact") == artifact), "") + refs[artifact] = producer if producer and producer not in refs.values() else f"{producer or 'query'}·{role}" + snaps = {} + for artifact in order: + content = snapshots[artifact] + snap = input_snapshot(content, ref=refs[artifact], artifact=artifact, node_id=refs[artifact].split("·")[0]) + snap["masked"] = await masked_of(content, artifact) + snaps[refs[artifact]] = snap + settings = await settings_of(run_snapshot(ctx.run)) + context = await context_of(list(snaps.values()), tool_ctx=tool_context(ctx.run, ctx.node.id, ctx.emit)) + request = {"op": "comparable", "mode": None, + "params": {"base_query": refs[order[0]], "target_query": refs[order[1]]}} + result = await analysis_ops.run(request, snaps.get, settings=settings, context=context) + verdict = (result.get("verdict") or {}).get("code") + if result.get("status") != "ok" or verdict not in VERDICTS["comparable"]: + raise _Skipped(str(result.get("reason") or "检查没有给出判定")) + artifact = await store(result, run_id=ctx.run.run_id, node_id=ctx.node.id) + return {"artifact": artifact, "verdict": verdict, "phrase": str(result.get("phrase") or ""), + "reasons": comparability_reasons(result)} + + +async def _dialect_of(source: str, comparing: _Comparing) -> str | None: + """数据源的类型（规范化 SQL 按它的方言解析）；合并查询在 SQLite 里执行。认不出为 None（通用读法）。""" + from app.engine.merge_query import MERGE_SOURCE + + if source == MERGE_SOURCE: + return "sqlite" + if source not in comparing.kinds: + from sqlalchemy import select + + from app.db.base import SessionLocal + from app.db.models import DataSource + + async with SessionLocal() as session: + comparing.kinds[source] = (await session.execute( + select(DataSource.kind).where(DataSource.name == source))).scalar_one_or_none() + return comparing.kinds[source] + + +def _period_order(a: tuple[str, ...], b: tuple[str, ...]) -> tuple[int, int]: + """两份同形 SQL 的字面量 → (基期的下标, 本期的下标)。第一处不同、两边都认得出的日期（YYYY-MM-DD、YYYYMMDD，带不带 + 时间）或同样长短的期次写法（2026-W41、202610）定先后，早的是基期；都认不出的，表达式里先出现的那份（下标 0）是本期。""" + for x, y in zip(a, b): + if x == y: + continue + left, right = x.strip("'\""), y.strip("'\"") + dx, dy = _DAY.match(left), _DAY.match(right) + if dx and dy: + if dx.groups() != dy.groups(): + return (0, 1) if dx.groups() < dy.groups() else (1, 0) + continue + if len(left) == len(right) and re.fullmatch(r"[\dW\-/Qq]+", left) and re.fullmatch(r"[\dW\-/Qq]+", right): + return (0, 1) if left < right else (1, 0) + return 1, 0 + + +def comparability_reasons(result: dict[str, Any]) -> list[str]: + """可比性结论里 fail、warn 的原因，一句一个（contract.MetricComparability.reasons）。结果表里有 status 一列和一列说明 + （reason / text / detail / message）的照表取；没有的按判定语「两期不可比：原因；原因」拆。可比的没有原因。""" + table = result.get("table") or {} + columns = [str(c) for c in table.get("columns") or []] + text_col = next((c for c in ("reason", "text", "detail", "message") if c in columns), None) + if "status" in columns and text_col is not None: + si, ti = columns.index("status"), columns.index(text_col) + found = [str(row[ti]) for row in table.get("rows") or [] + if isinstance(row, list) and len(row) > max(si, ti) and row[si] in ("fail", "warn") and row[ti]] + if found: + return found + phrase = str(result.get("phrase") or "") + if (result.get("verdict") or {}).get("code") == "comparable" or "：" not in phrase: + return [] + return [part.strip() for part in phrase.split("：", 1)[1].split("；") if part.strip()] + + #: 清单里输入取自「查询区间里这张表没有数据」的列的指标后面那句：直接读清单的模型节点不能把 0 当成真的是 0 LISTED_NO_DATA = "（这段时间表里没有数据：不要当成 0 解读）" #: 指标的 no_data_reason 打头那半句（后面接查询记下的说明） diff --git a/backend/tests/test_metric_comparability.py b/backend/tests/test_metric_comparability.py new file mode 100644 index 00000000..ad33d65b --- /dev/null +++ b/backend/tests/test_metric_comparability.py @@ -0,0 +1,168 @@ +"""口径卡的可比性挂钩（docs/design/analysis-round/00-总方案.md §3.4 第 3 条）：一个指标的输入落在两份不同的查询快照上、 +两份 SQL 同形只差字面量（本周对上周）时，口径卡执行时自动做一次可比性检查，结论落成工件、记在指标的 comparability 上。 + +- 同形只差日期 → 有 comparability，早的那期是基期（和表达式里谁先写无关）；同一对快照只查一次 +- SQL 结构不同（退款额 / 销售额）、两次是同一条 SQL → 不查、不记日志 +- 算子出错、算不了、还没有算子（第 1 波的 run 抛 NotImplementedError）→ 不挡口径卡，记一条 metric_comparability_skipped +""" +from __future__ import annotations + +import asyncio +import sqlite3 + +import pytest +from sqlalchemy import select + +from app.analysis.contract import CallError +from app.core import artifact_store +from app.data.engine import engines +from app.db.base import SessionLocal +from app.db.models import DataSource, Run, RunEvent +from app.engine.nodes.metrics import COMPARABILITY_SKIPPED, comparability_reasons +from app.engine.runner import run_manager +from tests.fixtures.analysis_ops import FakeOps + +TOOL = "db_query__shop" +THIS = "SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '2026-10-12' AND '2026-10-18'" +LAST = "SELECT SUM(amount) AS gmv FROM orders WHERE day BETWEEN '2026-10-05' AND '2026-10-11'" +REFUNDS = "SELECT SUM(refund) AS gmv FROM orders WHERE refund > 0 AND day BETWEEN '2026-10-12' AND '2026-10-18'" +WOW = "(cell(nodes.this_week, 0, 'gmv') - cell(nodes.last_week, 0, 'gmv')) / cell(nodes.last_week, 0, 'gmv')" + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + yield + await run_manager.shutdown() + + +@pytest.fixture(autouse=True) +async def shop(tmp_path): + path = tmp_path / "shop.db" + db = sqlite3.connect(path) + db.executescript("CREATE TABLE orders (id INTEGER PRIMARY KEY, day TEXT, region TEXT, amount REAL, refund REAL);") + db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?)", [ + (1, "2026-10-06", "华东", 500.0, 0.0), (2, "2026-10-08", "华南", 400.0, 20.0), + (3, "2026-10-13", "华东", 450.0, 0.0), (4, "2026-10-15", "华北", 300.0, 15.0)]) + db.commit() + db.close() + async with SessionLocal() as session: + row = (await session.execute(select(DataSource).where(DataSource.name == "shop"))).scalar_one_or_none() + if row is None: + row = DataSource(name="shop", kind="sqlite", database=str(path), readonly=True, enabled=True) + session.add(row) + row.database, row.kind, row.enabled, row.readonly, row.options = str(path), "sqlite", True, True, {} + await session.commit() + source_id = row.id + await engines.invalidate(source_id) + yield + await engines.invalidate(source_id) + + +def node(nid, ntype, **config): + return {"id": nid, "type": ntype, "position": {"x": 0, "y": 0}, "data": {"label": nid, "config": config}} + + +def graph(metrics, *, this=THIS, last=LAST): + nodes = [node("start", "input"), + node("this_week", "tool", tool=TOOL, args={"sql": this}), + node("last_week", "tool", tool=TOOL, args={"sql": last}), + node("card", "metrics", caliber="周报口径", metrics=metrics), + node("out", "output", fields=[{"name": "r", "value": "{{ nodes.card.text }}"}])] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def run_card(g) -> tuple[Run, dict, list[dict]]: + run = await run_manager.start(graph=g, input_payload={}) + 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: + break + assert row.status == "succeeded", row.error + async with SessionLocal() as session: + evts = [(e.type, e.node_id, e.data) for e in (await session.execute( + select(RunEvent).where(RunEvent.run_id == run.id).order_by(RunEvent.seq))).scalars()] + [finished] = [d for t, n, d in evts if t == "node.finished" and n == "card"] + card = artifact_store.load(finished["artifact"]) + logs = [d for t, n, d in evts if t == "log" and n == "card"] + return row, card, logs + + +def by_id(card): + return {m["id"]: m for m in card["metrics"]} + + +async def test_two_periods_of_the_same_query_get_a_comparability_check(monkeypatch): + ops = FakeOps(monkeypatch) + _, card, logs = await run_card(graph([ + {"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}, + {"id": "diff", "name": "销售额变化", "expression": "cell(nodes.last_week, 0, 'gmv') - cell(nodes.this_week, 0, 'gmv')"}, + {"id": "this", "name": "本周销售额", "expression": "cell(nodes.this_week, 0, 'gmv')"}])) + metrics = by_id(card) + [call] = ops.calls # 同一对快照只查一次 + # 早的那期是基期：表达式里先写的是本周还是上周都一样 + assert call["request"] == {"op": "comparable", "mode": None, + "params": {"base_query": "last_week", "target_query": "this_week"}} + base, target = call["snapshots"]["last_week"], call["snapshots"]["this_week"] + assert base["sql"] == LAST and target["sql"] == THIS and base["node_id"] == "last_week" + assert base["artifact"] == metrics["wow"]["inputs"][1]["artifact"] and base["masked"] == [] + found = metrics["wow"]["comparability"] + assert found == metrics["diff"]["comparability"] + stored = artifact_store.load(found["artifact"]) + assert stored["op"] == "comparable" and found == {"artifact": found["artifact"], "verdict": "comparable", + "phrase": "两期可比", "reasons": []} + # 只用到一份快照的指标不查 + assert "comparability" not in metrics["this"] + assert not [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED] + + +async def test_an_incomparable_pair_lists_its_reasons(monkeypatch): + FakeOps(monkeypatch, verdicts=["incomparable"]) + _, card, _ = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}])) + found = by_id(card)["wow"]["comparability"] + assert found["verdict"] == "incomparable" and found["phrase"] == "两期不可比：本期只有 3 天数据" + assert found["reasons"] == ["本期只有 3 天数据"] + + +async def test_different_structures_and_the_same_query_are_not_checked(monkeypatch): + ops = FakeOps(monkeypatch) + ratio = "cell(nodes.this_week, 0, 'gmv') / cell(nodes.last_week, 0, 'gmv')" + _, card, logs = await run_card(graph([{"id": "refund_rate", "name": "退款率", "expression": ratio}], last=REFUNDS)) + assert "comparability" not in by_id(card)["refund_rate"] and ops.calls == [] + _, card, logs2 = await run_card(graph([{"id": "same", "name": "同一条", "expression": ratio}], last=THIS)) + assert "comparability" not in by_id(card)["same"] and ops.calls == [] + assert not [log for log in [*logs, *logs2] if log.get("code") == COMPARABILITY_SKIPPED] + + +async def test_without_the_operator_the_card_still_runs(): + """第 1 波的 app.analysis.run 还抛 NotImplementedError：口径卡照常算，照实记一条没做成。""" + _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW, "decimals": 4}])) + wow = by_id(card)["wow"] + assert "comparability" not in wow and wow["value"] == pytest.approx((450 + 300 - 900) / 900, abs=1e-4) + [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED] + assert log["level"] == "warn" and log["metric"] == "wow" and "销售额环比" in log["message"] + assert "没做成可比性检查" in log["message"] + + +@pytest.mark.parametrize("ops_kw", [ + {"errors": [CallError("no_window", "认不出时间条件")]}, + {"errors": [RuntimeError("探查超时")]}, + {"statuses": ["refused"]}, +]) +async def test_failures_do_not_block_the_card(monkeypatch, ops_kw): + FakeOps(monkeypatch, **ops_kw) + _, card, logs = await run_card(graph([{"id": "wow", "name": "销售额环比", "expression": WOW}])) + assert "comparability" not in by_id(card)["wow"] + [log] = [log for log in logs if log.get("code") == COMPARABILITY_SKIPPED] + assert log["reason"] + + +def test_reasons_come_from_the_check_table_when_there_is_one(): + result = {"verdict": {"code": "conditional"}, "phrase": "两期有条件可比：两段的周末天数不同", + "table": {"columns": ["check", "status", "reason"], "labels": {}, + "rows": [["同口径", "pass", None], ["窗口长度", "warn", "窗口长度不同：7 天对 6 天"], + ["星期构成", "warn", "两段的周末天数不同：2 天对 1 天"], ["整期", "unknown", "核对不了"]]}} + assert comparability_reasons(result) == ["窗口长度不同：7 天对 6 天", "两段的周末天数不同：2 天对 1 天"] + assert comparability_reasons({"verdict": {"code": "comparable"}, "phrase": "两期可比"}) == []