From 2cf329648dd3e75f7fa4ff094e2853797c98f4ce Mon Sep 17 00:00:00 2001 From: ThearchyHelios Date: Tue, 6 Oct 2026 16:54:30 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0=E8=A1=A8=E6=A0=BC=E7=9A=84?= =?UTF-8?q?=E6=97=A5=E6=9C=9F=E5=88=97=E6=8C=89=E5=AF=BC=E5=85=A5=E6=97=B6?= =?UTF-8?q?=E9=85=8D=E6=96=B9=E9=87=8C=E7=9A=84=E9=80=BB=E8=BE=91=E7=B1=BB?= =?UTF-8?q?=E5=9E=8B=E8=AE=A4=EF=BC=9A=E5=8F=8C=E7=9B=B2=E5=A4=8D=E7=AE=97?= =?UTF-8?q?=E9=80=89=E4=BA=86=E6=97=B6=E9=97=B4=E5=88=97=E4=B8=8D=E5=90=8C?= =?UTF-8?q?=E7=9A=84=E4=B8=80=E8=BE=B9=E6=97=B6=EF=BC=8C=E4=B8=8A=E4=BC=A0?= =?UTF-8?q?=E6=BA=90=E4=B9=9F=E8=83=BD=E5=86=99=E4=B8=9A=E5=8A=A1=E6=97=A5?= =?UTF-8?q?=E6=9C=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit J1 给业务日期加了类型门槛（两边的时间列都在表结构里声明为日期 / 时间类型才写），而上传表格导入时日期一律按文字保存、表结构里总是 TEXT，上传源因此永远写不了业务日期。导入时配方里其实记着每一列的逻辑类型（导入确认时界面上标的「日期」）：列表块里类型为 DATE 的列、交叉表的日期轴。查过别处：构建回执里只有存储类型（列表的 DATE 列和文字列一样记 TEXT、role 都是 text），简单导入的回执只有 INTEGER / REAL / TEXT，所以只按配方认。 - dual_check_facts.recipe_dates：一份配方里逻辑类型是日期的列（{表: [列]}），表和列按 recipe_types.derive_tables 推，和建表同一份；日期轴写进它各分段的表，列表合计行另存的表不带日期列；类型按 _DATED_TYPE 认，只有时刻的不算 - dual_check_facts.upload_dates：只读，按数据源当前快照找配方（source_versions.snapshot_recipe_id，期 3 之前的快照取最后一期导入的），和 schema_cache 出自同一个版本；手工源、简单导入、快照或配方记录不在、配方解析不了一律给空 - 节点的 _source_info 带上 upload_dates，放进候选：审批卡的预告（preview / preview_joint）和裁决后的真写认的是发起审批时读的同一份，逐字一致；旧的 checkpoint 里没有这一项，照旧不写 - _dated_problem：上传源里按文字保存的列在 upload_dates 里的算日期列（按表结构认出的表名、列名去对，大小写不论）；不在的、读不出配方的照现在的原因说「来自上传的表格，表格里的日期一律按文字保存」；声明成别的类型的、只有时刻的照旧 测试：derive 一级补导入认作日期的写、只有一边是日期 / 读不出配方 / 记在别的表上的照旧说按文字保存、手工源不认 upload_dates、按表结构的名字对、preview_joint 同样认；新文件 test_dual_check_upload 经真实导入接口建上传源（暂存 → 改配方 → 试运行 → 提交、简单导入），配方里两列都是日期的审核人选盲算后业务日期记「支付日期」（人工确认、备注写运行），支付日期是文字的选哪边都不写、预告和跳过原因一字不差，配方记录改坏了的照旧不写；去掉认日期、去掉候选里的 upload_dates、去掉日期轴各做过一次，测试变红。 --- backend/app/engine/dual_check_facts.py | 115 ++++++++-- backend/app/engine/nodes/dual_check.py | 14 +- backend/tests/test_dual_check_facts.py | 97 +++++++- backend/tests/test_dual_check_upload.py | 287 ++++++++++++++++++++++++ 4 files changed, 489 insertions(+), 24 deletions(-) create mode 100644 backend/tests/test_dual_check_upload.py diff --git a/backend/app/engine/dual_check_facts.py b/backend/app/engine/dual_check_facts.py index ba227b3d..09a11808 100644 --- a/backend/app/engine/dual_check_facts.py +++ b/backend/app/engine/dual_check_facts.py @@ -9,8 +9,9 @@ 文本列（SQLite 常把日期存成文字）、整数列（Unix 秒、20260901 这类日期键）、名字带 time 的时长列都可能被说成「按它筛 时间」，认不准存的是不是日期，不记。只有时刻的 TIME 也不记：一天里的几点几分（MySQL 还常拿它存时长）不是日期，和 catalog.is_date_column（剖析和 SQL 检查共用的日期列判断）同一个口径——记成业务日期，SQL 检查会对按真实日期列统计的 - 查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，这一条门槛下它们的业务日期都不由这里记（原因照实说 - 是上传的表格按文字保存，不说成用户声明了 TEXT） + 查询一直报错误级。上传表格建的数据源，导入时日期一律按文字保存，表结构里总是 TEXT：这样的列看导入时配方里记的逻辑 + 类型（upload_dates，导入确认时界面上标的「日期」就是它），是日期的算日期列；配方读不出来（简单导入没有配方、配方 + 记录坏了）或者配方里不是日期的，照旧不记，原因照实说是上传的表格按文字保存，不说成用户声明了 TEXT - 一边多一个排除无效记录的条件 → 有效记录条件（valid_filter）：被采纳的一边有这个条件才记，记的就是这个条件 - 聚合只差去不去重、被采纳的一边去重 → 去重规则（dedup）。采纳了不去重的一边不记：那是这个指标自己的算法（数行）， 写成表的「不去重」会把别的指标带偏 @@ -144,11 +145,13 @@ 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}, uploaded}。scope 是这个指标自己的 - 范围：texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效记录条件用它 - 排除指标自己的筛选范围。uploaded 为真是上传表格建的数据源：业务日期不记时，文本列的原因照实说是上传的表格按文字 - 保存日期。schema_cache 是数据源的表结构：认表，业务日期还要看两边的列声明的类型。notes 是这个数据源的目录 - （{表名: notes}），认码值的意思用。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 + {equivalent, fallback, items: [FacetDiff.as_dict()]}, scope: {texts, columns}, uploaded, upload_dates}。scope 是 + 这个指标自己的范围：texts 是口径说明、所在分组的取值、本次运行的输入（指标名称总算在内），columns 是分组的列；有效 + 记录条件用它排除指标自己的筛选范围。uploaded 为真是上传表格建的数据源，upload_dates 是它导入时认作日期的列 + （{表名: [列名]}，由 upload_dates 从配方读出）：按文字保存的列在这里面的算日期列，不在的业务日期不记、原因照实说 + 是上传的表格按文字保存日期。两项都由节点发起审批时备好、放进候选：预告和审批之后的真写认的是同一份。schema_cache + 是数据源的表结构：认表，业务日期还要看两边的列声明的类型。notes 是这个数据源的目录（{表名: notes}），认码值的意思 + 用。只看差异说明，不重新比 SQL：审核人看到的就是这份说明。 """ name = str(candidate.get("metric") or "") side = candidate.get("choice") @@ -164,7 +167,9 @@ 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, uploaded=candidate.get("uploaded") is True) + dates = candidate.get("upload_dates") + fact, why = _business_date(item, side, schema_cache, uploaded=candidate.get("uploaded") is True, + dates=dates if isinstance(dates, Mapping) else None) elif aspect == "filter": fact, why = _valid_filter(item, side, candidate, schema_cache, notes or {}) elif aspect == "aggregate": @@ -187,11 +192,12 @@ def derive(candidate: Mapping[str, Any], schema_cache: Mapping[str, Any] | 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]: + uploaded: bool = False, dates: Mapping[str, Any] | None = None + ) -> tuple[tuple[str, str, Any] | None, str | None]: """两边按不同的列筛时间、被采纳的一边按哪一列：(表, business_date, {column}) 或 (None, 为什么不记)。两边的列都要在 - 表结构里声明为带日期的类型（_dated_problem），被采纳的一边先看：它有问题说它；没问题再看未被采纳的一边，另说清为什么 - 也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源还没有可用的表结构时（没探查、探查 - 失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。""" + 表结构里声明为带日期的类型、或者是上传表格导入时认作日期的列（_dated_problem），被采纳的一边先看：它有问题说它；没 + 问题再看未被采纳的一边，另说清为什么也要看它（审核人看到的是自己选的那一列，不说就看不懂另一列为什么要紧）。数据源 + 还没有可用的表结构时（没探查、探查失败）只说表结构不可用：那时并不知道有没有这一列，说「不在表结构里」是在误导。""" other = "b" if side == "a" else "a" detail = item.get("detail") if isinstance(item.get("detail"), Mapping) else {} @@ -212,7 +218,7 @@ def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str 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: + if (problem := _dated_problem(tables, ref, uploaded, dates)) is None: continue said = f"按时间筛的列「{ref['table']}.{ref['column']}」{problem}" if adopted: @@ -221,15 +227,20 @@ def _business_date(item: Mapping[str, Any], side: str, schema_cache: Mapping[str 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: +def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded: bool, + dates: Mapping[str, Any] | None = None) -> str | None: """按时间筛的这一列在表结构里是不是声明成了带日期的类型（_DATED_TYPE）：是的为 None，不是的交回它怎么了（查不到、没 声明类型、只有时刻、声明成了别的类型）连同结论，接在「按时间筛的列「表.列」」后面读。 表按 catalog.resolve_table_name 认，列先按原样、再按 name_key 认（差异说明里的名字是规范化过的，SQLite 不加引号的 名字一律小写）。按 name_key 认出好几列的说对应好几列：哪一列是它说不准。类型照表结构里记的说（「类型是」，不说 「声明为」）：introspect 记的是 SQLAlchemy 反射出来的类型，SQLite 里写成 TIMESTAMPTZ 的列反射出来是 NUMERIC；没写 - 类型的列反射出来是 NullType，记成「NULL」，按没有声明类型说。上传表格建的数据源（uploaded）日期一律存成文字，文字 - 类型不说「类型是 TEXT」：那不是用户声明的，导入确认时这一列标的还是「日期」，表结构也改不了。""" + 类型的列反射出来是 NullType，记成「NULL」，按没有声明类型说。 + + 上传表格建的数据源（uploaded）日期一律存成文字，表结构改不了，类型只能看导入时配方里记的（dates，见 upload_dates）： + 按文字保存的列在里面的就是日期列（按表结构里认出的表名、列名去对，大小写不论）。不在里面的（配方读不出来、配方里 + 不是日期）不说「类型是 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")] \ @@ -249,10 +260,79 @@ def _dated_problem(tables: Mapping[str, Any], ref: Mapping[str, Any], uploaded: if _TIME_ONLY_TYPE.fullmatch(declared): return f"在表结构里的类型是 {declared}，只有时刻，不含日期" if uploaded and any(word in declared.upper() for word in catalog._TEXT_TYPES): + if _imported_as_date(dates, str(found_table), str(found[0]["name"])): + return None return "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期" return f"在表结构里的类型是 {declared}，无法确认存的是日期" +def _imported_as_date(dates: Mapping[str, Any] | None, table: str, column: str) -> bool: + """表结构里的这一列是不是上传表格导入时认作日期的列（dates：{表名: [列名]}）。表、列都先按原样、再按 name_key 对： + 名字出自同一份配方，本来一字不差；按 name_key 对上好几个的不算，说不准是哪一个。""" + if not isinstance(dates, Mapping): + return False + + def one(names: Any, wanted: str) -> str | None: + names = [str(n) for n in names if isinstance(n, str)] + if wanted in names: + return wanted + loose = [n for n in names if name_key(n) == name_key(wanted)] + return loose[0] if len(loose) == 1 else None + + key = one(list(dates), table) + columns = dates.get(key) if key is not None else None + return isinstance(columns, (list, tuple)) and one(columns, column) is not None + + +# ---- 上传表格导入时认作日期的列：配方里记的逻辑类型 + +def recipe_dates(recipe: Any) -> dict[str, list[str]]: + """一份配方（recipe_types.Recipe）里逻辑类型是日期的列：{表名: [列名]}，没有日期列的表不列。 + + 就是导入确认时界面上标「日期」的那些（recipe_notes 给列写日期格式说明认的也是这两种）：交叉表的日期轴（Axis.type 恒为 + 日期），列表块里类型记为 DATE 的列。执行器把它们一律存成 YYYY-MM-DD 的文字，表结构里只剩 TEXT。哪张表有哪几列按 + recipe_types.derive_tables 推（建表用的同一份）：日期轴写进它各分段的表，列表合计行另存的表只有合计项和数字列。 + 列表的类型按 _DATED_TYPE 认，配方以后添了日期时间也算；只有时刻的不算。""" + from app.data.recipe_types import ListBlock, derive_tables + + tables, _ = derive_tables(recipe) + listed = {(block.table, column.name) for sheet in recipe.sheets for block in sheet.blocks + if isinstance(block, ListBlock) for column in block.columns if _DATED_TYPE.fullmatch(column.type)} + out: dict[str, list[str]] = {} + for table, columns in tables.items(): + names = [c.name for c in columns if c.role == "axis" or (table, c.name) in listed] + if names: + out[table] = list(dict.fromkeys(names)) + return out + + +async def upload_dates(session: Any, row: DataSource | None) -> dict[str, list[str]]: + """上传表格建的数据源当前版本里，导入时认作日期的列：{表名: [列名]}（recipe_dates）。只读，不碰导入流程。 + + 按当前快照找它的配方（source_versions.snapshot_recipe_id：快照记着按哪一版配方建的表，期 3 之前的快照取最后一期导入 + 的）：数据源的表结构（schema_cache）也出自当前快照，两者是同一个版本。读不出来的一律给空——手工登记的源、简单导入 + （没有配方，回执里只有存储类型）、快照或配方记录不在了、配方解析不了：业务日期照旧不记，原因说按文字保存。""" + from app.data.recipe_types import Recipe + from app.data.source_versions import snapshot_recipe_id + from app.db.models import SourceSnapshot, TableImport, TableRecipe + + if row is None or (row.origin or "manual") != "upload" or not row.current_snapshot_id: + return {} + snap = await session.get(SourceSnapshot, row.current_snapshot_id) + if snap is None or snap.source_id != row.id: + return {} + last = await session.get(TableImport, str(snap.imports[-1])) if snap.imports else None + recipe_id = snapshot_recipe_id(snap, {last.id: last} if last is not None else {}) + record = await session.get(TableRecipe, recipe_id) if recipe_id else None + if record is None or record.source_id != row.id: + return {} + try: + return recipe_dates(Recipe.model_validate(record.recipe)) + except ValueError: + logger.warning("上传表格的配方记录解析不了，业务日期按文字保存处理：source=%s recipe=%s", row.id, recipe_id) + return {} + + 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]: @@ -980,4 +1060,5 @@ async def _write(session: Any, row: DataSource, fact: Fact, *, run_id: str, node return None, [*reasons, f"「{fact.table}」的数据目录正被修改，{label}未能写入，请稍后在数据目录中确认"] -__all__ = ["FACT_ITEMS", "JOINT_LIMIT", "Fact", "derive", "note_of", "preview", "preview_joint", "record_choices"] +__all__ = ["FACT_ITEMS", "JOINT_LIMIT", "Fact", "derive", "note_of", "preview", "preview_joint", "recipe_dates", + "record_choices", "upload_dates"] diff --git a/backend/app/engine/nodes/dual_check.py b/backend/app/engine/nodes/dual_check.py index 04607ec8..fc5a0492 100644 --- a/backend/app/engine/nodes/dual_check.py +++ b/backend/app/engine/nodes/dual_check.py @@ -69,7 +69,7 @@ from app.db.models import DataSource from app.engine import deadline, recompute from app.engine.approval import read_decision from app.engine.context import NodeContext, NodeError -from app.engine.dual_check_facts import preview, preview_joint, record_choices +from app.engine.dual_check_facts import preview, preview_joint, record_choices, upload_dates from app.engine.evidence import build_catalog, is_number, next_exec from app.engine.expressions import ExpressionError, _cell_call, _chain_root, cell_value, column_kind, numeric_text, \ parse_expression @@ -1126,6 +1126,7 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], facts[row["metric_id"]] = { "metric_id": row["metric_id"], "metric": row["name"], "source": first.get("source"), "dialect": info.get("kind"), "uploaded": bool(info.get("uploaded")), + "upload_dates": info.get("upload_dates") or {}, "sql": {"a": single[0], "b": single[1]} if single else {}, "diff": {"equivalent": merged["equivalent"], "fallback": merged["fallback"] or single is None, "items": merged["items"]}, @@ -1165,19 +1166,22 @@ async def _review(state: GraphState, ctx: NodeContext, checked: dict[str, Any], async def _source_info(sources: _Sources, name: str) -> dict[str, Any]: """比较两条 SQL、预告目录写入要用的：方言、表结构、列的中文名、整份数据目录，以及是不是上传表格建的数据源 - （uploaded：导入时日期一律存成文字，业务日期不记时原因照实说，见 dual_check_facts._dated_problem）。数据源没了就 - 都不给，照样比原文。""" + （uploaded：导入时日期一律存成文字）和它导入时认作日期的列（upload_dates，按当前版本的配方读，见 + dual_check_facts.upload_dates）——业务日期据此认上传表格的日期列，认不出的原因照实说（_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": {}, "uploaded": False} + return {"kind": None, "schema": None, "names": {}, "notes": {}, "uploaded": False, "upload_dates": {}} async with SessionLocal() as session: entries = await catalog.read_catalog(session, row.id) + dates = await upload_dates(session, row) 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, "uploaded": (row.origin or "manual") == "upload"} + "names": catalog_labels(notes), "notes": notes, "uploaded": (row.origin or "manual") == "upload", + "upload_dates": dates} async def _explain(original: Any, blind: Any, info: dict[str, Any]) -> dict[str, Any]: diff --git a/backend/tests/test_dual_check_facts.py b/backend/tests/test_dual_check_facts.py index 54cc8bd1..b13851fa 100644 --- a/backend/tests/test_dual_check_facts.py +++ b/backend/tests/test_dual_check_facts.py @@ -4,8 +4,9 @@ 条件，聚合只差去不去重、被采纳的一边去重 → 去重规则；按被采纳那一边的写法记 - 业务日期只在两边的列都在表结构里声明为带日期的类型（DATE、DATETIME、TIMESTAMP，带精度、带时区的写法也算）时才写： 文本、整数（Unix 秒、20260901 这类日期键）、只有时刻的 TIME、表结构里查不到、没声明类型（SQLite 探出来是 NULL）的 - 不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源说清日期是按文字保存的； - 数据源还没有可用的表结构时说表结构不可用 + 不写，原因说清是哪一列、哪一边（未被采纳的那一边另说清为什么也要看它）；上传表格建的数据源，按文字保存的列看导入时 + 配方里认作日期的列（候选的 upload_dates）：是的算日期列，不是的、读不出配方的说清日期是按文字保存的；数据源还没有 + 可用的表结构时说表结构不可用 - 差了不止一处、有兜底说明、两条 SQL 规范化后一致、差在别的方面、被采纳的一边没有可记的写法、只有一边按时间筛：不写， 照实说原因。同一列上的时间范围不同、跨两张表的条件，差异说明只给兜底，同样不写；说明以后真把跨两张表的条件说成一条 筛选条件，derive 自己那道「只涉及一张表」的检查也拦得下 @@ -367,6 +368,98 @@ def test_an_uploaded_table_says_its_dates_are_stored_as_text(): candidate(BY_ORDER_DATE, RIGHT, "b"), texts)[1] +#: 上传表格的列按文字保存、配方里又没认作日期时的原因（同上一个用例） +STORED_AS_TEXT = "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期" + + +def uploaded(original: str, blind: str, choice: str, dates: dict | None) -> dict: + """上传表格建的数据源的候选：upload_dates 是节点从当前版本的配方读出的、导入时认作日期的列（{表: [列]}）。""" + return {**candidate(original, blind, choice), "uploaded": True, + **({"upload_dates": dates} if dates is not None else {})} + + +def test_an_uploaded_column_the_import_took_as_a_date_is_a_date(): + """上传表格建的数据源，表结构里的日期列总是 TEXT（导入时一律按文字保存），可导入时配方里这两列的类型是「日期」（候选的 + upload_dates）：照常记成业务日期，采纳哪一边记哪一列，预告是同一套判断。""" + texts = schema_with(order_date="TEXT", paid_date="TEXT") + both = {"orders": ["order_date", "paid_date"]} + for choice, column in (("a", "order_date"), ("b", "paid_date")): + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, choice, both), texts) + assert why is None, why + assert (fact.table, fact.item, fact.value, fact.choice) == ("orders", "business_date", {"column": column}, choice) + said = preview(uploaded(BY_ORDER_DATE, RIGHT, choice, both), texts, {}) + assert said[choice] == {"write": True, "source": "shop", "table": "orders", "item": "business_date", + "value": {"column": column}} + + +@pytest.mark.parametrize("dates, column, owner", [ + # 配方读不出来（简单导入没有配方、配方记录坏了）：两列都照旧说按文字保存 + (None, "orders.paid_date", "b"), + ({}, "orders.paid_date", "b"), + # 配方里只有一列是日期，另一列是文字：没认作日期的那一列照旧说 + ({"orders": ["paid_date"]}, "orders.order_date", "a"), + ({"orders": ["order_date"]}, "orders.paid_date", "b"), + # 认作日期的是别的表的同名列：对不上这一列 + ({"customers": ["order_date", "paid_date"]}, "orders.paid_date", "b"), +]) +def test_an_uploaded_column_the_import_did_not_take_as_a_date_keeps_the_reason(dates, column, owner): + """配方里不是日期、或者读不出配方的列：照现在的原因不写（说按文字保存，不说「类型是 TEXT」），被采纳的一边先说； + 被采纳的一边是日期、未被采纳的不是的，说未被采纳的那一边、另说清为什么也要看它。""" + texts = schema_with(order_date="TEXT", paid_date="TEXT") + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", dates), texts) + assert fact is None + if owner == "b": + assert why == f"指标「销售额」被采纳的盲算按时间筛的列「{column}」{STORED_AS_TEXT}，不写入数据目录", why + else: + assert why == (f"指标「销售额」未被采纳的原流程按时间筛的列「{column}」{STORED_AS_TEXT}：{BOTH_DATES}，" + "不写入数据目录"), why + assert preview(uploaded(BY_ORDER_DATE, RIGHT, "b", dates), texts, {})["b"] == {"write": False, "reason": why} + + +def test_the_import_dates_only_speak_for_text_columns_of_an_uploaded_source(): + """upload_dates 只回答「按文字保存的这一列，导入时是不是日期」： + - 不是上传源（手工登记的库）的文本列照旧说类型，候选里带着 upload_dates 也不认 + - 上传源里声明成别的类型的列（整数）照旧说类型；只有时刻的仍不算日期""" + texts = schema_with(order_date="TEXT", paid_date="TEXT") + both = {"orders": ["order_date", "paid_date"]} + fact, why = derive({**candidate(BY_ORDER_DATE, RIGHT, "b"), "upload_dates": both}, texts) + assert fact is None and "「orders.paid_date」在表结构里的类型是 TEXT，无法确认存的是日期" in why + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", both), schema_with(order_date="INTEGER", paid_date="TEXT")) + assert fact is None and "未被采纳的原流程按时间筛的列「orders.order_date」在表结构里的类型是 INTEGER" in why + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", both), schema_with(order_date="TEXT", paid_date="TIME")) + assert fact is None and "「orders.paid_date」在表结构里的类型是 TIME，只有时刻，不含日期" in why + + +def test_the_import_dates_follow_the_names_in_the_table_structure(): + """差异说明里的名字是规范化过的（不加引号的一律小写），配方里、表结构里的写法可能带大写：按表结构认出的那一列去对 + upload_dates，大小写不同照样对得上。""" + upper = {"tables": {"Orders": {"name": "Orders", "columns": [ + {"name": "Status", "type": "TEXT"}, {"name": "Order_Date", "type": "TEXT"}, {"name": "Paid_Date", "type": "TEXT"}, + {"name": "Amount", "type": "REAL"}]}}} + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", {"Orders": ["Order_Date", "Paid_Date"]}), upper) + assert why is None and (fact.table, fact.value) == ("Orders", {"column": "paid_date"}) + fact, why = derive(uploaded(BY_ORDER_DATE, RIGHT, "b", {"orders": ["order_date", "paid_date"]}), upper) + assert why is None and (fact.table, fact.value) == ("Orders", {"column": "paid_date"}) + + +def test_the_joint_preview_reads_the_import_dates_too(): + """几个指标落到同一张上传表的业务日期上：preview_joint 和单个指标的预告、真写一样从候选认 upload_dates。""" + from app.engine.dual_check_facts import preview_joint + + texts = schema_with(order_date="TEXT", paid_date="TEXT") + both = {"orders": ["order_date", "paid_date"]} + picked = [{**uploaded(BY_ORDER_DATE, RIGHT, "a", both), "metric": "销售额", "metric_id": "gmv"}, + {**uploaded(BY_ORDER_DATE, RIGHT, "a", both), "metric": "订单金额", "metric_id": "total"}] + by = {tuple((m["metric_id"], m["choice"]) for m in e["members"]): e + for e in preview_joint(picked, {"shop": (texts, {})})} + assert by[(("gmv", "b"), ("total", "b"))]["outcomes"] == { + "gmv": {"write": True, "value": {"column": "paid_date"}}, + "total": {"write": True, "value": {"column": "paid_date"}}} + # 配方读不出来的同一组候选：一项都认不出，没有可预告的组合 + plain = [{**c, "upload_dates": {}} for c in picked] + assert preview_joint(plain, {"shop": (texts, {})}) == [] + + 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」。""" diff --git a/backend/tests/test_dual_check_upload.py b/backend/tests/test_dual_check_upload.py new file mode 100644 index 00000000..a538a646 --- /dev/null +++ b/backend/tests/test_dual_check_upload.py @@ -0,0 +1,287 @@ +"""上传表格建的数据源上的双盲复算：业务日期按导入时配方里记的逻辑类型认日期列（engine/dual_check_facts.upload_dates）。 + +上传表格导入时日期一律按文字保存，表结构里总是 TEXT，以前这类数据源的业务日期永远写不了。导入时配方里其实记着每一列 +的逻辑类型（导入确认时界面上标的「日期」）： + +- 配方里是日期的两列（列表块里类型为 DATE 的列、交叉表的日期轴）：审核人选了哪一边，业务日期就记哪一列；审批卡的预告 + 和真写一字不差 +- 配方里是文字的列不写；配方读不出来（简单导入没有配方、配方记录坏了）的照旧不写，原因都照实说是上传的表格按文字保存 +- 只认数据源当前版本的配方：表结构出自同一个版本 + +数据源一律经真实的导入接口建（暂存 → 改配方 → 试运行 → 提交，或者简单导入），工作簿是现造的：表名、地区、金额都是 +虚构的。盲算模型是 MockChatModel 剧本（tests/fixtures/dual_check_shop.Blind）。 +""" +from __future__ import annotations + +import datetime as dt +import json +import uuid +from pathlib import Path + +import pytest +from httpx import ASGITransport, AsyncClient +from openpyxl import Workbook +from sqlalchemy import delete, select + +from app.data.recipe_types import Recipe +from app.db.base import SessionLocal +from app.db.models import DataSource, Setting, SourceSnapshot, TableRecipe +from app.engine import spend +from app.engine.dual_check_facts import recipe_dates, upload_dates +from app.engine.runner import run_manager +from app.main import app +from tests.fixtures.dual_check_shop import ROWS, Blind +from tests.fixtures.sources import drop_source +from tests.fixtures.xlsx import save +from tests.test_dual_check_node import decide, events, graph, logs, notes_of, pending, start + +XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet" +API = "/api/datasources" +SHEET = "订单明细" +TABLE = "订单" +HEADER = ("订单号", "地区", "客户", "金额", "状态", "下单日期", "支付日期") +#: 同 dual_check_shop 的五笔单：退款行、下单日和支付日不在同一周的单，只是表名、列名、状态换成中文 +STATUS = {"paid": "已支付", "refunded": "已退款"} +WEEK = "BETWEEN '2026-09-01' AND '2026-09-07'" +#: 口径说明要的写法：按支付日、排除退款（1、2、5 号单，330.0） +RIGHT = f"SELECT SUM(\"金额\") AS gmv FROM \"订单\" WHERE \"状态\" <> '已退款' AND \"支付日期\" {WEEK}" +#: 只错在时间列：按下单日（2、4、5 号单，310.0） +BY_ORDER_DATE = f"SELECT SUM(\"金额\") AS gmv FROM \"订单\" WHERE \"状态\" <> '已退款' AND \"下单日期\" {WEEK}" +DEFINITION = "已支付订单的金额合计：按支付日期归属到当周，不含退款" +#: 配方里不是日期、读不出配方时的原因：照旧说按文字保存 +STORED_AS_TEXT = "来自上传的表格，表格里的日期一律按文字保存，无法确认存的是日期" +FLOW_RECIPE = json.loads((Path(__file__).parent / "fixtures" / "recipes" / "flow_recipe.json").read_text("utf-8")) + + +@pytest.fixture(autouse=True) +async def engine_up(): + await run_manager.setup() + async with SessionLocal() as session: + await session.execute(delete(Setting).where(Setting.key.in_(["verify", spend.SPEND_KEY]))) + await session.commit() + yield + await run_manager.shutdown() + + +@pytest.fixture +async def client(): + async with AsyncClient(transport=ASGITransport(app=app), base_url="http://test") as c: + yield c + + +def workbook() -> bytes: + """订单明细：第 1 行表头，日期写成 Excel 的日期格（不是文字）。""" + wb = Workbook() + ws = wb.active + ws.title = SHEET + ws.append(list(HEADER)) + for oid, region, customer, amount, status, ordered, paid, *_ in ROWS: + ws.append([oid, region, customer, amount, STATUS[status], + dt.date.fromisoformat(ordered), dt.date.fromisoformat(paid)]) + return save(wb) + + +def list_recipe(*, ordered: str = "DATE", paid: str = "DATE") -> dict: + """列表配方：两个日期列各按给定的类型导入（DATE 是界面上的「日期」，TEXT 是「文字」）。""" + types = {"订单号": "INTEGER", "地区": "TEXT", "客户": "INTEGER", "金额": "REAL", "状态": "TEXT", + "下单日期": ordered, "支付日期": paid} + return {"recipe_format": "agentlab-recipe/2", + "sheets": [{"id": "s1", "match": {"name": SHEET}, "blocks": [{ + "id": "orders", "layout": "list", "table": TABLE, + "columns": [{"header": h, "name": h, "type": types[h]} for h in HEADER]}]}], + "tables": [{"name": TABLE, "grain": ["订单号"]}]} + + +async def recipe_import(client, recipe: dict) -> tuple[str, str]: + """按配方导入：暂存 → 换成给定的配方 → 试运行 → 勾全部确认项提交。返回 (数据源名, id)。""" + name = f"dc_upload_{uuid.uuid4().hex[:8]}" + r = await client.post(f"{API}/imports/stage", files={"file": (f"{SHEET}_2026-09.xlsx", workbook(), XLSX)}, + data={"name": name}) + assert r.status_code == 201, r.text + staging = r.json()["id"] + r = await client.put(f"{API}/imports/{staging}/recipe", json={"recipe": recipe}) + assert r.status_code == 200 and r.json()["recipe_problems"] == [], r.text + r = await client.post(f"{API}/imports/{staging}/trial", json={}) + assert r.status_code == 200, r.text + trial = r.json()["trial"] + assert trial["status"] in ("passed", "needs_decision"), trial["problems"] + r = await client.post(f"{API}/imports/{staging}/commit", json={ + "trial_id": trial["trial_id"], "confirmations": [i["id"] for i in trial.get("confirm_items") or []]}) + assert r.status_code == 201, r.text + return name, r.json()["source"]["id"] + + +async def simple_import(client) -> tuple[str, str]: + """期 1 的简单导入：按表头行导入，没有配方。""" + name = f"dc_simple_{uuid.uuid4().hex[:8]}" + r = await client.post(f"{API}/upload", files={"file": (f"{SHEET}.xlsx", workbook(), XLSX)}, data={"name": name}) + assert r.status_code == 201, r.text + async with SessionLocal() as session: + source_id = (await session.execute(select(DataSource.id).where(DataSource.name == name))).scalar_one() + return name, source_id + + +async def dates_of(source_id: str) -> dict: + async with SessionLocal() as session: + return await upload_dates(session, await session.get(DataSource, source_id)) + + +async def break_recipe(source_id: str) -> None: + """把数据源当前版本的配方记录改坏（造「配方读不出来」）。""" + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + snap = await session.get(SourceSnapshot, row.current_snapshot_id) + record = await session.get(TableRecipe, snap.recipe_id) + record.recipe = {"recipe_format": "agentlab-recipe/0"} + await session.commit() + + +@pytest.fixture +async def dated(client): + """两个日期列在配方里都是「日期」的上传源。""" + name, source_id = await recipe_import(client, list_recipe()) + yield name, source_id + await drop_source(source_id) + + +@pytest.fixture +async def paid_as_text(client): + """支付日期在配方里是「文字」、下单日期是「日期」的上传源。""" + name, source_id = await recipe_import(client, list_recipe(paid="TEXT")) + yield name, source_id + await drop_source(source_id) + + +# -------------------------------------------------------------------------- +# 配方里哪些列是日期 +# -------------------------------------------------------------------------- + + +def test_the_date_columns_of_a_list_recipe_are_the_ones_typed_as_dates(): + assert recipe_dates(Recipe.model_validate(list_recipe())) == {TABLE: ["下单日期", "支付日期"]} + assert recipe_dates(Recipe.model_validate(list_recipe(paid="TEXT"))) == {TABLE: ["下单日期"]} + assert recipe_dates(Recipe.model_validate(list_recipe(ordered="TEXT", paid="TEXT"))) == {} + + +def test_the_axis_of_a_crosstab_is_a_date_column_of_every_table_it_writes(): + """交叉表的日期轴（类型恒为日期）写进各分段的表：宽表、长表、表内合计另存的表都有这一列。""" + assert recipe_dates(Recipe.model_validate(FLOW_RECIPE)) == { + "日客流": ["日期"], "时段客流": ["日期"], "时段客流_表内合计": ["日期"]} + + +def test_a_list_total_kept_as_a_table_has_no_date_column(): + """列表的合计行另存成表时只有合计项和数字列：日期列不在里面。""" + recipe = list_recipe() + recipe["sheets"][0]["blocks"][0]["rows"] = {"total_row": {"label_column": "地区", "pick": "合计", + "keep_as": "订单合计"}} + recipe["tables"].append({"name": "订单合计", "kind": "reported_total"}) + assert recipe_dates(Recipe.model_validate(recipe)) == {TABLE: ["下单日期", "支付日期"]} + + +# -------------------------------------------------------------------------- +# 读当前版本的配方 +# -------------------------------------------------------------------------- + + +async def test_an_upload_source_reads_the_date_columns_from_its_current_recipe(client, dated, paid_as_text): + """真实导入流程建的源：表结构里两列都是 TEXT，配方里是日期的那几列读得出来。""" + async with SessionLocal() as session: + schema = (await session.get(DataSource, dated[1])).schema_cache + types = {c["name"]: c["type"] for c in schema["tables"][TABLE]["columns"]} + assert (types["下单日期"], types["支付日期"]) == ("TEXT", "TEXT"), "背景：导入时日期一律按文字保存" + assert await dates_of(dated[1]) == {TABLE: ["下单日期", "支付日期"]} + assert await dates_of(paid_as_text[1]) == {TABLE: ["下单日期"]} + + +async def test_without_a_readable_recipe_no_column_is_taken_as_a_date(client, dated, tmp_path): + """读不出配方的都不认：简单导入没有配方，配方记录坏了，手工登记的库，不存在的数据源。""" + name, source_id = await simple_import(client) + try: + async with SessionLocal() as session: + row = await session.get(DataSource, source_id) + assert row.origin == "upload" and row.current_recipe_id is None + assert await dates_of(source_id) == {} + finally: + await drop_source(source_id) + await break_recipe(dated[1]) + assert await dates_of(dated[1]) == {} + async with SessionLocal() as session: + manual = DataSource(name=f"dc_manual_{uuid.uuid4().hex[:8]}", kind="sqlite", database=str(tmp_path / "x.db")) + assert await upload_dates(session, manual) == {} + assert await upload_dates(session, None) == {} + + +async def test_the_node_hands_the_import_dates_over_with_the_source(client, dated): + """节点认数据源来历的那一步（_source_info）带上 upload_dates，候选据此交给 derive。""" + from app.engine.nodes.dual_check import _source_info, _Sources + + info = await _source_info(_Sources(), dated[0]) + assert info["uploaded"] is True and info["upload_dates"] == {TABLE: ["下单日期", "支付日期"]} + assert (await _source_info(_Sources(), "dc_no_such_source"))["upload_dates"] == {} + + +# -------------------------------------------------------------------------- +# 双盲复算：审批卡的预告、裁决后的真写 +# -------------------------------------------------------------------------- + + +def upload_graph(name: str, sql: str) -> dict: + return graph(name, sql, dual={"definitions": {"gmv": DEFINITION}}) + + +async def test_dates_the_import_took_as_dates_are_written_as_the_business_date(monkeypatch, client, dated): + """两列在配方里都是「日期」：审批卡预告选哪边记哪一列，审核人选盲算，业务日期记「支付日期」，人工确认、备注写明出处。""" + name, source_id = dated + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(upload_graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + [item] = (await pending(run.id)).payload["summary"] + assert len(item["diff"]) == 1 and "下单日期" in item["diff"][0] and "支付日期" in item["diff"][0] + where = {"source": name, "table": TABLE, "item": "business_date"} + assert item["catalog"] == {"a": {"write": True, **where, "value": {"column": "下单日期"}}, + "b": {"write": True, **where, "value": {"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"] == [{**where, "value": {"column": "支付日期"}}] and wrote["skipped"] == [] + entry = await notes_of(source_id, TABLE) + stored = entry.notes["business_date"] + assert stored["value"] == {"column": "支付日期"} + assert (stored["source"], stored["status"]) == ("human", "confirmed") and run.id in stored["note"] + + +async def test_a_column_the_recipe_kept_as_text_is_not_written(monkeypatch, client, paid_as_text): + """支付日期在配方里是「文字」：选哪边都不记，原因照旧说按文字保存——被采纳的是它就说它，被采纳的下单日期是日期的，说 + 未被采纳的支付日期、另说清为什么也要看它。预告和裁决后跳过的原因一字不差，目录里什么都没写。""" + name, source_id = paid_as_text + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(upload_graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + said = (await pending(run.id)).payload["summary"][0]["catalog"] + column = f"「{TABLE}.支付日期」{STORED_AS_TEXT}" + assert said["b"] == {"write": False, "reason": f"指标「销售额」被采纳的盲算按时间筛的列{column}，不写入数据目录"} + assert said["a"] == {"write": False, "reason": f"指标「销售额」未被采纳的盲算按时间筛的列{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, TABLE) is None + + +async def test_without_a_readable_recipe_nothing_is_written(monkeypatch, client, dated): + """配方记录坏了、读不出来：两列都照旧说按文字保存，不写。""" + name, source_id = dated + await break_recipe(source_id) + Blind(monkeypatch, name, {"销售额": (RIGHT, "gmv")}) + run = await start(upload_graph(name, BY_ORDER_DATE)) + assert run.status == "interrupted", run.error + said = (await pending(run.id)).payload["summary"][0]["catalog"] + for side, label, column in (("a", "原流程", "下单日期"), ("b", "盲算", "支付日期")): + assert said[side] == {"write": False, "reason": f"指标「销售额」被采纳的{label}按时间筛的列「{TABLE}.{column}」" + f"{STORED_AS_TEXT}，不写入数据目录"} + done = await decide(client, run.id, {"gmv": "a"}) + 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["a"]["reason"]] + assert await notes_of(source_id, TABLE) is None