yilunjiang/AgentLab
1
0
Fork 0
mirror of https://github.com/ThearchyHelios/AgentLab.git synced 2026-11-15 00:51:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

合入期 4 整合 J1：双盲复算写业务日期加类型门槛（两边的时间列都在表结构里声明为日期 / 时间类型才写，原因说清是哪一列）；跟上 SQL 差异说明收尾的测试与预览夹具（time_range 改为兜底变体，新增 time_column）；按位置取自另一份快照遮罩列的输入判为无法复算；可验证报告里被改过、解析不出的口径卡按内容认遮罩

Browse Source
This commit is contained in:
ThearchyHelios
2026-10-06 16:23:26 +08:00
parent d7f6b4d756 b080302e77
commit 2da4d34fec
13 changed files with 957 additions and 118 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/bundle/build.py
+20 -5
View File
@@ -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")
backend/app/bundle/verifier.js
+13 -6
View File
@@ -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') {
backend/app/engine/dual_check_facts.py
+70 -7
View File
@@ -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]:
backend/app/engine/nodes/dual_check.py
+140 -36
View File
@@ -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
backend/tests/fixtures/dual_check_shop.py
Vendored
+6 -2
View File
@@ -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()
backend/tests/test_bundle_masked_cards.py
+75 -3
View File
@@ -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)
backend/tests/test_dual_check_facts.py
+175 -12
View File
@@ -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)
# --------------------------------------------------------------------------
backend/tests/test_dual_check_fixture.py
+36 -6
View File
@@ -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()
backend/tests/test_dual_check_node.py
+289 -14
View File
@@ -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、模型的往来按
设计带着原始结果（证据面板显示时再遮），不在此列。"""
backend/tests/test_positional_masks_elsewhere.py
+2 -2
View File
@@ -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"])
frontend/src/dev/fixtures/dual_check.json
+54 -14
View File
@@ -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,
scripts/check-bundle.mjs
+36 -1
View File
@@ -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', '所在的报告文档没有通过核对')],
scripts/check-verify.mjs
+41 -10
View File
@@ -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()