diff --git a/backend/app/bundle/build.py b/backend/app/bundle/build.py index 90fca722..8b110460 100644 --- a/backend/app/bundle/build.py +++ b/backend/app/bundle/build.py @@ -10,7 +10,9 @@ - 工件：每份报告文档（各报告节点最后一次自查记下的 doc_artifact）、它引用到的口径卡和查询快照，按原字节 base64 嵌入，浏览器重算 sha256 和 id 比；带遮罩的快照不嵌内容、只给 id。口径卡上记着每个输入的值和代入式， 有输入取自遮罩列（cell() 的格、Agent 字段、按位置取的 vars.res.rows[0][2]）的整张卡同样只给 id，认法和证据 - 面板、推演同一个（recompute.input_origin）。没被报告引用的工件不嵌。 + 面板、推演同一个（recompute.input_origin）；服务端文件被改过的卡照样按卡上记的输入认，解析不出一张卡的整张只给 + id。只给 id 的工件，服务端的文件和 id 对不上的另标 mismatch、解析不出来的另标 unparsable，核对页照实说（对不上的 + 判不通过）。没被报告引用的工件不嵌。 - 封存之后追加的事件（按需裁判、审批质询这类批注）单列：类型、序号、内容摘要，不进清单。 它只证明「内部一致」：数字、文档、运行记录彼此对得上，而且没被改过。真伪要拿页面上的封存哈希和系统里这次 @@ -116,17 +118,24 @@ class _Store: 认哪个输入取自哪份查询结果的哪几列，和证据面板、推演同一个认法（recompute.input_origin）；按位置取的一格顺着 取值链从写变量的节点在口径卡算这一次之前的产出起走。核对不了的宁可多挡：快照读不出来、取值链上的产出读不出来、 那份结果没存快照只带着数据源名、定位里没写列。认得出、可结果既没带快照的 id 也没带数据源名的（数据整形改过形状） - 认不出出处，不算。读不出来的卡本来就不嵌，这里不管。""" + 认不出出处，不算。 + + 卡本身按解析出来的内容认，不看哈希（同 content）：服务端的文件被改过，嵌进页面的照样是这份原字节，卡上记着的号码 + 也照样在里面。读得出字节、解析不出一张卡的（截断的 JSON、不是对象）认不出有哪些输入，整张按带遮罩算，和读不出来的 + 快照按带遮罩算（_snapshot_masked）是同一个道理。字节都读不出来的卡本来就不嵌（只给 id、标缺失），这里不管。""" key = f"card:{artifact}" if key not in self._masked: - card = self.content(artifact) if self.intact(artifact) else None - outputs = _card_outputs(events, artifact) if isinstance(card, dict) else {} + card = self.content(artifact) + if not isinstance(card, dict): + self._masked[key] = self.raw(artifact) is not None + return self._masked[key] + outputs = _card_outputs(events, artifact) def output(node_id: Any) -> Any: found = outputs.get(str(node_id or "")) return self.content(found) if found and self.intact(found) else None - items = [item for metric in (card or {}).get("metrics") or [] if isinstance(metric, dict) + items = [item for metric in card.get("metrics") or [] if isinstance(metric, dict) for item in metric.get("inputs") or [] if isinstance(item, dict)] self._masked[key] = any(self._input_masked(item, output) for item in items) return self._masked[key] @@ -329,6 +338,12 @@ class _Artifacts: elif (kind == "query" and self.store.masked(artifact)) \ or (kind == "metric_set" and self.store.card_masked(artifact, self.events)): item["masked"] = True + # 不嵌内容，浏览器就算不出这份文件对不对得上：服务端自己看到的照实带上，核对页据此判不通过、说清原因。 + # 以前被改过的卡原样嵌进去，浏览器自己算出对不上；现在不嵌了，这个证据不能跟着丢 + if not self.store.intact(artifact): + item["mismatch"] = True + if not isinstance(self.store.content(artifact), dict): + item["unparsable"] = True else: # 原样嵌入，不在这里复验哈希：被改过的文件照实交出去，浏览器自己算出对不上 item["b64"] = base64.b64encode(bytes(raw)).decode("ascii") diff --git a/backend/app/bundle/verifier.js b/backend/app/bundle/verifier.js index 8184979f..38b8d154 100644 --- a/backend/app/bundle/verifier.js +++ b/backend/app/bundle/verifier.js @@ -106,6 +106,8 @@ artifactUndecodable: '内容无法解码', artifactUnparsable: '哈希一致，但内容无法解析', masked: '含遮罩列，未随文件附带，只能核对到哈希', + maskedMismatch: '导出时服务端的这份文件与工件 id（哈希）不一致，可能被改过；按含遮罩列处理，未随文件附带', + maskedUnparsable: '导出时服务端的这份文件无法解析，认不出是否含遮罩列，按含遮罩列处理，未随文件附带', maskedOdd: '只有查询快照和口径卡会因遮罩不随文件附带，这份工件却标着遮罩：导出时不会出现这种情况，文件可能被改过', artifactMissing: '导出时服务端已没有这份工件的文件', artifactAbsent: '内容未随文件附带：导出时不会出现这种情况，文件可能被改过', @@ -452,7 +454,7 @@ out.manifest.push({ kind: 'shown', shown: shownSeal, result: shownSeal === seal ? PASS : FAIL, reason: shownSeal === seal ? T.shownOk : T.shownMismatch }) - // ③ 工件。服务端给每份工件的是三样之一：原字节、遮罩（只有查询快照）、服务端已没有文件 + // ③ 工件。服务端给每份工件的是三样之一：原字节、遮罩（查询快照、口径卡，可另带 mismatch、unparsable）、服务端已没有文件 const arts = new Map() for (const a of Array.isArray(data.artifacts) ? data.artifacts.filter(isObj) : []) { const item = { a, result: NA, reason: '', json: null, loose: null, parsed: false } @@ -485,12 +487,17 @@ } } } else if (a.masked) { - // 查询快照带遮罩列；口径卡上有输入取自遮罩列（卡上记着输入的值和代入式）。两种都只给 id，报告文档不会 - if (a.kind === 'query' || a.kind === 'metric_set') { - item.reason = T.masked - } else { + // 查询快照带遮罩列；口径卡上有输入取自遮罩列（卡上记着输入的值和代入式）。两种都只给 id，报告文档不会。 + // 内容不随文件附带，浏览器算不出它对不对得上：导出时服务端看到自己的文件和 id 对不上（mismatch）、解析不出来 + // （unparsable）的照实说，对不上的判不通过——不然被改过的文件看起来和正常的遮罩一样，篡改的证据就丢了 + if (a.kind !== 'query' && a.kind !== 'metric_set') { item.result = FAIL item.reason = T.maskedOdd + } else if (a.mismatch) { + item.result = FAIL + item.reason = T.maskedMismatch + } else { + item.reason = a.unparsable ? T.maskedUnparsable : T.masked } } else if (a.missing) { item.reason = T.artifactMissing @@ -617,7 +624,7 @@ if (!art) return { ...info, result: FAIL, reason: T.numArtifactNotListed } if (art.result === FAIL) return { ...info, result: NA, reason: T.numArtifactFailed } if (art.result !== PASS) { - return { ...info, result: NA, reason: art.a.masked ? T.masked : art.a.missing ? T.artifactMissing : T.numArtifactUnparsable } + return { ...info, result: NA, reason: art.a.masked ? art.reason : art.a.missing ? T.artifactMissing : T.numArtifactUnparsable } } if (!chained.has(entry.artifact)) return { ...info, result: NA, reason: T.numNotChained } if (cite.kind === 'metric') { diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index d33b839c..ba227b3d 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -4,7 +4,13 @@ 对得上的事实只有三种（data/catalog.py 的表级项）： - 两边按不同的时间列筛时间 → 业务日期（business_date）：被采纳的一边按哪一列筛，业务日期就记哪一列。只有一边按时间 - 筛的不算：审核人判断的是要不要限定时间，没在两列之间选过 + 筛的不算：审核人判断的是要不要限定时间，没在两列之间选过。两边的列都得在表结构里声明为带日期的类型（DATE、 + DATETIME、TIMESTAMP、TIMESTAMPTZ，带精度、带时区的写法也算）：差异说明认时间列只看列名和常量像不像日期， + 文本列（SQLite 常把日期存成文字）、整数列（Unix 秒、20260901 这类日期键）、名字带 time 的时长列都可能被说成「按它筛 + 时间」，认不准存的是不是日期，不记。只有时刻的 TIME 也不记：一天里的几点几分（MySQL 还常拿它存时长）不是日期，和 + catalog.is_date_column（剖析和 SQL 检查共用的日期列判断）同一个口径——记成业务日期，SQL 检查会对按真实日期列统计的 + 查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，这一条门槛下它们的业务日期都不由这里记（原因照实说 + 是上传的表格按文字保存，不说成用户声明了 TEXT） - 一边多一个排除无效记录的条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 - 聚合只差去不去重、被采纳的一边去重 → 去重规则（dedup）。采纳了不去重的一边不记：那是这个指标自己的算法（数行）， 写成表的「不去重」会把别的指标带偏 @@ -55,6 +61,7 @@ from typing import Any from sqlalchemy import select from app.data import catalog +from app.data.names import name_key from app.db.base import SessionLocal from app.db.models import DataSource @@ -92,6 +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，可以带精度 +#: （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) @@ -129,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 是分组的列；有效记录条件用它排除 - 指标自己的筛选范围。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") @@ -148,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) + 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": @@ -170,7 +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) -> 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, 为什么不记)。两边的列都要在 + 表结构里声明为带日期的类型（_dated_problem），被采纳的一边先看：它有问题说它；没问题再看未被采纳的一边，另说清为什么 + 也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源还没有可用的表结构时（没探查、探查 + 失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。""" other = "b" if side == "a" else "a" detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} @@ -187,9 +208,51 @@ def _business_date(item: Mapping[str, Any], side: str) -> tuple[tuple[str, str, return None, "有一边按好几列筛时间，无法确定业务日期按哪一列" if (mine[0]["table"], mine[0]["column"]) == (theirs[0]["table"], theirs[0]["column"]): return None, "两边按同一列筛时间，没有可记的业务日期" + 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 _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 认出好几列的说对应好几列：哪一列是它说不准。类型照表结构里记的说（「类型是」，不说 + 「声明为」）：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 "不在数据源的表结构里，无法确认存的是日期" + if len(found) > 1: + return "在表结构里对应好几列，无法确认存的是日期" + declared = " ".join(str(found[0].get("type") or "").split()) + 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], schema_cache: Mapping[str, Any] | None, notes: Mapping[str, Mapping[str, Any]]) -> tuple[tuple[str, str, Any] | None, str | None]: diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 94a32f0f..f9b82a93 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -13,10 +13,19 @@ 连取值也不给；遮罩列不给。计算式只有一项来自查询、或者写了口径说明时不给列名；几项都来自查询又没写口径说明的，列名 里看不出日期、口径的取舍时才给，好让它分得清哪一项是哪一项。 +**遮罩列上的输入不复算。** 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，指标自己的筛选范围不写）。 @@ -326,11 +335,13 @@ def _skipped(ctx: NodeContext, reason: str) -> dict[str, Any]: class _Sources: - """一次复算里要读的查询快照和数据源：同一份只读一次。""" + """一次复算里要读的查询快照、数据源和快照要遮的列：同一份只读一次、只算一次。""" def __init__(self) -> None: self.snapshots: dict[str, tuple[Any, str | None]] = {} self.rows: dict[str, DataSource | None] = {} + self.masks: dict[tuple[str, str | None], frozenset[str]] = {} + self._source_masks: Any = None def snapshot(self, artifact: str) -> tuple[dict[str, Any] | None, str | None]: if artifact not in self.snapshots: @@ -351,18 +362,25 @@ class _Sources: select(DataSource).where(DataSource.name == name))).scalar_one_or_none() return self.rows[name] - async def hidden(self, artifact: str) -> frozenset[str]: - """这份快照要遮的列（小写）：查询当时记下的，加上数据源现在设的。宁可多遮。""" - from app.data.engine import masked_columns + async def hidden(self, artifact: str, source: str | None = None) -> frozenset[str]: + """这份快照要遮的列（小写），和证据面板同一套（api/evidence.py 的 _InputMasks._hidden）：查询当时记下的，加上 + 数据源现在设的；合并查询按它合并的那几个数据源算，换了名字的结果列按来历遮。快照读不出来时按 source（取值链上 + 那份结果里带着的数据源名）现在设的算：遮罩不能因为快照坏了就失效。宁可多遮。""" + from app.api.evidence import _Masks, _merge_masks, _recorded_masks # api.evidence 导入了 engine 里的模块 + from app.engine.merge_query import MERGE_SOURCE - snap, _ = self.snapshot(artifact) - recorded = (snap or {}).get("mask_columns") - out = {c.lower() for c in recorded if isinstance(c, str)} if isinstance(recorded, list) else set() - name = (snap or {}).get("source") - if isinstance(name, str) and name: - row = await self.source(name) - out |= {c.lower() for c in masked_columns(row.options if row is not None else None)} - return frozenset(out) + key = (artifact, source) + if key not in self.masks: + if self._source_masks is None: + self._source_masks = _Masks() + snap, _ = self.snapshot(artifact) + if snap is not None and snap.get("source") == MERGE_SOURCE: + hidden, _ = await _merge_masks(snap, self._source_masks) + else: + current, _ = await self._source_masks.of((snap or {}).get("source") or source) + hidden = set(current) | (_recorded_masks(snap) if snap is not None else set()) + self.masks[key] = frozenset(hidden) + return self.masks[key] def _loader(state: GraphState) -> tuple[Callable[[str], Any], dict[str, str]]: @@ -406,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, @@ -457,34 +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): + 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"]] = { @@ -495,6 +524,55 @@ async def _plan_metric(state: GraphState, metric: dict[str, Any], options: _Opti return row, jobs +async def _positional_masks(item: dict[str, Any], sources: _Sources, load: Callable[[str], Any], + outputs: dict[str, str], masks: dict[str, frozenset[str]]) -> 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: + 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 f"输入 {path} 取自设置了遮罩的列，不能用于复算", False + _, why = sources.snapshot(origin.artifact) + 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] ) -> tuple[dict[str, Any], str | None]: """一个来自查询快照的输入：(盲算要用的信息, None) 或 (_, 为什么不能单独复算)。""" @@ -559,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) @@ -572,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)): @@ -604,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: @@ -630,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 @@ -868,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"]) @@ -884,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 @@ -952,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 ""))) @@ -981,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"]}, @@ -1020,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]: @@ -1121,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 565ca784..6f67f5b2 100644 --- a/backend/tests/fixtures/dual_check_shop.py +++ b/backend/tests/fixtures/dual_check_shop.py @@ -9,6 +9,10 @@ - 付款客户数（按支付日、排除退款）：按行计数 3，去重计数 2 另有渠道（channel）和测试单标记（is_test，都是 0）两列：写数据目录的测试拿它们凑「指标自己的筛选范围」和标记列。 + +下单日、支付日两列声明为 DATE（SQLite 照样存成文字，按字符串比的先后和按日期比的一样）：双盲复算只在两边的时间列都 +声明为带日期的类型时才把选择记成业务日期。build(date_type="TEXT") 建一份声明成文本的、build(date_type="") 建一份 +不写类型的，看不记的那几种。 """ from __future__ import annotations @@ -50,11 +54,11 @@ BY_REGION = (f"SELECT region, SUM(amount) AS gmv FROM orders WHERE status <> 're DEFINITION = "已支付订单的金额合计：按支付日期（paid_date）归属到当周，不含退款" -def build(path: Path) -> Path: +def build(path: Path, *, date_type: str = "DATE") -> Path: db = sqlite3.connect(path) db.executescript( "CREATE TABLE orders (id INTEGER PRIMARY KEY, region TEXT, customer_id INTEGER, amount REAL, status TEXT, " - "order_date TEXT, paid_date TEXT, channel TEXT, is_test INTEGER);") + f"order_date {date_type}, paid_date {date_type}, channel TEXT, is_test INTEGER);") db.executemany("INSERT INTO orders VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)", ROWS) db.commit() db.close() diff --git a/backend/tests/test_bundle_masked_cards.py b/backend/tests/test_bundle_masked_cards.py index 11f8d917..0803a655 100644 --- a/backend/tests/test_bundle_masked_cards.py +++ b/backend/tests/test_bundle_masked_cards.py @@ -4,6 +4,13 @@ vars.res.rows[0][1]），在这份要发给别人的文件里照样看得到。认哪个输入取自哪份查询结果的哪几列，和证据面板、推演同一个 认法（recompute.input_origin）；核对不了的宁可多挡。 +服务端的卡文件被改过（哈希对不上）也照卡上记的输入认：嵌进去的是原字节，号码照样在里面。读得出字节、却解析不出一张卡 +（截断的 JSON、不是对象）的，认不出输入，整张按带遮罩算、不嵌：和读不出来的快照按带遮罩算是同一个道理。 + +不嵌的工件，服务端把它知道的情况照实带上：自己的文件和工件 id（哈希）对不上的标 mismatch，解析不出来的标 unparsable， +核对页据此判不通过、照实说原因（以前原字节嵌进去，浏览器自己算出对不上；现在不嵌了，这个证据不能跟着丢）。带遮罩的 +查询快照同样标。 + 工件、事件都是手搭的（会员、号码都是编的），照 tests/test_bundle_build.py 的样子。 """ from __future__ import annotations @@ -13,6 +20,7 @@ import base64 import pytest from app.bundle.build import build_bundle, render_html +from app.core.artifact_store import content_hash from tests.test_bundle_build import GRAPH, MASKED_ID, RUN, SNAP_MASKED, STORE, doc_of, stored #: SNAP_MASKED 里的号码（phone 设了遮罩） @@ -27,13 +35,19 @@ def card_with(inputs, substituted="len('000-0001')", expression=f"len({CELL})"): "substituted": substituted, "inputs": inputs}]} -def bundle_of(card, *, outputs=None, source_masks=None): - """一次手搭的运行：会员查询（带遮罩的快照）→ 数据整形（outputs 里的产出）→ 口径卡 → 报告。""" +def bundle_of(card, *, outputs=None, source_masks=None, tamper=None, rehash=False, files_changed=None): + """一次手搭的运行：会员查询（带遮罩的快照）→ 数据整形（outputs 里的产出）→ 口径卡 → 报告。tamper 给了就把服务端的 + 卡文件换成 tamper(原字节)，id 照旧（文件被改过、哈希对不上）；rehash 为真时 id 按换过的字节算（服务端存的就是这一份， + 哈希对得上）。files_changed 换掉服务端别的文件（{工件 id: 新字节}，id 照旧）。""" card_id, card_raw = stored(card) + if tamper is not None: + card_raw = tamper(card_raw) + if rehash: + card_id = content_hash(card_raw) doc = doc_of(extra_catalog={"m:gmv": {"alias": "m:gmv", "kind": "metric", "artifact": card_id, "node_id": "card", "locator": {"metric": "gmv"}, "value": 1280.5}}) doc_id, doc_raw = stored(doc) - files = {**STORE, card_id: card_raw, doc_id: doc_raw} + files = {**STORE, card_id: card_raw, doc_id: doc_raw, **(files_changed or {})} events = [ (1, "run.started", None, {"nodes": 7}), (2, "tool.end", "members", {"tool": "db_query__crm", "query_artifact": MASKED_ID}), @@ -105,3 +119,61 @@ def test_a_card_reading_a_cell_by_position_is_judged_the_same_way(parsed, masked assert PHONE not in render_html(bundle) else: assert PHONE in base64.b64decode(entry["b64"]).decode() + + +def test_a_card_file_changed_on_the_server_is_still_judged_by_its_inputs(): + """服务端的卡文件被改过（这里改了口径版本，哈希对不上）：照样按卡上记的输入认遮罩，记着号码的只给 id。以前先看哈希、 + 对不上就当卡上没有输入，原字节照样嵌进页面，号码进了文件。不嵌了，就另标 mismatch：核对页判这一项不通过，不和 + 没被改过的遮罩卡一样说成正常的无法核对。""" + card = card_with([cell_input("phone", PHONE)]) + bundle, entry = bundle_of(card, tamper=lambda raw: raw.replace(b'"caliber_version":"v1"', b'"caliber_version":"v9"')) + assert entry["id"] == stored(card)[0] + assert entry == {"id": entry["id"], "kind": "metric_set", "node_id": "card", "masked": True, "mismatch": True} + assert PHONE not in render_html(bundle) + + +def test_a_changed_card_without_masked_inputs_is_still_embedded_for_the_browser_to_judge(): + """改过的卡上没有遮罩列的值：照常原样嵌进去，浏览器自己算出哈希对不上（只是遮罩的判断不再看哈希）。""" + card = card_with([cell_input("spend", 320.0)], substituted="320.0", expression="cell(nodes.members, 0, 'spend')") + changed = stored(card)[1].replace(b'"caliber_version":"v1"', b'"caliber_version":"v9"') + _, entry = bundle_of(card, tamper=lambda raw: changed) + assert "masked" not in entry and base64.b64decode(entry["b64"]) == changed + + +@pytest.mark.parametrize("damage", [ + # 截断的 JSON：号码还在已经写下的那半截里 + lambda raw: raw[: raw.index(b"000-0001") + len(b"000-0001")], + # 解析得出、却不是一张卡（对象） + lambda raw: b'["000-0001"]', +]) +def test_a_card_file_that_does_not_parse_as_a_card_is_not_embedded(damage): + """服务端的卡文件读得出字节、解析不出一张卡：认不出卡上有哪些输入，整张按带遮罩算、只给 id，不原样嵌进页面。字节和 + id 对不上（mismatch），也解析不出来（unparsable）：两样都照实标上。""" + card = card_with([cell_input("phone", PHONE)]) + bundle, entry = bundle_of(card, tamper=damage) + assert entry == {"id": entry["id"], "kind": "metric_set", "node_id": "card", "masked": True, "mismatch": True, + "unparsable": True} + assert PHONE not in render_html(bundle) + + +def test_a_card_stored_as_something_unparsable_is_marked_so(): + """服务端存的这一份本来就解析不出一张卡（哈希对得上，不是被改过）：整张按带遮罩算、不嵌，只标 unparsable——核对页照实 + 说认不出是否含遮罩列，不说成「含遮罩列」，也不说被改过。""" + card = card_with([cell_input("phone", PHONE)]) + bundle, entry = bundle_of(card, tamper=lambda raw: raw[: raw.index(b"000-0001") + len(b"000-0001")], rehash=True) + assert entry == {"id": entry["id"], "kind": "metric_set", "node_id": "card", "masked": True, "unparsable": True} + assert PHONE not in render_html(bundle) + + +def test_a_masked_snapshot_changed_on_the_server_is_marked_so(): + """带遮罩的查询快照在服务端被改过（哈希对不上）：照旧只给 id，另标 mismatch，核对页判不通过。以前和没被改过的遮罩快照 + 一模一样，总结论还是「可核对的项全部通过」。没被改过的照旧只标遮罩。""" + card = card_with([cell_input("spend", 320.0)], substituted="320.0", expression="cell(nodes.members, 0, 'spend')") + snapshot = lambda bundle: next(a for a in bundle["artifacts"] if a["id"] == MASKED_ID) # noqa: E731 + bundle, _ = bundle_of(card) + assert snapshot(bundle) == {"id": MASKED_ID, "kind": "query", "node_id": "members", "masked": True} + changed = stored(SNAP_MASKED)[1].replace("林一".encode(), "林二".encode()) + bundle, _ = bundle_of(card, files_changed={MASKED_ID: changed}) + assert snapshot(bundle) == {"id": MASKED_ID, "kind": "query", "node_id": "members", "masked": True, + "mismatch": True} + assert PHONE not in render_html(bundle) diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index de39d7d9..54cc8bd1 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -2,8 +2,13 @@ - 只差一个方面、又对得上目录事实时才写：两边按不同的时间列筛 → 业务日期，一边多一个排除无效记录的条件 → 有效记录 条件，聚合只差去不去重、被采纳的一边去重 → 去重规则；按被采纳那一边的写法记 +- 业务日期只在两边的列都在表结构里声明为带日期的类型（DATE、DATETIME、TIMESTAMP，带精度、带时区的写法也算）时才写： + 文本、整数（Unix 秒、20260901 这类日期键）、只有时刻的 TIME、表结构里查不到、没声明类型（SQLite 探出来是 NULL）的 + 不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源说清日期是按文字保存的； + 数据源还没有可用的表结构时说表结构不可用 - 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛：不写， - 照实说原因 + 照实说原因。同一列上的时间范围不同、跨两张表的条件，差异说明只给兜底，同样不写；说明以后真把跨两张表的条件说成一条 + 筛选条件，derive 自己那道「只涉及一张表」的检查也拦得下 - 指标自己的筛选范围（只看华东、只看线上）不是有效记录条件：取值出现在指标名称、口径说明、所在分组、运行输入里的， 条件用到分组的列的，不是在排除退款、作废、测试这类无效记录的，都不写 - 写的是人工填写、已确认，备注写明运行、节点、指标；同样的说法已确认过就不再写（不升版本），确认过的不同说法不覆盖； @@ -22,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, @@ -40,8 +46,8 @@ SCHEMA = {"tables": { "orders": {"name": "orders", "columns": [ {"name": "id", "type": "INTEGER"}, {"name": "region", "type": "TEXT"}, {"name": "customer_id", "type": "INTEGER"}, {"name": "amount", "type": "REAL"}, - {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "TEXT"}, - {"name": "paid_date", "type": "TEXT"}, {"name": "channel", "type": "TEXT"}, + {"name": "status", "type": "TEXT"}, {"name": "order_date", "type": "DATE"}, + {"name": "paid_date", "type": "DATE"}, {"name": "channel", "type": "TEXT"}, {"name": "is_test", "type": "INTEGER"}], "primary_key": ["id"]}, "customers": {"name": "customers", "columns": [ {"name": "id", "type": "INTEGER"}, {"name": "level", "type": "TEXT"}, {"name": "status", "type": "TEXT"}], @@ -169,13 +175,33 @@ def test_table_names_are_dropped_from_the_recorded_filter(): assert why is None and fact.value == "status <> 'refunded'" +#: 订单关联客户；盲算另多一个比较两张表的列的条件 +JOINED = (f"SELECT SUM(o.amount) AS gmv FROM orders o JOIN customers c ON o.customer_id = c.id " + f"WHERE o.paid_date {WEEK}") + + def test_a_condition_across_two_tables_is_not_recorded(): - joined = (f"SELECT SUM(o.amount) AS gmv FROM orders o JOIN customers c ON o.customer_id = c.id " - f"WHERE o.paid_date {WEEK}") - picked = candidate(joined, f"{joined} AND o.status <> c.status", "b") - assert [i["aspect"] for i in picked["diff"]["items"]] == ["filter"] + """几张表交错的条件，差异说明只给兜底（归纳不成具体差异）：审核人选的是整体哪边对，不写。""" + picked = candidate(JOINED, f"{JOINED} AND o.status <> c.status", "b") + assert picked["diff"]["fallback"] is True and [i["aspect"] for i in picked["diff"]["items"]] == ["other"] fact, why = derive(picked, SCHEMA) - assert fact is None and "不是只涉及一张表的列" in why + assert fact is None and "未能归纳" in why and "销售额" in why + + +def test_a_filter_spanning_two_tables_is_not_a_valid_filter(): + """差异说明真把跨两张表的条件说成一条筛选条件（以后说明放宽了、或者别的来源交来的说明）：derive 自己也拦得下—— + 有效记录条件记在一张表名下，用到两张表的列的写进哪张都不对。差异说明直接构造：从 SQL 走现在只有兜底。""" + picked = candidate(JOINED, f"{JOINED} AND o.status <> c.status", "b") + picked["diff"] = {"equivalent": False, "fallback": False, "items": [{ + "aspect": "filter", "a": None, "b": "o.status <> c.status", + "text": "盲算只保留 o.status <> c.status 的行，原流程没有这个条件", + "detail": {"columns": [{"table": "customers", "column": "status"}, {"table": "orders", "column": "status"}]}, + }]} + fact, why = derive(picked, SCHEMA) + assert fact is None and "不是只涉及一张表的列" in why and "销售额" in why + # 同样的条件只涉及一张表时这道检查放行（后面照常看是不是在排除无效记录）：拦下它的确实是「两张表」 + picked["diff"]["items"][0]["detail"] = {"columns": [{"table": "orders", "column": "status"}]} + assert "不是只涉及一张表的列" not in (derive(picked, SCHEMA)[1] or "") def test_counting_distinct_becomes_the_dedup_rule(): @@ -207,16 +233,153 @@ def test_equivalent_sql_or_fallback_or_other_aspects_are_not_recorded(): assert [i["aspect"] for i in joined["diff"]["items"]] == ["aggregate"] assert "不只是去不去重" in derive(joined, SCHEMA)[1] + # 同一列上的时间范围不同：差异说明只给兜底（说不清差的是哪一段），不写 ranged = candidate(RIGHT, RIGHT.replace("'2026-09-07'", "'2026-09-06'"), "b") - assert [i["aspect"] for i in ranged["diff"]["items"]] == ["time_range"] - assert "时间范围" in derive(ranged, SCHEMA)[1] + assert ranged["diff"]["fallback"] is True and [i["aspect"] for i in ranged["diff"]["items"]] == ["other"] + assert "未能归纳" in derive(ranged, SCHEMA)[1] + + # 只差一个方面、可那个方面对不上目录里的三项（关联方式）：不写，原因说清差在哪 + left = (f"SELECT SUM(o.amount) AS gmv FROM orders o LEFT JOIN customers c ON c.id = o.customer_id " + f"WHERE o.paid_date {WEEK}") + joins = candidate(left, left.replace("LEFT JOIN", "JOIN"), "b") + assert joins["diff"]["fallback"] is False and [i["aspect"] for i in joins["diff"]["items"]] == ["join"] + fact, why = derive(joins, SCHEMA) + assert fact is None and "两边差在关联" in why and "对不上数据目录里的业务日期、有效记录条件或去重规则" in why assert "没有采用任何一边" in derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "choice": "neither"}, SCHEMA)[1] def test_a_table_missing_from_the_schema_is_not_recorded(): - fact, why = derive(candidate(BY_ORDER_DATE, RIGHT, "b"), {"tables": {"refunds": {"columns": []}}}) - assert fact is None and "不在数据源的表结构里" in why + """认出了事实（排除退款的有效记录条件），表却不在数据源的表结构里：不写。""" + fact, why = derive(candidate(WITH_REFUNDS, RIGHT, "b"), {"tables": {"refunds": {"columns": []}}}) + assert fact is None and "涉及的表「orders」不在数据源的表结构里" in why + + +def schema_with(**types: str | None) -> dict: + """SCHEMA 的订单表里几列换成给定的声明类型；给 None 的列从表结构里拿掉。""" + orders = SCHEMA["tables"]["orders"] + columns = [{**c, "type": types[c["name"]]} if c["name"] in types else c + for c in orders["columns"] if not (c["name"] in types and types[c["name"]] is None)] + return {"tables": {**SCHEMA["tables"], "orders": {**orders, "columns": columns}}} + + +@pytest.mark.parametrize("order_date, paid_date", [ + ("DATE", "DATE"), + ("TIMESTAMP", "TIMESTAMP"), + ("datetime", "DateTime(6)"), + ("TIMESTAMPTZ", "timestamp(3) with time zone"), + ("TIMESTAMP WITHOUT TIME ZONE", "TIMESTAMP(6) WITH LOCAL TIME ZONE"), +]) +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)，无法确认存的是日期"), + # 没被采纳的一边也得是：它要是按 Unix 秒、20260901 这类整数筛的，审核人并没有在两个日期列之间选 + ({"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，无法确认存的是日期"), + # 只有时刻的 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(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) + 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": {}, "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 + 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(): + """差异说明里的表名、列名是规范化过的（小写）：表结构里写成大写的照样认得出，类型照它声明的看。""" + upper = {"tables": {"ORDERS": {"name": "ORDERS", "columns": [ + {"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 + + +def test_the_preview_follows_the_type_check(): + """审批卡上的预告和真写同一套判断：一边的列是文本，选哪一边都预告不写（认不出事实，不带写到哪），原因同 derive。""" + schema = schema_with(paid_date="TEXT") + 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"] + + +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 9b44c95d..0ee341c9 100644 --- a/backend/tests/test_dual_check_fixture.py +++ b/backend/tests/test_dual_check_fixture.py @@ -9,11 +9,18 @@ check-verify 在夹具上能过，真实数据上却一处都不高亮（2026-10 差异都没有」，delta_pct 是 (盲算值 − 原流程值) ÷ |原流程值| × 100 保留两位。夹具的数据源当 SQLite 比（方言 sqlite）， 两边的叫法是「原流程」「盲算」，同节点调用时一样。 +**变体。** variants 换掉「销售额」的那一对 SQL，挑的是收尾版差异说明（只说时间列、多出来的筛选、聚合与去重、关联 +四类，别的只给兜底）下各有代表性的几种：按哪一列筛时间不同（time_column，两列都声明为日期类型，预告写业务日期）、 +只有一边按时间筛、一边多出来的筛选（排除无效记录的 filter_extra 预告写有效记录条件，指标自己范围的 filter_scope 不写）、 +差两个筛选条件、关联方式、聚合只差去重（dedup，预告写去重规则）、说不清的（fallback：同一列上的时间范围不同，只给兜底， +审批卡默认展开逐行对照）、一边解析失败。 + **写目录的预告。** 服务端在摘要里另带每个指标的 catalog（采用原流程、采用盲算各会往数据目录里写什么， dual_check_facts.preview，约定之外另加的），审批卡照它说；老服务端没有时卡上按差异的形状有条件地说，两种说法都按 「服务端到底写不写」核对（2026-10 复核：卡上说会记入、服务端却不写的有好几种）。夹具里的预告（catalog_preview 按指标、 -variants 和 peers 各自的 catalog）也是服务端的真实输出：按下面的 SCHEMA、空目录、指标自己的范围留空算。双盲复算的 -服务端合入之前没有这个模块，逐项比对那几项跳过，形状和门槛照样查。 +variants 和 peers 各自的 catalog）也是服务端的真实输出：按下面的 SCHEMA、空目录、指标自己的范围留空算。SCHEMA 里 +paid_at、created_at 声明为 TIMESTAMP：业务日期只在两边的时间列都声明为带日期的类型时才写。双盲复算的服务端合入之前 +没有这个模块，逐项比对那几项跳过，形状和门槛照样查。 **同一次裁决里几个指标落到同一项上。** 预告是每个指标单独算的，服务端写目录前却要把同一次裁决里落到同一张表同一项上的 说法归拢（业务日期、去重规则说法不同的一项都不写，有效记录条件合成一句写，驳回过的拿出来不写）。服务端把每种组合按 @@ -54,16 +61,17 @@ DIALECT = "sqlite" LABELS = ("原流程", "盲算") #: 摘要里每个指标最多列几条差异（同双盲复算节点的 SUMMARY_DIFFS） SUMMARY_DIFFS = 3 -#: 预告用的数据源名和表结构（虚构）：服务端要认得出表、认得出聚合的列属于哪张表，才会预告写哪一项 +#: 预告用的数据源名和表结构（虚构）：服务端要认得出表、认得出聚合的列属于哪张表，才会预告写哪一项；时间列声明为 +#: TIMESTAMP，业务日期才会预告写（文本、整数、只有时刻的时间列不记） SOURCE = "shop" SCHEMA: dict[str, Any] = {"tables": { "orders": {"name": "orders", "columns": [ {"name": "id", "type": "INTEGER"}, {"name": "user_id", "type": "INTEGER"}, {"name": "amount", "type": "REAL"}, - {"name": "status", "type": "TEXT"}, {"name": "channel", "type": "TEXT"}, {"name": "paid_at", "type": "TEXT"}, - {"name": "created_at", "type": "TEXT"}], "primary_key": ["id"]}, + {"name": "status", "type": "TEXT"}, {"name": "channel", "type": "TEXT"}, {"name": "paid_at", "type": "TIMESTAMP"}, + {"name": "created_at", "type": "TIMESTAMP"}], "primary_key": ["id"]}, "refunds": {"name": "refunds", "columns": [ {"name": "id", "type": "INTEGER"}, {"name": "order_id", "type": "INTEGER"}, {"name": "amount", "type": "REAL"}, - {"name": "created_at", "type": "TEXT"}], "primary_key": ["id"]}, + {"name": "created_at", "type": "TIMESTAMP"}], "primary_key": ["id"]}, }} #: 写目录时认得出的三项（dual_check_facts.FACT_ITEMS） FACT_ITEMS = ("business_date", "valid_filter", "dedup") @@ -183,6 +191,28 @@ def test_variants_carry_the_real_diff(): assert swap["diff"] == _diff(original, blind), f"{group}.{name}" +#: 变体里各有代表性的几种（见模块说明「变体」）：名字 → (差异条目的方面, 预告会写的边和项) +REPRESENTATIVE = { + "time_column": (["time"], {"a": "business_date", "b": "business_date"}), + "filter_extra": (["filter"], {"a": "valid_filter"}), + "dedup": (["aggregate"], {"a": "dedup"}), + "fallback": (["other"], {}), +} + + +def test_variants_keep_their_representative_shapes(): + """检查脚本拿这几种变体看卡上怎么说。差异说明或写目录的规则一改，变体可能悄悄变成别的形状（换了时间列却只给兜底、 + 预告不再写业务日期），那几项检查就什么也没测：这里守住每种的方面和预告会写的项，说不清的那种还得有逐行对照。不依赖 + 服务端的模块，一直查。""" + fx = _load() + for name, (aspects, writes) in REPRESENTATIVE.items(): + swap = fx["variants"][name] + assert [i["aspect"] for i in swap["diff"]["items"]] == aspects, name + assert {side: said["item"] for side, said in swap["catalog"].items() if said["write"]} == writes, name + fallback = fx["variants"]["fallback"]["diff"] + assert fallback["fallback"] is True and fallback["text_diff"], fallback + + def test_text_diff_is_not_the_raw_text(): """守住这次的教训：至少有一处规范化之后的行和原文的任何一行都不一样（真实输出就是这样），夹具不再照原文手写。""" fx = _load() diff --git a/backend/tests/test_dual_check_node.py b/backend/tests/test_dual_check_node.py index 6ce5a013..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）按无法复算记、原因点名 + 是哪个输入，不盲算；认不出取自哪一列的按位置取的格不写进计算式；按位置取的格不改变容器输入的判法 """ from __future__ import annotations @@ -28,9 +31,10 @@ 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 from app.engine.runner import run_manager from app.main import app from tests.fixtures.dual_check_shop import ( @@ -567,6 +571,78 @@ async def test_the_same_choice_twice_writes_the_catalog_once(monkeypatch, shop, # -------------------------------------------------------------------------- +@pytest.fixture +async def text_dated_shop(tmp_path): + """同一个库，只是下单日、支付日两列声明成文本（SQLite 常这样存日期）。""" + name, source_id = await add_source(build(tmp_path / "text_dated.db", date_type="TEXT")) + yield name, source_id + await drop_source(source_id) + + +async def test_time_columns_declared_as_text_are_not_written_as_the_business_date(monkeypatch, text_dated_shop, client): + """两边按不同的列筛时间，可这两列在表结构里声明成文本：认不准存的是不是日期，选哪边都不记成业务日期。审批卡上的 + 预告照实说（认不出可记的事实，不带写到哪），裁决后时间线上跳过的原因和预告一字不差。声明成 DATE 的同一个库照常记 + （test_choosing_the_blind_result_degrades_and_writes_the_catalog）。""" + name, source_id = text_dated_shop + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 1 and "order_date" in item["diff"][0] and "paid_date" in item["diff"][0] + said = item["catalog"] + # 两列都是文本：各说被采纳的那一边的列 + for side, label, column in (("a", "原流程", "order_date"), ("b", "盲算", "paid_date")): + assert 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") + 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_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")}) @@ -969,42 +1045,101 @@ async def test_masked_group_columns_are_not_shown_to_the_blind_agent(monkeypatch async def test_masked_group_values_stay_out_of_the_card_and_the_catalog_log(monkeypatch, masked_shop, client): """原流程按遮罩列（region）分组、口径卡取华东那一行：差异说明收窄到这一行时，不能拿这一行的分组值补等值条件—— - 那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄，照整条比。""" + 那样「region = '华东'」就进了审批卡、对照工件、事件和目录日志。遮罩的分组列上不收窄，照整条比。 + + 盲算没分组、按下单日筛：收窄了的话，两边差在时间列和原流程多出来的「region = '华东'」，说明和对照里都带着这个取值； + 照整条比，时间列照样说得出，原流程的分组、多选的列归纳不了（_compared_whole）。""" name, _ = masked_shop first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} - Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")}) run = await start(graph(name, BY_REGION, metrics=[first], dual={"definitions": {}})) assert run.status == "interrupted", run.error payload = (await pending(run.id)).payload + [item] = payload["summary"] + assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310") card = artifact_store.load(payload["card_artifact"]) done = await decide(client, run.id, {"first": "a"}) assert done.status == "succeeded", done.error evs = await events(run.id) _no_masked_value(payload, card, done, evs) - # 照整条比：原流程分组、盲算没分组都列出来，不止差一个方面，选了也不写目录 - assert len(payload["summary"][0]["diff"]) > 1 + _compared_whole(payload, card) + # 差异说明里有未能归纳的部分：选了也不写目录 [wrote] = logs(evs, "dual_check_catalog") assert wrote["written"] == [] and "不止差一个方面" in wrote["skipped"][0]["reason"] async def test_a_renamed_masked_group_column_is_not_narrowed_either(monkeypatch, masked_shop, client): """原流程把遮罩列改了名再分组（region AS area）：结果列名 area 不在遮罩里，分组值却还是 region 的取值。收窄时看的 - 是分组键用到的列，一样不拿这一行的分组值补等值条件，「华东」不进审批卡、对照工件、产出和事件。""" + 是分组键用到的列，一样不拿这一行的分组值补等值条件，「华东」不进审批卡、对照工件、产出和事件，照整条比。""" name, _ = masked_shop renamed = (f"SELECT region AS area, SUM(amount) AS gmv FROM orders WHERE status <> 'refunded' " f"AND paid_date {WEEK} GROUP BY region ORDER BY region") first = {**GMV, "id": "first", "name": "首组销售额", "expression": "cell(nodes.fetch, 0, 'gmv')"} - Blind(monkeypatch, name, {"首组销售额": (RIGHT, "gmv")}) + blind = Blind(monkeypatch, name, {"首组销售额": (BY_ORDER_DATE, "gmv")}) run = await start(graph(name, renamed, metrics=[first], dual={"definitions": {}})) assert run.status == "interrupted", run.error payload = (await pending(run.id)).payload - assert payload["summary"][0]["original_rendered"].startswith("300") + [item] = payload["summary"] + assert item["original_rendered"].startswith("300") and item["blind_rendered"].startswith("310") card = artifact_store.load(payload["card_artifact"]) done = await decide(client, run.id, {"first": "a"}) assert done.status == "succeeded", done.error _no_masked_value(payload, card, done, await events(run.id)) - # 照整条比：原流程分组、盲算没分组都列出来 - assert len(payload["summary"][0]["diff"]) > 1 + _compared_whole(payload, card) + # 交给盲算的这一行的分组取值也不给：结果列 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: + """差异说明照整条比的痕迹（原流程按地区分组取一行，盲算没分组、按下单日筛）：时间列照样说得出；原流程的分组、多选的 + 地区列归纳不了，另起一条「另有未能归纳的差异」；规范化后的对照里还留着原流程的 GROUP BY。收窄了的话分组去掉、换成 + 等值条件，两边都是简单形状，说明里就是时间列和多出来的那个条件，这几样都不会有。""" + [metric] = card["metrics"] + diff = metric["diff"] + assert [i["aspect"] for i in diff["items"]] == ["time", "other"], diff["items"] + assert diff["fallback"] is True and any("GROUP BY" in line for line in diff["text_diff"]), diff["text_diff"] + [item] = payload["summary"] + assert len(item["diff"]) == 2 and "order_date" in item["diff"][0] and "另有未能归纳的差异" in item["diff"][1], item + assert item["fallback"] is False async def test_a_masked_group_column_of_the_blind_query_is_not_narrowed_either(monkeypatch, masked_shop, client): @@ -1021,6 +1156,146 @@ 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])"} + + +def with_regions(spec: dict, source: str) -> dict: + """取数之后另查一次按地区分组的结果（第二份快照），数据整形把它解析进 vars.res：口径卡按位置取它的格。""" + nodes = list(spec["nodes"]) + at = next(i for i, n in enumerate(nodes) if n["id"] == "fetch") + 1 + nodes[at:at] = [node("regions", "tool", tool=f"db_query__{source}", args={"sql": BY_REGION}), + node("parse", "transform", mode="json", template="{{ nodes.regions }}", assign_to="res")] + return {"nodes": nodes, "edges": [{"source": a["id"], "target": b["id"]} for a, b in zip(nodes, nodes[1:])]} + + +async def test_a_masked_cell_read_by_position_from_another_snapshot_is_not_blind_computed(monkeypatch, masked_shop): + """口径卡上一个指标：cell() 取第一次查询（A，不碰遮罩列）的销售额，另按位置取第二次查询（B，按遮罩列 region 分组）的 + 首组名称（vars.res.rows[0][0]）。B 不是 cell() 用到的快照：以前预检只为 A 算遮罩，认不出按位置取的那一格在遮罩列上， + 指标照常盲算，「华东」作为字面量写进了交给盲算模型的计算式。现在按需算 B 的遮罩：这个指标按无法复算记缺口、不盲算， + 遮罩列的取值不进盲算的任务、节点产出和节点发的事件。""" + name, _ = masked_shop + blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")}) + run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name)) + assert run.status == "succeeded", run.error + evs = await events(run.id) + out = dual_output(run, evs) + [row] = out["metrics"] + assert row["metric_id"] == "mix" and row["status"] == "unchecked", row + # 和 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 + assert "华东" not in json.dumps(out, ensure_ascii=False) + assert not [e.type for e in evs if e.node_id == "dual" and not e.type.startswith(("tool.", "llm.")) + and "华东" in json.dumps(e.data, ensure_ascii=False)] + + +async def test_an_unmasked_cell_read_by_position_from_another_snapshot_is_blind_computed(monkeypatch, shop): + """同样的图、数据源没设遮罩：预检照常通过，指标进入盲算（按位置取的那一格作为字面量写进计算式），两边一致。拦下上面 + 那个指标的确实是遮罩。""" + name, _ = shop + blind = Blind(monkeypatch, name, {"销售额加首组名称字数": (RIGHT, "gmv")}) + run = await start(with_regions(graph(name, RIGHT, metrics=[MIX], dual={"definitions": {}}), name)) + assert run.status == "succeeded", run.error + out = dual_output(run, await events(run.id)) + assert out["status"] == "agreed" and out["metrics"][0]["status"] == "agreed", out + [task] = blind.tasks + assert "len('华东')" in task + + +#: 按位置取的那一格：数据整形解析好的第二份查询结果（首组地区名），快照 id 是编的 +POSITIONAL = {"path": "vars.res.rows[0][0]", "value": "华东", "node_id": "parse", "via": "transform", "status": "ok"} +LOST = "cd" * 32 + + +def parsed_regions(source: str) -> dict: + return {"columns": ["region", "gmv"], "rows": [["华东", 300.0], ["华北", 30.0]], "artifact": LOST, "source": source} + + +async def test_a_cell_read_by_position_from_an_unreadable_snapshot_is_not_trusted(masked_shop): + """按位置取的那一格所在的快照读不出来（被改过、文件不在了）：查询当时记下的遮罩无从核对。数据源现在设的遮罩遮着这一列 + 的，照遮罩算（交给重算的预检按遮罩说）；没遮的也不当成没遮，按无法复算记、原因说清。""" + from app.engine.nodes.dual_check import _positional_masks, _Sources + + name, _ = masked_shop + for source, expected in ((name, "输入 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, 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"]) diff --git a/frontend/src/dev/fixtures/dual_check.json b/frontend/src/dev/fixtures/dual_check.json index 6967faf3..f39233a5 100644 --- a/frontend/src/dev/fixtures/dual_check.json +++ b/frontend/src/dev/fixtures/dual_check.json @@ -565,33 +565,43 @@ "daily_max_usd": 2.0 }, "variants": { - "time_range": { + "time_column": { "sql": [ - "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE paid_at >= '2026-09-01'\n AND paid_at < '2026-09-08'", - "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE paid_at >= '2026-09-01'\n AND paid_at < '2026-09-09'" + "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE created_at >= '2026-09-01'\n AND created_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE paid_at >= '2026-09-01'\n AND paid_at < '2026-09-08'" ], "diff": { "equivalent": false, "items": [ { - "aspect": "time_range", - "text": "原流程的时间范围是 paid_at >= '2026-09-01' AND paid_at < '2026-09-08'，盲算是 paid_at >= '2026-09-01' AND paid_at < '2026-09-09'" + "aspect": "time", + "text": "原流程按 created_at 筛时间，盲算按 paid_at" } ], "fallback": false, "text_diff": [ - "- orders.paid_at >= '2026-09-01' AND orders.paid_at < '2026-09-08'", - "+ orders.paid_at >= '2026-09-01' AND orders.paid_at < '2026-09-09'" + "- orders.created_at >= '2026-09-01' AND orders.created_at < '2026-09-08'", + "+ orders.paid_at >= '2026-09-01' AND orders.paid_at < '2026-09-08'" ] }, "catalog": { "a": { - "write": false, - "reason": "指标「销售额」两边差在时间范围，对不上数据目录里的业务日期、有效记录条件或去重规则，不写入数据目录" + "write": true, + "source": "shop", + "table": "orders", + "item": "business_date", + "value": { + "column": "created_at" + } }, "b": { - "write": false, - "reason": "指标「销售额」两边差在时间范围，对不上数据目录里的业务日期、有效记录条件或去重规则，不写入数据目录" + "write": true, + "source": "shop", + "table": "orders", + "item": "business_date", + "value": { + "column": "paid_at" + } } } }, @@ -635,7 +645,7 @@ "items": [ { "aspect": "filter", - "text": "原流程排除了 status = 'test' 的行，盲算没有" + "text": "原流程排除了 status = 'test' 的行和 status 为空的行，盲算没有" } ], "fallback": false, @@ -702,7 +712,7 @@ }, { "aspect": "filter", - "text": "原流程排除了 status = 'refunded' 的行，盲算没有" + "text": "原流程排除了 status = 'refunded' 的行和 status 为空的行，盲算没有" } ], "fallback": false, @@ -789,6 +799,36 @@ } } }, + "fallback": { + "sql": [ + "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE paid_at >= '2026-09-01'\n AND paid_at < '2026-09-08'", + "SELECT SUM(amount) AS gmv\nFROM orders\nWHERE paid_at >= '2026-09-01'\n AND paid_at < '2026-09-09'" + ], + "diff": { + "equivalent": false, + "items": [ + { + "aspect": "other", + "text": "两条 SQL 的写法不同，未能归纳成具体差异，请对照原文" + } + ], + "fallback": true, + "text_diff": [ + "- orders.paid_at >= '2026-09-01' AND orders.paid_at < '2026-09-08'", + "+ orders.paid_at >= '2026-09-01' AND orders.paid_at < '2026-09-09'" + ] + }, + "catalog": { + "a": { + "write": false, + "reason": "指标「销售额」两边的 SQL 不止差一个方面，或有未能归纳的差异，无法确定审核人认可的是哪一处，不写入数据目录" + }, + "b": { + "write": false, + "reason": "指标「销售额」两边的 SQL 不止差一个方面，或有未能归纳的差异，无法确定审核人认可的是哪一处，不写入数据目录" + } + } + }, "parse": { "sql": [ "SELECT SUM(amount) AS gmv\nFROM orders", @@ -903,7 +943,7 @@ "items": [ { "aspect": "filter", - "text": "原流程排除了 orders.status = 'cancelled' 的行，盲算没有" + "text": "原流程排除了 orders.status = 'cancelled' 的行和 orders.status 为空的行，盲算没有" } ], "fallback": false, diff --git a/scripts/check-bundle.mjs b/scripts/check-bundle.mjs index 08c0bf22..ef9a589c 100644 --- a/scripts/check-bundle.mjs +++ b/scripts/check-bundle.mjs @@ -17,7 +17,9 @@ // 经继续运行跑到成功的早期运行、两条出具 // 判定、服务端缺文件、裸 NaN、别的运行的文档、口径卡有输入取自遮罩列），逐份断言各项结果和原因文案；在上面篡改 ④ // 引用链（换文档、拿掉报告、多一份没人引用的工件）、⑥ 运行信息（运行结果、类别、出具档位）、拿掉被引用的口径卡、把 -// 报告文档标成遮罩、只改页面顶部的封存哈希；宽屏点开表格里右对齐的数字不被面板盖住；不执行脚本时封存哈希、运行 ID 照样看得到，状态写「尚未核对」； +// 报告文档标成遮罩、把遮罩卡标成服务端的文件与 id 不一致或解析不出来（服务端对被改过、解析不了的遮罩卡就这么标）、 +// 只改页面顶部的封存哈希；宽屏点开表格里右对齐的数字不被面板盖住；不执行脚本时封存哈希、运行 ID 照样看得到，状态写 +// 「尚未核对」； // - 记录页的导出按钮（预览用例 ?syn=bundle-provenance-*，page.route 假装导出接口）：已封存时可点、下载的文件名同服务端； // 未封存时不可点并说明原因（左边写着「已封存」却不能点时原因写在屏幕上）；服务端回 409 时提示原因；失败结束的运行 // 提示之后继续运行会重新封存。 @@ -675,6 +677,39 @@ await section('样本页：篡改', async () => { doc.masked = true return '报告文档去掉内容、标遮罩' }, (s) => [...turnsRed('artifacts')(s), ['原因写明', hasItem(s, 'artifacts', 'fail', '只有查询快照和口径卡会因遮罩不随文件附带')]]) + // 遮罩卡在服务端被改过时，服务端照样只给 id、另标 mismatch（tests/test_bundle_masked_cards.py 守着服务端这一边）： + // 不能因为不嵌内容就和正常的遮罩卡一样「可核对的项全部通过」 + await tamper(samples['masked-card'], '遮罩卡标成服务端的文件与 id 不一致', (data) => { + byKind(data, 'metric_set').mismatch = true + return '口径卡加 mismatch' + }, (s) => { + const card = s.items.find((i) => i.check === 'artifacts' && i.text.includes('口径卡')) + const metricNumber = s.items.find((i) => i.check === 'numbers' && i.text.includes('销售额')) + return [ + ...turnsRed('artifacts', ['events', 'manifest', 'chain'])(s), + ['口径卡写明导出时服务端的文件与 id 不一致、可能被改过、未随文件附带', card?.result === 'fail' + && card.text.includes('导出时服务端的这份文件与工件 id（哈希）不一致，可能被改过') && card.text.includes('未随文件附带'), + card?.text], + ['引用这张卡的数字无法核对，写明引用的工件没有通过核对', metricNumber?.result === 'na' + && metricNumber.text.includes('引用的工件没有通过核对'), metricNumber?.text], + ] + }) + // 服务端标了解析不出来、哈希又对得上的遮罩卡：照实说认不出是否含遮罩列，是正常的无法核对（同哈希一致却解析不了的 + // 嵌入工件），不说成「含遮罩列」 + await tamper(samples['masked-card'], '遮罩卡标成服务端解析不出来', (data) => { + byKind(data, 'metric_set').unparsable = true + return '口径卡加 unparsable' + }, (s) => { + const card = s.items.find((i) => i.check === 'artifacts' && i.text.includes('口径卡')) + const metricNumber = s.items.find((i) => i.check === 'numbers' && i.text.includes('销售额')) + return [ + ['口径卡无法核对，写明服务端的文件无法解析、认不出是否含遮罩列', card?.result === 'na' + && card.text.includes('导出时服务端的这份文件无法解析，认不出是否含遮罩列'), card?.text], + ['引用它的数字原因同上', metricNumber?.result === 'na' && metricNumber.text.includes('认不出是否含遮罩列'), + metricNumber?.text], + ['总览照常通过', s.overall === 'pass', `${s.overall}：${s.text}`], + ] + }) await tamper(base, '报告文档的一个字节', (data) => flipByte(byKind(data, 'report_doc')), (s) => [ ...turnsRed('artifacts')(s), ['数字不予核对', s.checks.numbers === 'na' && hasItem(s, 'numbers', 'na', '所在的报告文档没有通过核对')], diff --git a/scripts/check-verify.mjs b/scripts/check-verify.mjs index 7cb95b43..9a5b139f 100644 --- a/scripts/check-verify.mjs +++ b/scripts/check-verify.mjs @@ -382,12 +382,14 @@ await section('审批卡：没有预告时，「记入数据目录」只说必 // 不写。卡上不能说得比服务端宽——每句只说必要条件（「仅当」开头），这几种不写的照实说，免得审核人为了不让目录被写错 // 而避开本来对的一边。变体的 catalog 是服务端对这一对 SQL 的真实预告，按它核对 const { CATALOG_TABLE_FIELD_LABEL: LABEL } = await import(await load('/src/lib/terms.ts')) - // [变体, 说明, 服务端不写的那一种卡上照实说的那句（只在卡上提数据目录时有）] + // [变体, 说明, 服务端不写的那一种卡上照实说的那句（只在卡上提数据目录时有）]。变体挑的是收尾版差异说明下各有代表性 + // 的几种（backend/tests/test_dual_check_fixture.py 的模块说明）：同一列上的时间范围不同只给兜底（fallback） const cases = [ - ['time_range', '只差时间范围（不是换了时间列）'], + ['fallback', '说不清的（同一列上的时间范围不同，只给兜底）'], ['two_filters', '差两个筛选条件'], ['join', '差在关联方式'], ['parse', '一边解析失败'], + ['time_column', '换了时间列（两列都声明为日期类型）', '只有一边按时间筛时不记'], ['time_one_side', '只有一边按时间筛', '只有一边按时间筛时不记'], ['filter_extra', '只多一个排除测试单的条件', '只看某个地区、某个渠道这类限定指标范围的条件不记'], ['filter_scope', '只多一个渠道条件', '只看某个地区、某个渠道这类限定指标范围的条件不记'], @@ -412,11 +414,22 @@ await section('审批卡：没有预告时，「记入数据目录」只说必 // 「不记」后面紧跟一句「记为人工确认的事实」，读起来像在说不记的那一项 check('……「人工确认」说的是写入的项', note.includes('写入的项记为人工确认的事实') && !note.includes('不记。记为'), note) } - if (name === 'time_one_side' || name === 'dedup') { + if (name === 'time_column' || name === 'time_one_side' || name === 'dedup') { // 同一次裁决里几个指标给同一张表记的业务日期、去重规则不同时，服务端一项都不写（有效记录条件是合成一句写）。 // 没有预告时卡上看不出别的指标落到哪一项上，照实说这一条 check('……同一次裁决里几个指标给同一张表记的说法不同时都不记', note.includes('同一次裁决里') && note.includes('不同时都不记'), note) } + if (name === 'time_column') { + // 服务端两边都会写（各记被采纳那一边的列）：卡上的必要条件说的就是按不同的列筛时间 + check('……两边按不同的列筛时间，采用的一边所用的时间列才记为业务日期', writes.length === 2 + && note.includes('按不同的列筛时间') && note.includes('采用的一边所用的时间列'), note) + } + if (name === 'fallback') { + // 收尾版差异说明归纳不了同一列上的时间范围：卡上写那句固定的兜底说明，默认展开对照，审核人逐行对照两条 SQL + check('……写那句固定的兜底说明，不当成一条具体差异', (await text(gmv.locator('[data-dual-fallback]'))) === '两条 SQL 的写法不同，未能归纳成具体差异' + && await gmv.locator('[data-dual-diff-item]').count() === 0, await text(gmv.locator('[data-dual-fallback]'))) + check('……默认展开对照', (await gmv.locator('[data-dual-compare-toggle]').getAttribute('aria-expanded')) === 'true') + } if (name === 'time_one_side') { // 正上方那条差异说「盲算没有按时间筛选」：卡上不能再说两边只差在按哪一列筛时间 check('……不和正上方那条差异自相矛盾', (await text(gmv.locator('[data-dual-diff-item]'))).includes('没有按时间筛选') @@ -436,7 +449,9 @@ await section('审批卡：没有预告时，「记入数据目录」只说必 } const diff = fx.variants[name].diff if (diff.text_diff.length) { - await gmv.locator('[data-dual-compare-toggle]').click() + // 只有兜底说明时对照默认就展开着：再点一下就收起来了 + const toggle = gmv.locator('[data-dual-compare-toggle]') + if ((await toggle.getAttribute('aria-expanded')) !== 'true') await toggle.click() const m = await marksIn(gmv) const real = sidesOf(diff) check(`……规范化后的行按真实输出分两栏高亮（${real.del.length} + ${real.add.length} 行）`, @@ -449,7 +464,7 @@ await section('审批卡：没有预告时，「记入数据目录」只说必 } // 几个输入都差在时间列：节点把两处的差异合成一份（每条前面写明是哪个输入），不止一条，不写目录 - const t = fx.variants.time_range.diff.items[0] + const t = fx.variants.time_column.diff.items[0] const merged = { equivalent: false, fallback: false, text_diff: fx.card.metrics[0].diff.text_diff, items: [{ aspect: 'time', text: `输入「gmv_web」：${fx.card.metrics[0].diff.items[0].text}` }, { aspect: 'time', text: `输入「gmv_app」：${t.text}` }], @@ -520,23 +535,39 @@ await section('审批卡：服务端预告了写目录时照它说，当前选 await shot(base.page, 'approval-preview-dark') // 几个变体：预告说两边都不写的，卡上就不提（没有预告时这几种会有条件地提）；写的那一边说清写到哪、写成什么 - for (const [name, what] of [['time_one_side', '只有一边按时间筛'], ['filter_scope', '只多一个渠道条件'], ['filter_extra', '只多一个排除测试单的条件'], - ['dedup', '只差去不去重'], ['time_range', '只差时间范围']]) { + for (const [name, what] of [['time_column', '换了时间列'], ['time_one_side', '只有一边按时间筛'], ['filter_scope', '只多一个渠道条件'], + ['filter_extra', '只多一个排除测试单的条件'], ['dedup', '只差去不去重'], ['fallback', '说不清的（只给兜底）']]) { const { page, errors } = await preview('dual-check-approval', { query: `variant=${name}&catalog=1`, artifact: cardWith(name) }) await page.waitForSelector('[data-dual-check][data-state="ready"]') const m = metric(page, 'gmv') const expect = fx.variants[name].catalog const writes = ['a', 'b'].filter((s) => expect[s].write) - check(`${what}（变体）：摘要换成这一对 SQL 的差异`, (await text(m.locator('[data-dual-diff-item]'))) === fx.variants[name].diff.items[0].text, - await text(m.locator('[data-dual-diff-item]'))) + if (fx.variants[name].diff.items.some((i) => i.aspect !== 'other' && i.aspect !== 'parse')) { + check(`${what}（变体）：摘要换成这一对 SQL 的差异`, (await text(m.locator('[data-dual-diff-item]'))) === fx.variants[name].diff.items[0].text, + await text(m.locator('[data-dual-diff-item]'))) + } else { + check(`${what}（变体）：摘要换成兜底说明，不当成一条具体差异`, (await text(m.locator('[data-dual-fallback]'))) === '两条 SQL 的写法不同，未能归纳成具体差异' + && await m.locator('[data-dual-diff-item]').count() === 0, await text(m.locator('[data-dual-fallback]'))) + } if (!writes.length) { check(`${what}：服务端两边都不写，卡上不提数据目录`, await m.locator('[data-dual-catalog]').count() === 0, await text(m.locator('[data-dual-catalog]'))) } else { const got = await previewClauses(m) + // 两边都写、写的是同一张表的同一项（换了时间列：同一张表的业务日期）时，后一条省掉哪张表的哪一项 + const same = writes.length === 2 && ['source', 'table', 'item'].every((k) => expect.a[k] === expect.b[k]) const ok = ['a', 'b'].every((s) => (expect[s].write - ? got[s]?.write && got[s].text.includes(LABEL[expect[s].item]) && got[s].text.includes(`会记为「${previewValue(expect[s].value)}」`) + ? got[s]?.write && got[s].text.includes(`会记为「${previewValue(expect[s].value)}」`) + && (s === 'b' && same ? got[s].text === `${CHOICE_LABEL.b}时会记为「${previewValue(expect.b.value)}」` : got[s].text.includes(LABEL[expect[s].item])) : !got[s]?.write && got[s]?.reason && got[s].text === blockedText(CHOICE_LABEL[s], expect[s].reason))) check(`${what}：写的一边说清写到哪、会记为什么，不写的一边照服务端的原话说不会记入的原因`, ok, `${got.a?.text} ‖ ${got.b?.text}`) + if (name === 'time_column') { + // 两列都声明为日期类型：选哪一边都记被采纳那一边的列，加重的那一条跟着选择走 + await m.locator('label', { hasText: '采用盲算的结果' }).click() + const cur = await previewClauses(m) + check('换了时间列，选盲算：加重的那一条说业务日期会记为盲算那一列', cur.b?.current && cur.b.write && !cur.a.current + && cur.b.text === `采用盲算的结果时会记为「${previewValue(expect.b.value)}」` + && cur.a.text === `采用原流程的结果时，「orders」（shop）的业务日期会记为「${previewValue(expect.a.value)}」`, `${cur.a?.text} ‖ ${cur.b?.text}`) + } if (name === 'dedup') { // 采用去重的原流程记去重规则，采用不去重的盲算不记：两种选择分别看加重的那一句 await m.locator('label', { hasText: '采用原流程的结果' }).click()