Files
AgentLab/backend/app/data/recipe_diff.py
ThearchyHelios 2cdbf9a733 Excel 导入改造期 3:网格交互、自助修复、按期累积——每月的报表可以累积成一份跨月的表,漂移了点修复按钮就能接着导入,网格上框一块就能写进配方,版本页可以回滚、移除一期、清除原件
按期累积(新 recipe_accumulate;table_versions 扩展)
- 按统计期出的交叉表报表首次导入时默认「按期累积」,由人确认;一期一个构建,两期及以上物化成一份新的 SQLite 快照,不用视图
- 统计期完全相同要勾「替换该期」,部分重叠拒收,中间有空缺要确认;物化前逐期核对库哈希和行号,并集上另加行数守恒、主键唯一、日期落在所属统计期三条核对
- 配方演进分三档:新增列(早期各期为空值)、退役的列和表保留旧值(只在这次才退役时要勾)、不兼容变化只能显式「重新开始累积」
- 快照 id 在累积时含目标配方和模式;同样的输入物化两次库哈希相同,同一文件换名再传直接复用
- 说明按期合并,任何一期没核对过就不写「已核对」;各期维度取值不同、退役列、合计表都有提示,经过数字检查

修复按钮(新 recipe_fixes;执行器给问题带结构化参数)
- 加入标签、移除标签、声明为合计行、分段标题改名、编辑恒等式成员、按锚点忽略行/列/区域外文字、声明占位符、声明隐藏行处理、工作表改名
- 服务端算补丁,先预览新旧配方对照、再应用,可以撤销;补丁封闭,不引入正则、自由文本常量和写死坐标;修复进确认清单
- 只改分段标题、新标题仍含原来选的词时沿用已确认的常量,不算破坏性变更;修复后只查改动过的段的认领,没改的关系交给试运行核对

框选转锚点(新 recipe_select)
- 网格上框选(鼠标、键盘、区域输入),按八种用途换算成按文字认的锚点,立刻干跑比对;配方里不存坐标,泛化不了时列出原因
- 整张表挪位后同一配方照样认得;框进合计行、夹着小计、表下说明都有明确的处理

确认与对照(recipe_confirm、recipe_diff、新 recipe_compare)
- 改配方后确认清单只列有变化的项(同时按本期数据比较),不用每修一次重勾十来项;新旧配方对照界面;按规则重新起草并把名字对齐到现行配方
- 改配方后已体现在配方里的回答保留,矛盾的列出来请人重答
- 「排除的行」写进回执和导入清单:隐藏行、跳过的空行、按配方忽略的行、停止后的文字行、只核对不另存的合计行

版本页(新 source_versions、api/source_versions;前端 VersionsDialog)
- 快照与导入记录列表、启用旧版本(可写理由)、查看清单、移除一期、作废接受、清除原件;写操作带界面看到的当前版本,并发改动回 409,不静默丢期
- 除当前版本和被运行引用的版本外,再保留最近 3 个版本供回滚;遮罩列丢失要明确确认

期 1、期 2 遗留
- 重传撞上被改过的构建文件:能用本次上传恢复就隔离旧文件再恢复,否则回 409,不再是 500
- 跳过空行的列表不再把表下隔一行的说明读成数据;不像说明的单格行要人确认;受影响的配方构建换 id
- 版本存储加进程级文件锁:第二个进程拿不到锁时不清理、不回收,写操作回 503(部署本来就是单进程)
2026-10-02 07:38:44 +08:00

977 lines
50 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""差异卡(期 2,WP-5b;P2-SPEC 7.6):上传新一期时,拿本期的导入回执和上一期的比,列出哪里和上次不一样。
**为什么要有差异卡。** 配方能通过试运行,只说明这一期的文件「对得上结构」;标签写法、块的先后、合计由
公式变成写死的数、区域外多了一句含数字的说明,这些都不会让试运行失败,却可能改变数据的口径(E3、E7、
R3)。差异卡把它们逐条摆出来:只是信息的(统计期、行数)标成 info,可能改变口径的标成「需确认」,
对应的确认项(`diff:<kind>:<键>`)由 recipe_confirm 收进必勾清单。
**输入形状。** prev / cur 都是「导入回执」``{"receipt": 回执, "checks": [核对摘要], "period": PeriodOut}``,
JSON 形状的 dict(dataclass 也收,先转成 dict)。回执就是 ``dataclasses.asdict(Extraction)`` 去掉
problems 之后的样子;逐项比较时用到其中的 RECEIPT_KEYS(tables、labels、block_order、derived_form、axes、
outside_text、ignored_columns、hidden、placeholders、canonicalized、sheets、lineage、full_calc_on_load;
列表的列顺序只能靠溯源里各列首格的列号推出来),外加顶层的 checks。
**缺键就报错(ReceiptIncomplete),不按空处理。** 早先的写法是「缺的键按空处理、比不了的项不出」,结果
调用方少带一个键(清单里没存 axes、cur 用了只有前 50 条规范写法对照的 TrialOut.receipt),需确认的
axis_form、label_writing、canon_values 就静默消失了(H2)。所以:
- prev、cur 都必须有 ``receipt``,回执里必须有 tables;
- 逐项比较时(上一期也是按配方导入的、配方没变),两边的回执都必须带齐 RECEIPT_KEYS、顶层带 checks;
规范写法对照被截断(canonicalized_total 大于条数)、或是部分干跑(partial)的结果,同样报错;
- 判断不了配方变没变(没传 recipe_changed,两边又没都带配方哈希)时报错,不默认「没变」去跨配方逐项比;
- 逐项比较时必须给本期配方:列表表头写法的确认项按**块**出,表 → 块只能从配方里查。
cur 取自本次试运行;prev 取自**现行导入的导入清单**(7.7),不取 TableBuild.report:构建可能被复用,
回执里没有 C1、C2,文件名也不是那一次导入的(评审甲-16)。上一期是简单导入(kind=switch,prev 是期 1 的
TableBuild.report,没有 ledger 键)或配方变了(redraft、改配方)时,两边的分段、块、表不一定对得上,
只给 info 类的 rows、period;破坏性变更由 recipe_confirm 按新旧 derive_tables 另算(breaking:*)。
**数字遮盖后的模板**(digit_template):canon 之后把每一段连续的 ASCII 数字换成 ``{}``。文件名、统计期格、
区域外文字的「写法变了」都按它比,每月必变的日期因此不算变化。
**期 3**(P3-SPEC 9.5):
- 给了累积计划(accumulate)时,按计划另出 period_added、period_replaced、period_gap(需确认)、period_backfill
(本期统计期是人工录入时需确认)、columns_added、columns_retired、labels_vary。它们来自计划,与配方变没变无关;
- 追加或替换该期时,区域外文字的差异(outside_*)**配方变了也照比**(outside_items,也公开给调用方单独算
ConfirmContext.outside_diff):本期和此前各期进同一列,单位写在区域外时变化只表现在这里,
recipe_confirm 的 accumulate_unit_risk 靠它(P3-SPEC 2.5);
- 逐项比较时按回执的 rows_excluded(排除的行)比「按配方忽略的行」:锚点集合或行数变了出 ignored_rows(需确认,
``diff:ignored_rows:<工作表>``,与期 2 ignored_columns 的 ``diff:ignored:<块>`` 不共用前缀)。rows_excluded
不在 RECEIPT_KEYS 里:上一期(期 3 之前的导入)没有这个字段时出 info「上一期未记录排除的行」,不当作空列表比。
纯函数,不碰数据库、不读文件。
"""
from __future__ import annotations
import dataclasses
import re
from typing import Any
from app.data.names import canon
from app.data.recipe_parsers import match_key, period_residue
from app.data.recipe_types import DiffItem, ListBlock, Recipe
_DIGITS = re.compile(r"[0-9]+")
#: 差异卡的全部 kind,按界面上的先后(7.6 的表;outside_moved 是表里「只是位置变了的算 info」那一项)
DIFF_KINDS: tuple[str, ...] = (
"period", "period_added", "period_replaced", "period_gap", "period_backfill", "rows", "columns_added",
"columns_retired", "labels_vary", "placeholders", "checks", "file_name", "full_calc", "context_text",
"context_source_added", "column_order", "label_writing", "header_writing", "canon_values", "row_order",
"block_order", "derived_form", "axis_form", "outside_added", "outside_removed", "outside_changed",
"outside_moved", "ignored_columns", "ignored_rows", "hidden",
)
#: 需确认的 kind → 确认项 id 的前缀(id = 前缀 + 键,键的格式见契约 recipe_types 的模块 docstring)
CONFIRM_PREFIX: dict[str, str] = {
"label_writing": "diff:label_writing:",
"header_writing": "diff:header_writing:",
"canon_values": "diff:canon:",
"row_order": "diff:row_order:",
"block_order": "diff:block_order:",
"derived_form": "diff:derived_form:",
"axis_form": "diff:axis_form:",
"outside_added": "diff:outside:",
"outside_removed": "diff:outside:",
"outside_changed": "diff:outside:",
"ignored_columns": "diff:ignored:",
"hidden": "diff:hidden:",
# 期 3。period_backfill、ignored_rows 只在特定情况下需确认(_item 的 confirm=False 时是 info)
"period_gap": "diff:period_gap:",
"period_backfill": "diff:period_backfill:",
"ignored_rows": "diff:ignored_rows:",
}
_STATUS = {"passed": "通过", "mismatch": "不一致", "unverifiable": "无法核对", "info": "提示"}
_DERIVED_FORM = {"formula": "公式", "literal": "写死的数", "mixed": "公式与写死的数混用"}
_AXIS_FORM = {"text": "文字", "date": "日期格", "mixed": "文字与日期格混用"}
_SHORT = 40
_A1 = re.compile(r"\$?([A-Za-z]{1,3})\$?([0-9]{1,7})$")
_WS = re.compile(r"\s+")
#: 逐项比较时两边回执都必须有的键(7.7 的导入清单 receipt 照 asdict(Extraction) 去掉 problems 写就都有)
RECEIPT_KEYS: tuple[str, ...] = (
"tables", "labels", "block_order", "derived_form", "axes", "outside_text", "ignored_columns", "hidden",
"placeholders", "canonicalized", "sheets", "lineage", "full_calc_on_load",
)
class ReceiptIncomplete(ValueError):
"""导入回执缺了比较要用的键、被截断或只是部分干跑:调用方的错,报出来,不按空处理(见模块 docstring)。
side 是 prev / cur / extraction,missing 是缺的键(截断、部分干跑时写成 canonicalized_total、partial)。
"""
def __init__(self, side: str, missing: list[str]) -> None:
self.side = side
self.missing = list(missing)
super().__init__(f"{side} 不完整,缺少比较要用的内容:{'、'.join(self.missing)}")
# --------------------------------------------------------------------------
# 小工具(recipe_confirm 也用)
# --------------------------------------------------------------------------
def digit_template(text: Any) -> str:
"""数字遮盖后的模板:「统计时间范围:2026年9月1日至2026年9月30日」→「统计时间范围:{}年{}月{}日至{}年{}月{}日」。
先 canon(全角数字经 NFKC 变成 ASCII、横线统一、空白折叠),再把每一段连续的 ASCII 数字换成 {}。
"""
return _DIGITS.sub("{}", canon("" if text is None else str(text)))
def same_period_sentence(old: Any, new: Any) -> bool:
"""两期的统计期格是不是「每月重复的同一句」:数字遮盖后的模板相同,且只去掉统计期那一段之后剩下的文字
(period_residue:其余数字照留)相同。任一期认不出统计期(None)就不算同一句。
只比遮盖模板不够(AU-3):「注:2026年9月数据只含3个分区」与「注:2026年10月数据只含2个分区」模板相同,
但统计期以外的口径数字变了。确认项(outside_digits 免确认的条件 (b))与差异卡(outside_changed)共用这一个判定。
"""
if digit_template(old) != digit_template(new):
return False
ra, rb = period_residue(old), period_residue(new)
return ra is not None and rb is not None and match_key(ra) == match_key(rb)
def plain(x: Any) -> Any:
"""dataclass(含嵌套)转成 JSON 形状;dict、list 逐层处理,其余原样。调用方传 Extraction 或 asdict 都行。"""
if dataclasses.is_dataclass(x) and not isinstance(x, type):
return dataclasses.asdict(x)
if isinstance(x, dict):
return {k: plain(v) for k, v in x.items()}
if isinstance(x, (list, tuple)):
return [plain(v) for v in x]
return x
def coord(sheet: Any, cell: Any) -> str:
"""「工作表!A1」。cell 已经带工作表名的原样返回;不知道工作表时只给 A1。"""
cell = str(cell or "")
if "!" in cell or not sheet:
return cell
return f"{sheet}!{cell}"
def split_coord(text: Any) -> tuple[str | None, int | None, int | None]:
"""「工作表!C5」→ (工作表, 行, 列号);认不出的部分为 None。"""
s = str(text or "")
sheet, _, a1 = s.rpartition("!")
m = _A1.match(a1.strip())
if m is None:
return (sheet or None), None, None
col = 0
for ch in m.group(1).upper():
col = col * 26 + (ord(ch) - 64)
return (sheet or None), int(m.group(2)), col
def col_letter(n: int) -> str:
out = ""
while n > 0:
n, r = divmod(n - 1, 26)
out = chr(65 + r) + out
return out
def short(text: Any, limit: int = _SHORT) -> str:
"""给标题用的短写法:空白折叠,超长的截断加「…」,全文放 detail。不做 canon:界面上照原文显示,
canon 的 NFKC 会把全角标点变成半角。"""
s = _WS.sub(" ", "" if text is None else str(text)).strip()
return s if len(s) <= limit else s[:limit] + "…"
def rows_text(rows: Any) -> str:
"""[28, 29, 30] →「第 28–30 行」;不连续的用顿号;空的给空串。"""
nums = sorted({int(r) for r in rows or [] if isinstance(r, int) or str(r).isdigit()})
if not nums:
return ""
if len(nums) > 1 and nums[-1] - nums[0] == len(nums) - 1:
return f"第 {nums[0]}–{nums[-1]} 行"
return "第 " + "、".join(str(n) for n in nums) + " 行"
def receipt_of(doc: Any) -> dict[str, Any]:
doc = plain(doc) or {}
rec = doc.get("receipt") if isinstance(doc, dict) else None
return rec if isinstance(rec, dict) else {}
def require_receipt(doc: Any, side: str, keys: tuple[str, ...] = RECEIPT_KEYS, *,
top: tuple[str, ...] = ("checks",)) -> None:
"""检查导入回执带齐了 keys(回执里)和 top(顶层),没有被截断、不是部分干跑;不满足就抛 ReceiptIncomplete。"""
doc = plain(doc)
if not isinstance(doc, dict) or not isinstance(doc.get("receipt"), dict):
raise ReceiptIncomplete(side, ["receipt"])
rec = doc["receipt"]
missing = [k for k in top if k not in doc] + [k for k in keys if k not in rec]
if "canonicalized" in keys and "canonicalized" in rec and "canonicalized_total" in rec:
# TrialOut.receipt 里的 canonicalized 只有前 50 条;拿它比会漏掉新写法
if int(rec.get("canonicalized_total") or 0) > len(rec.get("canonicalized") or []):
missing.append("canonicalized_total")
if rec.get("partial"):
missing.append("partial")
if missing:
raise ReceiptIncomplete(side, missing)
def recipe_sha_of(doc: Any) -> str | None:
"""导入回执上的配方哈希:顶层 recipe_sha256,或导入清单的 recipe.sha256(7.7)。"""
doc = plain(doc) or {}
if not isinstance(doc, dict):
return None
sha = doc.get("recipe_sha256")
if not sha and isinstance(doc.get("recipe"), dict):
sha = doc["recipe"].get("sha256")
return str(sha) if sha else None
def period_of(doc: Any) -> dict[str, Any] | None:
"""导入回执的统计期:顶层 period 优先(人工录入只记在那里),没有再看回执里的。"""
doc = plain(doc) or {}
p = doc.get("period") if isinstance(doc, dict) else None
if not isinstance(p, dict):
p = receipt_of(doc).get("period")
return p if isinstance(p, dict) else None
def checks_of(doc: Any) -> list[dict[str, Any]]:
doc = plain(doc) or {}
rows = doc.get("checks") if isinstance(doc, dict) else None
return [c for c in rows or [] if isinstance(c, dict) and c.get("id")]
def check_status(doc: Any, check_id: str) -> str | None:
for c in checks_of(doc):
if c.get("id") == check_id:
return str(c.get("status") or "")
return None
def default_sheet(rec: dict[str, Any]) -> str | None:
"""只有一张配方工作表时它的实际名字:统计期格的坐标万一没带工作表名,靠它补上。"""
matched = (rec.get("sheets") or {}).get("matched") or {}
names = {str(v) for v in matched.values() if v}
if len(names) == 1:
return next(iter(names))
sheets = {str(o.get("sheet")) for o in rec.get("outside_text") or [] if isinstance(o, dict) and o.get("sheet")}
return next(iter(sheets)) if len(sheets) == 1 else None
def outside_of(rec: dict[str, Any]) -> list[dict[str, Any]]:
"""区域外文字(含带附加文字的统计期格),坐标统一成「工作表!A1」。"""
out = []
for o in rec.get("outside_text") or []:
if not isinstance(o, dict):
continue
out.append({"coord": coord(o.get("sheet"), o.get("cell")), "text": str(o.get("text") or ""),
"kind": str(o.get("kind") or "text"), "period_source": bool(o.get("period_source"))})
return out
def period_texts(doc: Any, rec: dict[str, Any]) -> dict[str, str]:
"""统计期来源格:坐标 → 全文。"""
p = period_of(doc) or {}
sheet = default_sheet(rec)
return {coord(sheet, k): str(v or "") for k, v in (p.get("texts") or {}).items()}
def period_annotated(doc: Any, rec: dict[str, Any]) -> dict[str, str]:
"""带附加文字的统计期格:坐标 → period_residue。"""
p = period_of(doc) or {}
sheet = default_sheet(rec)
return {coord(sheet, k): str(v or "") for k, v in (p.get("annotated") or {}).items()}
# --------------------------------------------------------------------------
# 差异卡
# --------------------------------------------------------------------------
def diff_reports(prev: dict[str, Any] | None, cur: dict[str, Any], *, prev_file_name: str | None,
cur_file_name: str, recipe: Recipe | dict[str, Any] | None = None,
recipe_changed: bool | None = None, accumulate: Any = None) -> list[DiffItem]:
"""上一期 → 本期的差异卡。prev 为 None(首次)返回 []。需确认的排在前面,同类按 DIFF_KINDS 的顺序。
上一期的回执没有 ledger 键(简单导入,kind=switch)或配方变了(redraft、改配方)时只给 rows、period;
否则逐项比较。规格之外多了两个参数:
- recipe:本期的配方,逐项比较时必须给。列表的表头写法变化要按**块**出确认项
(``diff:header_writing:<块>``),回执里只有表,「表 → 块」只能从配方里查。
- recipe_changed:配方与上一期是否不同。不传时看 prev、cur 两边的配方哈希(顶层 ``recipe_sha256``,
或导入清单的 ``recipe.sha256``);两边没都带、又没传时报错,不默认「没变」。传了、两边也都带了哈希
而两者矛盾时同样报错。
- accumulate(期 3):累积计划(AccumulatePlan 或它的 asdict)。给了、而且是按期累积的追加或替换该期时,按计划
出统计期、列、维度取值的差异项;配方变了也比区域外文字(outside_items,两边回执缺 OUTSIDE_KEYS 时报错),
见模块 docstring。
"""
if prev is None:
return []
prev, cur = plain(prev) or {}, plain(cur) or {}
for side, doc in (("prev", prev), ("cur", cur)):
require_receipt(doc, side, ("tables",), top=())
p_rec, c_rec = receipt_of(prev), receipt_of(cur)
plan = plain(accumulate)
items: list[DiffItem] = []
items += _diff_period(prev, cur)
items += _diff_accumulate(plan, cur)
items += _diff_rows(p_rec, c_rec)
full = "ledger" in p_rec and not _recipe_changed(prev, cur, recipe_changed)
if not full and accumulating(plan):
# 按期累积的追加、替换该期:本期和此前各期进同一列,单位写在区域外时(lab.c01 的「单位:万元」),变化只
# 表现为区域外文字的差异(P3-SPEC 2.5,评审一-M10),accumulate_unit_risk 靠它。配方变了也照比:同一暂存区里
# 应用过一次修复,单位从「人次」改成「万人次」也不能没有提示(评审意见 H2)
items += outside_items(prev, cur)
if full:
require_receipt(prev, "prev")
require_receipt(cur, "cur")
if recipe is None:
raise ValueError("逐项比较需要本期配方:列表表头写法的确认项按块出,表与块的对应只能从配方里查")
blocks = _list_blocks(recipe)
items += _diff_placeholders(p_rec, c_rec)
items += _diff_checks(prev, cur)
items += _diff_file_name(prev_file_name, cur_file_name)
items += _diff_full_calc(p_rec, c_rec)
items += _diff_context_text(prev, cur, p_rec, c_rec)
items += _diff_column_order(p_rec, c_rec)
items += _diff_labels(p_rec, c_rec)
items += _diff_header_writing(p_rec, c_rec, blocks)
items += _diff_canon(p_rec, c_rec)
items += _diff_block_order(p_rec, c_rec)
items += _diff_forms(p_rec, c_rec)
items += _diff_outside(prev, cur, p_rec, c_rec)
items += _diff_ignored(p_rec, c_rec)
items += _diff_ignored_rows(p_rec, c_rec)
items += _diff_hidden(p_rec, c_rec)
order = {k: i for i, k in enumerate(DIFF_KINDS)}
items.sort(key=lambda d: (not d.requires_confirm, order.get(d.kind, len(order))))
return items
#: 比较区域外文字要用的回执键(outside_items)
OUTSIDE_KEYS: tuple[str, ...] = ("outside_text", "sheets")
def accumulating(plan: Any) -> bool:
"""累积计划是不是「本期和此前各期进同一个版本」:mode=accumulate 且 action 为追加或替换该期。first、restart、
replace 的结果只有本期,rejected 不提交。"""
plan = plain(plan)
return (isinstance(plan, dict) and plan.get("mode") == "accumulate"
and plan.get("action") in ("append", "replace_period"))
def outside_items(prev: dict[str, Any], cur: dict[str, Any]) -> list[DiffItem]:
"""上一期 → 本期区域外文字的差异项(outside_added / outside_removed / outside_changed / outside_moved /
context_source_added),与 diff_reports 逐项比较时的那几条相同。
公开出来是给按期累积用的(P3-SPEC 9.3 ConfirmContext.outside_diff):accumulate_unit_risk 看区域外文字涉及单位
或口径的变化,而 diff_reports 在配方变了时不做逐项比较。diff_reports 收到追加或替换该期的累积计划时已经调用
这里;调用方另算 outside_diff 时也用它。两边回执都必须带 OUTSIDE_KEYS,缺了报 ReceiptIncomplete,不按空处理
(按空处理时单位变化就静默消失了)。
"""
prev, cur = plain(prev) or {}, plain(cur) or {}
require_receipt(prev, "prev", OUTSIDE_KEYS, top=())
require_receipt(cur, "cur", OUTSIDE_KEYS, top=())
return _diff_outside(prev, cur, receipt_of(prev), receipt_of(cur))
def _recipe_changed(prev: dict[str, Any], cur: dict[str, Any], flag: bool | None) -> bool:
"""配方与上一期是否不同。传了 flag 就用它(两边都带哈希时还要对得上);没传就比两边的配方哈希。"""
a, b = recipe_sha_of(prev), recipe_sha_of(cur)
if a and b:
if flag is not None and flag != (a != b):
raise ValueError("recipe_changed 与 prev、cur 上的配方哈希不一致")
return a != b
if flag is None:
raise ValueError("无法判断配方是否与上一期相同:请传 recipe_changed,或在 prev、cur 上都带配方哈希")
return flag
def _item(kind: str, label: str, detail: str = "", key: str | None = None, *, confirm: bool = True) -> DiffItem:
"""confirm=False:这一类一般需确认,这一条只是信息(period_backfill 的统计期不是人工录入、ignored_rows 的
上一期没有记录)。"""
prefix = CONFIRM_PREFIX.get(kind)
if prefix is None or not confirm:
return DiffItem(kind=kind, label=label, detail=detail)
return DiffItem(kind=kind, label=label, detail=detail, requires_confirm=True, confirm_id=f"{prefix}{key}")
def _period_span(p: dict[str, Any] | None) -> tuple[str, str] | None:
if not p or not p.get("start") or not p.get("end"):
return None
return str(p["start"]), str(p["end"])
def _diff_period(prev: dict[str, Any], cur: dict[str, Any]) -> list[DiffItem]:
a, b = _period_span(period_of(prev)), _period_span(period_of(cur))
if a == b:
return []
if a is None:
return [_item("period", f"统计期:本期 {b[0]} 至 {b[1]}", "上一期没有记录统计期")]
if b is None:
return [_item("period", f"统计期:上一期 {a[0]} 至 {a[1]},本期未能确定")]
return [_item("period", f"统计期 {a[0]} 至 {a[1]} → {b[0]} 至 {b[1]}")]
def _table_rows(rec: dict[str, Any]) -> dict[str, int]:
out: dict[str, int] = {}
for t in rec.get("tables") or []:
if isinstance(t, dict) and t.get("name") is not None:
out[str(t["name"])] = int(t.get("rows") or 0)
return out
def _placeholder_counts(rec: dict[str, Any]) -> dict[str, int]:
return {str(k): int(v or 0) for k, v in (rec.get("placeholders") or {}).items()}
def _diff_rows(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
a, b = _table_rows(p_rec), _table_rows(c_rec)
if a == b:
return []
parts = []
for name in list(dict.fromkeys([*b, *a])):
if name in a and name in b:
if a[name] != b[name]:
parts.append(f"「{name}」{a[name]} → {b[name]}")
elif name in b:
parts.append(f"「{name}」本期新增,{b[name]} 行")
else:
parts.append(f"「{name}」本期没有,上一期 {a[name]} 行")
pa, pb = _placeholder_counts(p_rec), _placeholder_counts(c_rec)
notes = [f"占位符「{t}」{pa.get(t, 0)} → {pb.get(t, 0)} 格"
for t in dict.fromkeys([*pa, *pb]) if pa.get(t, 0) != pb.get(t, 0)]
return [_item("rows", "行数:" + ";".join(parts), ";".join(notes))]
def _diff_placeholders(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
# 只比原文集合:每月个数都会变(31 天 → 30 天),个数写在 rows 的细节里
a = {t for t, n in _placeholder_counts(p_rec).items() if n > 0}
b = {t for t, n in _placeholder_counts(c_rec).items() if n > 0}
if a == b:
return []
parts = []
if b - a:
parts.append("新出现" + "".join(f"「{t}」" for t in sorted(b - a)))
if a - b:
parts.append("不再出现" + "".join(f"「{t}」" for t in sorted(a - b)))
return [_item("placeholders", "占位符:" + ";".join(parts))]
def _diff_checks(prev: dict[str, Any], cur: dict[str, Any]) -> list[DiffItem]:
# 比 (id, 状态) 的集合:核对数每期都变,不算
a = {c["id"]: str(c.get("status") or "") for c in checks_of(prev)}
b = {c["id"]: str(c.get("status") or "") for c in checks_of(cur)}
if set(a.items()) == set(b.items()):
return []
parts = []
for cid in dict.fromkeys([*b, *a]):
if cid in a and cid in b:
if a[cid] != b[cid]:
parts.append(f"{cid} {_STATUS.get(a[cid], a[cid])} → {_STATUS.get(b[cid], b[cid])}")
elif cid in b:
parts.append(f"{cid} 本期新出现({_STATUS.get(b[cid], b[cid])})")
else:
parts.append(f"{cid} 本期不再出现")
return [_item("checks", "核对结果:" + ";".join(parts))]
def _base_name(name: str | None) -> str:
return str(name or "").replace("\\", "/").rsplit("/", 1)[-1]
def _diff_file_name(prev_name: str | None, cur_name: str) -> list[DiffItem]:
if prev_name is None:
return []
a, b = _base_name(prev_name), _base_name(cur_name)
if digit_template(a) == digit_template(b):
return []
return [_item("file_name", f"文件名的写法变了:「{short(a)}」→「{short(b)}」")]
def _diff_full_calc(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
"""本期设了打开时重算就出(规格 7.6「这一期设了」,不论上一期设没设):导入用的是保存值,每期都提醒。"""
if not c_rec.get("full_calc_on_load"):
return []
before = "上一期也设置了" if p_rec.get("full_calc_on_load") else "上一期没有设置"
return [_item("full_calc", "本期文件设置了打开时重新计算公式", f"导入使用文件里保存的值;{before}")]
def _diff_context_text(prev: dict[str, Any], cur: dict[str, Any], p_rec: dict[str, Any],
c_rec: dict[str, Any]) -> list[DiffItem]:
"""两期都是统计期来源、而且都只是在说统计期的格(不在区域外文字里):写法变了、C1 不是 mismatch 时出 info(D12)。
带附加文字的统计期格同时是区域外文字,写法变化归 outside_*(要确认),不在这里重复。
"""
if check_status(cur, "C1") == "mismatch":
return []
a, b = period_texts(prev, p_rec), period_texts(cur, c_rec)
in_outside = {o["coord"] for o in outside_of(p_rec)} | {o["coord"] for o in outside_of(c_rec)}
out = []
for cell in b:
if cell in a and cell not in in_outside and digit_template(a[cell]) != digit_template(b[cell]):
out.append(_item("context_text", f"统计期格 {cell} 的写法变了:「{short(a[cell])}」→「{short(b[cell])}」",
f"上一期:{a[cell]};本期:{b[cell]}"))
return out
def _is_list_table(t: dict[str, Any]) -> bool:
# 交叉表产出的表一定有轴列(宽表、长表、表内合计都以它开头);列表合计表是 reported_total
cols = [c for c in t.get("columns") or [] if isinstance(c, dict)]
return str(t.get("kind") or "data") == "data" and not any(c.get("role") == "axis" for c in cols)
def _tables(rec: dict[str, Any]) -> dict[str, dict[str, Any]]:
return {str(t["name"]): t for t in rec.get("tables") or [] if isinstance(t, dict) and t.get("name") is not None}
def _column_positions(rec: dict[str, Any], table: str) -> dict[str, int]:
"""溯源里各列首格的列号:列表一列一段(向下),首格的列号就是这一列在工作表里的位置。"""
out: dict[str, int] = {}
for col, runs in ((rec.get("lineage") or {}).get(table) or {}).items():
for run in runs or []:
if isinstance(run, (list, tuple)) and len(run) >= 3:
_, _, c = split_coord(run[2])
if c is not None:
out[str(col)] = min(out.get(str(col), c), c)
return out
def _diff_column_order(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
pa, pb = _tables(p_rec), _tables(c_rec)
out = []
for name, t in pb.items():
if name not in pa or not _is_list_table(t) or not _is_list_table(pa[name]):
continue
a, b = _column_positions(p_rec, name), _column_positions(c_rec, name)
common = set(a) & set(b)
if len(common) < 2:
continue
oa = sorted(common, key=lambda c: a[c])
ob = sorted(common, key=lambda c: b[c])
if oa != ob:
out.append(_item("column_order", f"表「{name}」的列顺序变了",
f"上一期:{'、'.join(oa)};本期:{'、'.join(ob)}"))
return out
def _segment_labels(rec: dict[str, Any]) -> dict[str, tuple[list[str], list[str]]]:
out: dict[str, tuple[list[str], list[str]]] = {}
for s in rec.get("labels") or []:
if isinstance(s, dict) and s.get("segment") is not None:
out[str(s["segment"])] = ([str(x) for x in s.get("raw") or []],
[str(x) for x in s.get("canonical") or []])
return out
def _diff_labels(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
"""label_writing(同一规范写法的原文变了,D11、D22、D24)与 row_order(分段内顺序变了,D19)。
原文逐字比:en dash 的「7–8」和「7-8」canon 之后相同,正是要报的那种写法变化(E3)。
"""
a, b = _segment_labels(p_rec), _segment_labels(c_rec)
writing, order = [], []
for seg, (raw_b, can_b) in b.items():
if seg not in a:
continue
raw_a, can_a = a[seg]
map_a = dict(zip(reversed(can_a), reversed(raw_a)))
map_b = dict(zip(reversed(can_b), reversed(raw_b)))
pairs = [(map_a[c], map_b[c]) for c in dict.fromkeys(can_b) if c in map_a and map_a[c] != map_b[c]]
if pairs:
shown = ";".join(f"「{x}」→「{y}」" for x, y in pairs[:3]) + ("等" if len(pairs) > 3 else "")
writing.append(_item("label_writing", f"分段「{seg}」的标签写法变了:{shown}",
f"共 {len(pairs)} 个标签;按规范写法存储的值不变", key=seg))
common = set(can_a) & set(can_b)
seq_a = [c for c in dict.fromkeys(can_a) if c in common]
seq_b = [c for c in dict.fromkeys(can_b) if c in common]
if seq_a != seq_b:
order.append(_item("row_order", f"分段「{seg}」的行顺序变了",
f"上一期:{'、'.join(seq_a)};本期:{'、'.join(seq_b)}", key=seg))
return writing + order
def _list_blocks(recipe: Recipe | dict[str, Any]) -> dict[str, str]:
"""列表的表名 → 块 id(只看本期配方)。"""
r = recipe if isinstance(recipe, Recipe) else Recipe.model_validate(recipe)
out: dict[str, str] = {}
for sheet in r.sheets:
for block in sheet.blocks:
if isinstance(block, ListBlock):
out.setdefault(block.table, block.id)
return out
def _diff_header_writing(p_rec: dict[str, Any], c_rec: dict[str, Any], blocks: dict[str, str]) -> list[DiffItem]:
pa, pb = _tables(p_rec), _tables(c_rec)
out = []
for name, t in pb.items():
if name not in pa or not _is_list_table(t):
continue
if name not in blocks:
# 回执里的列表表在配方里找不到块:传进来的配方不是产出这份回执的那份
raise ValueError(f"本期配方里没有写入表「{name}」的列表块,配方与回执对不上")
ha = {str(c.get("name")): c.get("header") for c in pa[name].get("columns") or [] if isinstance(c, dict)}
pairs = []
for c in t.get("columns") or []:
if not isinstance(c, dict):
continue
old, new = ha.get(str(c.get("name"))), c.get("header")
if old is not None and new is not None and str(old) != str(new):
pairs.append((str(old), str(new)))
if pairs:
shown = ";".join(f"「{short(x)}」→「{short(y)}」" for x, y in pairs[:3]) + ("等" if len(pairs) > 3 else "")
out.append(_item("header_writing", f"表「{name}」的表头写法变了:{shown}",
f"共 {len(pairs)} 列;列名和导入的值不变", key=blocks[name]))
return out
def _diff_canon(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
"""列表的主键列、canonical 存储的 TEXT 列出现新的「原文 ≠ 规范写法」对(来自标签的值不在 canonicalized 里)。"""
def pairs(rec: dict[str, Any]) -> list[tuple[str, str, str, str]]:
return [(str(e.get("table")), str(e.get("column")), str(e.get("raw")), str(e.get("canonical")))
for e in rec.get("canonicalized") or [] if isinstance(e, dict)]
seen = set(pairs(p_rec))
grouped: dict[tuple[str, str], list[tuple[str, str]]] = {}
for t, c, raw, can in pairs(c_rec):
if (t, c, raw, can) not in seen:
grouped.setdefault((t, c), []).append((raw, can))
out = []
for (t, c), new in grouped.items():
shown = ";".join(f"「{short(r)}」按「{short(k)}」保存" for r, k in new[:3]) + ("等" if len(new) > 3 else "")
out.append(_item("canon_values", f"表「{t}」列「{c}」出现新的写法:{shown}",
f"共 {len(new)} 种", key=f"{t}.{c}"))
return out
def _sheet_ids(rec: dict[str, Any]) -> dict[str, str]:
"""实际工作表名 → 配方工作表 id:工作表改了名,块的先后照样能和上一期对上。"""
return {str(v): str(k) for k, v in ((rec.get("sheets") or {}).get("matched") or {}).items()}
def _diff_block_order(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
a = {str(k): [str(x) for x in v or []] for k, v in (p_rec.get("block_order") or {}).items()}
b = {str(k): [str(x) for x in v or []] for k, v in (c_rec.get("block_order") or {}).items()}
ida, idb = _sheet_ids(p_rec), _sheet_ids(c_rec)
by_id_a = {ida[k]: v for k, v in a.items() if k in ida}
out = []
for sheet, seq_b in b.items():
seq_a = a.get(sheet)
if seq_a is None and sheet in idb:
seq_a = by_id_a.get(idb[sheet])
if seq_a is None:
continue
common = set(seq_a) & set(seq_b)
fa = [k for k in seq_a if k in common]
fb = [k for k in seq_b if k in common]
if fa != fb:
out.append(_item("block_order", f"工作表「{sheet}」上各部分的先后变了",
f"上一期:{'、'.join(fa)};本期:{'、'.join(fb)}", key=sheet))
return out
def _diff_forms(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
out = []
a = {str(k): str(v) for k, v in (p_rec.get("derived_form") or {}).items()}
for seg, form in ((c_rec.get("derived_form") or {}).items()):
seg, form = str(seg), str(form)
if seg in a and a[seg] != form:
out.append(_item("derived_form",
f"分段「{seg}」的合计由{_DERIVED_FORM.get(a[seg], a[seg])}改为"
f"{_DERIVED_FORM.get(form, form)}",
"合计仍按明细核对", key=seg))
def axes(rec: dict[str, Any]) -> dict[str, str]:
return {str(x.get("block")): str(x.get("form") or "") for x in rec.get("axes") or []
if isinstance(x, dict) and x.get("block") is not None}
aa, ab = axes(p_rec), axes(c_rec)
for blk, form in ab.items():
if blk in aa and aa[blk] != form:
out.append(_item("axis_form",
f"「{blk}」的日期表头由{_AXIS_FORM.get(aa[blk], aa[blk])}改为{_AXIS_FORM.get(form, form)}",
key=blk))
return out
def _diff_outside(prev: dict[str, Any], cur: dict[str, Any], p_rec: dict[str, Any],
c_rec: dict[str, Any]) -> list[DiffItem]:
"""区域外文字(含带附加文字的统计期格)的增、删、改,以及「区域外纯文字变成统计期来源」(D17、D27)。
顺序:先按全文配对(同格的先配;只是位置变了的出 info);再看同一格从纯文字变成统计期来源、满足条件的
归入 context_source_added;剩下同一格的按数字遮盖后的模板比(只是日期变了不算改);其余算增、删。
"""
p_out, c_out = outside_of(p_rec), outside_of(c_rec)
p_per, c_per = period_texts(prev, p_rec), period_texts(cur, c_rec)
c_ann = period_annotated(cur, c_rec)
c1_ok = check_status(cur, "C1") == "passed"
info: list[DiffItem] = []
confirm: list[DiffItem] = []
def take(pool: list[dict[str, Any]], pred: Any) -> dict[str, Any] | None:
for i, x in enumerate(pool):
if pred(x):
return pool.pop(i)
return None
# 1. 全文配对
p_left, c_left = list(p_out), []
for c in c_out:
if take(p_left, lambda p, c=c: p["coord"] == c["coord"] and canon(p["text"]) == canon(c["text"])) is None:
c_left.append(c)
rest = []
for c in c_left:
p = take(p_left, lambda p, c=c: canon(p["text"]) == canon(c["text"]))
if p is None:
rest.append(c)
else:
info.append(_item("outside_moved", f"区域外文字「{short(c['text'])}」从 {p['coord']} 移到 {c['coord']}"))
c_left = rest
# 2. 纯文字变成统计期来源
for p in list(p_left):
if p["kind"] != "text" or p["coord"] not in c_per:
continue
c = next((x for x in c_left if x["coord"] == p["coord"]), None)
# 本期这一格还在区域外文字里,按契约就是带附加文字的统计期格(period_source 漏标也一样)
annotated = p["coord"] in c_ann or c is not None
residue = c_ann.get(p["coord"])
if annotated and residue is None and c is not None:
residue = period_residue(c["text"])
if not c1_ok or (annotated and (residue is None or match_key(residue) != match_key(p["text"]))):
continue
p_left.remove(p)
if c is not None:
c_left.remove(c)
new_text = c_per[p["coord"]]
info.append(_item("context_source_added",
f"{p['coord']} 现在也写明了统计期:「{short(p['text'])}」→「{short(new_text)}」",
"统计期与其他来源一致;附加的文字与上一期这一格相同"))
# 3. 同一格:写法(数字遮盖后的模板)变了才算改。两期都是统计期来源的格,还要比去掉统计期之后的文字
# (其余数字照留):附加文字里的口径数字变了也算改(AU-3)
def changed(cell: str, old: str, new: str) -> None:
if cell in p_per and cell in c_per:
differ = not same_period_sentence(old, new)
else:
differ = digit_template(old) != digit_template(new)
if differ:
confirm.append(_item("outside_changed", f"{cell} 的区域外文字改了:「{short(old)}」→「{short(new)}」",
f"上一期:{old};本期:{new}", key=cell))
for p in list(p_left):
c = take(c_left, lambda x, p=p: x["coord"] == p["coord"])
if c is not None:
p_left.remove(p)
changed(p["coord"], p["text"], c["text"])
elif p["coord"] in c_per:
# 这一格本期只是在说统计期了,但不满足 context_source_added 的条件
p_left.remove(p)
changed(p["coord"], p["text"], c_per[p["coord"]])
for c in list(c_left):
if c["coord"] in p_per:
# 上一期只是统计期格,本期多了附加文字
c_left.remove(c)
changed(c["coord"], p_per[c["coord"]], c["text"])
# 4. 其余是增、删
for p in p_left:
confirm.append(_item("outside_removed", f"区域外文字「{short(p['text'])}」本期没有了(上一期在 {p['coord']})",
p["text"], key=p["coord"]))
for c in c_left:
confirm.append(_item("outside_added", f"区域外新增文字「{short(c['text'])}」({c['coord']})",
c["text"], key=c["coord"]))
return confirm + info
def _diff_ignored(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
a = {str(k): [str(x) for x in v or []] for k, v in (p_rec.get("ignored_columns") or {}).items()}
b = {str(k): [str(x) for x in v or []] for k, v in (c_rec.get("ignored_columns") or {}).items()}
out = []
for blk in dict.fromkeys([*b, *a]):
ka = {match_key(x) for x in a.get(blk, [])}
kb = {match_key(x) for x in b.get(blk, [])}
if ka != kb:
now = "".join(f"「{short(x, 20)}」" for x in b.get(blk, [])) or "无"
before = "".join(f"「{short(x, 20)}」" for x in a.get(blk, [])) or "无"
out.append(_item("ignored_columns", f"「{blk}」忽略的列变了:本期忽略{now}",
f"上一期忽略{before}", key=blk))
return out
def _hidden_parts(h: dict[str, Any]) -> tuple[set[int] | None, set[str] | None]:
rows = cols = None
if str(h.get("policy_rows") or "") in ("include", "exclude"):
rows = {int(r) for r in h.get("rows") or [] if isinstance(r, int) or str(r).isdigit()}
if str(h.get("policy_cols") or "") == "include":
cols = {col_letter(c) if isinstance(c, int) else str(c).upper() for c in h.get("cols") or []}
return rows, cols
def _diff_hidden(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
"""照常导入(include)或排除(exclude)隐藏行列时,隐藏的是哪些行列变了。
规格只写了 include;exclude 下被排除的行变了同样会静默改变导入的数据,一并报(见交付说明)。
reject_if_any 时有隐藏行列就拒收,不会走到这里。
"""
a = {str(k): v for k, v in (p_rec.get("hidden") or {}).items() if isinstance(v, dict)}
b = {str(k): v for k, v in (c_rec.get("hidden") or {}).items() if isinstance(v, dict)}
out = []
for sheet in dict.fromkeys([*b, *a]):
rb, cb = _hidden_parts(b.get(sheet, {}))
ra, ca = _hidden_parts(a.get(sheet, {}))
if rb is None and cb is None and ra is None and ca is None:
continue
parts = []
if (ra or set()) != (rb or set()) and (ra is not None or rb is not None):
parts.append("隐藏行:" + (rows_text(rb) or "本期没有"))
if (ca or set()) != (cb or set()) and (ca is not None or cb is not None):
parts.append("隐藏列:" + ("、".join(sorted(cb or ())) + " 列" if cb else "本期没有"))
if parts:
out.append(_item("hidden", f"工作表「{sheet}」的隐藏行列变了:" + ";".join(parts), key=sheet))
return out
# --------------------------------------------------------------------------
# 期 3:按期累积的计划、排除的行
# --------------------------------------------------------------------------
def _refs(entries: Any) -> list[str]:
"""[{"table", "column" | None}] →「表「日客流」的列「分区丙」」「表「X」」。"""
out = []
for e in entries or []:
if isinstance(e, dict) and e.get("table"):
out.append(f"表「{e['table']}」的列「{e['column']}」" if e.get("column") else f"表「{e['table']}」")
return out
def _diff_accumulate(plan: Any, cur: dict[str, Any]) -> list[DiffItem]:
"""按期累积的计划给出的差异项(P3-SPEC 2.3、2.5、9.5)。只在 mode=accumulate、action 是追加或替换该期时出:
first / restart / replace 的结果只有本期,这些项无从说起(restart、模式切换另有必勾的确认项)。"""
if not accumulating(plan):
return []
action = plan.get("action")
out: list[DiffItem] = []
period = _period_span(plan.get("period"))
parts = [p for p in plan.get("parts") or [] if isinstance(p, dict)]
if action == "append" and period is not None:
out.append(_item("period_added", f"按期累积:本期 {period[0]} 至 {period[1]} 加入当前版本,启用后共 {len(parts)} 期"))
if action == "replace_period":
rep = _period_span(plan.get("replaces")) or period
if rep is not None:
# 替换由必勾项 period_replace 确认,这里只是信息
out.append(_item("period_replaced", f"按期累积:替换已有的一期 {rep[0]} 至 {rep[1]}"))
for gap in plan.get("gaps") or []:
g = _period_span(gap)
if g is not None:
out.append(_item("period_gap", f"启用后各期之间有空缺:{g[0]} 至 {g[1]} 没有数据",
"比较不同期之前,请先确认空缺的这段时间是否本来就不需要", key=f"{g[0]}~{g[1]}"))
if plan.get("backfill") and period is not None:
human = (period_of(cur) or {}).get("source") == "human"
if human:
# 人工录入时日期的年份也按统计期补全:录错年份,同一份数据会按错误的年份进来(评审一-m12)
out.append(_item("period_backfill", "本期统计期为人工录入,且早于已有各期:请核对年份和月份无误",
f"本期 {period[0]} 至 {period[1]},按统计期排在已有各期之前", key=f"{period[0]}~{period[1]}"))
else:
out.append(_item("period_backfill", f"本期 {period[0]} 至 {period[1]} 早于已有各期,按统计期排在最前",
confirm=False))
added = _refs(plan.get("added"))
if added:
# 写「其他各期」不写「此前各期」:补传早期、替换中间某一期时,没有这些列的是更晚的期
out.append(_item("columns_added", "新增:" + "、".join(added) + ",其他各期没有这些数据,为空值"))
retired = _refs(plan.get("retired_new"))
if retired:
# 退役由必勾项 retire:* 确认,这里只是信息
out.append(_item("columns_retired", "自本期起不再导入:" + "、".join(retired) + ",早期各期保留原值"))
out += _labels_vary(plan, period)
return out
def _labels_vary(plan: dict[str, Any], period: tuple[str, str] | None) -> list[DiffItem]:
"""各期维度取值的差异(label_sets)里本期那一项(评审一-M2)。
label_sets 的 missing 是相对全部各期(含本期,补传早期时还含更晚的期)的并集少了哪些,extra 是比各期的交集多了
哪些,所以只能说「与其他各期相比」:本期没有的,其他期里至少有一期有;本期多的,其他期里至少有一期没有。不写
「此前各期有 / 没有」——8 月没有「7-8」、9 月有,上传 10 月(有「7-8」)时说「此前各期没有」就错了。细节里按期
列出其他各期各缺哪些取值,人能看出是哪几期不一样。"""
out: list[DiffItem] = []
for entry in plan.get("label_sets") or []:
if not isinstance(entry, dict):
continue
col = entry.get("column") or entry.get("segment") or ""
periods = [p for p in entry.get("periods") or [] if isinstance(p, dict)]
mine = next((p for p in periods if _period_span(p) == period), None)
if mine is None:
continue
parts = []
if mine.get("missing"):
parts.append(f"本期「{col}」没有" + "".join(f"「{short(x, 20)}」" for x in mine["missing"]))
if mine.get("extra"):
parts.append(f"本期「{col}」有" + "".join(f"「{short(x, 20)}」" for x in mine["extra"]) + ",但并非各期都有")
if not parts:
continue
others = []
for p in periods:
span = _period_span(p)
if p is mine or span is None:
continue
gone = "".join(f"「{short(x, 20)}」" for x in p.get("missing") or [])
others.append(f"{span[0]} 至 {span[1]} " + (f"没有{gone}" if gone else "取值齐全"))
detail = f"表「{entry.get('table')}」:跨期比较时各期覆盖的取值不同"
if others:
detail += ";其他各期:" + ";".join(others)
out.append(_item("labels_vary", "与其他各期相比," + ";".join(parts), detail))
return out
#: 按配方忽略的行(rows_excluded 的这两种原因):锚点集合或行数和上一期不同就要确认(P3-SPEC 3.2 ⑥)
_IGNORE_REASONS = ("ignored_rows", "ignored_outside")
def _ignored_by_sheet(rec: dict[str, Any]) -> dict[str, tuple[set[tuple[str, str]], int, list[str]]]:
"""工作表 → (锚点集合 {(原因, match_key)}, 行数, 锚点原文)。"""
out: dict[str, tuple[set[tuple[str, str]], int, list[str]]] = {}
for e in rec.get("rows_excluded") or []:
e = plain(e)
if not isinstance(e, dict) or e.get("reason") not in _IGNORE_REASONS:
continue
keys, n, raws = out.setdefault(str(e.get("sheet") or ""), (set(), 0, []))
anchor = str(e.get("anchor") or "")
keys.add((str(e["reason"]), match_key(anchor)))
if anchor and anchor not in raws:
raws.append(anchor)
rows = sum(int(b) - int(a) + 1 for a, b in e.get("rows") or [])
out[str(e.get("sheet") or "")] = (keys, n + rows, raws)
return out
def _excluded_total(rec: dict[str, Any]) -> int:
return sum(int(b) - int(a) + 1 for e in (plain(x) for x in rec.get("rows_excluded") or [])
if isinstance(e, dict) for a, b in e.get("rows") or [])
def _diff_ignored_rows(p_rec: dict[str, Any], c_rec: dict[str, Any]) -> list[DiffItem]:
"""按配方忽略的行(ignore_rows、ignore_outside)和上一期不同:需确认 diff:ignored_rows:<工作表>。
上一期回执里没有 rows_excluded(期 3 之前的导入)时不当作空列表比:那样本期一有排除的行就成了「新增」,
而上一期其实只是没记录。改出 info「上一期未记录排除的行,本期排除了 N 行」(本期没有排除的行就不出)。"""
if "rows_excluded" not in p_rec:
n = _excluded_total(c_rec)
if n <= 0:
return []
return [_item("ignored_rows", f"上一期未记录排除的行,本期排除了 {n} 行", "排除的行逐项列在本期回执里",
confirm=False)]
a, b = _ignored_by_sheet(p_rec), _ignored_by_sheet(c_rec)
out = []
for sheet in dict.fromkeys([*b, *a]):
ka, na, ra = a.get(sheet, (set(), 0, []))
kb, nb, rb = b.get(sheet, (set(), 0, []))
if ka == kb and na == nb:
continue
now = "".join(f"「{short(x, 20)}」" for x in rb) or "无"
before = "".join(f"「{short(x, 20)}」" for x in ra) or "无"
out.append(_item("ignored_rows", f"工作表「{sheet}」按配方忽略的行变了:本期 {nb} 行(锚点{now})",
f"上一期 {na} 行(锚点{before})", key=sheet))
return out