diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index 112ed028..ba227b3d 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -4,10 +4,13 @@ 对得上的事实只有三种（data/catalog.py 的表级项）： - 两边按不同的时间列筛时间 → 业务日期（business_date）：被采纳的一边按哪一列筛，业务日期就记哪一列。只有一边按时间 - 筛的不算：审核人判断的是要不要限定时间，没在两列之间选过。两边的列都得在表结构里声明为日期 / 时间类型（DATE、 - DATETIME、TIMESTAMP、TIMESTAMPTZ、TIME，带精度、带时区的写法也算）：差异说明认时间列只看列名和常量像不像日期， + 筛的不算：审核人判断的是要不要限定时间，没在两列之间选过。两边的列都得在表结构里声明为带日期的类型（DATE、 + DATETIME、TIMESTAMP、TIMESTAMPTZ，带精度、带时区的写法也算）：差异说明认时间列只看列名和常量像不像日期， 文本列（SQLite 常把日期存成文字）、整数列（Unix 秒、20260901 这类日期键）、名字带 time 的时长列都可能被说成「按它筛 - 时间」，认不准存的是不是日期，不记 + 时间」，认不准存的是不是日期，不记。只有时刻的 TIME 也不记：一天里的几点几分（MySQL 还常拿它存时长）不是日期，和 + catalog.is_date_column（剖析和 SQL 检查共用的日期列判断）同一个口径——记成业务日期，SQL 检查会对按真实日期列统计的 + 查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，这一条门槛下它们的业务日期都不由这里记（原因照实说 + 是上传的表格按文字保存，不说成用户声明了 TEXT） - 一边多一个排除无效记录的条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 - 聚合只差去不去重、被采纳的一边去重 → 去重规则（dedup）。采纳了不去重的一边不记：那是这个指标自己的算法（数行）， 写成表的「不去重」会把别的指标带偏 @@ -96,11 +99,14 @@ _MARKS_INVALID = _INVALID_WORDS | {"del", "removed", "hidden", "archived", "inte _MARKS_VALID = frozenset({"valid", "enabled", "active", "effective"}) #: 取值和指标名称、口径说明比对时，数字至少这么长才比（0、1 这种标记值到处都是） _SCOPE_DIGITS = 2 -#: 表结构里声明的日期 / 时间类型（业务日期只记这样的列）：DATE、DATETIME、TIMESTAMP、TIMESTAMPTZ、TIME、TIMETZ，可以带 -#: 精度（TIMESTAMP(3)、DATETIME(6)）、带时区（TIMESTAMP WITH TIME ZONE、TIME WITHOUT TIME ZONE、Oracle 的 WITH LOCAL -#: TIME ZONE），大小写不论。整个类型名都得是这个样子：DATE_KEY、INTERVAL 这类名字里带着 date、time 的不算 -_TEMPORAL_TYPE = re.compile(r"(?:date|datetime|timestamp|timestamptz|time|timetz)(?:\s*\(\s*\d+\s*\))?" - r"(?:\s+with(?:out)?\s+(?:local\s+)?time\s+zone)?", re.IGNORECASE) +#: 表结构里声明的带日期的类型（业务日期只记这样的列）：DATE、DATETIME、TIMESTAMP、TIMESTAMPTZ，可以带精度 +#: （TIMESTAMP(3)、DATETIME(6)）、带时区（TIMESTAMP WITH TIME ZONE、Oracle 的 WITH LOCAL TIME ZONE），大小写不论。整个 +#: 类型名都得是这个样子：DATE_KEY、INTERVAL 这类名字里带着 date、time 的不算 +_DATED_TYPE = re.compile(r"(?:date|datetime|timestamp|timestamptz)(?:\s*\(\s*\d+\s*\))?" + r"(?:\s+with(?:out)?\s+(?:local\s+)?time\s+zone)?", re.IGNORECASE) +#: 只有时刻的类型（TIME、TIMETZ、TIME(3)、TIME WITH TIME ZONE）：不记，原因另说「只有时刻」 +_TIME_ONLY_TYPE = re.compile(r"(?:time|timetz)(?:\s*\(\s*\d+\s*\))?(?:\s+with(?:out)?\s+time\s+zone)?", + re.IGNORECASE) @dataclass(frozen=True) @@ -138,10 +144,11 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, """一个候选（审核人在某个指标上选了 a 或 b）→ (要写的目录项, None) 或 (None, 为什么不写)。 candidate：{metric 指标名, choice "a" | "b", source 数据源名, dialect, sql: {a, b}（比较的两条 SQL）, diff: - {equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}}。scope 是这个指标自己的范围： - texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效记录条件用它排除 - 指标自己的筛选范围。schema_cache 是数据源的表结构：认表，业务日期还要看两边的列声明的类型。notes 是这个数据源的 - 目录（{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 + {equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}, uploaded}。scope 是这个指标自己的 + 范围：texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效记录条件用它 + 排除指标自己的筛选范围。uploaded 为真是上传表格建的数据源：业务日期不记时，文本列的原因照实说是上传的表格按文字 + 保存日期。schema_cache 是数据源的表结构：认表，业务日期还要看两边的列声明的类型。notes 是这个数据源的目录 + （{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 """ name = str(candidate.get("metric") or "") side = candidate.get("choice") @@ -157,7 +164,7 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, item = items[0] aspect = str(item.get("aspect") or "") if aspect == "time": - fact, why = _business_date(item, side, schema_cache) + fact, why = _business_date(item, side, schema_cache, uploaded=candidate.get("uploaded") is True) elif aspect == "filter": fact, why = _valid_filter(item, side, candidate, schema_cache, notes or {}) elif aspect == "aggregate": @@ -179,10 +186,12 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | None, metric_id=str(candidate.get("metric_id") or "")), None -def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str, Any] | None - ) -> tuple[tuple[str, str, Any] | None, str | None]: +def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str, Any] | None, *, + uploaded: bool = False) -> tuple[tuple[str, str, Any] | None, str | None]: """两边按不同的列筛时间、被采纳的一边按哪一列：(表, business_date, {column}) 或 (None, 为什么不记)。两边的列都要在 - 表结构里声明为日期 / 时间类型（_temporal_problem），被采纳的一边先看：它有问题说它，没问题再看另一边。""" + 表结构里声明为带日期的类型（_dated_problem），被采纳的一边先看：它有问题说它；没问题再看未被采纳的一边，另说清为什么 + 也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源还没有可用的表结构时（没探查、探查 + 失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。""" other = "b" if side == "a" else "a" detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} @@ -199,37 +208,49 @@ def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str return None, "有一边按好几列筛时间，无法确定业务日期按哪一列" if (mine[0]["table"], mine[0]["column"]) == (theirs[0]["table"], theirs[0]["column"]): return None, "两边按同一列筛时间，没有可记的业务日期" - for label, ref in ((SIDE_LABEL[side], mine[0]), (SIDE_LABEL[other], theirs[0])): - if (problem := _temporal_problem(schema_cache, ref)) is not None: - return None, f"{label}按时间筛的列「{ref['table']}.{ref['column']}」{problem}，无法确认是日期 / 时间类型" + tables = schema_cache.get("tables") if isinstance(schema_cache, Mapping) else None + if not isinstance(tables, Mapping) or not tables: + return None, "所在数据源还没有可用的表结构（尚未探查或探查失败），无法确认两边按时间筛的列存的是日期" + for adopted, ref in ((True, mine[0]), (False, theirs[0])): + if (problem := _dated_problem(tables, ref, uploaded)) is None: + continue + said = f"按时间筛的列「{ref['table']}.{ref['column']}」{problem}" + if adopted: + return None, f"被采纳的{SIDE_LABEL[side]}{said}" + return None, f"未被采纳的{SIDE_LABEL[other]}{said}：两边都是日期列时，审核人选的才是业务日期按哪一列" return (str(mine[0]["table"]), "business_date", {"column": str(mine[0]["column"])}), None -def _temporal_problem(schema_cache: Mapping[str, Any] | None, ref: Mapping[str, Any]) -> str | None: - """按时间筛的这一列在表结构里是不是声明成了日期 / 时间类型（_TEMPORAL_TYPE）：是的为 None，不是的交回它怎么了 - （查不到、没声明类型、声明成了别的类型），接在列名后面读。 +def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded: bool) -> str | None: + """按时间筛的这一列在表结构里是不是声明成了带日期的类型（_DATED_TYPE）：是的为 None，不是的交回它怎么了（查不到、没 + 声明类型、只有时刻、声明成了别的类型）连同结论，接在「按时间筛的列「表.列」」后面读。 表按 catalog.resolve_table_name 认，列先按原样、再按 name_key 认（差异说明里的名字是规范化过的，SQLite 不加引号的 - 名字一律小写）。按 name_key 认出好几列的说对应好几列：哪一列是它说不准。""" - cache = schema_cache if isinstance(schema_cache, Mapping) else None - table = catalog.resolve_table_name(cache, str(ref["table"])) - tables = cache.get("tables") if cache is not None else None - meta = tables.get(table) if isinstance(tables, Mapping) and table is not None else None + 名字一律小写）。按 name_key 认出好几列的说对应好几列：哪一列是它说不准。类型照表结构里记的说（「类型是」，不说 + 「声明为」）：introspect 记的是 SQLAlchemy 反射出来的类型，SQLite 里写成 TIMESTAMPTZ 的列反射出来是 NUMERIC；没写 + 类型的列反射出来是 NullType，记成「NULL」，按没有声明类型说。上传表格建的数据源（uploaded）日期一律存成文字，文字 + 类型不说「类型是 TEXT」：那不是用户声明的，导入确认时这一列标的还是「日期」，表结构也改不了。""" + found_table = catalog.resolve_table_name({"tables": tables}, str(ref["table"])) + meta = tables.get(found_table) if found_table is not None else None columns = [c for c in meta.get("columns") or [] if isinstance(c, Mapping) and c.get("name")] \ if isinstance(meta, Mapping) else [] column = str(ref["column"]) found = [c for c in columns if str(c["name"]) == column] \ or [c for c in columns if name_key(str(c["name"])) == name_key(column)] if not found: - return "不在数据源的表结构里" + return "不在数据源的表结构里，无法确认存的是日期" if len(found) > 1: - return "在表结构里对应好几列" + return "在表结构里对应好几列，无法确认存的是日期" declared = " ".join(str(found[0].get("type") or "").split()) - if not declared: - return "在表结构里没有声明类型" - if not _TEMPORAL_TYPE.fullmatch(declared): - return f"在表结构里声明为 {declared}" - return None + if not declared or declared.upper() == "NULL": + return "在表结构里没有声明类型，无法确认存的是日期" + if _DATED_TYPE.fullmatch(declared): + return None + if _TIME_ONLY_TYPE.fullmatch(declared): + return f"在表结构里的类型是 {declared}，只有时刻，不含日期" + if uploaded and any(word in declared.upper() for word in catalog._TEXT_TYPES): + return "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期" + return f"在表结构里的类型是 {declared}，无法确认存的是日期" def _valid_filter(item: Mapping[str, Any], side: str, candidate: Mapping[str, Any], diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 59213c71..f9b82a93 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -16,12 +16,16 @@ **遮罩列上的输入不复算。** cell() 取自遮罩列的输入不能单独复算；别的输入（计算式里写成当时代进去的值）里按位置取自 查询结果的一格（vars.res.rows[0][2]），顺着取值链认出它取自哪份快照的哪一列（recompute.input_origin，证据面板、推演、 可验证报告同一个认法），按需算那份快照要遮的列（和证据面板同一套，见 _Sources.hidden）——它不一定是 cell() 用到的 -那几份。取自遮罩列、遮没遮核对不了的，整个指标按无法复算记：遮罩值不能当字面量写进交给盲算的计算式。 +那几份。取自遮罩列、遮没遮核对不了的，整个指标按无法复算记：遮罩值不能当字面量写进交给盲算的计算式。伸进查询结果的行、 +却认不出取自哪一列的（下标不是字面量、数据整形改过形状的结果没带快照的 id 和数据源名），计算式里写成「其他输入」。 +交给盲算的分组取值同样按 SQL 认出取自遮罩列的结果列（region AS area、子查询里改了名的），和审批质询同一个认法 +（_value_hidden）。 **一致就放行，不一致交给人。** 不一致时逐个对不上的输入说清两条 SQL 差在哪（data/sql_semantics.explain_sql_diff）： 两边先收窄到这一格（只留算出这一列的那一项；分组的按这一行的分组值改成等值条件，HAVING 一并去掉），一条查询算好几列、 分组后取一行时才不会混进「选出的列」「分组」这类和这一格无关的差别。分组的列设了遮罩时不收窄：这一行的分组值写进 -SQL，就跟着差异说明进了审批卡、对照工件和目录日志。完整对照落成一件工件（审批 payload 经 runner._safe 会截断， +SQL，就跟着差异说明进了审批卡、对照工件和目录日志；遮罩列改了名再分组的（region AS area，在子查询、WITH 里改的也算） +按 SQL 认出来（_value_hidden），一样不收窄。完整对照落成一件工件（审批 payload 经 runner._safe 会截断， 长 SQL 放不下），用 replay.ask 挂起。审核人逐个指标选采用原流程（a）、采用盲算（b）或两边都不对（neither）：a 没有 缺口；b、neither、没选的写进产出的 issuance_gaps，成果节点的出具契约据此降档（on_reject 为 fail 时节点直接失败）。 两边只差一个方面、又对得上数据目录事实的，选择写进数据目录（engine/dual_check_facts.py，指标自己的筛选范围不写）。 @@ -420,10 +424,9 @@ async def _check(state: GraphState, ctx: NodeContext, targets: list[dict[str, An job["metrics"] = sharers[key] results, model, same, usage = await _blind(ctx, jobs, options, settings, _producer_models(state, rows)) for row, metric in zip(rows, targets): + masks, queries = row.pop("_masks") if row["status"] != UNCHECKED: - _compare(row, metric, results, options, load, outputs, row.pop("_masks")) - else: - row.pop("_masks", None) + _compare(row, metric, results, options, load, outputs, masks, queries) agreed = sum(1 for r in rows if r["status"] == SAME) if agreed and not any(r["status"] == DIFFERENT for r in rows): ctx.emit(EventType.LOG, level="info", code="dual_check_agreed", metrics=agreed, @@ -471,36 +474,46 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti "blind": {"value": None, "rendered": None}, "delta_pct": None, "inputs": [], "differing": [], } tops = _tops(metric) + # masks 是每份用到的快照要遮的列（cell() 的、按位置取的格落在的），queries 只是 cell() 用到的那几份：重算时容器 + # 输入按内容对快照只对这几份。按位置取的格落在的快照算遮罩只为认这一格，不当成指标用到的结果——放进去的话，同一个 + # 指标里装着它整份行的容器（计算式里写成「其他输入」，值不交给盲算）会对上它，它有一列遮罩就整个挡下 masks: dict[str, frozenset[str]] = {} + queries: list[str] = [] + withheld: set[str] = set() 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) + why, shown = await _positional_masks(item, sources, load, outputs, masks) + problem = problem or why + if not shown: + withheld.add(str(item.get("path"))) continue info, why = await _query_input(item, sources, aliases, tops) masks[item["artifact"]] = await sources.hidden(item["artifact"]) + queries.append(item["artifact"]) if why: problem = problem or why continue row["inputs"].append(info) - row["_masks"] = masks + row["_masks"] = (masks, tuple(dict.fromkeys(queries))) if problem is None: # 什么都不改先重算一遍：和卡上记的对不上（输入还原不了、用到了遮罩列），换了盲算的输入也算不准 check = recompute.recompute(metric, recompute.Changes(), recompute.Sources( - load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks)))) + load=load, outputs=outputs, masked=masks, queries=row["_masks"][1])) if check["status"] != "ok" or check["changed"]: problem = f"无法按原表达式重算这个指标（{check.get('reason') or '重算结果与口径卡不一致'}）" if problem: row.update(status=UNCHECKED, reason=problem) return row, {} labels = {inp["path"]: f"输入{i}" for i, inp in enumerate(row["inputs"], 1)} - formula = _formula(metric, labels) + formula = _formula(metric, labels, withheld) definition = options.definitions.get(row["metric_id"]) or "" jobs: dict[str, dict[str, Any]] = {} for inp in row["inputs"]: snap, _ = sources.snapshot(inp["artifact"]) - hidden = (masks.get(inp["artifact"]) or frozenset()) | await _constant_columns(snap or {}, sources) + hidden = await _value_hidden(snap or {}, masks.get(inp["artifact"]) or frozenset(), sources) \ + | await _constant_columns(snap or {}, sources) # 列名是原流程的取舍：只有一项来自查询、或者写了口径说明时用不着；几项又没口径说明的，看不出取舍才给 hint = inp["column"] if len(row["inputs"]) > 1 and not definition and _neutral(inp["column"]) else None jobs[inp["key"]] = { @@ -512,27 +525,52 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti 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() 用到的几份快照时，这一格落在另一份快照上就认不出，遮罩值 - 会当字面量写进交给盲算的计算式。 + outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> tuple[str | None, bool]: + """不是 cell()、Agent 字段的输入里，按位置取自一份查询结果的一格（vars.res.rows[0][2]）：(为什么不能复算, 卡上记的值 + 能不能写进交给盲算的计算式)。认出它取自哪份快照的哪一列（recompute.input_origin，证据面板、推演、可验证报告同一个 + 认法），按需算那份快照要遮的列记进 masks。只算 cell() 用到的几份快照时，这一格落在另一份快照上就认不出，遮罩值会当 + 字面量写进交给盲算的计算式。 - 那份快照读不出来、数据源现在设的遮罩又没遮这一列时，查询当时记下的遮罩无从核对：交回原因，按无法复算记，宁可多遮。 - 取值链上的产出读不出来、那份结果没存快照的（遮没遮核对不了），重算的预检自己会拦下，这里不管。""" + 取自遮罩列的直接交回点名的原因，和 cell() 取自遮罩列的说法一致（指标有好几个输入时审核人才找得到是哪一格）。那份 + 快照读不出来、数据源现在设的遮罩又没遮这一列时，查询当时记下的遮罩无从核对：交回原因，按无法复算记，宁可多遮。 + 取值链上的产出读不出来、那份结果没存快照只带着数据源名的（遮没遮核对不了），重算的预检自己会拦下，这里不管。 + 伸进查询结果的行、却认不出取自哪一列的（下标不是字面量；数据整形改过形状，结果既没带快照的 id 也没带数据源名）， + 照常复算，只是值不写进计算式（写成「其他输入」）：认不出就核对不了遮没遮。""" def output(node_id: Any) -> Any: key = outputs.get(str(node_id or "")) return load(key) if key else None + path = str(item.get("path") or "") origin = recompute.input_origin(item, output) - if origin is None or origin.unverified is not None or not origin.artifact: - return None + if origin is None: + return None, not _reaches_rows(path) + if origin.unverified is not None or not origin.artifact: + return None, False 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 + return f"输入 {path} 取自设置了遮罩的列，不能用于复算", False _, why = sources.snapshot(origin.artifact) - return f"输入 {item.get('path')} 所在的{why}，核对不了这一格是否取自设置了遮罩的列" if why else None + if why: + return f"输入 {path} 所在的{why}，无法核对这一格是否取自设置了遮罩的列", False + return None, True + + +def _reaches_rows(path: str) -> bool: + """取值路径伸进了一份查询结果的行（vars.res.rows[vars.i][0]、nodes.parse['rows'][1]）：按位置取的一格。""" + try: + node = ast.parse(path, mode="eval").body + except SyntaxError: + return False + while isinstance(node, (ast.Attribute, ast.Subscript)): + inner = node.value + if isinstance(node, ast.Subscript) and ( + (isinstance(inner, ast.Attribute) and inner.attr == "rows") + or (isinstance(inner, ast.Subscript) and isinstance(inner.slice, ast.Constant) + and inner.slice.value == "rows")): + return True + node = inner + return False async def _query_input(item: dict[str, Any], sources: _Sources, aliases: dict[str, str], tops: list[str] @@ -599,9 +637,10 @@ class _Swap(ast.NodeTransformer): visit_Attribute = visit_Subscript = visit_Name = _chain -def _formula(metric: dict[str, Any], labels: dict[str, str]) -> str: +def _formula(metric: dict[str, Any], labels: dict[str, str], withheld: set[str] | frozenset[str] = frozenset()) -> str: """交给盲算 agent 的计算式：来自查询的输入写成「输入1」「输入2」，别的输入写当时代进去的值（容器写「其他输入」）。 - 不露出取数节点的 id、cell() 的行列：那是原流程的取法，不是业务定义。""" + 不露出取数节点的 id、cell() 的行列：那是原流程的取法，不是业务定义。withheld 里的输入（按位置取的一格认不出取自哪 + 一列，遮没遮核对不了，见 _positional_masks）也写「其他输入」。""" expr = str(metric.get("expression") or "") try: tree, _, _ = parse_expression(expr) @@ -612,6 +651,8 @@ def _formula(metric: dict[str, Any], labels: dict[str, str]) -> str: def shown(path: str) -> ast.AST | None: if path in labels: return ast.Name(id=labels[path], ctx=ast.Load()) + if path in withheld: + return ast.Name(id="其他输入", ctx=ast.Load()) value = values.get(path) if value is None or isinstance(value, (bool, int, float)) or ( isinstance(value, str) and len(value) <= _DIM_CHARS and not _BRIEF.match(value)): @@ -644,7 +685,8 @@ def _row_pairs(snapshot: Mapping[str, Any], row: int, column: str) -> list[tuple def _dims(snapshot: Mapping[str, Any], row: int, column: str, hidden: frozenset[str]) -> list[str]: """结果分了组时，交给盲算 agent 的这一行的分组取值（地区、日期这类维度）：它要知道算的是哪一组。只给取值、不给 - 列名（原流程按哪一列分的组正是要它自己判断的），遮罩列、原 SQL 里写死的文字列（hidden，小写）不给。""" + 列名（原流程按哪一列分的组正是要它自己判断的），取自遮罩列的结果列（_value_hidden）、原 SQL 里写死的文字列（hidden， + 小写）不给。""" out: list[str] = [] for name, value in _row_pairs(snapshot, row, column): if name.lower() in hidden: @@ -670,6 +712,21 @@ def _neutral(column: str) -> bool: return bool(column.strip()) and not any(word in lowered for word in _LOADED_NAME) +async def _value_hidden(snapshot: Mapping[str, Any], hidden: frozenset[str], sources: _Sources) -> frozenset[str]: + """这份快照里取值来自遮罩列的名字（小写）：遮罩列本身，加上 SQL 里由它改名、套函数、经子查询或 WITH 转一道得来的 + 结果列和 SQL 里的名字（region AS area 的 area）。按列名遮只认得出前一种，后一种的取值照样是遮罩列的：交给盲算的 + 分组取值（_dims）、收窄时补的等值条件（_narrowed）都按这一组挡。和审批质询认改了名的遮罩列同一个认法 + （inquiry._derived_hidden）：SQL 解析不了、里面又写着遮罩列的，整份快照的列都算。hidden 为空（没设遮罩）时为空。""" + from app.engine.inquiry import _derived_hidden # inquiry 导入了 engine.nodes 里的模块 + + if not hidden: + return frozenset() + name = snapshot.get("source") + found = await sources.source(name) if isinstance(name, str) and name else None + columns, names = _derived_hidden(dict(snapshot), set(hidden), found.kind if found is not None else None) + return frozenset(hidden) | frozenset(c.lower() for c in (*columns, *names)) + + async def _constant_columns(snapshot: Mapping[str, Any], sources: _Sources) -> frozenset[str]: """原 SQL 最外层写死的文字列（'下单日' AS basis）的列名（小写）：那是原流程自己的标注，不是分组。解析不了为空。""" import sqlglot @@ -908,8 +965,9 @@ def _delta_pct(original: Any, blind: Any) -> float | None: def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any], options: _Options, - load: Callable[[str], Any], outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> None: - """盲算的输入代回原式，和原值比。row 就地改成一致 / 不一致 / 无法复算。""" + load: Callable[[str], Any], outputs: dict[str, str], masks: dict[str, frozenset[str]], + queries: tuple[str, ...]) -> None: + """盲算的输入代回原式，和原值比。row 就地改成一致 / 不一致 / 无法复算。masks、queries 同预检（_plan_metric）。""" values: dict[str, Any] = {} for inp in row["inputs"]: found = results.get(inp["key"]) @@ -924,7 +982,7 @@ def _compare(row: dict[str, Any], metric: dict[str, Any], results: dict[str, Any return values[inp["path"]] = found["value"] outcome = recompute.recompute(metric, recompute.Changes(inputs=values), recompute.Sources( - load=load, outputs=outputs, masked=masks, queries=tuple(dict.fromkeys(masks)))) + load=load, outputs=outputs, masked=masks, queries=queries)) if outcome["status"] != "ok": row.update(status=UNCHECKED, reason=f"用盲算的输入按原表达式算不出结果（{outcome.get('reason') or '无法计算'}）") return @@ -992,10 +1050,13 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], continue blind_snap, _ = sources.snapshot(str(blind["artifact"])) if blind.get("artifact") else (None, None) 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"] + # 收窄按取值来自遮罩列的名字挡（含改了名的）：分组值写进 SQL 就跟着差异说明进了审批卡 + a_hidden = await _value_hidden(original_snap or {}, await sources.hidden(inp["artifact"]), sources) + b_hidden = await _value_hidden(blind_snap or {}, blind_hidden, sources) + a_sql = _narrowed(inp["sql"], inp["column"], _record(original_snap, inp["row"]), kind, a_hidden) \ + 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") + b_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 ""))) @@ -1021,7 +1082,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], 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": info.get("kind"), + "dialect": info.get("kind"), "uploaded": bool(info.get("uploaded")), "sql": {"a": single[0], "b": single[1]} if single else {}, "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None, "items": merged["items"]}, @@ -1060,18 +1121,20 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], async def _source_info(sources: _Sources, name: str) -> dict[str, Any]: - """比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名和整份数据目录。数据源没了就都不给，照样比原文。""" + """比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名、整份数据目录，以及是不是上传表格建的数据源 + （uploaded：导入时日期一律存成文字，业务日期不记时原因照实说，见 dual_check_facts._dated_problem）。数据源没了就 + 都不给，照样比原文。""" 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": {}, "notes": {}} + return {"kind": None, "schema": None, "names": {}, "notes": {}, "uploaded": False} 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(notes), "notes": notes} + "names": catalog_labels(notes), "notes": notes, "uploaded": (row.origin or "manual") == "upload"} async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]: @@ -1161,9 +1224,10 @@ def _narrowed(sql: Any, column: Any, record: Mapping[str, Any] | None, kind: str 满足 HAVING，留着只会多出一条和这一格无关的差异。 差异说明拿收窄后的两条比，才只说这一格是怎么算出来的。认不准的（SELECT *、DISTINCT、窗口函数、集合运算、分组键 - 对不上结果列、这一行没有分组值）返回 None，调用方照整条比。hidden 是这份快照要遮的列（小写）：分组键的结果列、 - 或者它用到的列在里面时也返回 None——分组值写进 SQL 就跟着差异说明进了审批卡、对照工件和目录日志，遮罩列的值 - 不能出现在那些地方。""" + 对不上结果列、这一行没有分组值）返回 None，调用方照整条比。hidden 是这份快照里取值来自遮罩列的名字（小写， + _value_hidden：遮罩列，加上 SQL 里由它改名、经子查询或 WITH 转一道得来的）：分组键的结果列、或者它用到的列在里面 + 时也返回 None——分组值写进 SQL 就跟着差异说明进了审批卡、对照工件和目录日志，遮罩列的值不能出现在那些地方。只按 + 遮罩列名比的话，FROM (SELECT region AS area …) GROUP BY area 外层只看得到 area，差异说明还会把别名还原成遮罩列名。""" import sqlglot from sqlglot import exp diff --git a/backend/tests/fixtures/dual_check_shop.py b/backend/tests/fixtures/dual_check_shop.py index 3d14ab28..6f67f5b2 100644 --- a/backend/tests/fixtures/dual_check_shop.py +++ b/backend/tests/fixtures/dual_check_shop.py @@ -11,7 +11,8 @@ 另有渠道（channel）和测试单标记（is_test，都是 0）两列：写数据目录的测试拿它们凑「指标自己的筛选范围」和标记列。 下单日、支付日两列声明为 DATE（SQLite 照样存成文字，按字符串比的先后和按日期比的一样）：双盲复算只在两边的时间列都 -声明为日期 / 时间类型时才把选择记成业务日期。build(date_type="TEXT") 建一份声明成文本的，看不记的那一种。 +声明为带日期的类型时才把选择记成业务日期。build(date_type="TEXT") 建一份声明成文本的、build(date_type="") 建一份 +不写类型的，看不记的那几种。 """ from __future__ import annotations diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index e845a710..54cc8bd1 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -2,8 +2,10 @@ - 只差一个方面、又对得上目录事实时才写：两边按不同的时间列筛 → 业务日期，一边多一个排除无效记录的条件 → 有效记录 条件，聚合只差去不去重、被采纳的一边去重 → 去重规则；按被采纳那一边的写法记 -- 业务日期只在两边的列都在表结构里声明为日期 / 时间类型（DATE、DATETIME、TIMESTAMP、TIME，带精度、带时区的写法也算） - 时才写：文本、整数（Unix 秒、20260901 这类日期键）、表结构里查不到、没声明类型的不写，原因说清是哪一列 +- 业务日期只在两边的列都在表结构里声明为带日期的类型（DATE、DATETIME、TIMESTAMP，带精度、带时区的写法也算）时才写： + 文本、整数（Unix 秒、20260901 这类日期键）、只有时刻的 TIME、表结构里查不到、没声明类型（SQLite 探出来是 NULL）的 + 不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源说清日期是按文字保存的； + 数据源还没有可用的表结构时说表结构不可用 - 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛：不写， 照实说原因。同一列上的时间范围不同、跨两张表的条件，差异说明只给兜底，同样不写；说明以后真把跨两张表的条件说成一条 筛选条件，derive 自己那道「只涉及一张表」的检查也拦得下 @@ -25,6 +27,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.db.models import DataSource from app.engine.dual_check_facts import derive, preview, record_choices from tests.fixtures.dual_check_shop import ( BOTH_WRONG, @@ -266,42 +269,69 @@ def schema_with(**types: str | None) -> dict: ("datetime", "DateTime(6)"), ("TIMESTAMPTZ", "timestamp(3) with time zone"), ("TIMESTAMP WITHOUT TIME ZONE", "TIMESTAMP(6) WITH LOCAL TIME ZONE"), - ("time", "TIME WITH TIME ZONE"), ]) -def test_a_business_date_is_written_when_both_columns_are_declared_as_dates_or_times(order_date, paid_date): - """两边的列都在表结构里声明为日期 / 时间类型（带精度、带时区、大小写不同都认）：照常记被采纳的那一列。""" +def test_a_business_date_is_written_when_both_columns_are_declared_as_dates(order_date, paid_date): + """两边的列都在表结构里声明为带日期的类型（日期、日期时间、时间戳，带精度、带时区、大小写不同都认）：照常记被采纳的 + 那一列。""" fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), schema_with(order_date=order_date, paid_date=paid_date)) assert why is None and (fact.item, fact.value) == ("business_date", {"column": "paid_date"}) +#: 未被采纳的一边的列有问题时，说清为什么也要看它 +BOTH_DATES = "两边都是日期列时，审核人选的才是业务日期按哪一列" + + @pytest.mark.parametrize("types, column, said", [ # SQLite 常把日期存成文字：认不准存的是不是日期 - ({"paid_date": "TEXT"}, "orders.paid_date", "在表结构里声明为 TEXT"), - ({"paid_date": "VARCHAR(10)"}, "orders.paid_date", "在表结构里声明为 VARCHAR(10)"), + ({"paid_date": "TEXT"}, "orders.paid_date", "在表结构里的类型是 TEXT，无法确认存的是日期"), + ({"paid_date": "VARCHAR(10)"}, "orders.paid_date", "在表结构里的类型是 VARCHAR(10)，无法确认存的是日期"), # 没被采纳的一边也得是：它要是按 Unix 秒、20260901 这类整数筛的，审核人并没有在两个日期列之间选 - ({"order_date": "INTEGER"}, "orders.order_date", "在表结构里声明为 INTEGER"), - ({"order_date": "BIGINT"}, "orders.order_date", "在表结构里声明为 BIGINT"), + ({"order_date": "INTEGER"}, "orders.order_date", "在表结构里的类型是 INTEGER，无法确认存的是日期"), + ({"order_date": "BIGINT"}, "orders.order_date", "在表结构里的类型是 BIGINT，无法确认存的是日期"), # 名字里带着 date、time 的别的类型也不算 - ({"paid_date": "DATE_KEY"}, "orders.paid_date", "在表结构里声明为 DATE_KEY"), - ({"paid_date": "INTERVAL"}, "orders.paid_date", "在表结构里声明为 INTERVAL"), - ({"paid_date": ""}, "orders.paid_date", "在表结构里没有声明类型"), - ({"paid_date": None}, "orders.paid_date", "不在数据源的表结构里"), + ({"paid_date": "DATE_KEY"}, "orders.paid_date", "在表结构里的类型是 DATE_KEY，无法确认存的是日期"), + ({"paid_date": "INTERVAL"}, "orders.paid_date", "在表结构里的类型是 INTERVAL，无法确认存的是日期"), + # 只有时刻的 TIME：一天里的几点几分（MySQL 里还常拿它存时长）不是日期，和 catalog.is_date_column 同一个口径 + ({"paid_date": "TIME"}, "orders.paid_date", "在表结构里的类型是 TIME，只有时刻，不含日期"), + ({"order_date": "time with time zone"}, "orders.order_date", + "在表结构里的类型是 time with time zone，只有时刻，不含日期"), + ({"paid_date": "TIMETZ"}, "orders.paid_date", "在表结构里的类型是 TIMETZ，只有时刻，不含日期"), + ({"paid_date": "TIME(3)"}, "orders.paid_date", "在表结构里的类型是 TIME(3)，只有时刻，不含日期"), + ({"paid_date": ""}, "orders.paid_date", "在表结构里没有声明类型，无法确认存的是日期"), + # SQLite 里没写类型的列，SQLAlchemy 探出来是 NullType，表结构里记的是「NULL」：同样是没有声明类型 + ({"paid_date": "NULL"}, "orders.paid_date", "在表结构里没有声明类型，无法确认存的是日期"), + ({"paid_date": None}, "orders.paid_date", "不在数据源的表结构里，无法确认存的是日期"), ]) -def test_a_business_date_is_not_written_unless_both_columns_are_dates_or_times(types, column, said): - """任一边的列不是声明的日期 / 时间类型（文本、整数、查不到、没声明类型）：不写，原因说清是哪一列。选哪一边都一样： - 被采纳的那一边有问题说它，没问题就说另一边的。""" +def test_a_business_date_is_not_written_unless_both_columns_are_dates(types, column, said): + """任一边的列不是带日期的声明类型（文本、整数、只有时刻、查不到、没声明类型）：不写，原因说清是哪一列。被采纳的那一边 + 有问题说它；被采纳的一边没问题，就说未被采纳的那一边，另说清为什么也要看它。order_date 是原流程（a）的列， + paid_date 是盲算（b）的。""" + owner = "a" if column == "orders.order_date" else "b" + label = {"a": "原流程", "b": "盲算"}[owner] for choice in ("a", "b"): fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, choice), schema_with(**types)) assert fact is None, (types, choice) - assert why.startswith("指标「销售额」") and why.endswith("，无法确认是日期 / 时间类型，不写入数据目录"), why - assert f"按时间筛的列「{column}」{said}" in why, why + if choice == owner: + assert why == f"指标「销售额」被采纳的{label}按时间筛的列「{column}」{said}，不写入数据目录", why + else: + assert why == f"指标「销售额」未被采纳的{label}按时间筛的列「{column}」{said}：{BOTH_DATES}，不写入数据目录", why -@pytest.mark.parametrize("schema", [None, {}, {"tables": {}}, {"tables": {"customers": SCHEMA["tables"]["customers"]}}]) +@pytest.mark.parametrize("schema", [None, {}, {"tables": {}}, {"tables": {}, "failed": True, "error": "timeout"}]) def test_a_business_date_is_not_written_without_the_table_structure(schema): - """没有表结构（还没探查、探查失败、表不在里面）：确认不了两列的类型，不写。""" + """数据源还没有可用的表结构（还没探查、探查失败）：两列的类型无从确认，不写。原因说表结构不可用，不说这一列不在表结构 + 里——那时并不知道有没有这一列。""" fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), schema) - assert fact is None and "按时间筛的列「orders.paid_date」不在数据源的表结构里，无法确认是日期 / 时间类型" in why + assert fact is None + assert why == ("指标「销售额」所在数据源还没有可用的表结构（尚未探查或探查失败），无法确认两边按时间筛的列存的是日期，" + "不写入数据目录"), why + + +def test_a_table_missing_from_an_available_table_structure_is_said_to_be_missing(): + """表结构是有的，里面只是没有这张表：照实说这一列不在表结构里。""" + fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), {"tables": {"customers": SCHEMA["tables"]["customers"]}}) + assert fact is None + assert why == "指标「销售额」被采纳的盲算按时间筛的列「orders.paid_date」不在数据源的表结构里，无法确认存的是日期，不写入数据目录" def test_the_type_check_follows_the_names_in_the_table_structure(): @@ -310,7 +340,7 @@ def test_the_type_check_follows_the_names_in_the_table_structure(): {"name": "STATUS", "type": "TEXT"}, {"name": "ORDER_DATE", "type": "DATE"}, {"name": "PAID_DATE", "type": "TEXT"}, {"name": "AMOUNT", "type": "REAL"}]}}} fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "a"), upper) - assert fact is None and "「orders.paid_date」在表结构里声明为 TEXT" in why + assert fact is None and "「orders.paid_date」在表结构里的类型是 TEXT" in why def test_the_preview_follows_the_type_check(): @@ -319,7 +349,37 @@ def test_the_preview_follows_the_type_check(): said = preview(candidate(BY_ORDER_DATE, RIGHT, "a"), schema, {}) for side in ("a", "b"): assert said[side] == {"write": False, "reason": derive(candidate(BY_ORDER_DATE, RIGHT, side), schema)[1]} - assert "「orders.paid_date」在表结构里声明为 TEXT" in said[side]["reason"] + assert "「orders.paid_date」在表结构里的类型是 TEXT" in said[side]["reason"] + + +def test_an_uploaded_table_says_its_dates_are_stored_as_text(): + """上传表格建的数据源（候选带 uploaded）：导入时日期一律按文字保存，表结构里的类型总是 TEXT。照样不写（认不准存的是 + 不是日期），可原因不说「类型是 TEXT」——那不是用户声明的，导入确认时这一列标的还是「日期」，表结构也改不了——说清 + 是上传的表格按文字保存日期。整数列照常说类型；不是上传的数据源，同样的文本列也照常说类型。""" + texts = schema_with(order_date="TEXT", paid_date="TEXT") + for choice, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")): + fact, why = derive({**candidate(BY_ORDER_DATE, RIGHT, choice), "uploaded": True}, texts) + assert fact is None and why == (f"指标「销售额」被采纳的{label}按时间筛的列「orders.{column}」来自上传的表格，" + "表格里的日期一律按文字保存，无法确认存的是日期，不写入数据目录"), why + fact, why = derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "uploaded": True}, schema_with(order_date="INTEGER")) + assert fact is None and "未被采纳的原流程按时间筛的列「orders.order_date」在表结构里的类型是 INTEGER" in why + assert "被采纳的盲算按时间筛的列「orders.paid_date」在表结构里的类型是 TEXT" in derive( + candidate(BY_ORDER_DATE, RIGHT, "b"), texts)[1] + + +async def test_an_undeclared_column_read_from_a_real_database_is_said_to_have_no_type(tmp_path): + """真探查一个 SQLite 库（introspect 存的是 str(列的类型)）：没写类型的列记成「NULL」（SQLAlchemy 的 NullType）， + 说成没有声明类型，不说「类型是 NULL」。""" + name, source_id = await add_source(build(tmp_path / "untyped.db", date_type="")) + try: + async with SessionLocal() as session: + schema = (await session.get(DataSource, source_id)).schema_cache + assert {c["name"]: c["type"] for c in schema["tables"]["orders"]["columns"]}["paid_date"] == "NULL" + fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), schema) + assert fact is None and why == ("指标「销售额」被采纳的盲算按时间筛的列「orders.paid_date」在表结构里没有声明类型，" + "无法确认存的是日期，不写入数据目录"), why + finally: + await drop_source(source_id) # -------------------------------------------------------------------------- diff --git a/backend/tests/test_dual_check_fixture.py b/backend/tests/test_dual_check_fixture.py index 9f7c0a0b..0ee341c9 100644 --- a/backend/tests/test_dual_check_fixture.py +++ b/backend/tests/test_dual_check_fixture.py @@ -19,7 +19,7 @@ check-verify 在夹具上能过，真实数据上却一处都不高亮（2026-10 dual_check_facts.preview，约定之外另加的），审批卡照它说；老服务端没有时卡上按差异的形状有条件地说，两种说法都按 「服务端到底写不写」核对（2026-10 复核：卡上说会记入、服务端却不写的有好几种）。夹具里的预告（catalog_preview 按指标、 variants 和 peers 各自的 catalog）也是服务端的真实输出：按下面的 SCHEMA、空目录、指标自己的范围留空算。SCHEMA 里 -paid_at、created_at 声明为 TIMESTAMP：业务日期只在两边的时间列都声明为日期 / 时间类型时才写。双盲复算的服务端合入之前 +paid_at、created_at 声明为 TIMESTAMP：业务日期只在两边的时间列都声明为带日期的类型时才写。双盲复算的服务端合入之前 没有这个模块，逐项比对那几项跳过，形状和门槛照样查。 **同一次裁决里几个指标落到同一项上。** 预告是每个指标单独算的，服务端写目录前却要把同一次裁决里落到同一张表同一项上的 @@ -62,7 +62,7 @@ LABELS = ("原流程", "盲算") #: 摘要里每个指标最多列几条差异（同双盲复算节点的 SUMMARY_DIFFS） SUMMARY_DIFFS = 3 #: 预告用的数据源名和表结构（虚构）：服务端要认得出表、认得出聚合的列属于哪张表，才会预告写哪一项；时间列声明为 -#: TIMESTAMP，业务日期才会预告写（文本、整数的时间列不记） +#: TIMESTAMP，业务日期才会预告写（文本、整数、只有时刻的时间列不记） SOURCE = "shop" SCHEMA: dict[str, Any] = {"tables": { "orders": {"name": "orders", "columns": [ diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index f6802b25..8c5587c2 100644 --- a/backend/tests/test_dual_check_node.py +++ b/backend/tests/test_dual_check_node.py @@ -7,13 +7,16 @@ - 不一致：挂起，审批 payload 和对照工件按约定，payload 另带实际生效的 on_reject、摘要里每个指标带选 a、选 b 各会往 数据目录里写什么的预告；a / b / neither 三种选择的产出和缺口，没选的照实记，驳回、on_reject 为 fail 时失败 - 差异说明只比这一格所在的那一列（一条查询算好几列、分组后取一行的，收窄了再比，HAVING 一并去掉）；分组的列设了 - 遮罩时（改了名再分组的也算）不收窄、照整条比，遮罩列的取值不进审批卡、对照工件、产出和事件 + 遮罩时（改了名再分组的、在子查询或 WITH 里改了名的也算）不收窄、照整条比，遮罩列的取值不进审批卡、对照工件、产出、 + 事件和盲算的任务 - 只差一个方面时写数据目录（业务日期、有效记录条件），差了不止一处不写，人工确认过的不同说法不覆盖，指标自己的 筛选范围（取值在运行输入里、用到分组的列）不写，两条 SQL 规范化后一致的不写；人工驳回过的条件不会借着和别的 - 指标并成一项写回去，真写的和审批卡上的预告一致；时间列在表结构里声明成文本的不记成业务日期，预告照实说 + 指标并成一项写回去，真写的和审批卡上的预告一致；时间列在表结构里声明成文本的不记成业务日期，预告照实说（上传表格 + 建的数据源说清日期是按文字保存的） - 全局关闭直通、节点覆盖；设置按发起时的快照；重放时盲算不再执行、事件不重复；报告的写作目录不含盲算的查询 -- 触顶、分组结果的维度（只给取值、遮罩列不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列上的 - 盲算结果不采用；按位置取自另一份快照遮罩列的输入（cell() 取 A、vars.res.rows[0][0] 取 B）按无法复算记，不盲算 +- 触顶、分组结果的维度（只给取值、取自遮罩列的不给）、带着取舍的列名不交给盲算、两个指标用到同一格只盲算一次、遮罩列 + 上的盲算结果不采用；按位置取自另一份快照遮罩列的输入（cell() 取 A、vars.res.rows[0][0] 取 B）按无法复算记、原因点名 + 是哪个输入，不盲算；认不出取自哪一列的按位置取的格不写进计算式；按位置取的格不改变容器输入的判法 """ from __future__ import annotations @@ -28,7 +31,7 @@ from sqlalchemy import delete, select from app.core import artifact_store from app.data import catalog from app.db.base import SessionLocal -from app.db.models import Approval, Run, RunEvent, Setting +from app.db.models import Approval, DataSource, Run, RunEvent, Setting from app.engine import spend from app.engine.launch import launch_run from app.engine.nodes.dual_check import GAP_UNCHECKED @@ -589,9 +592,8 @@ async def test_time_columns_declared_as_text_are_not_written_as_the_business_dat 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"] + assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「orders.{column}」" + "在表结构里的类型是 TEXT，无法确认存的是日期，不写入数据目录"} done = await decide(client, run.id, {"gmv": "b"}) assert done.status == "succeeded", done.error [wrote] = logs(await events(run.id), "dual_check_catalog") @@ -599,6 +601,48 @@ async def test_time_columns_declared_as_text_are_not_written_as_the_business_dat assert await notes_of(source_id) is None +async def test_an_uploaded_source_says_its_dates_are_stored_as_text(monkeypatch, text_dated_shop, client): + """上传表格建的数据源（origin=upload），导入时日期一律存成文字：审批卡的预告和裁决后时间线上跳过的原因，都说上传的 + 表格按文字保存日期，不说「类型是 TEXT」（那不是用户声明的，导入确认时这一列标的是「日期」）。上传源的查询要走版本 + 快照，这里只把节点认数据源来历的那一步（_source_info）换成上传源，查询照手工源走；_source_info 自己认 origin 的那一步 + 另见下一个用例。""" + from app.engine.nodes import dual_check + + name, source_id = text_dated_shop + real = dual_check._source_info + + async def uploaded(sources, source): + return {**await real(sources, source), "uploaded": True} + + monkeypatch.setattr(dual_check, "_source_info", uploaded) + 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"] + said = item["catalog"] + for side, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")): + assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「orders.{column}」" + "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是" + "日期，不写入数据目录"} + 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_the_source_info_knows_an_uploaded_source(shop): + from app.engine.nodes.dual_check import _source_info, _Sources + + name, source_id = shop + assert (await _source_info(_Sources(), name))["uploaded"] is False + async with SessionLocal() as session: + (await session.get(DataSource, source_id)).origin = "upload" + await session.commit() + assert (await _source_info(_Sources(), name))["uploaded"] is True + assert (await _source_info(_Sources(), "dc_no_such_source"))["uploaded"] is False + + async def test_a_refund_filter_becomes_the_valid_row_filter(monkeypatch, shop, client): name, source_id = shop Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) @@ -1031,7 +1075,7 @@ async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, 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, {"首组销售额": (BY_ORDER_DATE, "gmv")}) + blind = 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 @@ -1042,6 +1086,47 @@ async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, assert done.status == "succeeded", done.error _no_masked_value(payload, card, done, await events(run.id)) _compared_whole(payload, card) + # 交给盲算的这一行的分组取值也不给：结果列 area 的值取自遮罩列 region（以前 _dims 只按结果列名认遮罩） + [task] = blind.tasks + assert "华东" not in task, task + + +#: 外层按 area 分组、取每组的销售额 +_BY_AREA = f"SELECT area, SUM(amount) AS gmv FROM {{}} WHERE status <> 'refunded' AND paid_date {WEEK} GROUP BY area ORDER BY area" +#: 遮罩列在子查询、WITH 里改了名（直接起别名、套函数、写在列名表里、隔了两层），外层再按新名字分组：外层的分组键 area +#: 看不出取自遮罩列 region +RENAMED_INSIDE = { + "subquery": _BY_AREA.format("(SELECT region AS area, amount, status, paid_date FROM orders) t"), + "cte": "WITH t AS (SELECT region AS area, amount, status, paid_date FROM orders) " + _BY_AREA.format("t"), + "function": "WITH t AS (SELECT UPPER(region) AS area, amount, status, paid_date FROM orders) " + _BY_AREA.format("t"), + "column_list": ("WITH t(area, amount, status, paid_date) AS (SELECT region, amount, status, paid_date FROM orders) " + + _BY_AREA.format("t")), + "nested": _BY_AREA.format("(SELECT area, amount, status, paid_date FROM (SELECT region AS area, amount, status, " + "paid_date FROM orders) a) b"), +} + + +@pytest.mark.parametrize("shape", list(RENAMED_INSIDE)) +async def test_a_masked_group_column_renamed_inside_is_not_narrowed_or_shown(monkeypatch, masked_shop, client, shape): + """遮罩列在子查询或 WITH 里改了名（region AS area），外层按 area 分组：外层只看得到 area，以前照常拿这一行的分组值 + 补上「area = '华东'」，差异说明把别名还原成「region = '华东'」，进了审批卡、对照工件和目录日志；交给盲算的这一行的 + 分组取值里也有「华东」。现在按 SQL 认出取值来自遮罩列的名字（同审批质询认改了名的遮罩列）：不收窄、照整条比， + 盲算任务里也不给这一组的取值。""" + name, _ = masked_shop + first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + blind = Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")}) + run = await start(graph(name, RENAMED_INSIDE[shape], 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) + [task] = blind.tasks + assert "华东" not in task, task def _compared_whole(payload: dict, card: dict) -> None: @@ -1071,6 +1156,24 @@ 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)) +async def test_a_with_clause_that_does_not_touch_masked_columns_is_still_narrowed(monkeypatch, masked_shop): + """数据源设了遮罩，原流程的 WITH 里却没用到遮罩列、按渠道分组：照常收窄到这一行（补上 channel = 'offline'），差异 + 说明说得出这个条件。认改了名的遮罩列是按 SQL 追到底的，不是见到子查询、WITH 就一律不收窄。""" + name, _ = masked_shop + by_channel = (f"WITH t AS (SELECT channel, amount, status, paid_date FROM orders) SELECT channel, SUM(amount) AS gmv " + f"FROM t WHERE status <> 'refunded' AND paid_date {WEEK} GROUP BY channel ORDER BY channel") + first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} + blind = Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + run = await start(graph(name, by_channel, metrics=[first], dual={"definitions": {}})) + assert run.status == "interrupted", run.error + [item] = (await pending(run.id)).payload["summary"] + assert item["original_rendered"].startswith("230") and item["blind_rendered"].startswith("330") + assert any("channel = 'offline'" in d for d in item["diff"]), item["diff"] + # 渠道不是遮罩列：交给盲算的这一行的分组取值照给 + [task] = blind.tasks + assert "「offline」" in task, task + + #: 销售额（第一份快照的一格）加上首组地区名的字数（第二份快照按位置取的一格） MIX = {"id": "mix", "name": "销售额加首组名称字数", "unit": "元", "decimals": 1, "format": "thousands", "expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows[0][0])"} @@ -1098,7 +1201,8 @@ async def test_a_masked_cell_read_by_position_from_another_snapshot_is_not_blind 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"] + # 和 cell() 取自遮罩列的说法一致，点名是哪个输入：指标有好几个输入时审核人才找得到是哪一格 + assert row["reason"] == "输入 vars.res.rows[0][0] 取自设置了遮罩的列，不能用于复算", 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 @@ -1135,17 +1239,63 @@ async def test_a_cell_read_by_position_from_an_unreadable_snapshot_is_not_truste 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] 所在的查询快照与哈希不一致，" - "疑似被修改，核对不了这一格是否取自设置了遮罩的列")): + for source, expected in ((name, "输入 vars.res.rows[0][0] 取自设置了遮罩的列，不能用于复算"), + ("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 + why, shown = await _positional_masks(POSITIONAL, sources, state.get, {"parse": "state:parse"}, masks) + assert (why, shown) == (expected, False), source assert ("region" in masks[LOST]) is (source == name), masks +async def test_a_cell_whose_origin_cannot_be_told_is_not_written_into_the_formula(masked_shop): + """按位置取的一格认不出取自哪份结果的哪一列——下标不是字面量（rows[vars.i]）、数据整形改过形状的结果既没带快照的 id + 也没带数据源名——遮没遮无从核对：计算式里写成「其他输入」，不把卡上记的值交给盲算模型。认得出、又没遮的照常写值； + 不伸进查询结果的行的输入（运行输入、常量）也照常写。""" + from app.engine.nodes.dual_check import _formula, _positional_masks, _Sources + + name, _ = masked_shop + reshaped = {"columns": ["region", "gmv"], "rows": [["华东", 300.0], ["华北", 30.0]]} + cases = [ + ({"path": "vars.res.rows[vars.i][0]", "value": "华东", "node_id": "parse", "via": "transform"}, + {**reshaped, "artifact": LOST, "source": name}, False), + ({**POSITIONAL}, reshaped, False), + ({"path": "vars.limit", "value": 3, "node_id": "parse", "via": "transform"}, reshaped, True), + ] + for item, parsed, expected in cases: + state = {"state:parse": parsed} + why, shown = await _positional_masks(item, _Sources(), state.get, {"parse": "state:parse"}, {}) + assert (why, shown) == (None, expected), item["path"] + metric = {"expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows[vars.i][0]) + vars.limit", + "inputs": [{"path": "cell(nodes.fetch, 0, 'gmv')", "value": 330.0}, + {"path": "vars.res.rows[vars.i][0]", "value": "华东"}, {"path": "vars.i", "value": 0}, + {"path": "vars.limit", "value": 3}]} + formula = _formula(metric, {"cell(nodes.fetch, 0, 'gmv')": "输入1"}, {"vars.res.rows[vars.i][0]"}) + assert formula == "输入1 + len(其他输入) + 3", formula + + +#: 销售额，加上第二份快照的行数（容器）和首组的销售额（按位置取的一格，不在遮罩列上） +MIX_COUNTS = {"id": "mix", "name": "销售额加地区数和首组销售额", "unit": "元", "decimals": 1, "format": "thousands", + "expression": "cell(nodes.fetch, 0, 'gmv') + len(vars.res.rows) + vars.res.rows[0][1]"} + + +async def test_reading_a_cell_by_position_does_not_change_how_a_container_input_is_judged(monkeypatch, masked_shop): + """按位置取的那一格在第二份快照（B）的没遮的列上，同一个指标另用了 B 的行数（容器，计算式里写成「其他输入」，值不交给 + 盲算）。按位置取的那一格只为认遮罩算 B 的遮罩，不把 B 当成 cell() 用到的快照：以前 B 因此进了重算的快照清单，容器的 + 完整值对上了 B，B 有一列遮罩就整个按「用到了遮罩列」挡下——只写 len(vars.res.rows) 能核、加上这一格就不能核了。""" + name, _ = masked_shop + blind = Blind(monkeypatch, name, {MIX_COUNTS["name"]: (RIGHT, "gmv")}) + run = await start(with_regions(graph(name, RIGHT, metrics=[MIX_COUNTS], 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(其他输入) + 300.0" in task and "华东" not in task, task + + def _no_masked_value(payload: dict, card: dict, done: Run, evs: list[RunEvent], value: str = "华东") -> None: """遮罩列上的取值不出现在审批 payload、对照工件、节点产出和节点发的事件里。查询本身的 tool.end、模型的往来按 设计带着原始结果（证据面板显示时再遮），不在此列。""" diff --git a/backend/tests/test_positional_masks_elsewhere.py b/backend/tests/test_positional_masks_elsewhere.py index 654f822b..8c91aa31 100644 --- a/backend/tests/test_positional_masks_elsewhere.py +++ b/backend/tests/test_positional_masks_elsewhere.py @@ -54,7 +54,7 @@ async def dual_output(run_id: str) -> dict: async def test_the_dual_check_does_not_blind_compute_a_metric_reading_a_masked_cell_by_position(shop): """mix = round(cell(nodes.fetch, 1, 'amount') + len(vars.res.rows[1][2]), 1)：cell() 那一格不在遮罩列上，按位置取的 - 号码在。预检判它用到了遮罩列，按无法复算记缺口，原因不说「推演」；没遮罩时预检照常通过、指标进入盲算。""" + 号码在。认出它取自遮罩列，按无法复算记缺口，原因点名是哪个输入、不说「推演」；没遮罩时预检照常通过、指标进入盲算。""" import asyncio from app.db.base import SessionLocal @@ -76,7 +76,7 @@ async def test_the_dual_check_does_not_blind_compute_a_metric_reading_a_masked_c masked = await run({"mask_columns": ["phone"]}) [mix] = masked["metrics"] assert mix["metric_id"] == "mix" and mix["status"] == "unchecked" - assert mix["reason"] == "无法按原表达式重算这个指标（这个指标用到了设置了遮罩的列）" + assert mix["reason"] == "输入 vars.res.rows[1][2] 取自设置了遮罩的列，不能用于复算" assert "推演" not in json.dumps(masked, ensure_ascii=False) assert not [p for p in PHONES if p in json.dumps(masked, ensure_ascii=False)] assert any("遮罩" in gap for gap in masked["issuance_gaps"])