diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index c1ad8ae1..ecf2b1eb 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -23,10 +23,17 @@ is_test = 0、is_valid = 1、deleted_at IS NULL）；只涉及一张表。拿不 **怎么写。** 照对话提案的规矩核对（plan_patch：列在不在表结构里、取值格式），记为人工填写、已确认，备注写明来自哪次 运行、哪个节点、哪个指标。目录里这一项已经是人工确认过的另一种说法时不覆盖，照实交回原因；已经是同样的说法就什么都 -不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边也按它筛了的，和新条件合起来写 -（已确认的不改，照实说）；被采纳的一边没按它筛的，两者的关系拿不准，不动。推断的去重规则说的未必是同一件事，也不 -替换。同一次裁决里几个指标对同一项给出不同说法的，一个都不写。整个写入由节点放在 ask() 之后的一个 @task 里：审批 -恢复重放、接着运行时不会再写一遍。 +不写（再写一遍也不升版本）。有效记录条件是一整句条件：目录里已有的条件被采纳的一边（几个指标并成一项时是每一个） +也按它筛了的，和新条件合起来写（已确认的不改，照实说）；有一个没按它筛的，两者的关系拿不准，不动。推断的去重规则 +说的未必是同一件事，也不替换。整个写入由节点放在 ask() 之后的一个 @task 里：审批恢复重放、接着运行时不会再写一遍。 + +**同一次裁决里几个指标落到同一项上。** 说法一样的写一次，备注里列出每个指标。有效记录条件是能用 AND 叠加的排除条件 +（这里只收排除无效记录的），几个指标各自采纳的条件都有人的选择撑着，合成一句写——「排除退款」和「排除测试单」并不 +矛盾。业务日期一张表只有一个，去重规则说的未必是同一件事，说法不同就一个都不写。 + +**预告。** 审批卡上要让审核人看得出选了之后会往目录里写什么（preview）：按发起审批时的目录，选原流程、选盲算各会 +写什么、不写的为什么。和真写是同一套判断（derive 认事实、_plan 对目录），所以只在两处可能和最后写的不一样：审批停下 +期间目录被人改了，或者同一次裁决里几个指标落到同一项上（上一段）。 catalog.save_patch 不收备注，这里用它的同一套原料（read_entry、apply_patch、write_entry 的乐观锁）写，只多把备注放进项里。 """ @@ -35,7 +42,7 @@ from __future__ import annotations import logging import re from collections.abc import Mapping -from dataclasses import dataclass +from dataclasses import dataclass, replace from typing import Any from sqlalchemy import select @@ -79,7 +86,8 @@ _SCOPE_DIGITS = 2 @dataclass(frozen=True) class Fact: """要写进目录的一项：哪个数据源的哪张表、哪一项、写什么。metric、choice 只用来写备注和说明；sql、dialect 是被采纳 - 那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。""" + 那一边的写法和方言，有效记录条件和目录里已有的条件合并时要看它还按哪些条件筛了。also 是同一次裁决里落到同一项上、 + 并进这一项的其余几个指标（各自的指标、选择、写法），value 是合起来之后的说法。""" source: str table: str @@ -89,6 +97,18 @@ class Fact: choice: str sql: str = "" dialect: str | None = None + also: tuple[Fact, ...] = () + + +def _members(fact: Fact) -> tuple[Fact, ...]: + """这一项背后的每个指标的选择：并过的连同并进来的。""" + return (fact, *fact.also) + + +def _names(fact: Fact) -> str: + """说明里的指标叫法：「销售额」，几个指标并成一项时「销售额」、「订单金额」。""" + names = list(dict.fromkeys(m.metric for m in _members(fact))) + return "、".join(f"「{n}」" for n in names[:4]) + (f"等 {len(names)} 个" if len(names) > 4 else "") def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, @@ -402,28 +422,45 @@ def _conditions_of(sql: str, read: str | None) -> set[str] | None: return {_normalized(part, read) for where in tree.find_all(exp.Where) for part in _and_parts(where.this)} +def _and_join(base: str | None, values: list[str], dialect: str | None) -> str | None: + """几句条件用 AND 接成一句：已经有的（base 里的、前面接过的，按 _normalized 比）不再接；最外层是 OR 的加括号， + 不然后面接的条件只并到 OR 的后一半上。每句保留原来的写法。有一句解析不了返回 None。""" + from sqlglot import exp + + read = _read(dialect) + texts: list[str] = [] + seen: set[str] = set() + for text in [*([base] if base else []), *values]: + tree = _parse(str(text), read) + if tree is None: + return None + keys = {_normalized(p, read) for p in _and_parts(tree)} + if keys <= seen: + continue + seen |= keys + texts.append(f"({text})" if isinstance(tree, exp.Or) else str(text)) + return " AND ".join(texts) + + def _combined_filter(existing: str, fact: Fact) -> tuple[str | None, str | None]: """目录里已有的有效记录条件和这次的新条件合起来：(合起来的整句条件, None) 或 (None, 为什么不动)。 有效记录条件是一整句（copilot 提案也要求「原有条件也要带上」）。只写新条件会把两边都按着筛的原有条件静默替换掉； - 原有条件被采纳的一边没按它筛，审核人的选择说明不了它对不对，也不能替人删掉。""" + 原有条件有一个被采纳的一边没按它筛，审核人的选择说明不了它对不对（合起来写就等于替人确认了它），也不能替人删掉。""" label = catalog.TABLE_FIELD_LABEL["valid_filter"] read = _read(fact.dialect) - have_tree, new_tree = _parse(existing, read), _parse(str(fact.value), read) - if have_tree is None or new_tree is None: + have_tree = _parse(existing, read) + combined = _and_join(existing, [str(m.value) for m in _members(fact)], fact.dialect) \ + if have_tree is not None else None + if combined is None: return None, f"「{fact.table}」已有的{label}「{existing}」无法解析，与这次的条件合不起来，未改动" - have = [_normalized(p, read) for p in _and_parts(have_tree)] - used = _conditions_of(fact.sql, read) or set() - if not set(have) <= used: - return None, (f"「{fact.table}」已有的{label}「{existing}」，被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" - "没有按它筛选，两者的关系无法确定，未改动。如需更改，请到数据目录中修改") - if {_normalized(p, read) for p in _and_parts(new_tree)} <= set(have): - return existing, None - from sqlglot import exp - - # 原有条件最外层是 OR 的，接 AND 之前要加括号，不然新条件只并到 OR 的后一半上 - head = f"({existing})" if isinstance(have_tree, exp.Or) else existing - return f"{head} AND {fact.value}", None + have = {_normalized(p, read) for p in _and_parts(have_tree)} + for member in _members(fact): + if not have <= (_conditions_of(member.sql, read) or set()): + return None, (f"「{fact.table}」已有的{label}「{existing}」，指标「{member.metric}」被采纳的" + f"{SIDE_LABEL.get(member.choice, member.choice)}没有按它筛选，两者的关系无法确定，未改动。" + "如需更改，请到数据目录中修改") + return combined, None def _dedup(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], @@ -491,9 +528,13 @@ def _aggregate_pair(a: Any, b: Any, dialect: Any) -> tuple[str, tuple[bool, bool def note_of(fact: Fact, *, run_id: str, node_title: str) -> str: - """目录项上的备注：来自哪次运行、哪个节点、哪个指标、采纳了哪一边。""" - return (f"双盲复算（运行 {run_id}，节点「{node_title}」）：审核人在指标「{fact.metric}」上" - f"采用了{SIDE_LABEL.get(fact.choice, fact.choice)}的写法") + """目录项上的备注：来自哪次运行、哪个节点、哪些指标、各采纳了哪一边。""" + parts = [] + for side, label in SIDE_LABEL.items(): + names = list(dict.fromkeys(m.metric for m in _members(fact) if m.choice == side)) + if names: + parts.append(f"在指标{'、'.join(f'「{n}」' for n in names)}上采用了{label}的写法") + return f"双盲复算（运行 {run_id}，节点「{node_title}」）：审核人" + "，".join(parts) def _value_text(item: str, value: Any) -> str: @@ -509,21 +550,38 @@ def _same_fact(item: str, a: Any, b: Any) -> bool: return isinstance(a, str) and isinstance(b, str) and " ".join(a.split()) == " ".join(b.split()) -def _contradiction(facts: list[Fact]) -> str: +def _differing(facts: list[Fact]) -> str: """同一次裁决里对同一项给出的不同说法，说清是谁、各说了什么。""" first = facts[0] label = catalog.TABLE_FIELD_LABEL.get(first.item, first.item) said = "；".join(f"指标「{f.metric}」采纳的是「{_value_text(f.item, f.value)}」" for f in facts[:4]) - return (f"同一次裁决里对「{first.table}」的{label}给出的选择互相矛盾（{said}），无法确定哪一个是数据事实，" + return (f"同一次裁决里对「{first.table}」的{label}给出了不同的说法（{said}），无法确定哪一个是数据事实，" "都未写入数据目录") +def _merged(members: list[Fact]) -> tuple[Fact | None, str | None]: + """同一次裁决里落到同一项上的几个说法合成一项：(合起来的那一项, None) 或 (None, 为什么都不写)。 + + 说法都一样的写一次。有效记录条件用 AND 合成一句（derive 只收排除无效记录的条件，几句叠起来不会互相打架）。 + 业务日期一张表只有一个、去重规则说的未必是同一件事，说法不同就都不写。""" + first, rest = members[0], tuple(members[1:]) + if not rest: + return first, None + if first.item == "valid_filter": + value = _and_join(None, [str(m.value) for m in members], first.dialect) + return (replace(first, value=value, also=rest), None) if value is not None else (None, _differing(members)) + if all(_same_fact(first.item, m.value, first.value) for m in rest): + return replace(first, also=rest), None + return None, _differing(members) + + async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, node_title: str, actor: str | None) -> dict[str, list[dict[str, Any]]]: """把审核人的选择写进数据目录：{written: [{source, table, item}], skipped: [{reason}]}。永远不抛：数据源没了、 目录写不进去，都落在 skipped 里照实说。 - 先把每个候选认成目录项，再按（数据源、表、项）归拢：同一项上说法互相矛盾的整组不写，剩下的逐个写。""" + 先把每个候选认成目录项，再按（数据源、表、项）归拢成一项（_merged：说法一样的、有效记录条件合起来写，说法 + 不同的整组不写），逐项写。""" skipped: list[dict[str, Any]] = [] facts: list[Fact] = [] for candidate in candidates: @@ -549,15 +607,15 @@ async def record_choices(candidates: list[Mapping[str, Any]], *, run_id: str, no groups.setdefault((fact.source, fact.table, fact.item), []).append(fact) written: list[dict[str, Any]] = [] for members in groups.values(): - if any(not _same_fact(f.item, f.value, members[0].value) for f in members[1:]): - skipped.append({"reason": _contradiction(members)}) + fact, why = _merged(members) + if fact is None: + skipped.append({"reason": why}) continue - for fact in members: - done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor) - if done: - written.append({"source": fact.source, "table": fact.table, "item": fact.item}) - else: - skipped.append({"reason": why}) + done, why = await _record(fact, run_id=run_id, node_title=node_title, actor=actor) + if done: + written.append({"source": fact.source, "table": fact.table, "item": fact.item}) + else: + skipped.append({"reason": why}) return {"written": written, "skipped": skipped} @@ -566,69 +624,103 @@ async def _record(fact: Fact, *, run_id: str, node_title: str, actor: str | None async with SessionLocal() as session: row = (await session.execute(select(DataSource).where(DataSource.name == fact.source))).scalar_one_or_none() if row is None: - return False, f"数据源「{fact.source}」已不存在，指标「{fact.metric}」的选择未写入数据目录" + return False, f"数据源「{fact.source}」已不存在，指标{_names(fact)}的选择未写入数据目录" return await _write(session, row, fact, note=note_of(fact, run_id=run_id, node_title=node_title), actor=actor) except Exception: # noqa: BLE001 - 目录写不进去不能毁掉审批的结果：缺口照样按选择记 logger.exception("双盲复算的选择写入数据目录失败（数据源 %s）", fact.source) - return False, f"指标「{fact.metric}」的选择写入数据目录时出错，未写入" + return False, f"指标{_names(fact)}的选择写入数据目录时出错，未写入" + + +def _plan(notes: Mapping[str, Any], fact: Fact, tables: Mapping[str, Any] | None) -> tuple[Any, str | None]: + """按目录里这张表现在的说法，算出这一项要写成什么：(要写的取值, None) 或 (None, 为什么不写)。不读库、不写库： + 真写（_write）和审批卡上的预告（preview）用的是这同一套判断。""" + label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) + names = _names(fact) + stored = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None + if stored is not None and stored.get("status") == "rejected" and _same_fact(fact.item, stored.get("value"), + fact.value): + # 同一个说法以前被人明确驳回过：两次人工判断对不上，不替人拿主意 + return None, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回，" + f"与这次指标{names}的选择相反，未写入。如需更改，请到数据目录中修改") + live = stored if stored is not None and stored.get("status") != "rejected" else None + value = fact.value + if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \ + and live["value"].get("column") == value.get("column"): + # 业务日期按哪一列是这次选出来的事实；目录里同一列另写了规则、时区的，原样留着 + value = live["value"] + elif fact.item == "valid_filter" and isinstance(live, dict) and isinstance(live.get("value"), str) \ + and live["value"].strip() and not _same_fact(fact.item, live["value"], value): + combined, why = _combined_filter(live["value"].strip(), fact) + if combined is None: + return None, why + if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]): + adopted = f"被采纳的{SIDE_LABEL.get(fact.choice, fact.choice)}" if not fact.also else "这次几个指标被采纳的写法" + return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，{adopted}" + f"另外多出条件「{fact.value}」，未改动。如需补上，请到数据目录中修改") + value = combined + elif fact.item == "dedup" and isinstance(live, dict) and live.get("status") != "confirmed" \ + and not _same_fact(fact.item, live.get("value"), value): + # 推断的去重规则说的未必是同一件事（按哪一列去重、哪种汇总要去重），替换掉就把它丢了 + return None, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」，" + f"与这次指标{names}采纳的「{value}」不同，无法确定两者的关系，未覆盖。" + "如需更改，请到数据目录中修改") + plan = catalog.plan_patch(notes, [{"path": fact.item, "value": value}], table=fact.table, tables=tables) + if plan.problems: + return None, f"「{fact.table}」的{label}未写入数据目录：{plan.problems[0]}" + change = plan.changes[0] + if change.state == "same": + return None, f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」，无需重复写入" + if change.before_status == "confirmed": + return None, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」，" + f"与这次指标{names}采纳的「{_value_text(fact.item, value)}」不同，未覆盖。" + "如需更改，请到数据目录中修改") + return change.after, None + + +def preview(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, + notes: Mapping[str, Mapping[str, Any]] | None = None) -> dict[str, dict[str, Any]]: + """审批卡上的预告：审核人在这个指标上选原流程（a）、选盲算（b），按目前的数据目录各会往目录里写什么。 + + candidate 同 derive（choice 不用给）；notes 是这个数据源的目录（{表名: notes}）。返回 {a: …, b: …}，每边是 + {write: true, source, table, item, value}（value 是真要写进去的说法：和目录里已有的有效记录条件合起来的，是合起来 + 之后的整句）或 {write: false, reason}。和恢复后真写是同一套判断，只在模块说明里说的两处可能不一样。永远不抛。""" + notes = notes or {} + tables = schema_cache.get("tables") if isinstance(schema_cache, Mapping) else None + out: dict[str, dict[str, Any]] = {} + for side in SIDE_LABEL: + try: + fact, why = derive({**candidate, "choice": side}, schema_cache, notes) + value = None + if fact is not None: + value, why = _plan(notes.get(fact.table) or {}, fact, tables) + except Exception: # noqa: BLE001 - 预告算不出来不能拦住审批：照实说，以写入后的结果为准 + logger.exception("双盲复算预告数据目录的写入失败（指标 %s）", candidate.get("metric")) + fact, why = None, f"指标「{candidate.get('metric')}」的选择核对数据目录时出错，无法预告是否写入" + out[side] = {"write": True, "source": fact.source, "table": fact.table, "item": fact.item, "value": value} \ + if fact is not None and why is None else {"write": False, "reason": why} + return out async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: str | None) -> tuple[bool, str | None]: - """写一项。(写了没有, 没写的原因)。""" + """写一项。(写了没有, 没写的原因)。写什么、写不写由 _plan 按读到的目录定；版本对不上（别人刚改过）重读再来。""" label = catalog.TABLE_FIELD_LABEL.get(fact.item, fact.item) - side = SIDE_LABEL.get(fact.choice, fact.choice) tables = (row.schema_cache or {}).get("tables") if isinstance(row.schema_cache, dict) else None for _ in range(_RETRIES): entry = await catalog.read_entry(session, row.id, fact.table) notes = entry.notes if entry else {} version = entry.version if entry else 0 - stored = notes.get(fact.item) if isinstance(notes.get(fact.item), dict) else None - if stored is not None and stored.get("status") == "rejected" and _same_fact(fact.item, stored.get("value"), - fact.value): - # 同一个说法以前被人明确驳回过：两次人工判断对不上，不替人拿主意 - return False, (f"「{fact.table}」的{label}「{_value_text(fact.item, fact.value)}」曾被人工驳回，" - f"与这次指标「{fact.metric}」的选择相反，未写入。如需更改，请到数据目录中修改") - live = stored if stored is not None and stored.get("status") != "rejected" else None - value = fact.value - if fact.item == "business_date" and live is not None and isinstance(live.get("value"), dict) \ - and live["value"].get("column") == value.get("column"): - # 业务日期按哪一列是这次选出来的事实；目录里同一列另写了规则、时区的，原样留着 - value = live["value"] - elif fact.item == "valid_filter" and isinstance(live, dict) and isinstance(live.get("value"), str) \ - and live["value"].strip() and not _same_fact(fact.item, live["value"], value): - combined, why = _combined_filter(live["value"].strip(), fact) - if combined is None: - return False, why - if live.get("status") == "confirmed" and not _same_fact(fact.item, combined, live["value"]): - return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{live['value'].strip()}」，被采纳的" - f"{side}另外多一个条件「{fact.value}」，未改动。如需补上，请到数据目录中修改") - value = combined - elif fact.item == "dedup" and isinstance(live, dict) and live.get("status") != "confirmed" \ - and not _same_fact(fact.item, live.get("value"), value): - # 推断的去重规则说的未必是同一件事（按哪一列去重、哪种汇总要去重），替换掉就把它丢了 - return False, (f"「{fact.table}」的{label}已有推断的说法「{_value_text(fact.item, live.get('value'))}」，" - f"与这次指标「{fact.metric}」采纳的「{value}」不同，无法确定两者的关系，未覆盖。" - "如需更改，请到数据目录中修改") - changes = [{"path": fact.item, "value": value, "reason": note}] - plan = catalog.plan_patch(notes, changes, table=fact.table, tables=tables) - if plan.problems: - return False, f"「{fact.table}」的{label}未写入数据目录：{plan.problems[0]}" - change = plan.changes[0] - if change.state == "same": - return False, (f"「{fact.table}」的{label}已是人工确认过的「{_value_text(fact.item, value)}」，" - "无需重复写入") - if change.before_status == "confirmed": - return False, (f"「{fact.table}」的{label}已有人工确认过的说法「{_value_text(fact.item, change.before)}」，" - f"与这次指标「{fact.metric}」采纳的「{_value_text(fact.item, value)}」不同，未覆盖。" - "如需更改，请到数据目录中修改") + value, why = _plan(notes, fact, tables) + if why is not None: + return False, why at = catalog.now_iso() try: - merged = catalog.apply_patch(notes, changes, table=fact.table, tables=tables, at=at) + merged = catalog.apply_patch(notes, [{"path": fact.item, "value": value, "reason": note}], + table=fact.table, tables=tables, at=at) except catalog.CatalogInvalid as e: return False, f"「{fact.table}」的{label}未写入数据目录：{e}" # 不论原来是谁提的：这一项现在出自人的选择，记为人工填写、已确认，备注写明出处 - merged[fact.item] = catalog.make_item(change.after, "human", "confirmed", note=note, at=at) + merged[fact.item] = catalog.make_item(value, "human", "confirmed", note=note, at=at) try: await catalog.write_entry(session, row.id, fact.table, merged, if_version=version, actor=actor) except catalog.CatalogConflict: @@ -640,4 +732,4 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, note: str, actor: return False, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认" -__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "record_choices"] +__all__ = ["FACT_ITEMS", "Fact", "derive", "note_of", "preview", "record_choices"] diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 25e2a736..8d30c94c 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -14,11 +14,14 @@ 里看不出日期、口径的取舍时才给，好让它分得清哪一项是哪一项。 **一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）： -两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件），一条查询算好几列、分组后取一行时 -才不会混进「选出的列」「分组」这类和这一格无关的差别。完整对照落成一件工件（审批 payload 经 runner._safe 会截断， +两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件，HAVING 一并去掉），一条查询算好几列、 +分组后取一行时才不会混进「选出的列」「分组」这类和这一格无关的差别。分组的列设了遮罩时不收窄：这一行的分组值写进 +SQL，就跟着差异说明进了审批卡、对照工件和目录日志。完整对照落成一件工件（审批 payload 经 runner._safe 会截断， 长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有 缺口；b、neither、没选的写进产出的 issuance_gaps，成果节点的出具契约据此降档（on_reject 为 fail 时节点直接失败）。 两边只差一个方面、又对得上数据目录事实的，选择写进数据目录（engine/dual_check_facts.py，指标自己的筛选范围不写）。 +选了会怎样要在卡上看得出：payload 在约定的几项之外带上实际生效的 on_reject，摘要里每个指标带 catalog——选 a、选 b +按发起审批时的目录各会往目录里写什么、不写的为什么（dual_check_facts.preview，和真写同一套判断）。 盲算触顶、出错、交不出数，或者输入本身没法单独复算的指标按「无法复算」记缺口，发 dual_check_unchecked。 **产出的状态。** 全部一致是 agreed；没有不一致、但有无法复算的：一个都没核成是 unchecked，核成了一部分是 partial； @@ -54,6 +57,7 @@ from app.db.models import DataSource from app.engine import recompute from app.engine.approval import read_decision from app.engine.context import NodeContext, NodeError +from app.engine.dual_check_facts import preview, record_choices from app.engine.evidence import build_catalog, is_number, next_exec from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \ parse_expression @@ -907,10 +911,14 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], """不一致的指标逐个说清差在哪，完整对照落成工件，组好审批的 payload。返回 {payload, facts, card_artifact}。 两条 SQL 先收窄到这一格再比（_narrowed）：原流程一条查询算好几列、分组后取一行时，整条拿去比会混进「选出的列」 - 「分组」这类和这一格无关的差别，审批卡上只列前几条，审核人看到的多半是噪声，只差一个方面的也就写不进目录。 + 「分组」这类和这一格无关的差别，审批卡上只列前几条，审核人看到的多半是噪声，只差一个方面的也就写不进目录。两边 + 各按自己那份快照要遮的列收窄（原流程的、盲算的快照各记着查询当时的遮罩，再加数据源现在设的）。 facts 是每个不一致指标写数据目录要用的东西（dual_check_facts.derive 的候选，少一个 choice）：只有一个输入对不上时是 那一处完整的差异条目（带两边比较的写法和涉及的表、列），几个输入都对不上时只说「不止一处」；scope 是这个指标自己的 - 范围（口径说明、所在分组、运行输入、分组的列），有效记录条件据此排除指标自己的筛选范围。 + 范围（口径说明、所在分组、运行输入、分组的列），有效记录条件据此排除指标自己的筛选范围。scope 只在节点里用、不进 + payload 和工件（所在分组的取值里可能有遮罩列的）。 + payload 在约定的几项之外带 on_reject（实际生效的处置），摘要里每个指标带 catalog（选 a、选 b 各会往数据目录里写 + 什么，dual_check_facts.preview）：审核人在卡上要看得出没采用原流程会不会让运行失败、选了会不会记成人工确认的目录项。 """ rows = [r for r in checked["rows"] if r["status"] == DIFFERENT] results = checked["results"] @@ -938,9 +946,11 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], if inp["key"] not in row["differing"]: continue blind_snap, _ = sources.snapshot(str(blind["artifact"])) if blind.get("artifact") else (None, None) - a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind) or inp["sql"] - b_sql = _narrowed(blind.get("sql"), blind.get("column"), _record(blind_snap, blind.get("row")), kind) \ - or blind.get("sql") + blind_hidden = await sources.hidden(str(blind["artifact"])) if blind.get("artifact") else frozenset() + a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind, + await sources.hidden(inp["artifact"])) or inp["sql"] + b_sql = _narrowed(blind.get("sql"), blind.get("column"), _record(blind_snap, blind.get("row")), kind, + blind_hidden) or blind.get("sql") diff = await _explain(a_sql, b_sql, infos[inp["source"]]) per_input.append((inp, blind, diff)) compared.append((a_sql, str(b_sql or ""))) @@ -960,22 +970,24 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], "items": [{"aspect": i.get("aspect"), "text": i.get("text")} for i in merged["items"]], "fallback": merged["fallback"], "text_diff": merged["text_diff"]}, }) - summary.append({ - "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], - "original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"], - "delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]], - "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]), - }) first = per_input[0][0] if per_input else (row["inputs"][0] if row["inputs"] else {}) single = compared[0] if len(per_input) == 1 else None + info = infos.get(str(first.get("source"))) or {} facts[row["metric_id"]] = { "metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"), - "dialect": (infos.get(str(first.get("source"))) or {}).get("kind"), + "dialect": info.get("kind"), "sql": {"a": single[0], "b": single[1]} if single else {}, "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None, "items": merged["items"]}, "scope": {"texts": [t for t in dict.fromkeys(scope_texts) if t], "columns": sorted(scope_columns)}, } + summary.append({ + "metric_id": row["metric_id"], "name": row["name"], "unit": row["unit"], + "original_rendered": row["original"]["rendered"], "blind_rendered": row["blind"]["rendered"], + "delta_pct": row["delta_pct"], "diff": [str(i.get("text")) for i in merged["items"][:SUMMARY_DIFFS]], + "fallback": not any(i.get("aspect") not in ("other", "parse") for i in merged["items"]), + "catalog": preview(facts[row["metric_id"]], info.get("schema"), info.get("notes")), + }) content = {"metrics": metrics, "model": checked.get("model"), "same_model": bool(checked.get("same_model"))} try: artifact = await artifact_store.put_json(content, kind="dual_check", run_id=ctx.run.run_id, @@ -985,7 +997,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], payload = { "kind": "dual_check", "node_id": ctx.node.id, "mode": "input", "title": f"双盲复算：{len(rows)} 个指标两边结果不一致，请选择采用哪一边的结果", - "card_artifact": artifact, "summary": summary, + "card_artifact": artifact, "summary": summary, "on_reject": options.on_reject, } names = "、".join(f"「{r['name']}」" for r in rows[:5]) + (f"等 {len(rows)} 个" if len(rows) > 5 else "") ctx.emit(EventType.LOG, level="warn", code="dual_check_disagreed", @@ -995,17 +1007,18 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], async def _source_info(sources: _Sources, name: str) -> dict[str, Any]: - """比较两条 SQL 要用的：方言、表结构、列的中文名（数据目录里的）。数据源没了就都不给，照样比原文。""" + """比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名和整份数据目录。数据源没了就都不给，照样比原文。""" from app.data import catalog from app.data.sql_semantics import catalog_labels row = await sources.source(name) if row is None: - return {"kind": None, "schema": None, "names": {}} + return {"kind": None, "schema": None, "names": {}, "notes": {}} async with SessionLocal() as session: entries = await catalog.read_catalog(session, row.id) + notes = {table: entry.notes for table, entry in entries.items()} return {"kind": row.kind, "schema": row.schema_cache if isinstance(row.schema_cache, dict) else None, - "names": catalog_labels({table: entry.notes for table, entry in entries.items()})} + "names": catalog_labels(notes), "notes": notes} async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]: @@ -1088,12 +1101,16 @@ def _sql_literal(value: Any) -> Any: return None -def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str | None) -> str | None: +def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str | None, + hidden: frozenset[str] = frozenset()) -> str | None: """把一条查询收窄到这一格：最外层 SELECT 只留算出这一列的那一项；分组的去掉分组，按这一行的分组值补上等值条件 - （华东那一行 → WHERE … AND region = '华东'），排序、取行数跟着去掉。 + （华东那一行 → WHERE … AND region = '华东'），HAVING、排序、取行数跟着去掉——这一行既然在结果里，它所在的组就 + 满足 HAVING，留着只会多出一条和这一格无关的差异。 差异说明拿收窄后的两条比，才只说这一格是怎么算出来的。认不准的（SELECT *、DISTINCT、窗口函数、集合运算、分组键 - 对不上结果列、这一行没有分组值）返回 None，调用方照整条比。""" + 对不上结果列、这一行没有分组值）返回 None，调用方照整条比。hidden 是这份快照要遮的列（小写）：分组键的结果列、 + 或者它用到的列在里面时也返回 None——分组值写进 SQL 就跟着差异说明进了审批卡、对照工件和目录日志，遮罩列的值 + 不能出现在那些地方。""" import sqlglot from sqlglot import exp @@ -1128,6 +1145,8 @@ def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str if name is None or name not in values: return None expr = projections[index].this if isinstance(projections[index], exp.Alias) else projections[index] + if name in hidden or any(col.name.lower() in hidden for col in expr.find_all(exp.Column)): + return None if values[name] is None: conditions.append(exp.Is(this=expr.copy(), expression=exp.Null())) continue @@ -1135,7 +1154,7 @@ def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str if literal is None: return None conditions.append(exp.EQ(this=expr.copy(), expression=literal)) - for arg in ("group", "order", "limit", "offset"): + for arg in ("group", "having", "order", "limit", "offset"): tree.set(arg, None) tree = tree.where(*conditions, copy=False) elif tree.args.get("limit") is None and tree.args.get("offset") is None: @@ -1233,7 +1252,6 @@ def _resolved_text(rows: list[dict[str, Any]], choices: dict[str, str]) -> str: async def _catalog(ctx: NodeContext, candidates: list[dict[str, Any]]) -> dict[str, Any]: from app.data.catalog import TABLE_FIELD_LABEL - from app.engine.dual_check_facts import record_choices result = await record_choices(candidates, run_id=ctx.run.run_id, node_title=ctx.node.title, actor=ctx.actor()) written, skipped = result["written"], result["skipped"] diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index 9f263441..d75bafe6 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -7,7 +7,10 @@ - 指标自己的筛选范围（只看华东、只看线上）不是有效记录条件：取值出现在指标名称、口径说明、所在分组、运行输入里的， 条件用到分组的列的，不是在排除退款、作废、测试这类无效记录的，都不写 - 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖； - 目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动；同一次裁决里互相矛盾的选择都不写 + 目录里已有的有效记录条件被采纳的一边也用着的，合起来写，没用着的不动 +- 同一次裁决里几个指标落到同一项上：说法一样的写一次；有效记录条件（都在排除无效记录）用 AND 合成一句写；业务日期 + 说法不同的都不写 +- 审批卡上的预告（preview）：选 a、选 b 各会写什么、不写的为什么，按目录里已有的说法算，和真写同一套判断 差异条目一律由 sql_semantics.explain_sql_diff 真算出来，和节点交给审核人看的是同一份。表名、数字都是虚构的。 """ @@ -18,7 +21,7 @@ import pytest from app.data import catalog from app.data.sql_semantics import explain_sql_diff from app.db.base import SessionLocal -from app.engine.dual_check_facts import derive, record_choices +from app.engine.dual_check_facts import derive, preview, record_choices from tests.fixtures.dual_check_shop import ( BOTH_WRONG, BUYERS, @@ -237,6 +240,13 @@ async def seed(source_id: str, notes: dict) -> None: await catalog.write_entry(session, source_id, "orders", notes, if_version=0, actor="数据组") +async def seed_again(source_id: str, notes: dict) -> None: + """把目录这张表整份改回 notes（在现有版本上写）。""" + async with SessionLocal() as session: + entry = await catalog.read_entry(session, source_id, "orders") + await catalog.write_entry(session, source_id, "orders", notes, if_version=entry.version, actor="数据组") + + async def test_a_choice_is_written_as_a_confirmed_human_item_with_its_origin(shop): name, source_id = shop out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source=name)], run_id="run-42", @@ -359,7 +369,8 @@ async def test_a_rejected_item_is_not_brought_back_but_a_rejected_other_value_do async def test_contradictory_choices_in_one_decision_write_nothing(shop): - """同一次裁决里，一个指标采纳了按下单日、另一个采纳了按支付日：写哪个都是替人拿主意，都不写。""" + """同一次裁决里，一个指标采纳了按下单日、另一个采纳了按支付日：一张表只有一个业务日期，写哪个都是替人拿主意， + 都不写。""" name, source_id = shop out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "a", metric="销售额", source=name), candidate(BY_ORDER_DATE, RIGHT, "b", metric="复核销售额", source=name)], @@ -367,10 +378,94 @@ async def test_contradictory_choices_in_one_decision_write_nothing(shop): assert out["written"] == [] [skipped] = out["skipped"] assert "「销售额」" in skipped["reason"] and "「复核销售额」" in skipped["reason"] - assert "order_date" in skipped["reason"] and "paid_date" in skipped["reason"] and "互相矛盾" in skipped["reason"] + assert "order_date" in skipped["reason"] and "paid_date" in skipped["reason"] + assert "给出了不同的说法" in skipped["reason"] and "无法确定哪一个是数据事实" in skipped["reason"] assert await entry_of(source_id) is None +async def test_complementary_valid_filters_in_one_decision_are_combined(shop): + """同一次裁决里，一个指标采纳了排除退款、另一个采纳了排除测试单：两条都在排除无效记录，各有人的选择撑着， + 不互相矛盾，用 AND 合成一句写。""" + name, source_id = shop + out = await record_choices([ + candidate(SUM_WEEK, extra("status <> 'refunded'"), "b", metric="销售额", source=name), + candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name), + # 第三个指标采纳的是和第一个一样的条件：合并时不重复 + candidate(COUNT_WEEK, extra("status <> 'refunded'", COUNT_WEEK), "b", metric="订单数", source=name), + ], run_id="run-1", node_title="复核", actor=None) + assert out == {"written": [{"source": name, "table": "orders", "item": "valid_filter"}], "skipped": []} + entry = await entry_of(source_id) + item = entry.notes["valid_filter"] + assert item["value"] == "status <> 'refunded' AND is_test = 0" and entry.version == 1 + assert (item["source"], item["status"]) == ("human", "confirmed") + assert all(f"「{m}」" in item["note"] for m in ("销售额", "订单金额", "订单数")) and "run-1" in item["note"] + + +async def test_combined_filters_extend_an_inferred_condition_every_adopted_side_uses(shop): + """目录里推断的「channel <> 'test'」：几个被采纳的一边都按它筛了，合起来写；有一个没按它筛的，关系拿不准，不动。""" + name, source_id = shop + await seed(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")}) + base = extra("channel <> 'test'") + both = [candidate(base, extra("status <> 'refunded'", base), "b", metric="销售额", source=name), + candidate(base, extra("is_test = 0", base), "b", metric="订单金额", source=name)] + out = await record_choices(both, run_id="run-1", node_title="复核", actor=None) + assert out["written"] == [{"source": name, "table": "orders", "item": "valid_filter"}], out + assert (await entry_of(source_id)).notes["valid_filter"]["value"] == \ + "channel <> 'test' AND status <> 'refunded' AND is_test = 0" + + await seed_again(source_id, {"valid_filter": catalog.make_item("channel <> 'test'", "llm")}) + one_without = [both[0], candidate(SUM_WEEK, extra("is_test = 0"), "b", metric="订单金额", source=name)] + out = await record_choices(one_without, run_id="run-2", node_title="复核", actor=None) + assert out["written"] == [] + reason = out["skipped"][0]["reason"] + assert "「channel <> 'test'」" in reason and "指标「订单金额」被采纳的盲算没有按它筛选" in reason and "未改动" in reason + assert (await entry_of(source_id)).notes["valid_filter"]["value"] == "channel <> 'test'" + + +async def test_the_same_choice_from_two_metrics_is_written_once(shop): + name, source_id = shop + out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", metric="销售额", source=name), + candidate(BY_ORDER_DATE, RIGHT, "b", metric="订单金额", source=name)], + run_id="run-1", node_title="复核", actor=None) + assert out == {"written": [{"source": name, "table": "orders", "item": "business_date"}], "skipped": []} + entry = await entry_of(source_id) + assert entry.version == 1 and "「销售额」" in entry.notes["business_date"]["note"] + assert "「订单金额」" in entry.notes["business_date"]["note"] + + +# -------------------------------------------------------------------------- +# 审批卡上的预告：选哪一边各会写什么（和真写同一套判断） +# -------------------------------------------------------------------------- + + +def test_the_preview_says_what_each_choice_would_write(): + picked = candidate(BY_ORDER_DATE, RIGHT, "a") + assert preview(picked, SCHEMA, {}) == { + "a": {"write": True, "source": "shop", "table": "orders", "item": "business_date", + "value": {"column": "order_date"}}, + "b": {"write": True, "source": "shop", "table": "orders", "item": "business_date", + "value": {"column": "paid_date"}}, + } + # 只有被采纳的一边有条件时才写：选另一边照实说为什么不写 + said = preview(candidate(WITH_REFUNDS, RIGHT, "a"), SCHEMA, {}) + assert said["a"]["write"] is False and "被采纳的原流程没有这个条件" in said["a"]["reason"] + assert said["b"] == {"write": True, "source": "shop", "table": "orders", "item": "valid_filter", + "value": "status <> 'refunded'"} + + +def test_the_preview_reads_the_catalog_as_it_is(): + """目录里已有的说法照写入时的规矩算：确认过的不覆盖、同样的说法不重写、推断的有效记录条件合起来写。""" + confirmed = {"orders": {"business_date": catalog.make_item({"column": "order_date"}, "human", "confirmed")}} + said = preview(candidate(BY_ORDER_DATE, RIGHT, "a"), SCHEMA, confirmed) + assert said["a"]["write"] is False and "无需重复写入" in said["a"]["reason"] + assert said["b"]["write"] is False and "已有人工确认过的说法「order_date」" in said["b"]["reason"] + + inferred = {"orders": {"valid_filter": catalog.make_item("is_test = 0", "llm")}} + base = f"SELECT SUM(amount) AS gmv FROM orders WHERE is_test = 0 AND paid_date {WEEK}" + said = preview(candidate(base, extra("status <> 'refunded'", base), "a"), SCHEMA, inferred) + assert said["b"]["write"] is True and said["b"]["value"] == "is_test = 0 AND status <> 'refunded'" + + async def test_a_missing_source_or_an_unrecordable_choice_is_reported(shop): name, _ = shop out = await record_choices([candidate(BY_ORDER_DATE, RIGHT, "b", source="no_such_source"), diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index c68837cf..5ed4c487 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -4,13 +4,15 @@ - 一致：不挂起，产出 agreed，日志 dual_check_agreed；盲算的查询发 tool.end（role: verify），台账只有一条 dual_check， 用量计入运行；盲算 agent 的任务里没有原 SQL，也没有原流程结果的列名 - 一个都没核成是 unchecked、只核成一部分是 partial，日志 dual_check_unchecked 说清哪几个、为什么 -- 不一致：挂起，审批 payload 和对照工件按约定；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject - 为 fail 时失败 -- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比） +- 不一致：挂起，审批 payload 和对照工件按约定，payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往 + 数据目录里写什么的预告；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject 为 fail 时失败 +- 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比，HAVING 一并去掉）；分组的列设了 + 遮罩时不收窄，遮罩列的取值不进审批卡、对照工件、产出和事件 - 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的 - 筛选范围不写，两条 SQL 规范化后一致的不写 + 筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写 - 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询 -- 触顶、分组结果的维度（只给取值、遮罩列不给）、两个指标用到同一格只盲算一次、遮罩列上的盲算结果不采用 +- 触顶、分组结果的维度（只给取值、遮罩列不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的 + 盲算结果不采用 """ from __future__ import annotations @@ -234,6 +236,23 @@ async def test_the_blind_agent_is_not_told_which_date_column_the_original_groupe assert "order_date" not in task and "gmv" not in task +async def test_column_names_carrying_a_choice_are_not_given_as_hints(monkeypatch, shop): + """几项输入都来自查询、又没写口径说明时，列名才交给盲算区分计算式里的几项；列名里带着原流程的取舍的（按下单日、 + 已支付）不给：「gmv_by_order_date」等于告诉它按下单日归属。看不出取舍的（total）照给。""" + name, _ = shop + wide = (f"SELECT SUM(amount) AS gmv_by_order_date, COUNT(*) AS paid_orders, SUM(amount) AS total FROM orders " + f"WHERE status <> 'refunded' AND order_date {WEEK}") + aov = {"id": "aov", "name": "客单价", "unit": "元", "decimals": 2, "format": "plain", + "expression": "(cell(nodes.fetch, 0, 'gmv_by_order_date') + cell(nodes.fetch, 0, 'total')) " + "/ cell(nodes.fetch, 0, 'paid_orders')"} + blind = Blind(monkeypatch, name, {}) # 盲算都交不出数：这里只看任务原文 + await start(graph(name, wide, metrics=[aov], dual={"definitions": {}})) + assert len(blind.tasks) == 3 + said = "\n".join(blind.tasks) + assert "gmv_by_order_date" not in said and "paid_orders" not in said + assert "「total」" in said + + async def test_two_metrics_using_one_cell_are_checked_once(monkeypatch, shop): name, _ = shop share = {"id": "gmv_k", "name": "销售额（千元）", "unit": "千元", "decimals": 2, "format": "thousands", @@ -352,16 +371,25 @@ async def test_disagreement_pauses_with_the_agreed_payload_and_card(monkeypatch, approval = await pending(run.id) assert approval.mode == "input" and approval.node_id == "dual" payload = approval.payload - assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary"} + # 约定的六个键，另加 on_reject（实际生效的处置）：审核人在卡上要看得出没采用原流程会不会让运行失败 + assert set(payload) == {"kind", "node_id", "mode", "title", "card_artifact", "summary", "on_reject"} assert payload["kind"] == "dual_check" and payload["node_id"] == "dual" and payload["mode"] == "input" + assert payload["on_reject"] == "degrade" [item] = payload["summary"] assert set(item) == {"metric_id", "name", "unit", "original_rendered", "blind_rendered", "delta_pct", "diff", - "fallback"} + "fallback", "catalog"} assert (item["metric_id"], item["name"], item["unit"]) == ("gmv", "销售额", "元") assert item["original_rendered"].startswith("310") and item["blind_rendered"].startswith("330") assert item["delta_pct"] == round((330 - 310) / 310 * 100, 2) assert item["fallback"] is False and len(item["diff"]) == 1 assert "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] and "原流程" in item["diff"][0] + # 选哪一边会往数据目录里写什么，按发起审批时的目录预告：只差时间列一处，选哪边就记哪一列 + assert item["catalog"] == { + "a": {"write": True, "source": name, "table": "orders", "item": "business_date", + "value": {"column": "order_date"}}, + "b": {"write": True, "source": name, "table": "orders", "item": "business_date", + "value": {"column": "paid_date"}}, + } card = artifact_store.load(payload["card_artifact"]) assert set(card) == {"metrics", "model", "same_model"} and card["same_model"] is False @@ -470,6 +498,8 @@ async def test_on_reject_fail_stops_the_run(monkeypatch, shop, client): name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) run = await start(graph(name, BY_ORDER_DATE, dual={"on_reject": "fail"})) + # 审批卡上看得出：这一步没采用原流程，运行就会停下 + assert (await pending(run.id)).payload["on_reject"] == "fail" done = await decide(client, run.id, {"gmv": "b"}) assert done.status == "failed" and "审核人没有采用原流程的结果" in (done.error or "") and "销售额" in done.error evs = await events(run.id) @@ -546,6 +576,10 @@ async def test_a_confirmed_catalog_entry_is_not_overwritten(monkeypatch, shop, c actor="数据组") Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) run = await start(graph(name, BY_ORDER_DATE)) + # 预告按目录里已有的说法算：选原流程是同一个说法、不用再写，选盲算不会覆盖已确认的 + preview = (await pending(run.id)).payload["summary"][0]["catalog"] + assert preview["a"]["write"] is False and "无需重复写入" in preview["a"]["reason"] + assert preview["b"]["write"] is False and "已有人工确认过的说法「order_date」" in preview["b"]["reason"] done = await decide(client, run.id, {"gmv": "b"}) [wrote] = logs(await events(done.id), "dual_check_catalog") assert wrote["written"] == [] @@ -564,6 +598,9 @@ async def test_a_metric_scope_filter_is_not_written_as_a_valid_filter(monkeypatc run = await start(graph(name, original, metrics=[east], dual={"definitions": {"east": "华东地区已支付订单的金额"}})) [item] = (await pending(run.id)).payload["summary"] assert len(item["diff"]) == 1 and "region = '华东'" in item["diff"][0] + # 预告和真写的是同一套判断：两边选哪个都不写，原因照实 + assert item["catalog"]["a"]["write"] is False and "可能是指标自己的筛选范围" in item["catalog"]["a"]["reason"] + assert item["catalog"]["b"]["write"] is False and "被采纳的盲算没有这个条件" in item["catalog"]["b"]["reason"] done = await decide(client, run.id, {"east": "a"}) assert done.status == "succeeded", done.error [wrote] = logs(await events(run.id), "dual_check_catalog") @@ -631,6 +668,61 @@ async def test_a_grouped_query_is_compared_on_the_row_it_was_taken_from(monkeypa assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"} +async def test_a_having_clause_is_dropped_when_narrowing_to_one_row(monkeypatch, shop, client): + """原流程分组后另有 HAVING：取的这一行既然在结果里，它所在的组就满足 HAVING，对这一格是噪声。收窄时连 HAVING + 一起去掉，差异只剩真正的那一处（时间列），写得进目录。""" + name, source_id = shop + by_region = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND order_date {WEEK} " + "GROUP BY region HAVING SUM(amount) > 10 ORDER BY region") + east_paid = RIGHT.replace("WHERE ", "WHERE region = '华东' AND ") + east = {**GMV, "id": "east", "name": "华东销售额"} + Blind(monkeypatch, name, {"华东销售额": (east_paid, "gmv")}) + run = await start(graph(name, by_region, metrics=[east], dual={"definitions": {}})) + [item] = (await pending(run.id)).payload["summary"] + assert item["original_rendered"].startswith("200") and item["blind_rendered"].startswith("300") + assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] + await decide(client, run.id, {"east": "b"}) + assert (await notes_of(source_id)).notes["business_date"]["value"] == {"column": "paid_date"} + + +async def test_a_condition_on_a_grouped_column_is_not_written(monkeypatch, shop, client, tmp_path): + """原流程按 is_test 分组、口径卡取 is_test = 0 那一行：收窄后原流程多出「is_test = 0」，看着像排除测试单的有效 + 记录条件，可它用的是原流程分组的列——是这个指标按组取的范围，不写目录。取值 0 太短、不在任何文字里，只有分组的列 + 拦得下它。""" + name, source_id = shop + db = sqlite3.connect(tmp_path / "shop.db") + db.execute("INSERT INTO orders VALUES (6, '华东', 105, 40.0, 'paid', '2026-09-03', '2026-09-03', 'online', 1)") + db.commit() + db.close() + by_flag = (f"SELECT is_test, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' AND paid_date {WEEK} " + "GROUP BY is_test ORDER BY is_test") + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, by_flag, dual={"definitions": {}})) + [item] = (await pending(run.id)).payload["summary"] + assert item["original_rendered"].startswith("330") and item["blind_rendered"].startswith("370") + assert len(item["diff"]) == 1 and "is_test = 0" in item["diff"][0] + done = await decide(client, run.id, {"gmv": "a"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "分组的列" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + +async def test_a_condition_matching_a_run_input_is_not_written(monkeypatch, shop, client): + """排除退款的条件看着像有效记录条件，可取值正是本次运行的输入（这一期要排除的状态）：是这个指标这一期的筛选范围， + 不是整张表的事实，不写目录。没有这个输入时同样的选择会写（test_a_refund_filter_becomes_the_valid_row_filter）。""" + name, source_id = shop + spec = graph(name, RIGHT, dual={"definitions": {}}) + spec["nodes"][0]["data"]["config"]["fields"] = [{"name": "exclude_status"}] + Blind(monkeypatch, name, {"销售额": (WITH_REFUNDS, "gmv")}) + run = await start(spec, {"exclude_status": "refunded"}) + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 1 and "refunded" in item["diff"][0] + done = await decide(client, run.id, {"gmv": "a"}) + [wrote] = logs(await events(done.id), "dual_check_catalog") + assert wrote["written"] == [] and "本次运行的输入" in wrote["skipped"][0]["reason"] + assert await notes_of(source_id) is None + + async def test_two_differing_inputs_do_not_write_the_catalog(monkeypatch, shop, client): """一个指标的两个输入都对不上：审核人选的是整个指标哪边对，拆不出是哪一处，不写目录。""" name, source_id = shop @@ -733,6 +825,50 @@ async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch assert "华东" not in task and "region" not in task +async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monkeypatch, masked_shop, client): + """原流程按遮罩列（region）分组、口径卡取华东那一行：差异说明收窄到这一行时，不能拿这一行的分组值补等值条件—— + 那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄，照整条比。""" + name, _ = masked_shop + first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}})) + assert run.status == "interrupted", run.error + payload = (await pending(run.id)).payload + 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 + [wrote] = logs(evs, "dual_check_catalog") + assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] + + +async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client): + """盲算自己按遮罩列分组、交了华东那一行：同样不拿这一行的分组值改写盲算的 SQL。""" + name, _ = masked_shop + Blind(monkeypatch, name, {"销售额": (BY_REGION, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + payload = (await pending(run.id)).payload + assert payload["summary"][0]["blind_rendered"].startswith("300") + card = artifact_store.load(payload["card_artifact"]) + done = await decide(client, run.id, {"gmv": "b"}) + assert done.status == "succeeded", done.error + _no_masked_value(payload, card, done, await events(run.id)) + + +def _no_masked_value(payload: dict, card: dict, done: Run, evs: list[RunEvent], value: str = "华东") -> None: + """遮罩列上的取值不出现在审批 payload、对照工件、节点产出和节点发的事件里。查询本身的 tool.end、模型的往来按 + 设计带着原始结果（证据面板显示时再遮），不在此列。""" + seen = {"payload": payload, "card": card, "output": dual_output(done, evs), + "events": [e.data for e in evs if (e.node_id == "dual" or e.type.startswith("human.") or e.type == "log") + and not e.type.startswith(("tool.", "llm."))]} + for where, content in seen.items(): + assert value not in json.dumps(content, ensure_ascii=False), where + + # -------------------------------------------------------------------------- # 设置：全局开关、节点覆盖、发起时的快照 # --------------------------------------------------------------------------