mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 08:01:58 +00:00
ThearchyHelios
2cdbf9a733
按期累积（新 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（部署本来就是单进程）
977 lines
50 KiB
Python
977 lines
50 KiB
Python
"""差异卡（期 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
|