diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 94a32f0f..59213c71 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -13,6 +13,11 @@ 连取值也不给;遮罩列不给。计算式只有一项来自查询、或者写了口径说明时不给列名;几项都来自查询又没写口径说明的,列名 里看不出日期、口径的取舍时才给,好让它分得清哪一项是哪一项。 +**遮罩列上的输入不复算。** cell() 取自遮罩列的输入不能单独复算;别的输入(计算式里写成当时代进去的值)里按位置取自 +查询结果的一格(vars.res.rows[0][2]),顺着取值链认出它取自哪份快照的哪一列(recompute.input_origin,证据面板、推演、 +可验证报告同一个认法),按需算那份快照要遮的列(和证据面板同一套,见 _Sources.hidden)——它不一定是 cell() 用到的 +那几份。取自遮罩列、遮没遮核对不了的,整个指标按无法复算记:遮罩值不能当字面量写进交给盲算的计算式。 + **一致就放行,不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪(data/sql_semantics.explain_sql_diff): 两边先收窄到这一格(只留算出这一列的那一项;分组的按这一行的分组值改成等值条件,HAVING 一并去掉),一条查询算好几列、 分组后取一行时才不会混进「选出的列」「分组」这类和这一格无关的差别。分组的列设了遮罩时不收窄:这一行的分组值写进 @@ -326,11 +331,13 @@ def _skipped(ctx: NodeContext, reason: str) -> dict[str, Any]: class _Sources: - """一次复算里要读的查询快照和数据源:同一份只读一次。""" + """一次复算里要读的查询快照、数据源和快照要遮的列:同一份只读一次、只算一次。""" def __init__(self) -> None: self.snapshots: dict[str, tuple[Any, str | None]] = {} self.rows: dict[str, DataSource | None] = {} + self.masks: dict[tuple[str, str | None], frozenset[str]] = {} + self._source_masks: Any = None def snapshot(self, artifact: str) -> tuple[dict[str, Any] | None, str | None]: if artifact not in self.snapshots: @@ -351,18 +358,25 @@ class _Sources: select(DataSource).where(DataSource.name == name))).scalar_one_or_none() return self.rows[name] - async def hidden(self, artifact: str) -> frozenset[str]: - """这份快照要遮的列(小写):查询当时记下的,加上数据源现在设的。宁可多遮。""" - from app.data.engine import masked_columns + async def hidden(self, artifact: str, source: str | None = None) -> frozenset[str]: + """这份快照要遮的列(小写),和证据面板同一套(api/evidence.py 的 _InputMasks._hidden):查询当时记下的,加上 + 数据源现在设的;合并查询按它合并的那几个数据源算,换了名字的结果列按来历遮。快照读不出来时按 source(取值链上 + 那份结果里带着的数据源名)现在设的算:遮罩不能因为快照坏了就失效。宁可多遮。""" + from app.api.evidence import _Masks, _merge_masks, _recorded_masks # api.evidence 导入了 engine 里的模块 + from app.engine.merge_query import MERGE_SOURCE - snap, _ = self.snapshot(artifact) - recorded = (snap or {}).get("mask_columns") - out = {c.lower() for c in recorded if isinstance(c, str)} if isinstance(recorded, list) else set() - name = (snap or {}).get("source") - if isinstance(name, str) and name: - row = await self.source(name) - out |= {c.lower() for c in masked_columns(row.options if row is not None else None)} - return frozenset(out) + key = (artifact, source) + if key not in self.masks: + if self._source_masks is None: + self._source_masks = _Masks() + snap, _ = self.snapshot(artifact) + if snap is not None and snap.get("source") == MERGE_SOURCE: + hidden, _ = await _merge_masks(snap, self._source_masks) + else: + current, _ = await self._source_masks.of((snap or {}).get("source") or source) + hidden = set(current) | (_recorded_masks(snap) if snap is not None else set()) + self.masks[key] = frozenset(hidden) + return self.masks[key] def _loader(state: GraphState) -> tuple[Callable[[str], Any], dict[str, str]]: @@ -461,6 +475,8 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti problem: str | None = None for item in metric.get("inputs") or []: if not _from_query(item): + if isinstance(item, dict): + problem = problem or await _positional_masks(item, sources, load, outputs, masks) continue info, why = await _query_input(item, sources, aliases, tops) masks[item["artifact"]] = await sources.hidden(item["artifact"]) @@ -495,6 +511,30 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti return row, jobs +async def _positional_masks(item: dict[str, Any], sources: _Sources, load: Callable[[str], Any], + outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> str | None: + """不是 cell()、Agent 字段的输入里,按位置取自一份查询结果的一格(vars.res.rows[0][2]):认出它取自哪份快照的哪一列 + (recompute.input_origin,证据面板、推演、可验证报告同一个认法),把那份快照要遮的列记进 masks,重算的预检据此认出它 + 取自遮罩列、按遮罩说(同一份快照上的那种一样)。只算 cell() 用到的几份快照时,这一格落在另一份快照上就认不出,遮罩值 + 会当字面量写进交给盲算的计算式。 + + 那份快照读不出来、数据源现在设的遮罩又没遮这一列时,查询当时记下的遮罩无从核对:交回原因,按无法复算记,宁可多遮。 + 取值链上的产出读不出来、那份结果没存快照的(遮没遮核对不了),重算的预检自己会拦下,这里不管。""" + def output(node_id: Any) -> Any: + key = outputs.get(str(node_id or "")) + return load(key) if key else None + + origin = recompute.input_origin(item, output) + if origin is None or origin.unverified is not None or not origin.artifact: + return None + hidden = await sources.hidden(origin.artifact, origin.source) + masks[origin.artifact] = hidden + if any(c.lower() in hidden for c in origin.columns): + return None + _, why = sources.snapshot(origin.artifact) + return f"输入 {item.get('path')} 所在的{why},核对不了这一格是否取自设置了遮罩的列" if why else None + + async def _query_input(item: dict[str, Any], sources: _Sources, aliases: dict[str, str], tops: list[str] ) -> tuple[dict[str, Any], str | None]: """一个来自查询快照的输入:(盲算要用的信息, None) 或 (_, 为什么不能单独复算)。""" diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 6ce5a013..f6802b25 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -7,13 +7,13 @@ - 不一致:挂起,审批 payload 和对照工件按约定,payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往 数据目录里写什么的预告;a / b / neither 三种选择的产出和缺口,没选的照实记,驳回、on_reject 为 fail 时失败 - 差异说明只比这一格所在的那一列(一条查询算好几列、分组后取一行的,收窄了再比,HAVING 一并去掉);分组的列设了 - 遮罩时(改了名再分组的也算)不收窄,遮罩列的取值不进审批卡、对照工件、产出和事件 + 遮罩时(改了名再分组的也算)不收窄、照整条比,遮罩列的取值不进审批卡、对照工件、产出和事件 - 只差一个方面时写数据目录(业务日期、有效记录条件),差了不止一处不写,人工确认过的不同说法不覆盖,指标自己的 筛选范围(取值在运行输入里、用到分组的列)不写,两条 SQL 规范化后一致的不写;人工驳回过的条件不会借着和别的 - 指标并成一项写回去,真写的和审批卡上的预告一致 + 指标并成一项写回去,真写的和审批卡上的预告一致;时间列在表结构里声明成文本的不记成业务日期,预告照实说 - 全局关闭直通、节点覆盖;设置按发起时的快照;重放时盲算不再执行、事件不重复;报告的写作目录不含盲算的查询 - 触顶、分组结果的维度(只给取值、遮罩列不给)、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的 - 盲算结果不采用 + 盲算结果不采用;按位置取自另一份快照遮罩列的输入(cell() 取 A、vars.res.rows[0][0] 取 B)按无法复算记,不盲算 """ from __future__ import annotations @@ -31,6 +31,7 @@ from app.db.base import SessionLocal from app.db.models import Approval, Run, RunEvent, Setting from app.engine import spend from app.engine.launch import launch_run +from app.engine.nodes.dual_check import GAP_UNCHECKED from app.engine.runner import run_manager from app.main import app from tests.fixtures.dual_check_shop import ( @@ -567,6 +568,37 @@ async def test_the_same_choice_twice_writes_the_catalog_once(monkeypatch, shop, # -------------------------------------------------------------------------- +@pytest.fixture +async def text_dated_shop(tmp_path): + """同一个库,只是下单日、支付日两列声明成文本(SQLite 常这样存日期)。""" + name, source_id = await add_source(build(tmp_path / "text_dated.db", date_type="TEXT")) + yield name, source_id + await drop_source(source_id) + + +async def test_time_columns_declared_as_text_are_not_written_as_the_business_date(monkeypatch, text_dated_shop, client): + """两边按不同的列筛时间,可这两列在表结构里声明成文本:认不准存的是不是日期,选哪边都不记成业务日期。审批卡上的 + 预告照实说(认不出可记的事实,不带写到哪),裁决后时间线上跳过的原因和预告一字不差。声明成 DATE 的同一个库照常记 + (test_choosing_the_blind_result_degrades_and_writes_the_catalog)。""" + name, source_id = text_dated_shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] + said = item["catalog"] + # 两列都是文本:各说被采纳的那一边的列 + for side, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")): + assert set(said[side]) == {"write", "reason"} and said[side]["write"] is False, said[side] + assert f"{label}按时间筛的列「orders.{column}」在表结构里声明为 TEXT,无法确认是日期 / 时间类型" \ + in said[side]["reason"] + done = await decide(client, run.id, {"gmv": "b"}) + assert done.status == "succeeded", done.error + [wrote] = logs(await events(run.id), "dual_check_catalog") + assert wrote["written"] == [] and [s["reason"] for s in wrote["skipped"]] == [said["b"]["reason"]] + assert await notes_of(source_id) is None + + async def test_a_refund_filter_becomes_the_valid_row_filter(monkeypatch, shop, client): name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) @@ -969,42 +1001,60 @@ async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monkeypatch, masked_shop, client): """原流程按遮罩列(region)分组、口径卡取华东那一行:差异说明收窄到这一行时,不能拿这一行的分组值补等值条件—— - 那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄,照整条比。""" + 那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄,照整条比。 + + 盲算没分组、按下单日筛:收窄了的话,两边差在时间列和原流程多出来的「region = '华东'」,说明和对照里都带着这个取值; + 照整条比,时间列照样说得出,原流程的分组、多选的列归纳不了(_compared_whole)。""" name, _ = masked_shop first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} - Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")}) run = await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}})) assert run.status == "interrupted", run.error payload = (await pending(run.id)).payload + [item] = payload["summary"] + assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310") card = artifact_store.load(payload["card_artifact"]) done = await decide(client, run.id, {"first": "a"}) assert done.status == "succeeded", done.error evs = await events(run.id) _no_masked_value(payload, card, done, evs) - # 照整条比:原流程分组、盲算没分组都列出来,不止差一个方面,选了也不写目录 - assert len(payload["summary"][0]["diff"]) > 1 + _compared_whole(payload, card) + # 差异说明里有未能归纳的部分:选了也不写目录 [wrote] = logs(evs, "dual_check_catalog") assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, masked_shop, client): """原流程把遮罩列改了名再分组(region AS area):结果列名 area 不在遮罩里,分组值却还是 region 的取值。收窄时看的 - 是分组键用到的列,一样不拿这一行的分组值补等值条件,「华东」不进审批卡、对照工件、产出和事件。""" + 是分组键用到的列,一样不拿这一行的分组值补等值条件,「华东」不进审批卡、对照工件、产出和事件,照整条比。""" name, _ = masked_shop renamed = (f"SELECT region AS area, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' " f"AND paid_date {WEEK} GROUP BY region ORDER BY region") first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} - Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")}) run = await start(graph(name, renamed, metrics=[first], dual={"definitions": {}})) assert run.status == "interrupted", run.error payload = (await pending(run.id)).payload - assert payload["summary"][0]["original_rendered"].startswith("300") + [item] = payload["summary"] + assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310") card = artifact_store.load(payload["card_artifact"]) done = await decide(client, run.id, {"first": "a"}) assert done.status == "succeeded", done.error _no_masked_value(payload, card, done, await events(run.id)) - # 照整条比:原流程分组、盲算没分组都列出来 - assert len(payload["summary"][0]["diff"]) > 1 + _compared_whole(payload, card) + + +def _compared_whole(payload: dict, card: dict) -> None: + """差异说明照整条比的痕迹(原流程按地区分组取一行,盲算没分组、按下单日筛):时间列照样说得出;原流程的分组、多选的 + 地区列归纳不了,另起一条「另有未能归纳的差异」;规范化后的对照里还留着原流程的 GROUP BY。收窄了的话分组去掉、换成 + 等值条件,两边都是简单形状,说明里就是时间列和多出来的那个条件,这几样都不会有。""" + [metric] = card["metrics"] + diff = metric["diff"] + assert [i["aspect"] for i in diff["items"]] == ["time", "other"], diff["items"] + assert diff["fallback"] is True and any("GROUP BY" in line for line in diff["text_diff"]), diff["text_diff"] + [item] = payload["summary"] + assert len(item["diff"]) == 2 and "order_date" in item["diff"][0] and "另有未能归纳的差异" in item["diff"][1], item + assert item["fallback"] is False async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client): @@ -1021,6 +1071,81 @@ async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(m _no_masked_value(payload, card, done, await events(run.id)) +#: 销售额(第一份快照的一格)加上首组地区名的字数(第二份快照按位置取的一格) +MIX = {"id": "mix", "name": "销售额加首组名称字数", "unit": "元", "decimals": 1, "format": "thousands", + "expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows[0][0])"} + + +def with_regions(spec: dict, source: str) -> dict: + """取数之后另查一次按地区分组的结果(第二份快照),数据整形把它解析进 vars.res:口径卡按位置取它的格。""" + nodes = list(spec["nodes"]) + at = next(i for i, n in enumerate(nodes) if n["id"] == "fetch") + 1 + nodes[at:at] = [node("regions", "tool", tool=f"db_query__{source}", args={"sql": BY_REGION}), + node("parse", "transform", mode="json", template="{{ nodes.regions }}", assign_to="res")] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def test_a_masked_cell_read_by_position_from_another_snapshot_is_not_blind_computed(monkeypatch, masked_shop): + """口径卡上一个指标:cell() 取第一次查询(A,不碰遮罩列)的销售额,另按位置取第二次查询(B,按遮罩列 region 分组)的 + 首组名称(vars.res.rows[0][0])。B 不是 cell() 用到的快照:以前预检只为 A 算遮罩,认不出按位置取的那一格在遮罩列上, + 指标照常盲算,「华东」作为字面量写进了交给盲算模型的计算式。现在按需算 B 的遮罩:这个指标按无法复算记缺口、不盲算, + 遮罩列的取值不进盲算的任务、节点产出和节点发的事件。""" + name, _ = masked_shop + blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")}) + run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name)) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + [row] = out["metrics"] + assert row["metric_id"] == "mix" and row["status"] == "unchecked", row + assert "遮罩" in row["reason"] and "推演" not in row["reason"], row["reason"] + assert out["status"] == "unchecked" and out["issuance_gaps"] == [GAP_UNCHECKED.format(name=MIX["name"], + reason=row["reason"])] + assert blind.tasks == [] and blind.runs == 0, blind.tasks + assert "华东" not in json.dumps(out, ensure_ascii=False) + assert not [e.type for e in evs if e.node_id == "dual" and not e.type.startswith(("tool.", "llm.")) + and "华东" in json.dumps(e.data, ensure_ascii=False)] + + +async def test_an_unmasked_cell_read_by_position_from_another_snapshot_is_blind_computed(monkeypatch, shop): + """同样的图、数据源没设遮罩:预检照常通过,指标进入盲算(按位置取的那一格作为字面量写进计算式),两边一致。拦下上面 + 那个指标的确实是遮罩。""" + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")}) + run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name)) + assert run.status == "succeeded", run.error + out = dual_output(run, await events(run.id)) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed", out + [task] = blind.tasks + assert "len('华东')" in task + + +#: 按位置取的那一格:数据整形解析好的第二份查询结果(首组地区名),快照 id 是编的 +POSITIONAL = {"path": "vars.res.rows[0][0]", "value": "华东", "node_id": "parse", "via": "transform", "status": "ok"} +LOST = "cd" * 32 + + +def parsed_regions(source: str) -> dict: + return {"columns": ["region", "gmv"], "rows": [["华东", 300.0], ["华北", 30.0]], "artifact": LOST, "source": source} + + +async def test_a_cell_read_by_position_from_an_unreadable_snapshot_is_not_trusted(masked_shop): + """按位置取的那一格所在的快照读不出来(被改过、文件不在了):查询当时记下的遮罩无从核对。数据源现在设的遮罩遮着这一列 + 的,照遮罩算(交给重算的预检按遮罩说);没遮的也不当成没遮,按无法复算记、原因说清。""" + from app.engine.nodes.dual_check import _positional_masks, _Sources + + name, _ = masked_shop + for source, expected in ((name, None), ("dc_no_such_source", "输入 vars.res.rows[0][0] 所在的查询快照与哈希不一致," + "疑似被修改,核对不了这一格是否取自设置了遮罩的列")): + sources = _Sources() + sources.snapshots[LOST] = (None, "查询快照与哈希不一致,疑似被修改") + masks: dict = {} + state = {"state:parse": parsed_regions(source)} + why = await _positional_masks(POSITIONAL, sources, state.get, {"parse": "state:parse"}, masks) + assert why == expected, source + assert ("region" in masks[LOST]) is (source == name), masks + + def _no_masked_value(payload: dict, card: dict, done: Run, evs: list[RunEvent], value: str = "华东") -> None: """遮罩列上的取值不出现在审批 payload、对照工件、节点产出和节点发的事件里。查询本身的 tool.end、模型的往来按 设计带着原始结果(证据面板显示时再遮),不在此列。"""