mirror of
https://github.com/ThearchyHelios/AgentLab.git
synced 2026-11-15 08:31:59 +00:00
ThearchyHelios
a137182145
为什么：每月一份的明细导出是最常见的场景，期 4 之前列表只能每期替换；夹在中间的小计会让规则起草失败；
表格对象这个现成的边界信号没用上；CSV 不能走配方。四项都复用现有的执行器、核对、累积、溯源，不另起一套。
- 列表按期累积：ListBlock.period 二选一——{column, check: within_context} 按日期列分期，试运行多一条结构类核对
D（每行的日期都在本期统计期内，空着也不行）；{synthesize: 统计期} 每行加「统计期起」「统计期止」两列，列表在
统计期定下来之前就抽完了，这两格先放占位、写库时换成真值。accumulate_blockers 放行绑定了统计期的列表；合计行、
小计行另存的表各期主键分不开，照旧不能累积；同一订单两期都出现由并集的主键核对拒收。U3 按列表的分期方式核对。
说明里「按哪一列筛选」用分期列，加的两列有列说明；空行之后文字行的说法补了并集版本（列表能累积以后要用到）。
- 穿插的小计：ListRows.subtotals {label_column, pick, scope: since_previous | group_column, group_column, keep_as}。
小计行不进数据，每个数字格一条合计格（分段键是块 id + ·小计），核对模块按它所管那组明细的 rowid 区间用 SQL
求和比对（新 T 编号，标题「小计行按各组明细求和」），整行记进排除的行（reason=subtotal），可以另存成
reported_total 表（小计项、分组取值、各数字列）；数字格全空报 subtotal_blank（结构类）；合计行的公式引用了小计格
时按它管的明细换算。静态校验：标签列、分组列要在列里，按组小计要给分组列，小计和合计的开头词不能互为前缀。
规则起草把第一列同一个合计字样、下面还有明细的行提议成小计（以前记 stray_totals 失败），框选同一条规则。
- 表格对象起草：工作表有表格对象时按它出列表块：表头行、列范围、数据行、汇总行（映射成 total_row，标签取汇总行
第一列的文字）都由表格对象给，锚点仍按表头文字、不记坐标；表名优先用表格对象的名字（含数字的不用）。不调用 AI。
- CSV 走配方：新增 csv_grid，按内容认出 CSV（不是 zip、不是 OLE、开头没有 NUL），编码和分隔符复用 tabular 读 CSV
的那一段，适配成一张名为「CSV」的工作表（SheetScan.path=csv:，WorkbookScan.format=csv）；只有写法毫无歧义的
整数、小数转成数，千分位、前导零、日期写法一律留作文字交给配方的规则。坐标沿用「工作表!A1」（偏离细案的
R{行}C{列}，理由见 csv_grid 的模块说明）。CSV 不做稀疏判定。导入向导和简单上传的结构提示都放行 CSV。
- 每个列表一道题「每期怎么更新」（q_list_period）：每期替换 / 按期累积并按某个日期列核对 / 按期累积并加统计期
起止两列，后两种同时把导入模式改成按期累积、缺统计期的工作表一并配上（问题集合按草稿生成、回答时不变，草稿里
的列表还不符合累积资格，q_mode 不会出现，所以并成一道题）。确认项新增 subtotals:<块>、list_period:<块>。
- 前端配方面板：列表块多「小计行」（标签列、开头词、管哪些明细、分组列、另存）和「按期累积时每一行属于哪一期」；
回执「排除的行」认 subtotal。
验证：新增 tests/test_csv_grid.py、test_recipe_p5.py（执行器、核对、校验、资格、并集物化、起草）、
test_recipe_p5_e2e.py（经接口：三份月度 CSV 累积、部分重叠拒收、同一期替换要确认；带穿插小计的 Excel 起草、
核对、提交）。去掉小计识别、统计期回填、D 核对分别回跑，新测试 10 条失败。旧测试里「夹在中间的小计一律失败」
「列表只能每期替换」的断言按新行为改写。
1469 lines
78 KiB
Python
1469 lines
78 KiB
Python
"""框选转锚点（期 3，WP-2，P3-SPEC 第 4 节）：网格上框一块区域、选它是什么，换算成按文字认的配方规则。
|
||
|
||
**坐标不进配方**（4.1）：配方 schema 里没有任何坐标字段（extra="forbid"），框选的坐标只在换算的那一刻用两次——
|
||
从格子里取锚点文字（表头、行标签、分段标题、合计词、轴右侧表头、同一行的文字），以及换算完干跑后把执行器认出的
|
||
区域和框比对（replay_compare）。所以下个月行列挪了位置，配方照样按文字找到它（4.6 第 2 条用挪位的文件钉住）。
|
||
|
||
**换算规则全部复用起草器**（期 2 规格 6.1，评审二-m12）：列表直接用起草器的列表计划和块构造函数
|
||
（recipe_suggest.list_block），框只决定表头行和列范围，列名、单位、类型、grain、合计行、跳过空行、占位符、合并
|
||
单元格填充都由起草器按框内的数据推断；交叉表直接取规则起草对整张工作表的结果。换算不了就明确说为什么
|
||
（EditResult.ok=False，problems 的 code 见 4.3），不退回写死坐标，也不猜。
|
||
|
||
客户端传来的文字只有两种会进补丁：理由，以及经过名字校验、撞名检查的表名（options.table，评审一-m6）。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Callable
|
||
|
||
from pydantic import ValidationError
|
||
|
||
from app.data import recipe_suggest as S
|
||
from app.data.names import canon, collide_key, name_problem, to_sql_name
|
||
from app.data.recipe_fixes import (
|
||
REASON_MAX,
|
||
EditRequestError,
|
||
all_ids,
|
||
finish,
|
||
has_digit,
|
||
keep_ok,
|
||
ptr,
|
||
quoted,
|
||
table_index,
|
||
total_blockers,
|
||
total_ops,
|
||
unique_id,
|
||
unique_name,
|
||
)
|
||
from app.data.recipe_parsers import (
|
||
LABEL_MAX,
|
||
candidate_words,
|
||
hour_range,
|
||
hour_range_total,
|
||
looks_numeric_text,
|
||
match_key,
|
||
month_day_or_date,
|
||
split_unit_suffix,
|
||
text_label,
|
||
)
|
||
from app.data.recipe_types import (
|
||
RECIPE_FORMAT,
|
||
Anchor,
|
||
CrosstabBlock,
|
||
Draft,
|
||
EditResult,
|
||
Extraction,
|
||
Grid,
|
||
ListBlock,
|
||
Problem,
|
||
Recipe,
|
||
ReplayCompare,
|
||
Selection,
|
||
)
|
||
from app.data.xlsx_scan import cell_ref, col_letter, parse_ref
|
||
|
||
#: 交叉表：框与规则草稿认出的交叉表至少重叠这么多（占框的比例）才算框住了它（4.2）
|
||
CROSSTAB_OVERLAP = 0.5
|
||
#: 锚点的长度上限（契约 IgnoreRow.label / IgnoreOutside.anchor 40 字、ListBlock.after_title 40 字、IgnoreColumn.header 80 字）
|
||
_ANCHOR_MAX, _HEADER_MAX = 40, 80
|
||
|
||
Rect = tuple[int, int, int, int]
|
||
|
||
|
||
# ==========================================================================
|
||
# 小工具
|
||
# ==========================================================================
|
||
|
||
|
||
def _rect(ref: str) -> Rect:
|
||
"""「C5」或「C5:F8」→ (r1, c1, r2, c2)，左上到右下（反着框也行）。"""
|
||
a, _, b = ref.partition(":")
|
||
r1, c1 = parse_ref(a)
|
||
r2, c2 = parse_ref(b) if b else (r1, c1)
|
||
return min(r1, r2), min(c1, c2), max(r1, r2), max(c1, c2)
|
||
|
||
|
||
def _a1(rect: Rect | None) -> str | None:
|
||
if rect is None:
|
||
return None
|
||
r1, c1, r2, c2 = rect
|
||
return cell_ref(r1, c1) if (r1, c1) == (r2, c2) else f"{cell_ref(r1, c1)}:{cell_ref(r2, c2)}"
|
||
|
||
|
||
def _rows_text(r1: int, r2: int) -> str:
|
||
return f"第 {r1} 行" if r1 == r2 else f"第 {r1}–{r2} 行"
|
||
|
||
|
||
def _intersects(a: Rect, b: Rect) -> bool:
|
||
return a[0] <= b[2] and b[0] <= a[2] and a[1] <= b[3] and b[1] <= a[3]
|
||
|
||
|
||
def _area(a: Rect) -> int:
|
||
return (a[2] - a[0] + 1) * (a[3] - a[1] + 1)
|
||
|
||
|
||
def _overlap(a: Rect, b: Rect) -> int:
|
||
if not _intersects(a, b):
|
||
return 0
|
||
return (min(a[2], b[2]) - max(a[0], b[0]) + 1) * (min(a[3], b[3]) - max(a[1], b[1]) + 1)
|
||
|
||
|
||
def _bbox(rects: list[Rect]) -> Rect | None:
|
||
if not rects:
|
||
return None
|
||
return min(r[0] for r in rects), min(r[1] for r in rects), max(r[2] for r in rects), max(r[3] for r in rects)
|
||
|
||
|
||
def _cells_bbox(cells: set[tuple[int, int]] | list[tuple[int, int]]) -> Rect | None:
|
||
cells = list(cells)
|
||
if not cells:
|
||
return None
|
||
return min(r for r, _ in cells), min(c for _, c in cells), max(r for r, _ in cells), max(c for _, c in cells)
|
||
|
||
|
||
def _text_of(grid: Grid, r: int, c: int) -> str | None:
|
||
"""格子里的文字（去首尾空白）：不是文字、是公式或只有空白时 None。"""
|
||
cell = grid.get(r, c)
|
||
if cell is None or cell.formula is not None or not isinstance(cell.value, str):
|
||
return None
|
||
t = cell.value.strip()
|
||
return t or None
|
||
|
||
|
||
def _textual(grid: Grid, r: int, c: int) -> bool:
|
||
"""这一格能不能作表头文字：是文字、不像数、不是日期写法。"""
|
||
cell = grid.get(r, c)
|
||
if cell is None:
|
||
return True
|
||
if cell.formula is not None or not isinstance(cell.value, str):
|
||
return False
|
||
return not looks_numeric_text(cell.value) and not S._date_kind(cell.value)
|
||
|
||
|
||
def _row_empty(grid: Grid, r: int, c1: int, c2: int) -> bool:
|
||
return all(grid.get(r, c) is None for c in range(c1, c2 + 1))
|
||
|
||
|
||
def _reason(sel: Selection) -> str:
|
||
"""忽略类框选的理由（必填，1–200 字）。理由进配方哈希：预览和应用必须用同一个理由。"""
|
||
raw = sel.options.get("reason")
|
||
text = raw.strip() if isinstance(raw, str) else ""
|
||
if not text:
|
||
raise EditRequestError("reason_required", "忽略要写明理由")
|
||
if len(text) > REASON_MAX:
|
||
raise EditRequestError("edit_invalid", f"理由不能超过 {REASON_MAX} 字")
|
||
return text
|
||
|
||
|
||
@dataclass
|
||
class _Ctx:
|
||
recipe: dict[str, Any] | None
|
||
parsed: Recipe | None
|
||
grid: Grid
|
||
sel: Selection
|
||
rect: Rect
|
||
ext: Extraction | None
|
||
retired: dict[str, list[str]]
|
||
#: 配方里对应这张工作表的下标（没有为 None）
|
||
si: int | None = None
|
||
problems: list[Problem] = field(default_factory=list)
|
||
|
||
@property
|
||
def sheet(self) -> str:
|
||
return self.grid.sheet
|
||
|
||
def coord(self, r: int, c: int) -> str:
|
||
return f"{self.sheet}!{cell_ref(r, c)}"
|
||
|
||
def fail(self, code: str, message: str, cells: list[str] | None = None) -> None:
|
||
box = f"{self.sheet}!{_a1(self.rect)}"
|
||
self.problems.append(Problem(code, "recipe", message, cells=(cells or [box])[:20], model_message=code))
|
||
|
||
|
||
def _failed(ctx: _Ctx) -> EditResult:
|
||
return EditResult(ok=False, kind="selection", key=f"{ctx.sel.as_}:", title=_TITLES.get(ctx.sel.as_, "框选"),
|
||
summary=[], ops=[], anchors=[], notes=[], problems=list(ctx.problems))
|
||
|
||
|
||
_TITLES = {"list": "框选为列表", "crosstab": "框选为交叉表", "segment": "框选为分段", "derived": "框选为合计行",
|
||
"section_title": "框选为分段标题", "ignore_rows": "忽略框选的行", "ignore_columns": "忽略框选的列",
|
||
"ignore_outside": "忽略框选行里导入区域之外的数字"}
|
||
|
||
|
||
def _sheet_of(recipe: dict[str, Any] | None, grid: Grid, ext: Extraction | None) -> int | None:
|
||
"""配方里对应这张网格的工作表：干跑认出的（sheets.matched：工作表 id → 实际名）优先，其次按名字。"""
|
||
if not recipe:
|
||
return None
|
||
sheets = recipe.get("sheets") or []
|
||
if ext is not None:
|
||
for sid, actual in ext.sheets.matched.items():
|
||
if match_key(actual) == match_key(grid.sheet):
|
||
hit = next((i for i, s in enumerate(sheets) if s.get("id") == sid), None)
|
||
if hit is not None:
|
||
return hit
|
||
key = match_key(grid.sheet)
|
||
return next((i for i, s in enumerate(sheets) if match_key((s.get("match") or {}).get("name")) == key), None)
|
||
|
||
|
||
@dataclass
|
||
class _Located:
|
||
bi: int
|
||
block: dict[str, Any]
|
||
rects: list[Rect]
|
||
cross: Any = None # recipe_suggest._CrossLoc
|
||
lst: Any = None # recipe_suggest._ListLoc
|
||
|
||
|
||
def _locate(ctx: _Ctx) -> list[_Located]:
|
||
"""这张工作表上配方各块认领的区域。干跑结果带块 id（RegionMark.block，期 3）时按它取；否则（期 2 的执行器、
|
||
没有干跑）用起草器的定位函数在网格上近似。交叉表另外总是用起草器定位一遍：标签列、轴行、各分段的行要用到。"""
|
||
if ctx.si is None or ctx.parsed is None:
|
||
return []
|
||
sheet = ctx.recipe["sheets"][ctx.si] # type: ignore[index]
|
||
model = ctx.parsed.sheets[ctx.si]
|
||
tagged: dict[str, list[Rect]] = {}
|
||
if ctx.ext is not None and any(m.block for m in ctx.ext.regions):
|
||
for m in ctx.ext.regions:
|
||
if m.block and match_key(m.sheet) == match_key(ctx.sheet):
|
||
tagged.setdefault(m.block, []).append(_rect(m.ref))
|
||
out: list[_Located] = []
|
||
# 已用的表头行只对表头相同的块排除（上下两张同表头的表各认各的）；左右并排的两张表表头在同一行，不能互相排除
|
||
taken: dict[frozenset[str], set[int]] = {}
|
||
for bi, (bd, bm) in enumerate(zip(sheet.get("blocks") or [], model.blocks)):
|
||
loc = _Located(bi, bd, list(tagged.get(bm.id, [])))
|
||
if isinstance(bm, CrosstabBlock):
|
||
loc.cross = S._locate_crosstab(bm, ctx.grid)
|
||
if not loc.rects and loc.cross is not None:
|
||
box = _cells_bbox(loc.cross.region)
|
||
loc.rects = [box] if box else []
|
||
elif isinstance(bm, ListBlock):
|
||
keys = frozenset(match_key(c.header) for c in bm.columns)
|
||
loc.lst = S._locate_list(bm, ctx.grid, taken.setdefault(keys, set()))
|
||
if loc.lst is not None:
|
||
taken[keys].update(loc.lst.header_rows)
|
||
if not loc.rects:
|
||
box = _cells_bbox(loc.lst.region)
|
||
loc.rects = [box] if box else []
|
||
out.append(loc)
|
||
return out
|
||
|
||
|
||
def _crosstab_at(ctx: _Ctx, located: list[_Located]) -> _Located | None:
|
||
"""框落在哪个交叉表块的标签列上：框含标签列、不越过标签列往左、在日期表头之下、不越过最后一个日期往右。"""
|
||
r1, c1, r2, c2 = ctx.rect
|
||
for loc in located:
|
||
x = loc.cross
|
||
if x is None:
|
||
continue
|
||
if c1 == x.label_col and c2 <= x.c2 and r1 > x.axis_row:
|
||
return loc
|
||
return None
|
||
|
||
|
||
def _labels(ctx: _Ctx, col: int) -> list[tuple[int, str]] | None:
|
||
"""框内各行标签列的文字；有行没有文字标签时报 selection_label_blank 并返回 None。"""
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
out: list[tuple[int, str]] = []
|
||
blank: list[int] = []
|
||
for r in range(r1, r2 + 1):
|
||
t = _text_of(ctx.grid, r, col)
|
||
if t is None:
|
||
blank.append(r)
|
||
else:
|
||
out.append((r, t))
|
||
if blank:
|
||
ctx.fail("selection_label_blank",
|
||
f"{'、'.join(f'第 {r} 行' for r in blank[:8])}的标签列（{col_letter(col)} 列）没有文字，无法按标签认：请调整框的上下范围，"
|
||
"或改选别的类型", [ctx.coord(r, col) for r in blank])
|
||
return None
|
||
return out
|
||
|
||
|
||
def _new_table_name(ctx: _Ctx, want: str, used: set[str]) -> str | None:
|
||
"""新表名：客户端给了 options.table 时过名字校验和撞名检查（含当前版本里退役的表名，评审一-m6），否则按 want
|
||
推出并加 _2 去重。不合法或撞名时报问题、返回 None。"""
|
||
given = ctx.sel.options.get("table")
|
||
if given is None or given == "":
|
||
return unique_name(to_sql_name(want, fallback=f"表{len(used) + 1}") if want else f"表{len(used) + 1}", used)
|
||
if not isinstance(given, str):
|
||
raise EditRequestError("edit_invalid", "表名要是文字")
|
||
why = name_problem(given)
|
||
if why:
|
||
ctx.fail("selection_name_invalid", f"表名不能用：{why}")
|
||
return None
|
||
if collide_key(given) in used:
|
||
ctx.fail("selection_name_taken", f"表名「{given}」与已有的表或已退役的表重名（名字不区分大小写和全角半角），请换一个名字")
|
||
return None
|
||
used.add(collide_key(given))
|
||
return given
|
||
|
||
|
||
def _used_tables(ctx: _Ctx, *, except_: set[str] | frozenset[str] = frozenset()) -> set[str]:
|
||
"""已用的表名（collide_key）：工作配方里的表和当前版本里退役的表；except_ 里的（被替换的块自己的表）不算。"""
|
||
used = {collide_key(t.get("name", "")) for t in (ctx.recipe or {}).get("tables") or []
|
||
if t.get("name") not in except_}
|
||
used |= {collide_key(t) for t in ctx.retired if t not in except_}
|
||
return used
|
||
|
||
|
||
# ==========================================================================
|
||
# 列表
|
||
# ==========================================================================
|
||
|
||
|
||
def _header_problems(ctx: _Ctx, rows: list[int], cols: list[int]) -> dict[int, str] | None:
|
||
"""表头：每列按执行器的拼法得到（合并区取左上格、去掉空的和与上一层相同的、用「_」拼接）。任一列为空、
|
||
是数字或日期、两列 match_key 相同，各报一条（4.2 列表第 1 条）。"""
|
||
texts = S._header_texts(ctx.grid, rows, cols)
|
||
# 空：表头那几行这一列一格都没有（是数字、日期的另报 not_text，不算空）
|
||
blank = [c for c in cols if c not in texts and all(ctx.grid.get(r, c) is None for r in rows)]
|
||
if blank:
|
||
ctx.fail("selection_header_blank",
|
||
f"{_rows_text(rows[0], rows[-1])}的 {'、'.join(col_letter(c) for c in blank)} 列表头为空：列表的每一列都要有表头。"
|
||
"请调整框的左右范围，或改选「忽略这些列」", [ctx.coord(rows[-1], c) for c in blank])
|
||
bad = sorted({c for c in cols for r in rows if not _textual(ctx.grid, r, c)})
|
||
if bad:
|
||
ctx.fail("selection_header_not_text",
|
||
f"{'、'.join(col_letter(c) for c in bad)} 列的表头是数字或日期，不像表头：框的第一行应是表头。"
|
||
"请把框的上边移到表头所在的行", [ctx.coord(rows[0], c) for c in bad])
|
||
seen: dict[str, int] = {}
|
||
dup: list[int] = []
|
||
for c in cols:
|
||
if c in texts:
|
||
k = match_key(texts[c])
|
||
if k in seen:
|
||
dup.append(c)
|
||
seen.setdefault(k, c)
|
||
if dup:
|
||
ctx.fail("selection_header_duplicate",
|
||
f"表头{quoted([texts[c] for c in dup])}出现了不止一次：同一张列表里表头不能重复。请调整框的左右范围",
|
||
[ctx.coord(rows[-1], c) for c in dup])
|
||
return None if (blank or bad or dup) else texts
|
||
|
||
|
||
def _header_candidates(grid: Grid, headers: list[str], n: int) -> list[int]:
|
||
"""整张网格上像这组表头的行（执行器的候选规则：过半表头命中）。"""
|
||
want = {match_key(h) for h in headers}
|
||
out = []
|
||
for t in grid.row_numbers():
|
||
rows = list(range(t, t + n))
|
||
cols = sorted({c for r in rows for c, _ in grid.row(r)})
|
||
keys = {match_key(x) for x in S._header_texts(grid, rows, cols).values()}
|
||
if len(want & keys) * 2 > len(want):
|
||
out.append(t)
|
||
return out
|
||
|
||
|
||
def _after_title(ctx: _Ctx, top: int, cols: list[int], headers: list[str], n: int) -> tuple[str, str] | None | bool:
|
||
"""表头不止一处时找 after_title：表头行往上、块的列范围内最近的、整张表里唯一的、不含数字的文字格（4.2 第 2 条，
|
||
评审一-M11：最近的唯一文字格常常是「2026年8月 销售月报」这种带年月的标题，写进配方等于写死期次）。
|
||
返回 (文字, 坐标)；不需要时 None；需要但找不到时 False。"""
|
||
others = [t for t in _header_candidates(ctx.grid, headers, n) if t != top]
|
||
if not others:
|
||
return None
|
||
above = [t for t in others if t < top]
|
||
floor = 1
|
||
if above:
|
||
# 上面那张表到它下方第一个空行为止：标题只在那之后找，不能拿上一张表的数据格当标题
|
||
floor = max(above) + n
|
||
while floor < top and not _row_empty(ctx.grid, floor, cols[0], cols[-1]):
|
||
floor += 1
|
||
counts: dict[str, int] = {}
|
||
for (r, c), cell in ctx.grid.cells.items():
|
||
if isinstance(cell.value, str) and cell.formula is None:
|
||
counts[match_key(cell.value)] = counts.get(match_key(cell.value), 0) + 1
|
||
for r in range(top - 1, floor - 1, -1):
|
||
for c in cols:
|
||
t = _text_of(ctx.grid, r, c)
|
||
if t and len(t) <= _ANCHOR_MAX and not has_digit(t) and counts.get(match_key(t)) == 1:
|
||
return t, ctx.coord(r, c)
|
||
return False
|
||
|
||
|
||
def _no_extra_columns(ctx: _Ctx, rows: list[int], cols: list[int], ignored: set[str]) -> bool:
|
||
"""执行器按表头行里包含锚点的最大连续非空段认列范围（期 2 规格 4.4）：框的左右紧挨着还有非空表头时，重放会把
|
||
它们也算进这张表（多出的列默认拒收），按文字无法表达「到这一列为止」，报 selection_extra_columns（4.2 第 3 条）。
|
||
被替换的块上已经按表头忽略的列不算（ignored：表头的 match_key）。"""
|
||
lo, hi = cols[0], cols[-1]
|
||
wide = S._header_texts(ctx.grid, rows, list(range(max(1, lo - 60), hi + 61)))
|
||
left, right = lo, hi
|
||
while left - 1 in wide:
|
||
left -= 1
|
||
while right + 1 in wide:
|
||
right += 1
|
||
extra = [c for c in range(left, right + 1) if (c < lo or c > hi) and match_key(wide[c]) not in ignored]
|
||
if not extra:
|
||
return True
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
ctx.fail("selection_extra_columns",
|
||
f"框的旁边紧挨着还有表头{quoted([wide[c] for c in extra])}（{'、'.join(col_letter(c) for c in extra)} 列）："
|
||
f"按文字定位时它们会被算进这张表。请把框扩大到 {_a1((r1, left, r2, right))}，或改选「忽略这些列」",
|
||
[ctx.coord(rows[-1], c) for c in extra])
|
||
return False
|
||
|
||
|
||
def _bottom_box(ctx: _Ctx, plan: Any, first: int) -> bool:
|
||
"""bottom="box"：框的下边就是数据的末尾（4.2 第 5 条）。合计行照起草器的规则认（期 2 规格 6.1 第 11 条）。
|
||
|
||
- 框内最后一个非空行 last 的第一列以合计词开头、上面还有数据行：它是合计行（rows.total_row），不是数据行。
|
||
框把合计行也框进来是很自然的操作，当成数据行导入的话求和翻倍，而且干跑、核对、重放比对都看不出来（H2）；
|
||
- 合计字样出现在框内别的位置（第一列但不在末尾，或在别的列上）：起草器在这里记起草失败，框选同样报错，
|
||
不把合计当明细导入；
|
||
- 合计行下面紧接着还有同一张表的行：它是夹在中间的小计（起草器的 _more_rows），同样报错；
|
||
- 末尾不是合计行时，看 last 的下一行（只看框的列范围）：全空或到网格底部就「遇到空行停止」，以合计词开头
|
||
就是合计行，都不是就报 selection_bottom_not_anchorable。看的是 last+1、不是框的下一行：框底多带几行空行
|
||
时结果与框到 last 相同，提示里建议的末尾行也从 last+1 往下数，不会让人把下方的备注、制表人行也框进来。"""
|
||
r1, c1, r2, c2 = ctx.rect
|
||
grid = ctx.grid
|
||
filled = [r for r in range(first, r2 + 1) if not _row_empty(grid, r, c1, c2)]
|
||
if not filled:
|
||
ctx.fail("selection_bottom_not_anchorable", "框里只有表头、没有数据行：请把框扩大到数据的实际末尾")
|
||
return False
|
||
last = filled[-1]
|
||
word = S._total_word(S._label_of(grid.get(last, c1)))
|
||
data = filled[:-1] if word is not None else filled
|
||
if not data:
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"框里只有表头和合计行（第 {last} 行），没有数据行：请把框扩大到数据所在的行")
|
||
return False
|
||
# 期 5：框里夹在中间、第一列以合计字样开头、下面紧接着还有同一张表的明细的行（起草器 _scan_block 的同一条
|
||
# 规则），照起草器提议成小计（_settle_subtotals）；提议不成的照旧报错
|
||
mid = [r for r in data[1:] if S._total_word(S._label_of(grid.get(r, c1))) is not None]
|
||
rest = [r for r in data if r not in mid]
|
||
if mid and rest and not any(grid.get(r, c) is not None and S._total_word(grid.get(r, c).value) is not None
|
||
for r in data for c in range(c1 + 1, c2 + 1)):
|
||
trial = S._ListPlan(header_rows=plan.header_rows, cols=plan.cols, headers=plan.headers, rows=rest,
|
||
subtotal_rows=list(mid),
|
||
subtotal_words=[S._total_word(S._label_of(grid.get(r, c1))) or "" for r in mid])
|
||
types = S._types_of(grid, rest, plan.cols)
|
||
if all(S._more_rows(grid, r, trial, S._bottom(grid), types) for r in mid):
|
||
S._settle_subtotals(grid, trial)
|
||
if trial.subtotal_word is not None:
|
||
plan.subtotal_rows, plan.subtotal_word = trial.subtotal_rows, trial.subtotal_word
|
||
data = [r for r in data if r not in mid]
|
||
if not _no_stray_totals(ctx, data, c1, c2):
|
||
return False
|
||
bottom = S._bottom(grid)
|
||
acc = S._ColTypes(grid, plan.cols)
|
||
for r in data:
|
||
acc.add(r)
|
||
plan.rows = data
|
||
gaps = [r for r in range(first, last + 1) if r not in filled]
|
||
if gaps:
|
||
# 框内有空行（c08 gap 那种）：跳过空行，执行器跳过空行后照样能碰到末尾的合计行；重放比对把实际的数据区摆出来
|
||
plan.skip_blank, plan.skipped = True, gaps
|
||
if word is not None:
|
||
return _total_at_end(ctx, plan, last, word, bottom, acc.types())
|
||
below = last + 1
|
||
if below > bottom or _row_empty(grid, below, c1, c2):
|
||
return True
|
||
word = S._total_word(S._label_of(grid.get(below, c1)))
|
||
if word is not None:
|
||
return _total_at_end(ctx, plan, below, word, bottom, acc.types())
|
||
end = below
|
||
while end <= bottom and not _row_empty(grid, end, c1, c2) and S._total_word(S._label_of(grid.get(end, c1))) is None:
|
||
end += 1
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"第 {below} 行还有数据：按文字无法表达「到第 {last} 行为止」。请把框扩大到数据的实际末尾（第 {end - 1} 行），"
|
||
"改选「下边界按规则推断」，或删掉文件里多出的行", [ctx.coord(below, c1)])
|
||
return False
|
||
|
||
|
||
def _no_stray_totals(ctx: _Ctx, data: list[int], c1: int, c2: int) -> bool:
|
||
"""数据行里不能有合计字样（起草器 _scan_block 的 stray_totals：第一列以合计词开头、或别的列整格以合计词开头）。
|
||
起草器在这里记起草失败；框选照样不收，否则合计会被当成明细导入、求和翻倍（H2）。"""
|
||
grid = ctx.grid
|
||
first_col = [r for r in data if S._total_word(S._label_of(grid.get(r, c1))) is not None]
|
||
other = [(r, c) for r in data for c in range(c1 + 1, c2 + 1)
|
||
if (cell := grid.get(r, c)) is not None and S._total_word(cell.value) is not None]
|
||
if first_col:
|
||
r = first_col[0]
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"第 {r} 行像合计行（「{S._label_of(grid.get(r, c1))}」），但它不在框的末尾：无法确定它汇总的是哪几行，"
|
||
f"按数据导入会把合计重复计数。合计行就是表的末尾时，请把框的下边移到第 {r} 行；它是夹在中间的小计时，"
|
||
"请修改文件", [ctx.coord(x, c1) for x in first_col])
|
||
return False
|
||
if other:
|
||
r, c = other[0]
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"第 {r} 行 {col_letter(c)} 列是合计字样（「{grid.get(r, c).value}」），但合计字样不在第一列，无法按合计行"
|
||
"核对，按数据导入会把合计重复计数：请修改文件，把合计字样写在第一列，或把框的下边移到这一行之上",
|
||
[ctx.coord(x, y) for x, y in other])
|
||
return False
|
||
return True
|
||
|
||
|
||
def _auto_settled(ctx: _Ctx, plan: Any) -> bool:
|
||
"""bottom="auto"：起草器收进来、却无法确定的行（紧挨着数据的新表头、不在末尾第一列的合计字样）照起草器记失败
|
||
（recipe_suggest._list_failures，同一套文案）。起草器的 _scan_block 会把这些行照样收进数据行，只是另记失败；
|
||
框选不看失败的话，夹在中间的小计会被当成明细导入、求和翻倍（H2）。"""
|
||
failures = S._list_failures(ctx.sheet, ctx.grid, plan)
|
||
if not failures:
|
||
return True
|
||
cells = [ctx.coord(r, c) for r, c in plan.stray_totals] + [ctx.coord(r, plan.cols[0]) for r, _f in plan.adjacent]
|
||
for f in failures:
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"{f.human}。按文字无法表达这张表到哪里为止，按数据导入会出错：请修改文件", cells)
|
||
return False
|
||
|
||
|
||
def _total_at_end(ctx: _Ctx, plan: Any, r: int, word: str, bottom: int, types: dict[int, str]) -> bool:
|
||
"""第 r 行是表末尾的合计行：下面紧接着还有同一张表的行时，它是夹在中间的小计（起草器的 _more_rows），
|
||
按合计行停在这里会把下面的明细丢掉，报错；否则记为合计行（执行器碰到它就停）。"""
|
||
if S._more_rows(ctx.grid, r, plan, bottom, types):
|
||
ctx.fail("selection_bottom_not_anchorable",
|
||
f"第 {r} 行像合计行（「{S._label_of(ctx.grid.get(r, plan.cols[0]))}」），但下面紧接着还有同一张表的数据："
|
||
"它是夹在中间的小计，按文字无法表达这张表到哪里为止，按数据导入又会把小计重复计数。请修改文件",
|
||
[ctx.coord(r, plan.cols[0])])
|
||
return False
|
||
plan.total_row, plan.total_word = r, word
|
||
return True
|
||
|
||
|
||
def _select_list(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
r1, c1, r2, c2 = ctx.rect
|
||
grid = ctx.grid
|
||
n = ctx.sel.options.get("header_rows", 1)
|
||
if not isinstance(n, int) or isinstance(n, bool) or not 1 <= n <= 3:
|
||
raise EditRequestError("edit_invalid", "表头行数只能是 1 到 3")
|
||
bottom = ctx.sel.options.get("bottom", "box")
|
||
if bottom not in ("box", "auto"):
|
||
raise EditRequestError("edit_invalid", "下边界只能「以框的下边为准」或「按规则推断」")
|
||
rows_h = list(range(r1, r1 + n))
|
||
cols = list(range(c1, c2 + 1))
|
||
if len(cols) < 1 or r2 < r1 + n:
|
||
ctx.fail("selection_bottom_not_anchorable", "框里只有表头、没有数据行：请把框扩大到数据的实际末尾")
|
||
return _failed(ctx)
|
||
# 重叠：恰好一个块时替换它（沿用块 id 和表名，免得关系、单位里的引用断掉），两个以上报错，不相交时新增
|
||
hits = [loc for loc in located if any(_intersects(ctx.rect, r) for r in loc.rects)]
|
||
if len(hits) > 1:
|
||
ctx.fail("selection_overlap", f"框与{quoted([h.block.get('id') for h in hits])}几块都相交：请只框一张表")
|
||
return _failed(ctx)
|
||
old = hits[0] if hits else None
|
||
if old is not None and old.block.get("layout") != "list":
|
||
ctx.fail("selection_overlap", f"框与交叉表「{old.block.get('id')}」相交：列表不能替换交叉表。请调整框的范围")
|
||
return _failed(ctx)
|
||
texts = _header_problems(ctx, rows_h, cols)
|
||
if texts is None:
|
||
return _failed(ctx)
|
||
ignored = {match_key(x.get("header")) for x in (old.block.get("ignore_columns") or [])} if old is not None else set()
|
||
if not _no_extra_columns(ctx, rows_h, cols, ignored):
|
||
return _failed(ctx)
|
||
plan = S._ListPlan(header_rows=rows_h, cols=cols, headers=dict(texts))
|
||
notes: list[str] = []
|
||
if bottom == "box":
|
||
if not _bottom_box(ctx, plan, r1 + n):
|
||
return _failed(ctx)
|
||
else:
|
||
S._scan_block(grid, plan, S._bottom(grid))
|
||
if not plan.rows:
|
||
ctx.fail("selection_bottom_not_anchorable", "表头之下没有认出数据行：请检查框的上边是否在表头所在的行")
|
||
return _failed(ctx)
|
||
if not _auto_settled(ctx, plan):
|
||
return _failed(ctx)
|
||
last = plan.total_row or plan.rows[-1]
|
||
if grid.truncated and last >= S._bottom(grid):
|
||
notes.append("数据延续到已读入范围之外，完整范围在试运行时核对")
|
||
if grid.truncated and plan.rows[-1] >= S._bottom(grid) and not notes:
|
||
notes.append("数据延续到已读入范围之外，完整范围在试运行时核对")
|
||
S._finish_block(grid, plan)
|
||
old_tables = set()
|
||
if old is not None:
|
||
old_tables = {old.block.get("table")}
|
||
tr = (old.block.get("rows") or {}).get("total_row") or {}
|
||
if tr.get("keep_as"):
|
||
old_tables.add(tr["keep_as"])
|
||
used = _used_tables(ctx, except_=old_tables)
|
||
given = ctx.sel.options.get("table")
|
||
table: str | None = None
|
||
if given not in (None, ""):
|
||
table = _new_table_name(ctx, "", used)
|
||
if table is None:
|
||
return _failed(ctx)
|
||
elif old is not None:
|
||
table = old.block.get("table")
|
||
block_id = old.block["id"] if old is not None else unique_id(f"列表{_list_count(ctx.recipe) + 1}",
|
||
all_ids(ctx.recipe or {}))
|
||
block, main, total_spec = S.list_block(ctx.sheet, plan, block_id=block_id, tables_used=used, table=table)
|
||
# 表头上方的标题：表头不止一处时才要
|
||
anchors = [Anchor("header", texts[c], ctx.coord(r1 + n - 1, c)) for c in cols]
|
||
found = _after_title(ctx, r1, cols, [texts[c] for c in cols], n)
|
||
if found is False:
|
||
ctx.fail("selection_anchor_ambiguous",
|
||
"这组表头在工作表里不止一处，而上方能区分这张表的文字都含年月或数字，下一期很可能对不上：请修改文件，"
|
||
"或给这张表加一行不含数字的标题")
|
||
return _failed(ctx)
|
||
if found:
|
||
block["after_title"] = found[0]
|
||
anchors.append(Anchor("after_title", found[0], found[1]))
|
||
if plan.total_row is not None:
|
||
anchors.append(Anchor("total_word", plan.total_word or "", ctx.coord(plan.total_row, c1)))
|
||
for c in cols:
|
||
if S._YEAR.search(texts[c]):
|
||
notes.append(f"表头「{texts[c]}」含年份，下一期年份变化时将无法匹配")
|
||
if old is not None and old.block.get("ignore_columns"):
|
||
# 被替换的块上原有的按表头忽略照旧保留：框选只决定表头行和列范围，不该悄悄丢掉人确认过的规则
|
||
block["ignore_columns"] = list(old.block["ignore_columns"])
|
||
for col in block["columns"]:
|
||
retired = next((x for x in ctx.retired.get(block["table"], []) if collide_key(x) == collide_key(col["name"])
|
||
and x != col["name"]), None)
|
||
if retired is not None:
|
||
notes.append(f"列「{col['name']}」与已退役的列「{retired}」同名（不区分大小写），沿用原来的写法「{retired}」")
|
||
col["name"] = retired
|
||
ops = _block_ops(ctx, old, block, main, total_spec, old_tables)
|
||
if ops is None:
|
||
return _failed(ctx)
|
||
total_rect = (plan.total_row, c1, plan.total_row, c2) if plan.total_row else None
|
||
expected = {"header": _a1((r1, c1, r1 + n - 1, c2)), "data": _a1((plan.rows[0], c1, plan.rows[-1], c2)),
|
||
"total": _a1(total_rect)}
|
||
verb = "替换现有的" if old is not None else "新增"
|
||
summary = [f"按框选{verb}列表「{block_id}」（表「{block['table']}」）：表头按文字{quoted([texts[c] for c in cols])}定位"]
|
||
if found:
|
||
summary.append(f"只在文字「{found[0]}」之后找表头")
|
||
summary.append("下边界：" + ("到合计行为止" if plan.total_row else "遇到空行停止")
|
||
+ ("，中间的空行跳过" if plan.skip_blank else ""))
|
||
if bottom == "auto":
|
||
summary.append(f"重放的数据区到第 {plan.rows[-1]} 行")
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"list:{block_id}",
|
||
title=f"按框选{verb}列表「{block_id}」：表头按文字定位", summary=summary, anchors=anchors,
|
||
notes=notes, expected=expected, block=block_id)
|
||
|
||
|
||
def _list_count(recipe: dict[str, Any] | None) -> int:
|
||
return sum(1 for s in (recipe or {}).get("sheets") or [] for b in s.get("blocks") or [] if b.get("layout") == "list")
|
||
|
||
|
||
def _block_ops(ctx: _Ctx, old: _Located | None, block: dict, main: dict, total_spec: dict | None,
|
||
old_tables: set[str]) -> list[dict[str, Any]] | None:
|
||
"""把一个列表块写进配方：替换旧块（连同表定义）、新增块，或者配方为空时用根替换新建配方（4.2 第 7 条）。"""
|
||
recipe = ctx.recipe
|
||
if not recipe:
|
||
tables = [main] + ([total_spec] if total_spec else [])
|
||
return [{"op": "replace", "path": "", "value": {
|
||
"recipe_format": RECIPE_FORMAT, "sheets": [{"id": "s1", "match": {"name": ctx.sheet[:31]}, "blocks": [block]}],
|
||
"tables": tables, "relations": []}}]
|
||
if ctx.si is None:
|
||
sid = _new_sheet_id(recipe)
|
||
ops: list[dict[str, Any]] = [{"op": "add", "path": "/sheets/-", "value": {
|
||
"id": sid, "match": {"name": ctx.sheet[:31]}, "blocks": [block]}}]
|
||
return ops + _add_tables(recipe, [main] + ([total_spec] if total_spec else []))
|
||
if old is None:
|
||
ops = [{"op": "add", "path": ptr("sheets", ctx.si, "blocks", "-"), "value": block}]
|
||
return ops + _add_tables(recipe, [main] + ([total_spec] if total_spec else []))
|
||
# 替换：表定义原位替换（保留用户写的说明），表名变了就是改名；旧块另存的合计表这次不再另存时删掉它的定义
|
||
# （别的块还在写入时不删）。先替换、再删（按下标从大到小）、最后追加，下标不会错位
|
||
ops = [{"op": "replace", "path": ptr("sheets", ctx.si, "blocks", old.bi), "value": block}]
|
||
old_total = ((old.block.get("rows") or {}).get("total_row") or {}).get("keep_as")
|
||
replaced: set[int] = set()
|
||
adds: list[dict] = []
|
||
for spec, was in ((main, old.block.get("table")), (total_spec, old_total)):
|
||
if spec is None:
|
||
continue
|
||
ti = table_index(recipe, was) if was else None
|
||
if ti is None:
|
||
adds.append(spec)
|
||
continue
|
||
replaced.add(ti)
|
||
ops.append({"op": "replace", "path": ptr("tables", ti),
|
||
"value": {**spec, "note": recipe["tables"][ti].get("note") or ""}})
|
||
removes = sorted({ti for name in old_tables if (ti := table_index(recipe, name)) is not None
|
||
and ti not in replaced and not _written_elsewhere(recipe, name, ctx.si, old.bi)}, reverse=True)
|
||
ops += [{"op": "remove", "path": ptr("tables", ti)} for ti in removes]
|
||
ops += [{"op": "add", "path": "/tables/-", "value": spec} for spec in adds]
|
||
return ops
|
||
|
||
|
||
def _add_tables(recipe: dict, specs: list[dict]) -> list[dict[str, Any]]:
|
||
return [{"op": "add", "path": "/tables/-", "value": s} for s in specs]
|
||
|
||
|
||
def _new_sheet_id(recipe: dict) -> str:
|
||
have = {s.get("id") for s in recipe.get("sheets") or []}
|
||
n = len(have) + 1
|
||
while f"s{n}" in have:
|
||
n += 1
|
||
return f"s{n}"
|
||
|
||
|
||
def _written_elsewhere(recipe: dict, table: str, si: int, bi: int) -> bool:
|
||
"""除了 (si, bi) 这个块，还有别的块写入这张表吗（替换块时不能删掉别人还在用的表定义）。"""
|
||
for sj, sheet in enumerate(recipe.get("sheets") or []):
|
||
for bj, block in enumerate(sheet.get("blocks") or []):
|
||
if (sj, bj) == (si, bi):
|
||
continue
|
||
if block.get("layout") == "list":
|
||
tr = (block.get("rows") or {}).get("total_row") or {}
|
||
if table in (block.get("table"), tr.get("keep_as")):
|
||
return True
|
||
for seg in block.get("segments") or []:
|
||
if table in (seg.get("table"), (seg.get("keep_as") or {}).get("table")):
|
||
return True
|
||
return False
|
||
|
||
|
||
# ==========================================================================
|
||
# 交叉表（整块）
|
||
# ==========================================================================
|
||
|
||
|
||
def _block_tables(block: dict) -> set[str]:
|
||
out: set[str] = set()
|
||
for seg in block.get("segments") or []:
|
||
if seg.get("table"):
|
||
out.add(seg["table"])
|
||
if (seg.get("keep_as") or {}).get("table"):
|
||
out.add(seg["keep_as"]["table"])
|
||
return out
|
||
|
||
|
||
def _rel_tables(rel: dict) -> set[str]:
|
||
if rel.get("kind") == "sum_eq":
|
||
return {rel.get("table")}
|
||
if rel.get("kind") == "not_comparable":
|
||
return {(rel.get("a") or {}).get("table"), (rel.get("b") or {}).get("table")}
|
||
return set()
|
||
|
||
|
||
def _select_crosstab(ctx: _Ctx, located: list[_Located], draft_fn: Callable[[], Draft] | None) -> EditResult:
|
||
"""框只用来指认是哪一块：draft_fn() 对整张工作表起草，取认领区域与框重叠最多（且 ≥50%）的交叉表块，连同它写入的表、
|
||
它认领的事实对应的关系、统计期上下文（配方里还没有时）一起并入。关系 id 与已有的重号时，后面的顺延（4.2）。"""
|
||
d = draft_fn() if draft_fn is not None else None
|
||
dr = d.recipe if d is not None else None
|
||
no = ("框住的范围里没有认出交叉表（日期表头加行标签）：请框在交叉表上，或改选「列表」")
|
||
if not dr:
|
||
ctx.fail("selection_no_crosstab", no)
|
||
return _failed(ctx)
|
||
try:
|
||
dmodel = Recipe.model_validate(dr)
|
||
except ValidationError:
|
||
ctx.fail("selection_no_crosstab", no)
|
||
return _failed(ctx)
|
||
best: tuple[int, dict, dict, Any] | None = None
|
||
for dsi, dsheet in enumerate(dmodel.sheets):
|
||
if match_key(dsheet.match.name) != match_key(ctx.sheet):
|
||
continue
|
||
for dbi, dblock in enumerate(dsheet.blocks):
|
||
if not isinstance(dblock, CrosstabBlock):
|
||
continue
|
||
loc = S._locate_crosstab(dblock, ctx.grid)
|
||
box = _cells_bbox(loc.region) if loc is not None else None
|
||
if box is None:
|
||
continue
|
||
ov = _overlap(ctx.rect, box)
|
||
if ov * 1.0 >= CROSSTAB_OVERLAP * _area(ctx.rect) and (best is None or ov > best[0]):
|
||
best = (ov, dr["sheets"][dsi], dr["sheets"][dsi]["blocks"][dbi], loc)
|
||
if best is None:
|
||
ctx.fail("selection_no_crosstab", no)
|
||
return _failed(ctx)
|
||
_ov, dsheet_d, dblock_d, dloc = best
|
||
region = _cells_bbox(dloc.region)
|
||
hits = [loc for loc in located if any(_intersects(region, r) for r in loc.rects)] # type: ignore[arg-type]
|
||
crosses = [loc for loc in located if loc.block.get("layout") == "crosstab"]
|
||
if len(hits) > 1 or (hits and hits[0].block.get("layout") != "crosstab"):
|
||
ctx.fail("selection_overlap", f"这块交叉表与{quoted([h.block.get('id') for h in hits])}相交：请先调整或删掉那些块")
|
||
return _failed(ctx)
|
||
old = hits[0] if hits else None
|
||
if old is None and crosses:
|
||
ctx.fail("selection_overlap",
|
||
f"这个工作表已经有交叉表「{crosses[0].block.get('id')}」：一个工作表只能有一个交叉表，请框在它上面以替换它")
|
||
return _failed(ctx)
|
||
recipe = ctx.recipe
|
||
block = copy.deepcopy(dblock_d)
|
||
btables = _block_tables(block)
|
||
rels = [copy.deepcopy(r) for r in dr.get("relations") or [] if _rel_tables(r) and _rel_tables(r) <= btables]
|
||
tables = [copy.deepcopy(t) for t in dr.get("tables") or [] if t.get("name") in btables]
|
||
# 被替换的交叉表只由它自己写入的表，连同引用这些表的关系一起换掉
|
||
drop = ({t for t in _block_tables(old.block) if not _written_elsewhere(recipe, t, ctx.si, old.bi)}
|
||
if recipe and old is not None and ctx.si is not None else set())
|
||
if recipe and old is not None:
|
||
# 替换已有的交叉表：先把起草器的名字按来源对齐到现行配方（6.3 的 align_to_contract），版式没变时得到的
|
||
# 就是原来的配方，表名、列名、分段 id、关系编号都不会因为换了一种定位方式而变
|
||
sheet = recipe["sheets"][ctx.si] # type: ignore[index]
|
||
cand = {"recipe_format": RECIPE_FORMAT, "mode": recipe.get("mode", "replace"),
|
||
"sheets": [{"id": sheet.get("id"), "match": sheet.get("match"),
|
||
"context": list(dsheet_d.get("context") or []), "blocks": [block]}],
|
||
"tables": tables, "relations": rels}
|
||
aligned, _report = S.align_to_contract(cand, recipe)
|
||
block = aligned["sheets"][0]["blocks"][0]
|
||
tables, rels = aligned["tables"], aligned["relations"]
|
||
btables = _block_tables(block)
|
||
if recipe:
|
||
old_ids = ({old.block.get("id")} | {s.get("id") for s in old.block.get("segments") or []}) if old else set()
|
||
used_ids = all_ids(recipe) - old_ids
|
||
used = _used_tables(ctx, except_=drop)
|
||
rename = {t: unique_name(t, used) for t in sorted(btables)}
|
||
seg_rename = {seg["id"]: unique_id(seg["id"], used_ids) for seg in block["segments"]}
|
||
block["id"] = old.block["id"] if old is not None else unique_id(block["id"], used_ids)
|
||
_rename_block(block, rename, seg_rename)
|
||
for t in tables:
|
||
t["name"] = rename.get(t["name"], t["name"])
|
||
# 关系 id 与留下来的关系重号时，后面的顺延
|
||
left = {str(r.get("id")) for r in recipe.get("relations") or [] if not (_rel_tables(r) & drop)}
|
||
next_n = max([int(i[1:]) for i in left if i[1:].isdigit()] + [0])
|
||
for r in rels:
|
||
_rename_rel(r, rename)
|
||
if r["id"] in left:
|
||
next_n += 1
|
||
while f"R{next_n}" in left:
|
||
next_n += 1
|
||
r["id"] = f"R{next_n}"
|
||
left.add(r["id"])
|
||
expected_rows = sorted({r for rows in dloc.seg_rows.values() for r in rows})
|
||
lc = dloc.label_col
|
||
expected = {"axis": _a1((dloc.axis_row, dloc.c1, dloc.axis_row, dloc.c2)),
|
||
"labels": _a1((expected_rows[0], lc, expected_rows[-1], lc)) if expected_rows else None,
|
||
"values": _a1((expected_rows[0], dloc.c1, expected_rows[-1], dloc.c2)) if expected_rows else None}
|
||
anchors = [Anchor("axis", _text_of(ctx.grid, dloc.axis_row, dloc.c1) or str(ctx.grid.get(dloc.axis_row, dloc.c1).value),
|
||
ctx.coord(dloc.axis_row, dloc.c1))]
|
||
for sid, tr in dloc.title_rows.items():
|
||
anchors.append(Anchor("section_title", _text_of(ctx.grid, tr, lc) or sid, ctx.coord(tr, lc)))
|
||
context = list(dsheet_d.get("context") or [])
|
||
if not recipe:
|
||
root = {"recipe_format": RECIPE_FORMAT, "mode": dr.get("mode", "replace"),
|
||
"sheets": [{"id": "s1", "match": dict(dsheet_d.get("match") or {"name": ctx.sheet[:31]}),
|
||
"context": context, "blocks": [block]}],
|
||
"tables": tables, "relations": rels}
|
||
ops: list[dict[str, Any]] = [{"op": "replace", "path": "", "value": root}]
|
||
elif ctx.si is None:
|
||
ops = [{"op": "add", "path": "/sheets/-", "value": {
|
||
"id": _new_sheet_id(recipe), "match": {"name": ctx.sheet[:31]}, "context": context, "blocks": [block]}}]
|
||
ops += _add_tables(recipe, tables) + [{"op": "add", "path": "/relations/-", "value": r} for r in rels]
|
||
else:
|
||
if old is not None:
|
||
ops = [{"op": "replace", "path": ptr("sheets", ctx.si, "blocks", old.bi), "value": block}]
|
||
else:
|
||
ops = [{"op": "add", "path": ptr("sheets", ctx.si, "blocks", "-"), "value": block}]
|
||
tis = sorted({ti for t in drop if (ti := table_index(recipe, t)) is not None}, reverse=True)
|
||
ris = sorted({i for i, r in enumerate(recipe.get("relations") or []) if _rel_tables(r) & drop}, reverse=True)
|
||
ops += [{"op": "remove", "path": ptr("tables", ti)} for ti in tis]
|
||
ops += [{"op": "remove", "path": ptr("relations", ri)} for ri in ris]
|
||
ops += _add_tables(recipe, tables) + [{"op": "add", "path": "/relations/-", "value": r} for r in rels]
|
||
if context and not recipe["sheets"][ctx.si].get("context"):
|
||
ops.append({"op": "add", "path": ptr("sheets", ctx.si, "context"), "value": context})
|
||
verb = "替换现有的" if old is not None else "新增"
|
||
summary = [f"按框选{verb}交叉表「{block['id']}」：日期表头、行标签、分段标题按文字定位",
|
||
f"写入表{quoted([t['name'] for t in tables])}"]
|
||
if rels:
|
||
summary.append(f"登记系统发现的关系{quoted([r['id'] for r in rels])}")
|
||
return finish(recipe, ops, kind="selection", key=f"crosstab:{block['id']}",
|
||
title=f"按框选{verb}交叉表「{block['id']}」：按日期表头和行标签定位", summary=summary,
|
||
anchors=anchors, expected=expected, block=block["id"])
|
||
|
||
|
||
def _rename_block(block: dict, tables: dict[str, str], segs: dict[str, str]) -> None:
|
||
for seg in block.get("segments") or []:
|
||
seg["id"] = segs.get(seg["id"], seg["id"])
|
||
if seg.get("table"):
|
||
seg["table"] = tables.get(seg["table"], seg["table"])
|
||
loc = seg.get("locate") or {}
|
||
if loc.get("segment"):
|
||
loc["segment"] = segs.get(loc["segment"], loc["segment"])
|
||
if seg.get("verify"):
|
||
seg["verify"]["against_table"] = tables.get(seg["verify"]["against_table"], seg["verify"]["against_table"])
|
||
if seg.get("keep_as"):
|
||
seg["keep_as"]["table"] = tables.get(seg["keep_as"]["table"], seg["keep_as"]["table"])
|
||
|
||
|
||
def _rename_rel(rel: dict, tables: dict[str, str]) -> None:
|
||
if rel.get("table"):
|
||
rel["table"] = tables.get(rel["table"], rel["table"])
|
||
for side in ("a", "b"):
|
||
if isinstance(rel.get(side), dict):
|
||
rel[side]["table"] = tables.get(rel[side]["table"], rel[side]["table"])
|
||
|
||
|
||
# ==========================================================================
|
||
# 分段、合计行、分段标题
|
||
# ==========================================================================
|
||
|
||
|
||
def _seg_rows(x: Any) -> dict[str, list[int]]:
|
||
return {sid: list(rows) for sid, rows in x.seg_rows.items()}
|
||
|
||
|
||
def _select_segment(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
loc = _crosstab_at(ctx, located)
|
||
if loc is None:
|
||
ctx.fail("selection_not_in_block", "框要落在交叉表的标签列上（可以带上右边的数据格），并在日期表头之下")
|
||
return _failed(ctx)
|
||
role = ctx.sel.options.get("role")
|
||
if role not in ("measures", "dimension"):
|
||
raise EditRequestError("edit_invalid", "框选为分段时要选是指标（每行一个指标）还是维度（每行一个取值）")
|
||
x = loc.cross
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
labels = _labels(ctx, x.label_col)
|
||
if labels is None:
|
||
return _failed(ctx)
|
||
taken = {r: sid for sid, rows in _seg_rows(x).items() for r in rows}
|
||
taken.update({r: sid for sid, r in x.title_rows.items()})
|
||
clash = sorted({taken[r] for r, _ in labels if r in taken})
|
||
if clash:
|
||
ctx.fail("selection_overlap", f"框里的行已经属于分段{quoted(clash)}：请只框还没有分段认领的行")
|
||
return _failed(ctx)
|
||
title_row = r1 - 1
|
||
title = _text_of(ctx.grid, title_row, x.label_col)
|
||
if title is None or not _row_empty(ctx.grid, title_row, x.c1, x.c2) or title_row in taken or len(title) > 40:
|
||
title, title_row = None, None
|
||
texts = [t for _, t in labels]
|
||
block = loc.block
|
||
axis = (block.get("axis") or {}).get("name", "日期")
|
||
used_ids = all_ids(ctx.recipe or {})
|
||
used = _used_tables(ctx)
|
||
notes: list[str] = []
|
||
if role == "measures":
|
||
base = to_sql_name(ctx.sheet, fallback=f"表{len(used) + 1}")[:45]
|
||
table = _new_table_name(ctx, f"{base}_按日", used)
|
||
if table is None:
|
||
return _failed(ctx)
|
||
cols_used = {collide_key(axis)}
|
||
measures: dict[str, str] = {}
|
||
units: dict[str, str] = {}
|
||
for i, lab in enumerate(texts):
|
||
col, unit, _raw = S._measure_column(lab, i + 2)
|
||
col = unique_name(col, cols_used)
|
||
measures[lab] = col
|
||
if unit:
|
||
units[col] = unit
|
||
seg = {"id": unique_id(table, used_ids), "role": "measures", "table": table,
|
||
"locate": {"by": "section_title", "title": title, "segment": None} if title
|
||
else {"by": "labels", "title": None, "segment": None},
|
||
"labels": {"expect": texts}, "measures": measures}
|
||
spec = {"name": table, "grain": [axis], "kind": "data", "units": units, "note": ""}
|
||
else:
|
||
parser = "hour_range" if all(hour_range(t) is not None for t in texts) else "text"
|
||
bad = [t for t in texts if parser == "text" and (text_label(t) is None or len(t) > LABEL_MAX)]
|
||
if bad:
|
||
ctx.fail("selection_label_unparsed", f"标签{quoted(bad)}为空或超过 {LABEL_MAX} 字，无法作维度的取值")
|
||
return _failed(ctx)
|
||
merged, blocked = _merge_target(ctx, block, parser, texts, title)
|
||
if merged is not None:
|
||
return _merged_segment(ctx, loc, merged, texts, title, title_row, labels)
|
||
if blocked:
|
||
notes.append(blocked)
|
||
want = split_unit_suffix(title)[0] if title else ""
|
||
n_fallback = len(used) + 1
|
||
plain = (to_sql_name(want, fallback=f"表{n_fallback}") if want else f"表{n_fallback}")[:48]
|
||
table = _new_table_name(ctx, plain, used)
|
||
if table is None:
|
||
return _failed(ctx)
|
||
dim = S._HOUR_DIM if parser == "hour_range" else S._TEXT_DIM
|
||
value = to_sql_name(plain[-2:], fallback="数值") if len(plain) >= 2 else "数值"
|
||
reserved = {collide_key(n) for n in (axis, dim, *(S._DERIVE if parser == "hour_range" else ()))}
|
||
if collide_key(value) in reserved or len(value) < 2:
|
||
value = "数值"
|
||
unit_raw = split_unit_suffix(title)[1] if title else None
|
||
seg = {"id": unique_id(table, used_ids), "role": "dimension", "table": table,
|
||
"locate": {"by": "section_title", "title": title, "segment": None} if title
|
||
else {"by": "labels", "title": None, "segment": None},
|
||
"labels": {"expect": texts},
|
||
"dim": {"name": dim, "parser": parser, "derive": dict(S._DERIVE) if parser == "hour_range" else {}},
|
||
"value": value, "const": {}, "stop_parser": None}
|
||
units = {value: unit_raw} if unit_raw and S.P.unit_known(unit_raw) else {}
|
||
spec = {"name": table, "grain": [axis, dim], "kind": "data", "units": units, "note": ""}
|
||
ops = [{"op": "add", "path": _segment_path(ctx, loc), "value": seg}, {"op": "add", "path": "/tables/-", "value": spec}]
|
||
kind = "指标" if role == "measures" else "维度"
|
||
return _segment_result(ctx, loc, seg, ops, kind=kind, texts=texts, title=title, title_row=title_row,
|
||
labels=labels, where=f"写入新表「{table}」", notes=notes)
|
||
|
||
|
||
def _segment_path(ctx: _Ctx, loc: _Located) -> str:
|
||
"""新分段插在按行号排在它后面的第一个分段之前；合计分段要紧跟它的基础分段，不能插进两者之间。"""
|
||
x = loc.cross
|
||
r1 = ctx.rect[0]
|
||
segs = loc.block.get("segments") or []
|
||
rows_of = _seg_rows(x)
|
||
idx = len(segs)
|
||
for i, s in enumerate(segs):
|
||
rows = rows_of.get(s.get("id")) or ([x.title_rows[s["id"]]] if s.get("id") in x.title_rows else [])
|
||
if rows and min(rows) > r1:
|
||
idx = i
|
||
break
|
||
while idx < len(segs) and segs[idx].get("role") == "derived":
|
||
idx += 1
|
||
return ptr("sheets", ctx.si, "blocks", loc.bi, "segments", idx)
|
||
|
||
|
||
def _segment_result(ctx: _Ctx, loc: _Located, seg: dict, ops: list[dict[str, Any]], *, kind: str, texts: list[str],
|
||
title: str | None, title_row: int | None, labels: list[tuple[int, str]], where: str,
|
||
notes: list[str], extra: list[str] | None = None) -> EditResult:
|
||
x = loc.cross
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
anchors = [Anchor("row_label", t, ctx.coord(r, x.label_col)) for r, t in labels]
|
||
if title:
|
||
anchors.insert(0, Anchor("section_title", title, ctx.coord(title_row, x.label_col))) # type: ignore[arg-type]
|
||
expected = {"labels": _a1((r1, x.label_col, r2, x.label_col)), "values": _a1((r1, x.c1, r2, x.c2)),
|
||
"title": _a1((title_row, x.label_col, title_row, x.label_col)) if title else None}
|
||
how = f"分段标题「{title}」" if title else f"标签{quoted(texts[:5])}" + ("等" if len(texts) > 5 else "")
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"segment:{seg['id']}",
|
||
title=f"按框选新增{kind}分段「{seg['id']}」：按{'分段标题' if title else '标签'}定位",
|
||
summary=[f"交叉表「{loc.block['id']}」新增{kind}分段「{seg['id']}」，按{how}定位，{where}", *(extra or [])],
|
||
anchors=anchors, notes=notes, expected=expected, block=loc.block.get("id"))
|
||
|
||
|
||
@dataclass
|
||
class _Merge:
|
||
table: str
|
||
template: dict # 那张表里第一个分段（dim、value 照抄它）
|
||
col: str # 常量列名
|
||
pick: str # 新分段的常量取值（候选词的第一个，同起草器第 8b 条）
|
||
|
||
|
||
def _dim_unit(title: str | None) -> str | None:
|
||
return split_unit_suffix(title)[1] if title else None
|
||
|
||
|
||
def _merge_target(ctx: _Ctx, block: dict, parser: str, texts: list[str],
|
||
title: str | None) -> tuple[_Merge | None, str]:
|
||
"""起草器第 5 条（P2-SPEC 6.1，recipe_suggest._group_dimensions）：dim.parser 相同、标签规范写法不相交、单位
|
||
相同的维度分段写进同一张表，各段用常量列区分。框选新的维度分段时照这条找可以并入的表（同一块里、按分段
|
||
顺序的第一张），常量取值照起草器第 8b 条取 candidate_words(新标题, 兄弟标题) 的第一个，并且要与已有的取值
|
||
不同（起草器「选不出互不相同的常量就各成一张表」）。返回 (可以并入的表, 说明)：
|
||
|
||
- 能并入：(_Merge, "")；
|
||
- 有 parser、单位、标签都对得上的表，却并不进去（客户端指定了新表名；那张表的分段没有常量列，并入要给现有
|
||
分段加列；分段标题里选不出与已有取值不同的词）：(None, 说明)，说明写进 notes——新建了表、没有并入，
|
||
要让人看见；
|
||
- 没有对得上的表：(None, "")。
|
||
只在同一块里找，并且那张表的写入方必须全在这一块里：跨块、跨工作表合表不是起草器的规则。"""
|
||
keys = {S._dim_key(parser, t) for t in texts}
|
||
unit = _dim_unit(title)
|
||
segs = [s for s in block.get("segments") or [] if s.get("role") == "dimension"]
|
||
order: list[str] = []
|
||
for s in segs:
|
||
if s.get("table") not in order:
|
||
order.append(s.get("table"))
|
||
for table in order:
|
||
group = [s for s in segs if s.get("table") == table]
|
||
if any((s.get("dim") or {}).get("parser") != parser or _dim_unit((s.get("locate") or {}).get("title")) != unit
|
||
for s in group):
|
||
continue
|
||
if any(keys & {S._dim_key(parser, e) for e in (s.get("labels") or {}).get("expect") or []} for s in group):
|
||
continue
|
||
if _written_by_others(ctx.recipe or {}, table, block):
|
||
continue
|
||
given = ctx.sel.options.get("table")
|
||
if given not in (None, ""):
|
||
return None, f"按指定的表名新建了表，没有并入同类的表「{table}」"
|
||
cols = {tuple((s.get("const") or {}).keys()) for s in group}
|
||
if len(cols) != 1 or len(next(iter(cols))) != 1:
|
||
return None, (f"没有并入同类的表「{table}」：它的分段没有用来区分分段的常量列，并入需要给现有分段加列。"
|
||
"需要并入时请在配方面板中修改")
|
||
col = next(iter(cols))[0]
|
||
picks = {canon(((s.get("const") or {}).get(col) or {}).get("pick") or "") for s in group}
|
||
sibs = [t for s in group if (t := (s.get("locate") or {}).get("title"))]
|
||
words = candidate_words(title, sibs) if title else []
|
||
if not words or canon(words[0]) in picks:
|
||
return None, (f"没有并入同类的表「{table}」：分段标题里没有能与已有分段区分的词作「{col}」的取值。"
|
||
"需要并入时请在配方面板中修改")
|
||
return _Merge(table, group[0], col, words[0]), ""
|
||
return None, ""
|
||
|
||
|
||
def _written_by_others(recipe: dict, table: str, block: dict) -> bool:
|
||
"""这张表还有这一块以外的写入方（别的块、列表、合计另存）。"""
|
||
for sheet in recipe.get("sheets") or []:
|
||
for b in sheet.get("blocks") or []:
|
||
if b is block:
|
||
continue
|
||
if b.get("layout") == "list":
|
||
tr = (b.get("rows") or {}).get("total_row") or {}
|
||
if table in (b.get("table"), tr.get("keep_as")):
|
||
return True
|
||
for seg in b.get("segments") or []:
|
||
if table in (seg.get("table"), (seg.get("keep_as") or {}).get("table")):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _merged_segment(ctx: _Ctx, loc: _Located, m: _Merge, texts: list[str], title: str | None,
|
||
title_row: int | None, labels: list[tuple[int, str]]) -> EditResult:
|
||
"""并入已有的表：dim、值列照抄那张表的分段，常量列取 m.pick，不新增表定义。"""
|
||
t = m.template
|
||
seg = {"id": unique_id(m.pick, all_ids(ctx.recipe or {})), "role": "dimension", "table": m.table,
|
||
"locate": {"by": "section_title", "title": title, "segment": None},
|
||
"labels": {"expect": texts}, "dim": copy.deepcopy(t["dim"]), "value": t["value"],
|
||
"const": {m.col: {"pick": m.pick}}, "stop_parser": None}
|
||
ops = [{"op": "add", "path": _segment_path(ctx, loc), "value": seg}]
|
||
return _segment_result(ctx, loc, seg, ops, kind="维度", texts=texts, title=title, title_row=title_row,
|
||
labels=labels, where=f"并入表「{m.table}」", notes=[],
|
||
extra=[f"「{m.col}」取「{m.pick}」（取自分段标题，与同表其他分段区分）"])
|
||
|
||
|
||
def _select_derived(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
loc = _crosstab_at(ctx, located)
|
||
if loc is None:
|
||
ctx.fail("selection_not_in_block", "框要落在交叉表的标签列上（可以带上右边的数据格），并在日期表头之下")
|
||
return _failed(ctx)
|
||
x = loc.cross
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
labels = _labels(ctx, x.label_col)
|
||
if labels is None:
|
||
return _failed(ctx)
|
||
bad = [t for _, t in labels if hour_range_total(t) is None]
|
||
if bad:
|
||
ctx.fail("selection_label_unparsed", f"标签{quoted(bad)}无法解析为带区间的合计（如「18-22时合计」）")
|
||
return _failed(ctx)
|
||
keep = ctx.sel.options.get("keep", True)
|
||
if not isinstance(keep, bool):
|
||
raise EditRequestError("edit_invalid", "合计行是否另存只能选是或否")
|
||
box_rows = set(range(r1, r2 + 1))
|
||
rows_of = _seg_rows(x)
|
||
others = {r for rows in rows_of.values() for r in rows}
|
||
segs = loc.block.get("segments") or []
|
||
gi = None
|
||
for i, s in enumerate(segs):
|
||
rows = set(rows_of.get(s.get("id")) or [])
|
||
# 框里的行必须紧跟这个分段的最后一行。起草器定位时会把紧跟的、有数据的合计行也收进分段（它还不知道该停），
|
||
# 所以「收进来的」也算紧跟
|
||
if s.get("role") == "dimension" and (r1 - 1) in rows and all(r in rows or r not in others for r in box_rows):
|
||
gi = i
|
||
break
|
||
if gi is None:
|
||
ctx.fail("selection_not_after_segment", "框里的行要紧跟在一个按时段展开的分段的最后一行之后")
|
||
return _failed(ctx)
|
||
why = total_blockers(ctx.recipe, ctx.si, loc.bi, gi) # type: ignore[arg-type]
|
||
if why:
|
||
ctx.fail("selection_not_after_segment", why)
|
||
return _failed(ctx)
|
||
texts = [t for _, t in labels]
|
||
ops, summary, _kept = total_ops(ctx.recipe, ctx.si, loc.bi, gi, texts, keep) # type: ignore[arg-type]
|
||
new_id = ops[1]["value"]["id"]
|
||
expected = {"labels": _a1((r1, x.label_col, r2, x.label_col)), "values": _a1((r1, x.c1, r2, x.c2))}
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"derived:{new_id}",
|
||
title=f"按框选把{quoted(texts)}改作合计核对", summary=summary,
|
||
anchors=[Anchor("total_word", t, ctx.coord(r, x.label_col)) for r, t in labels],
|
||
expected=expected, block=loc.block.get("id"))
|
||
|
||
|
||
def _select_section_title(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
r1, c1, r2, c2 = ctx.rect
|
||
loc = _crosstab_at(ctx, located)
|
||
if loc is None or (r1, c1) != (r2, c2):
|
||
ctx.fail("selection_not_in_block", "分段标题要框交叉表标签列上的一格")
|
||
return _failed(ctx)
|
||
seg_id = ctx.sel.options.get("segment")
|
||
segs = loc.block.get("segments") or []
|
||
gi = next((i for i, s in enumerate(segs) if s.get("id") == seg_id), None)
|
||
if gi is None or segs[gi].get("role") == "derived":
|
||
raise EditRequestError("edit_invalid", "要指明是这个交叉表里哪个分段（指标或维度分段）的标题")
|
||
seg = segs[gi]
|
||
text = _text_of(ctx.grid, r1, c1)
|
||
if text is None or len(text) > 40:
|
||
ctx.fail("selection_label_blank", "这一格没有文字（或超过 40 字），不能作分段标题")
|
||
return _failed(ctx)
|
||
sp = ("sheets", ctx.si, "blocks", loc.bi, "segments", gi)
|
||
ops: list[dict[str, Any]] = [{"op": "replace", "path": ptr(*sp, "locate"),
|
||
"value": {"by": "section_title", "title": text, "segment": None}}]
|
||
summary = [f"分段「{seg_id}」改按分段标题「{text}」定位"]
|
||
consts = seg.get("const") or {}
|
||
if consts and not keep_ok(seg, text):
|
||
# 常量沿用不了：常量列整列的值要换成新标题里的哪个词，只能由人选（4.1「不猜」）。框选不替人挑第一个
|
||
# 候选词——那是破坏性变更，按期累积时还要重新开始累积。④ 的修复按钮把候选词逐个列成选项，配方面板也能选
|
||
held = "、".join(f"「{k}」现在取「{v.get('pick')}」" for k, v in consts.items())
|
||
ctx.fail("selection_label_unparsed",
|
||
f"分段「{seg_id}」的常量列{held}，新标题「{text}」里没有这个词，无法沿用；改用哪个词需要人工选择。"
|
||
"请在配方面板中修改取值；该分段报了找不到分段标题时，也可以用问题旁的修复按钮选词",
|
||
[ctx.coord(r1, c1)])
|
||
return _failed(ctx)
|
||
if consts:
|
||
summary.append("、".join(f"「{k}」沿用「{v.get('pick')}」" for k, v in consts.items()))
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"section_title:{seg_id}",
|
||
title=f"按框选把「{text}」作为分段「{seg_id}」的分段标题", summary=summary,
|
||
anchors=[Anchor("section_title", text, ctx.coord(r1, c1))],
|
||
expected={"title": _a1(ctx.rect)}, block=loc.block.get("id"))
|
||
|
||
|
||
# ==========================================================================
|
||
# 忽略
|
||
# ==========================================================================
|
||
|
||
|
||
def _digits_fail(ctx: _Ctx, texts: list[str], cells: list[str]) -> None:
|
||
ctx.fail("selection_anchor_has_digits",
|
||
f"{quoted(texts)}含数字，下一期很可能对不上，无法按文字忽略：请修改文件，或把它框选为新的分段", cells)
|
||
|
||
|
||
def _select_ignore_rows(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
reason = _reason(ctx.sel)
|
||
loc = _crosstab_at(ctx, located)
|
||
if loc is None:
|
||
ctx.fail("selection_not_in_block", "要忽略的行要框在交叉表的标签列上（可以带上右边的数据格），并在日期表头之下")
|
||
return _failed(ctx)
|
||
x = loc.cross
|
||
labels = _labels(ctx, x.label_col)
|
||
if labels is None:
|
||
return _failed(ctx)
|
||
digits = [(r, t) for r, t in labels if has_digit(t) or len(t) > _ANCHOR_MAX]
|
||
if digits:
|
||
_digits_fail(ctx, [t for _, t in digits], [ctx.coord(r, x.label_col) for r, _ in digits])
|
||
return _failed(ctx)
|
||
have = {match_key(i.get("label")) for i in loc.block.get("ignore_rows") or []}
|
||
texts: list[str] = []
|
||
for _r, t in labels:
|
||
if match_key(t) not in have:
|
||
have.add(match_key(t))
|
||
texts.append(t)
|
||
bp = ("sheets", ctx.si, "blocks", loc.bi)
|
||
ops = [{"op": "add", "path": ptr(*bp, "ignore_rows", "-"), "value": {"label": t, "reason": reason}} for t in texts]
|
||
r1, _c1, r2, _c2 = ctx.rect
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"ignore_rows:{loc.block['id']}",
|
||
title=f"按框选按行标签忽略{quoted([t for _, t in labels])}",
|
||
summary=[f"交叉表「{loc.block['id']}」按行标签忽略{quoted(texts)}（理由：{reason}）"] if texts
|
||
else ["这些行已经按行标签忽略，配方不变"],
|
||
anchors=[Anchor("row_label", t, ctx.coord(r, x.label_col)) for r, t in labels],
|
||
expected={"ignored": _a1((r1, x.label_col, r2, x.label_col))}, block=loc.block.get("id"))
|
||
|
||
|
||
def _select_ignore_columns(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
reason = _reason(ctx.sel)
|
||
r1, c1, r2, c2 = ctx.rect
|
||
target: tuple[_Located, int, dict[int, str]] | None = None
|
||
for loc in located:
|
||
if loc.cross is not None and c1 > loc.cross.c2 and loc.cross.axis_row <= r2:
|
||
x = loc.cross
|
||
texts = {c: t for c in range(c1, c2 + 1) if (t := _text_of(ctx.grid, x.axis_row, c))}
|
||
target = (loc, x.axis_row, texts)
|
||
break
|
||
if loc.lst is not None and loc.lst.cols and loc.lst.cols[0] <= c1 and c2 <= loc.lst.cols[-1]:
|
||
hr = loc.lst.header_rows
|
||
texts = S._header_texts(ctx.grid, hr, list(range(c1, c2 + 1)))
|
||
target = (loc, hr[-1], texts)
|
||
break
|
||
if target is None:
|
||
ctx.fail("selection_not_in_block", "要忽略的列要框在交叉表最后一个日期右侧，或列表的表头范围之内")
|
||
return _failed(ctx)
|
||
loc, hrow, texts = target
|
||
cols = list(range(c1, c2 + 1))
|
||
blank = [c for c in cols if c not in texts and ctx.grid.get(hrow, c) is None]
|
||
if blank:
|
||
ctx.fail("selection_header_blank", f"{'、'.join(col_letter(c) for c in blank)} 列没有表头，无法按表头忽略",
|
||
[ctx.coord(hrow, c) for c in blank])
|
||
return _failed(ctx)
|
||
bad = [c for c in cols if c not in texts or not _textual(ctx.grid, hrow, c)
|
||
or month_day_or_date(texts[c]) is not None]
|
||
if bad:
|
||
ctx.fail("selection_header_not_text", f"{'、'.join(col_letter(c) for c in bad)} 列的表头是数字或日期，无法按表头忽略",
|
||
[ctx.coord(hrow, c) for c in bad])
|
||
return _failed(ctx)
|
||
digits = [c for c in cols if has_digit(texts[c]) or len(texts[c]) > _HEADER_MAX]
|
||
if digits:
|
||
_digits_fail(ctx, [texts[c] for c in digits], [ctx.coord(hrow, c) for c in digits])
|
||
return _failed(ctx)
|
||
have = {match_key(i.get("header")) for i in loc.block.get("ignore_columns") or []}
|
||
heads: list[str] = []
|
||
for c in cols:
|
||
if match_key(texts[c]) not in have:
|
||
have.add(match_key(texts[c]))
|
||
heads.append(texts[c])
|
||
bp = ("sheets", ctx.si, "blocks", loc.bi)
|
||
ops = [{"op": "add", "path": ptr(*bp, "ignore_columns", "-"), "value": {"header": h, "reason": reason}}
|
||
for h in heads]
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"ignore_columns:{loc.block['id']}",
|
||
title=f"按框选按表头忽略{quoted([texts[c] for c in cols])}",
|
||
summary=[f"块「{loc.block['id']}」按表头忽略{quoted(heads)}（理由：{reason}）"] if heads
|
||
else ["这些列已经按表头忽略，配方不变"],
|
||
anchors=[Anchor("header", texts[c], ctx.coord(hrow, c)) for c in cols],
|
||
expected={"ignored": _a1((hrow, c1, hrow, c2))}, block=loc.block.get("id"))
|
||
|
||
|
||
def _select_ignore_outside(ctx: _Ctx, located: list[_Located]) -> EditResult:
|
||
reason = _reason(ctx.sel)
|
||
if ctx.si is None:
|
||
ctx.fail("selection_not_in_block", "这个工作表还没有导入区域：请先框选要导入的表")
|
||
return _failed(ctx)
|
||
r1, c1, r2, c2 = ctx.rect
|
||
rects = [r for loc in located for r in loc.rects]
|
||
inside = lambda r, c: any(a[0] <= r <= a[2] and a[1] <= c <= a[3] for a in rects) # noqa: E731
|
||
found: list[tuple[int, int, str]] = []
|
||
bare: list[int] = []
|
||
for r in range(r1, r2 + 1):
|
||
hit = next(((c, t) for c in range(c1, c2 + 1) if not inside(r, c) and (t := _text_of(ctx.grid, r, c))
|
||
and not looks_numeric_text(t)), None)
|
||
if hit is None:
|
||
bare.append(r)
|
||
else:
|
||
found.append((r, hit[0], hit[1]))
|
||
if bare:
|
||
ctx.fail("selection_no_anchor", f"{'、'.join(f'第 {r} 行' for r in bare[:8])}在导入区域之外没有文字格，"
|
||
"没有可以作锚点的文字：只能修改文件",
|
||
[ctx.coord(r, c1) for r in bare])
|
||
return _failed(ctx)
|
||
digits = [(r, c, t) for r, c, t in found if has_digit(t) or len(t) > _ANCHOR_MAX]
|
||
if digits:
|
||
_digits_fail(ctx, [t for *_x, t in digits], [ctx.coord(r, c) for r, c, _t in digits])
|
||
return _failed(ctx)
|
||
have = {match_key(i.get("anchor")) for i in ctx.recipe["sheets"][ctx.si].get("ignore_outside") or []} # type: ignore[index]
|
||
texts: list[str] = []
|
||
for _r, _c, t in found:
|
||
if match_key(t) not in have:
|
||
have.add(match_key(t))
|
||
texts.append(t)
|
||
ops = [{"op": "add", "path": ptr("sheets", ctx.si, "ignore_outside", "-"), "value": {"anchor": t, "reason": reason}}
|
||
for t in texts]
|
||
sid = ctx.recipe["sheets"][ctx.si]["id"] # type: ignore[index]
|
||
return finish(ctx.recipe, ops, kind="selection", key=f"ignore_outside:{sid}",
|
||
title=f"按框选忽略同一行有{quoted([t for *_x, t in found])}的导入区域之外的数字",
|
||
summary=[f"工作表「{ctx.sheet}」：同一行有{quoted(texts)}时，忽略这一行导入区域之外的数字（理由：{reason}）"]
|
||
if texts else ["这些行已经按文字忽略，配方不变"],
|
||
anchors=[Anchor("outside_text", t, ctx.coord(r, c)) for r, c, t in found],
|
||
expected={"ignored": _a1(ctx.rect)}, block=None)
|
||
|
||
|
||
# ==========================================================================
|
||
# 入口
|
||
# ==========================================================================
|
||
|
||
|
||
def selection_edit(recipe: dict[str, Any] | None, grid: Grid, sel: Selection, *,
|
||
extraction: Extraction | None, draft_fn: Callable[[], Draft] | None = None,
|
||
retired_names: dict[str, list[str]] | None = None) -> EditResult:
|
||
"""一次框选 → EditResult（补丁、锚点、期望区域）。
|
||
|
||
- recipe：工作配方的**完整形式**（同 fix_ops），配方为空（规则起草失败）时是 None：list、crosstab 用根替换新建配方，
|
||
这样框选也是起草失败时不用 AI 的兜底；
|
||
- grid：该工作表的 read_grid(max_rows=draft_max_rows)，和起草用的同一份；
|
||
- extraction：当前工作配方最近一次干跑的结果（判断选区和已有的块是否重叠；带块 id 时按它取各块的区域）；
|
||
- draft_fn：只在 as=crosstab 时用，接口层注入一个对当前文件跑规则起草的闭包；
|
||
- retired_names：当前版本里退役的表名 → 该表退役的列名（取自快照清单的 columns；整张表退役时键在、值是它的列），
|
||
新表名的撞名检查要算上这些键，新列名与退役列只差大小写时沿用退役列的写法（SQLite 列名不分大小写，并存会撞）。
|
||
选项写错（header_rows 不在 1–3、role 不认识……）抛 EditRequestError（edit_invalid）；忽略类没写理由抛
|
||
EditRequestError（reason_required）。框和网格、配方的关系泛化不了时返回 ok=False，problems 写原因（4.3）。"""
|
||
rect = _rect(sel.ref)
|
||
parsed: Recipe | None = None
|
||
if recipe:
|
||
try:
|
||
parsed = Recipe.model_validate(recipe)
|
||
except ValidationError:
|
||
parsed = None
|
||
ctx = _Ctx(recipe=recipe or None, parsed=parsed, grid=grid, sel=sel, rect=rect, ext=extraction,
|
||
retired={str(k): list(v or []) for k, v in (retired_names or {}).items()})
|
||
rows = grid.row_numbers()
|
||
if not rows or (grid.truncated and rect[2] > rows[-1]):
|
||
ctx.fail("selection_out_of_grid", "框超出了已读入的范围（每张工作表只读入前 500 行）：请框在已显示的范围之内，"
|
||
"大表的列表可以改选「下边界按规则推断」")
|
||
return _failed(ctx)
|
||
if recipe and parsed is None and sel.as_ not in ("list", "crosstab"):
|
||
ctx.fail("selection_not_in_block", "当前的配方还有格式问题，无法在它上面框选：请先在配方面板中修改")
|
||
return _failed(ctx)
|
||
ctx.si = _sheet_of(ctx.recipe, grid, extraction)
|
||
located = _locate(ctx)
|
||
if sel.as_ == "list":
|
||
return _select_list(ctx, located)
|
||
if sel.as_ == "crosstab":
|
||
return _select_crosstab(ctx, located, draft_fn)
|
||
if ctx.si is None and sel.as_ != "ignore_outside":
|
||
ctx.fail("selection_not_in_block", "这个工作表还没有交叉表或列表：请先框选要导入的表")
|
||
return _failed(ctx)
|
||
handler = {"segment": _select_segment, "derived": _select_derived, "section_title": _select_section_title,
|
||
"ignore_rows": _select_ignore_rows, "ignore_columns": _select_ignore_columns,
|
||
"ignore_outside": _select_ignore_outside}.get(sel.as_)
|
||
if handler is None:
|
||
raise EditRequestError("edit_invalid", "框选没有指明选中的是什么，或取值不在可选范围内")
|
||
return handler(ctx, located)
|
||
|
||
|
||
# ==========================================================================
|
||
# 重放比对（4.5）
|
||
# ==========================================================================
|
||
|
||
#: 期望区域的键 → 干跑的格子角色（RegionMark.role）
|
||
_ROLES = {
|
||
"header": ("col_header",), "data": ("value",), "total": ("total_label", "total_value"),
|
||
"axis": ("col_header",), "labels": ("row_label", "derived_label", "section_title"),
|
||
"values": ("value", "derived_value"), "title": ("section_title",), "ignored": ("ignored", "ignored_column"),
|
||
}
|
||
_NAMES = {"header": "表头", "data": "数据区", "total": "合计行", "axis": "日期表头", "labels": "行标签",
|
||
"values": "数值区", "title": "分段标题", "ignored": "忽略的格"}
|
||
|
||
|
||
def _crop(rect: Rect | None, window: int | None) -> Rect | None:
|
||
if rect is None or window is None:
|
||
return rect
|
||
if rect[0] > window:
|
||
return None
|
||
return rect[0], rect[1], min(rect[2], window), rect[3]
|
||
|
||
|
||
def _covered(rect: Rect, marks: list[Rect]) -> bool:
|
||
"""rect 里的每一格都落在某个 mark 里（框是人框的，几十格以内，逐格判）。"""
|
||
return all(any(m[0] <= r <= m[2] and m[1] <= c <= m[3] for m in marks)
|
||
for r in range(rect[0], rect[2] + 1) for c in range(rect[1], rect[3] + 1))
|
||
|
||
|
||
def replay_compare(edit: EditResult, after: Extraction, *, window_rows: int | None = None) -> ReplayCompare:
|
||
"""框换算出的期望区域与应用后干跑认出的区域比对（4.5）。match=False 不阻止应用：界面把差异摆出来，按钮文字变成
|
||
「仍然应用（以重放结果为准）」。
|
||
|
||
actual 从 after.regions 里按 RegionMark.block 过滤出这个块的格子、按角色合成外接矩形。列表、交叉表按矩形相等判
|
||
（表头矩形相等、数据矩形的行列范围相等、期望有合计行时合计行就是框下一行）；分段、合计行、分段标题、忽略这几种
|
||
框只是一部分，按「框里的格都被认成了期望的角色」判。干跑结果里没有块 id（期 2 的执行器）时退回按列范围取。
|
||
window_rows：大表的交互干跑只跑前若干行时，期望和实际都先裁到这些行再比（评审三-M3）。"""
|
||
exp = {k: v for k, v in (edit.expected or {}).items()}
|
||
tagged = any(m.block for m in after.regions)
|
||
exp_rects = {k: _rect(v) for k, v in exp.items() if v}
|
||
cols = _bbox(list(exp_rects.values()))
|
||
|
||
def marks_for(key: str) -> list[Rect]:
|
||
roles = _ROLES.get(key, ())
|
||
out = []
|
||
for m in after.regions:
|
||
if m.role not in roles:
|
||
continue
|
||
if tagged and edit.block is not None and m.block != edit.block:
|
||
continue
|
||
r = _rect(m.ref)
|
||
if (not tagged or edit.block is None) and cols is not None and not (r[1] <= cols[3] and cols[1] <= r[3]):
|
||
continue
|
||
out.append(r)
|
||
return [x for x in (_crop(r, window_rows) for r in out) if x is not None]
|
||
|
||
whole = edit.key.split(":", 1)[0] in ("list", "crosstab")
|
||
actual: dict[str, str | None] = {}
|
||
diffs: list[str] = []
|
||
expected: dict[str, str | None] = {}
|
||
match = True
|
||
for key in exp:
|
||
e = _crop(exp_rects.get(key), window_rows)
|
||
expected[key] = _a1(e)
|
||
ms = marks_for(key)
|
||
if whole:
|
||
a = _bbox(ms)
|
||
else:
|
||
a = _bbox([m for m in ms if e is not None and _intersects(m, e)])
|
||
actual[key] = _a1(a)
|
||
name = _NAMES.get(key, key)
|
||
if e is None and a is None:
|
||
continue
|
||
if e is None:
|
||
if whole:
|
||
match = False
|
||
diffs.append(f"重放时多认出了{name}（{_a1(a)}），框里没有")
|
||
continue
|
||
if a is None:
|
||
match = False
|
||
diffs.append(f"重放时没有认出{name}（框里是 {_a1(e)}）")
|
||
continue
|
||
if whole:
|
||
if a != e:
|
||
match = False
|
||
diffs.append(_diff_text(name, e, a))
|
||
elif not _covered(e, ms):
|
||
match = False
|
||
diffs.append(f"框里的{name}（{_a1(e)}）在重放时没有全部认出，认出的是 {_a1(a)}")
|
||
return ReplayCompare(expected=expected, actual=actual, match=match, diffs=diffs, window_rows=window_rows)
|
||
|
||
|
||
def _diff_text(name: str, e: Rect, a: Rect) -> str:
|
||
"""人话：「重放的数据区到第 10 行，比框多 2 行（第 9–10 行）」。"""
|
||
parts = []
|
||
if a[2] > e[2]:
|
||
parts.append(f"重放的{name}到第 {a[2]} 行，比框多 {a[2] - e[2]} 行（{_rows_text(e[2] + 1, a[2])}）")
|
||
elif a[2] < e[2]:
|
||
parts.append(f"重放的{name}到第 {a[2]} 行，比框少 {e[2] - a[2]} 行（{_rows_text(a[2] + 1, e[2])}）")
|
||
if a[0] != e[0]:
|
||
parts.append(f"重放的{name}从第 {a[0]} 行开始，框是第 {e[0]} 行")
|
||
if (a[1], a[3]) != (e[1], e[3]):
|
||
parts.append(f"重放的{name}在 {col_letter(a[1])}–{col_letter(a[3])} 列，框是 {col_letter(e[1])}–{col_letter(e[3])} 列")
|
||
return "；".join(parts) or f"重放的{name}是 {_a1(a)}，框是 {_a1(e)}"
|