双盲复算复核修正(第 1 轮):业务日期只认带日期的类型、原因按哪一边说清;遮罩列在子查询或 WITH 里改了名不再收窄,交给盲算的分组取值按 SQL 认出取自遮罩列的;按位置取的格点名原因、认不出出处的不写进计算式、不再改变容器输入的判法

业务日期的类型门槛:只有时刻的 TIME / TIMETZ 不再算(一天里的几点几分不是日期,MySQL 还常拿它存时长;和 catalog.is_date_column 同一个口径,记成业务日期后 SQL 检查会对按真实日期列统计的查询一直报错误级),原因说「只有时刻,不含日期」。类型照表结构里记的说「类型是 X」(introspect 存的是 SQLAlchemy 反射出来的类型,SQLite 里写成 TIMESTAMPTZ 的反射成 NUMERIC),没写类型的列反射成「NULL」,按没有声明类型说。被采纳的一边有问题说「被采纳的盲算按时间筛的列……」,未被采纳的一边另说清为什么也要看它(两边都是日期列时,审核人选的才是业务日期按哪一列)。数据源还没有可用的表结构(未探查、探查失败)时说表结构不可用,不说这一列不在表结构里。上传表格建的数据源(origin=upload,节点认出后经候选的 uploaded 交给 derive)日期一律存成文字,原因说「来自上传的表格,表格里的日期一律按文字保存」,不说成用户声明了 TEXT;预告和真写逐字一致。

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