diff --git a/backend/app/engine/compare.py b/backend/app/engine/compare.py index 5481382e..6fb43657 100644 --- a/backend/app/engine/compare.py +++ b/backend/app/engine/compare.py @@ -233,6 +233,10 @@ def compare(base: Side, target: Side) -> dict[str, Any]: - queries：每对查询一项 {node_id, node_title, ordinal, structure, base_sql, target_sql, literals: [{base, target}]（只列不同的字面量）, base_rows, target_rows, base_truncated, target_truncated, base_artifact, target_artifact, differences: [结构上差在哪], problem?（快照读不出来、被改过）} + - analyses：分析卡每一项一条 {node_id, title, item, op, mode, base, target（各自 {status, verdict, direction, + phrase, artifact} 或 None）, change}：两次运行的分析卡按节点和项的标识对齐（analysis.end 事件，循环里执行了几次的取 + 最后一次），change 是 same（判定和判定语都一样）/ verdict（状态或判定变了）/ phrase（判定一样、判定语里的数变了）/ + added / removed。只列出两次的判定，不拆变化来源 - notes：整体的说明：连线、默认设置、运行输入的变化，口径卡执行了几次、读不出来，遮罩 """ return _Compare(base, target).result() @@ -330,6 +334,38 @@ def _queries(side: Side) -> list[_Query]: return out +def _analysis_items(side: Side) -> dict[str, tuple[dict[str, dict[str, Any]], int]]: + """封存范围内每张分析卡各项最后一次的结论（analysis.end，engine/nodes/analysis.py 每项一条）：{卡: ({项: 事件数据}, + 执行了几次)}，项按记下的先后。循环里执行了几次的，同一项后一次的盖过前一次。""" + out: dict[str, tuple[dict[str, dict[str, Any]], int]] = {} + for _, etype, node_id, data in side.events: + if not node_id or not isinstance(data, dict): + continue + if etype == EventType.ANALYSIS_END and isinstance(data.get("item"), str): + items, runs = out.setdefault(node_id, ({}, 0)) + items[data["item"]] = data + elif etype == EventType.NODE_FINISHED and node_id in out: + items, runs = out[node_id] + out[node_id] = (items, runs + 1) + return out + + +def _analysis_side(data: dict[str, Any] | None) -> dict[str, Any] | None: + if data is None: + return None + return {"status": data.get("status"), "verdict": data.get("verdict"), "direction": data.get("direction"), + "phrase": data.get("phrase") or "", "artifact": data.get("analysis_artifact")} + + +def _analysis_change(base: dict[str, Any] | None, target: dict[str, Any] | None) -> str: + if base is None or target is None: + return ADDED if base is None else REMOVED + if (base.get("status"), base.get("verdict"), base.get("direction")) != \ + (target.get("status"), target.get("verdict"), target.get("direction")): + return "verdict" + return "same" if (base.get("phrase") or "") == (target.get("phrase") or "") else "phrase" + + def _outputs(side: Side) -> dict[str, tuple[Any, ...]]: """每个节点在封存范围内交出的产出工件（node.finished 的 artifact），按执行的先后。工件按内容寻址：两次运行的同一个 节点交出的一模一样，工件 id 就一样。产出没落成工件的记 None，核对不了。""" @@ -803,8 +839,9 @@ class _Compare: summary = {name: 0 for name in CLASSES} for m in metrics: summary[m["classification"]] += 1 + analyses = self._analyses() return {"target": self.target.run, "base": self.base.run, "summary": summary, "template": self.template, - "metrics": metrics, "queries": queries, "notes": list(dict.fromkeys(self.notes))} + "metrics": metrics, "analyses": analyses, "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")) @@ -1395,6 +1432,25 @@ class _Compare: 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, inherited + # ---------------------------------------------------------------- 分析卡 + + def _analyses(self) -> list[dict[str, Any]]: + """两次运行的分析卡按节点、项的标识对齐（最小可用：列出两次的判定和判定语，不拆变化来源）。""" + before, after = _analysis_items(self.base), _analysis_items(self.target) + 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, ({}, 0)), after.get(node_id, ({}, 0)) + for side, (_, runs) in (("基准运行", b), ("这次运行", t)): + if runs > 1: + self.notes.append(f"分析卡「{self.title(node_id)}」在{side}中执行了 {runs} 次，按最后一次对比") + for item in [*t[0], *(i for i in b[0] if i not in t[0])]: + bi, ti = b[0].get(item), t[0].get(item) + meta = ti or bi or {} + out.append({"node_id": node_id, "title": self.title(node_id), "item": item, "op": meta.get("op"), + "mode": meta.get("mode"), "base": _analysis_side(bi), "target": _analysis_side(ti), + "change": _analysis_change(bi, ti)}) + return out + # ---------------------------------------------------------------- 指标 def _metrics(self) -> list[dict[str, Any]]: diff --git a/backend/tests/test_analysis_compare.py b/backend/tests/test_analysis_compare.py new file mode 100644 index 00000000..33da6a32 --- /dev/null +++ b/backend/tests/test_analysis_compare.py @@ -0,0 +1,42 @@ +"""运行对比里的分析卡（engine/compare.py 的 analyses）：两次运行的分析卡按节点和项的标识对齐，列出两次的判定和判定语 +（最小可用：不拆变化来源）。运行照 tests/fixtures/compare_runs.py 的虚构周报手写，另加分析卡每项一条 analysis.end。 +""" +from __future__ import annotations + +from app.engine import compare +from tests.fixtures.compare_runs import week + + +def _with_card(run, items, *, runs=1): + for _ in range(runs): + run.add("node.started", "insight", {"node_type": "analysis"}) + for item, verdict, phrase in items: + run.add("analysis.end", "insight", {"item": item, "op": "baseline", "mode": None, "status": "ok", + "verdict": verdict, "direction": "down", "phrase": phrase, + "analysis_artifact": f"{item}-{verdict}-{phrase}"}) + run.add("node.finished", "insight", {"duration_ms": 1, "attempt": 1}) + return run + + +def test_compare_aligns_analysis_items(): + base = _with_card(week("r1"), [("wow", "within_band", "在常规波动范围内"), ("trend", "rising", "呈上升趋势"), + ("gone", "no_trend", "没有明显趋势")]) + target = _with_card(week("r2"), [("wow", "outside_band", "超出常规波动范围"), ("trend", "rising", "呈上升趋势"), + ("new", "rising", "呈上升趋势")], runs=2) + result = compare.compare(base.side(), target.side()) + found = {a["item"]: a for a in result["analyses"]} + assert [a["item"] for a in result["analyses"]] == ["wow", "trend", "new", "gone"] + assert found["wow"]["change"] == "verdict" and found["wow"]["base"]["verdict"] == "within_band" + assert found["wow"]["target"] == {"status": "ok", "verdict": "outside_band", "direction": "down", + "phrase": "超出常规波动范围", "artifact": "wow-outside_band-超出常规波动范围"} + assert found["trend"]["change"] == "same" and found["new"]["change"] == "added" and found["new"]["base"] is None + assert found["gone"]["change"] == "removed" and found["gone"]["target"] is None + assert found["wow"]["title"] == "insight" and found["wow"]["op"] == "baseline" + assert "分析卡「insight」在这次运行中执行了 2 次，按最后一次对比" in result["notes"] + # 判定一样、判定语里的数变了 + target = _with_card(week("r3"), [("wow", "within_band", "在常规波动范围内（-3%）")]) + [wow] = compare.compare(_with_card(week("r4"), [("wow", "within_band", "在常规波动范围内")]).side(), + target.side())["analyses"] + assert wow["change"] == "phrase" + # 没有分析卡的两次运行：analyses 是空的 + assert compare.compare(week("r5").side(), week("r6").side())["analyses"] == []