test(elec): 新增「源表 xlsx vs 电费批次」逐行核对脚本

按源表自身的公式口径核对（本期抄表数为空 = 未抄表 = 不计费，
与用量列是否残留数字无关）；识别合并单元格向下沿用、总表行、摘录表；
并标出用量列「手填常量且本期为空」的可疑格子。

实测：9月批次与源表 20 处不一致（多计 10 行 ¥167,947.41、少计 601B ¥299.67）；
8月工作表为摘录，核对到的 8 行 0 差异。

Co-authored-by: CodeBuddy

.workbuddy/memory/2026-10-13.md

@@ -273,3 +273,54 @@ env -u http_proxy -u https_proxy -u HTTP_PROXY -u HTTPS_PROXY git push origin ma
 - **沙箱端到端**：把今天注入 2026-12-27 → 命中 → 自动出单 **27 份 PDF（`%PDF-` 全有效）** → 推进 33 行 AC/AJ → 发送（桩）。
 - **两个提醒**：① 电费是「人工录入 + 到点发送」，不录就没单可发；② 本机用 `start.bat`（非 safe）会开第二个定时器 → 与线上重复发送，本机只用 safe。
 - 新脚本已提交 `fdb5148`。
+
+---
+
+## 电费批次台账审计（2026-10-13 12:52）
+
+**结论：数据忠实于源表，但有三处实质缺陷。**
+
+- ✅ **9月批次 `20261009-120146`（60 行）逐格还原源表**：上期读数全对；48 行 pending（本期未抄）、12 行可计费；
+  用量×电价×系数=小计 全部复算通过；303 室无读数未入列；29 个单元 = 源表房间数。
+  8月批次 `20261009-115501`（62 行 / 12 单据）与源表摘录一致。
+  **源表「合计」为负（-959,819.73 等）是未抄完的中间态，无意义，不必对齐**。
+- ❌ **缺陷 1 · 出租方大范围错记「精锐」**（8月 5 个单元 / 9月 4 个单元）：
+  `301/302/303室` 台账 A = **科博**、`602/603室` = **瑞格嘉德**，批次却全记精锐。
+  根因：批次建于 10-09 11:55/12:01，当时名册大概只有精锐 → `resolve_owner` 走「名单兜底」。
+  **`rerender` 修不了**：电费重渲染用批次里存的 `owner`（`_txt(d.get("owner")) or resolve_owner(...)`）→ 存的精锐被沿用。
+  8月 12 份单据中 **302室（杭州逢林 ¥622.86）已实际发出**，出租方写错。
+- ❌ **缺陷 2 · 9月漏发 601B 单据（¥299.67）**：
+  `elec_parser.group_by_company` 的跳过规则是 `"空置" in comp`（过宽）。
+  台账 F 列 = `上海博睿孚信息科技发展有限公司（空置）`（源表给了完整读数 + 备注「套内，结算至2026年9月21日」→ 明确要出结算单）→ 被整单跳过。
+  同类写法台账共 2 处：601B室、309室「保洁阿姨（空置）」。
+  9月 6 个可计费单元只出 1 份单据，`document_amount 18356.55` vs `total_amount 168247.08`（差 149890.53）。
+- ❌ **缺陷 3 · 9月单据仍是旧 `.xlsx`**（10-12 起应统一 PDF；8月已重出 PDF）。
+- ⚠️ **待人工确认**：`603/605/606A/606B/701` 五单元源表「只有用量、无上下期读数」（用量正是 8月的本期读数）→
+  系统按用量全额计费 ¥167,947.41；需确认是「首次全额计费」还是源表漏填上期。
+- **干跑验证的修法**（不写数据）：① `owner` 置空让 `resolve_owner` 按台账重算；② 公司名去掉「（空置）」标记，
+  仅当去标记后为空才跳过。结果：8月仍 12 份（仅 302 出租方 精锐→科博）；9月变 2 份（603 归瑞格嘉德 + **新增 601B ¥299.67**），
+  4 个纯空置单元仍被正确排除。
+- **本机 vs 线上**：本机批次 3 个 / 线上 5 个（少 `20261012-150747` 11月房租、`20261012-170741` 12月房租）；
+  本机 8月批次在 10-13 10:50 被重出 PDF 时**逐份发送状态全丢**（12 份变「待发送」），线上仍是 xlsx + 12 份「已发送」+ sent_at。**本机批次应以线上为准。**
+
+---
+
+## 用真实源表 xlsx 复核电费批次（2026-10-13 13:25）
+
+用户提供 `桌面/电费统计.xlsx`（`8月` 摘录 4 单元 / `9月` 全量 75 行）。新增
+**`tools/check_elec_source.py`**（源表 vs 批次逐行核对，只读）固化口径。
+
+- 源表列：A楼层 B房间号 C单位名称(**VLOOKUP 外部 `汇总表`**) D电表名称 E上期 F本期 G用量 H电价 I系数 J小计 K合计 L账单 M发票 N备注。
+- **源表公式**：`G = IF(COUNT(E:F)=0,"",F-E)`、`J = IF(COUNT(E:F)=0,"",G*H*I)`。
+  → **本期抄表数(F)为空 = 未抄表 = 不出小计**，与用量列里有没有数字无关。
+- **9月批次结论：不一致（20 处 / 10 行）**
+  - 源表 应计费 **2 行（仅 601B ¥299.67）**；批次计费 **12 行**。
+  - **多计 10 行 ¥167,947.41**：`603/605/606A/606B/701` 各 2 行，E/F 全空（源表 J 算出空、合计行也不含），
+    但**用量列被手填了常量**（G 列 61 公式 + **10 常量** = 897.3/4305.2/647.7/24458/417.5/3008.8/312.8/2663.6/861.4/67096），
+    批次按 `compute_row` 规则 C「抄表数都空但填了用量 → 取该用量」当成可计费。
+  - **少计 1 行 ¥299.67**：`601B` 是唯一 F 有读数的行（221.5/2583.6 → 小计 171.99/127.68、合计 299.67，备注「套内，结算至2026年9月21日」），
+    因台账 F 列公司名含「（空置）」被 `group_by_company` 跳过。
+  - 净效果：唯一 1 份单据发给了 `603室 锦玉良源 ¥18,356.55` —— 对象与金额都不对（**10-21 09:00 会自动发出**）。
+- **8月工作表是摘录**（仅 301/306/315/608 共 8 行）→ 与批次 **0 差异** ✅。
+- **工具坑位**：解析源表时 `D`(电表名称) 列有合并单元格（`D16:D17`），必须向下沿用，否则丢行。
+- 出租方错记问题同前（301/302/303→科博，602/603→瑞格嘉德）。

tools/check_elec_source.py

+# -*- coding: utf-8 -*-
+"""拿源表 xlsx 核对系统内电费批次台账是否一致（只读，不改任何数据）。
+
+用法
+    python tools/check_elec_source.py <源表.xlsx>                  # 自动按抄表月份挑批次
+    python tools/check_elec_source.py <源表.xlsx> --sheet 9月
+    python tools/check_elec_source.py <源表.xlsx> --batch 20261009-120146
+
+口径（与源表自身的公式保持一致）
+    源表 G(用量) = IF(COUNT(E:F)=0,"",F-E)；J(小计) = IF(COUNT(E:F)=0,"",G*H*I)
+    → **本期抄表数(F)为空 = 未抄表 = 不应计费**，无论用量列里是否残留数字。
+    总表行（房间号含「总表」）不参与房间计费。
+
+退出码 0=一致，1=有差异，2=用法/文件问题。
+"""
+import argparse
+import json
+import os
+import sys
+
+HERE = os.path.dirname(os.path.abspath(__file__))
+ROOT = os.path.dirname(HERE)
+sys.path.insert(0, ROOT)
+
+try:
+    import openpyxl
+except ImportError:
+    print("需要 openpyxl：pip install openpyxl")
+    sys.exit(2)
+
+BATCH_DIR = os.path.join(ROOT, "data", "batches")
+METER_MAP = {"空调": "空调", "房间": "照明", "照明": "照明"}
+
+
+# ------------------------------------------------------------------ 源表解析
+def _norm_unit(v):
+    s = str(v).strip()
+    if s.endswith("总表"):
+        return None
+    if isinstance(v, int) or s.replace(".0", "").isdigit():
+        return s.replace(".0", "") + "室"
+    return s + "室"
+
+
+def parse_sheet(ws):
+    """工作表 → [行 dict]。合并单元格（房间号/电表名称）向下沿用。"""
+    rows = []
+    unit = None
+    meter = None
+    total_mode = False
+    for r in range(1, ws.max_row + 1):
+        B = ws.cell(r, 2).value
+        D = ws.cell(r, 4).value
+        E, F = ws.cell(r, 5).value, ws.cell(r, 6).value
+        G, J = ws.cell(r, 7).value, ws.cell(r, 10).value
+        a = ws.cell(r, 1).value
+        if B is not None:
+            total_mode = str(B).strip().endswith("总表")
+            if not total_mode:
+                unit = _norm_unit(B)
+        if total_mode or unit is None:
+            continue
+        if D is not None:
+            meter = METER_MAP.get(str(D).strip())      # 合并单元格：向下沿用
+            if meter is None:
+                meter = "__skip__"                    # 表头等非电表行
+        if meter in (None, "__skip__"):
+            continue
+        if E is None and F is None and G is None and J is None:
+            continue                              # 全空行（如 303 未入住）
+        if str(a or "").strip() == "合计":
+            continue
+        rows.append({
+            "row": r, "unit": unit, "meter": meter,
+            "prev": E, "cur": F, "qty": G, "amt": J,
+            "price": ws.cell(r, 8).value, "factor": ws.cell(r, 9).value,
+        })
+    return rows
+
+
+def stale_constants(ws):
+    """用量列里既不是公式、又没有本期读数的格子（多为上一期残留的常量）。"""
+    wbf = openpyxl.load_workbook(ws.parent.path, data_only=False)
+    wsf = wbf[ws.title]
+    out = []
+    for r in range(1, ws.max_row + 1):
+        v = wsf.cell(r, 7).value
+        if isinstance(v, (int, float)) and ws.cell(r, 6).value is None:
+            out.append((r, v))
+    return out
+
+
+# ------------------------------------------------------------------ 批次
+def pick_batch(sheet_title, batch_id=None):
+    if batch_id:
+        p = os.path.join(BATCH_DIR, batch_id + ".json")
+        return json.load(open(p, encoding="utf-8")) if os.path.exists(p) else None
+    for f in sorted(os.listdir(BATCH_DIR), reverse=True):
+        if not f.endswith(".json"):
+            continue
+        d = json.load(open(os.path.join(BATCH_DIR, f), encoding="utf-8"))
+        if d.get("kind") == "elec" and sheet_title in (d.get("title") or ""):
+            return d
+    return None
+
+
+# ------------------------------------------------------------------ 比对
+def close(a, b):
+    if a is None and b is None:
+        return True
+    if a is None or b is None:
+        return False
+    return abs(float(a) - float(b)) < 0.005
+
+
+def compare(sheet_rows, batch):
+    problems = []
+    by_unit = {}
+    for x in batch["details"]:
+        by_unit.setdefault((x["unit"], x["meter"]), []).append(x)
+    seen = {}
+
+    for s in sheet_rows:
+        k = (s["unit"], s["meter"])
+        i = seen.get(k, 0)
+        seen[k] = i + 1
+        xs = by_unit.get(k, [])
+        x = xs[i] if i < len(xs) else None
+        if x is None:
+            problems.append(("R%d" % s["row"], "%s/%s" % k, "批次没有这一行"))
+            continue
+        if not close(s["prev"], x.get("prev")):
+            problems.append(("R%d" % s["row"], "%s/%s" % k,
+                             "上期 源表 %s ≠ 批次 %s" % (s["prev"], x.get("prev"))))
+        if s["cur"] is None:
+            # 未抄表：应 pending、不应存读数/用量/金额
+            if not x.get("pending"):
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "源表本期为空（未抄表）但批次计费了 %.2f"
+                                 % (x.get("amount") or 0)))
+            if x.get("qty") is not None or x.get("cur") is not None:
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "源表本期为空却存了读数/用量 %s" % x.get("qty")))
+        else:
+            if x.get("pending"):
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "源表本期有读数 %s 但批次判为未抄表" % s["cur"]))
+                continue
+            if not close(s["cur"], x.get("cur")):
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "本期 源表 %s ≠ 批次 %s" % (s["cur"], x.get("cur"))))
+            if not close(s["qty"], x.get("qty")):
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "用量 源表 %s ≠ 批次 %s" % (s["qty"], x.get("qty"))))
+            exp = round(float(s["qty"]) * float(s["price"]) * float(s["factor"]), 2)
+            if not close(exp, x.get("amount")):
+                problems.append(("R%d" % s["row"], "%s/%s" % k,
+                                 "金额 应为 %.2f ≠ 批次 %s" % (exp, x.get("amount"))))
+
+    if len(sheet_rows) < len(batch["details"]):
+        return problems, True                     # 源表是摘录：不做「行数/多余行」反向比对
+    for k, xs in by_unit.items():
+        if len(xs) != seen.get(k, 0):
+            problems.append(("批次", "%s/%s" % k,
+                             "批次 %d 行 ≠ 源表 %d 行" % (len(xs), seen.get(k, 0))))
+    return problems, False
+
+
+def main():
+    ap = argparse.ArgumentParser()
+    ap.add_argument("xlsx")
+    ap.add_argument("--sheet")
+    ap.add_argument("--batch")
+    a = ap.parse_args()
+
+    wb = openpyxl.load_workbook(a.xlsx, data_only=True)
+    wb.path = a.xlsx
+    titles = a.sheet and [a.sheet] or wb.sheetnames
+    rc = 0
+    for t in titles:
+        ws = wb[t]
+        if a.sheet is None and "电费统计表" not in str(ws.cell(1, 1).value or ""):
+            continue
+        rows = parse_sheet(ws)
+        if not rows:
+            continue
+        batch = pick_batch(t, a.batch)
+        print("=" * 78)
+        print("工作表《%s》 %s  | 抄表日 %s" % (
+            t, ws.cell(1, 1).value, str(ws.cell(2, 1).value or "")[:22]))
+        if batch is None:
+            print("  ! 找不到对应电费批次（可加 --batch <id>）")
+            rc = 1
+            continue
+        print("批次 %s《%s》 明细 %d 行 / 单据 %d 份 / 可计费 %d 行 / 合计 %.2f" % (
+            batch["batch_id"], batch.get("title"),
+            len(batch["details"]), len(batch.get("documents") or []),
+            batch["totals"]["row_count"] - batch["totals"]["pending_count"],
+            batch["totals"]["total_amount"]))
+        n_pend = sum(1 for s in rows if s["cur"] is None)
+        n_bill = len(rows) - n_pend
+        print("源表 数据行 %d：应未抄表 %d / 应计费 %d" % (len(rows), n_pend, n_bill))
+        exp_amt = round(sum(float(s["qty"]) * float(s["price"]) * float(s["factor"])
+                            for s in rows if s["cur"] is not None), 2)
+        print("源表 应计费金额 = %.2f" % exp_amt)
+
+        stale = stale_constants(ws)
+        if stale:
+            print("  ⚠ 用量列有 %d 处「手填常量且本期为空」（源表公式会算成空，系统却可能据此计费）："
+                  % len(stale))
+            for r, v in stale[:12]:
+                print("      R%-3d 用量=%s" % (r, v))
+
+        probs, excerpt = compare(rows, batch)
+        if excerpt:
+            print("  ⓘ 源表是摘录（%d 行 < 批次 %d 行），只核对了源表出现的行，不判「批次多出行」。"
+                  % (len(rows), len(batch["details"])))
+        print("  差异 %d 处" % len(probs))
+        for loc, key, msg in probs:
+            print("    %-8s %-14s %s" % (loc, key, msg))
+        if not probs:
+            print("    （一致）")
+        else:
+            rc = 1
+        print()
+    sys.exit(rc)
+
+
+if __name__ == "__main__":
+    main()