"""Reproduce a narrowly scoped T2 merged-category experiment; never edit inputs.""" import argparse import copy import csv import hashlib import importlib.metadata import json import platform import shutil from collections import defaultdict from datetime import datetime, timezone from pathlib import Path import openpyxl import pdfplumber def sha256(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def require(condition, message): if not condition: raise ValueError(message) def fill_categories(rows, geometry): """Only fill blanks whose row centre lies inside the preceding merged cell.""" require(len(rows) == len(geometry), "Row geometry mismatch") result = copy.deepcopy(rows) anchor = None changes = [] seen = set() for index, (row, boxes) in enumerate(zip(result, geometry)): require(len(row) == 6 and len(boxes) == 6, "Expected six columns") require(row[0] and row[0] not in seen, "Missing or duplicate row ID") seen.add(row[0]) require(boxes[0] is not None, "Missing ID geometry") centre_y = (boxes[0][1] + boxes[0][3]) / 2 if row[1] not in (None, ""): require(boxes[1] is not None, "Category without cell geometry") anchor = (row[1], boxes[1]) else: require(anchor is not None, "Leading category blank") require(boxes[1] is None, "Blank has its own cell; not a merged continuation") require(anchor[1][1] < centre_y < anchor[1][3], "Blank outside merged span") row[1] = anchor[0] changes.append({"id": row[0], "column": "分類", "before": rows[index][1], "after": row[1], "merged_bbox": list(anchor[1])}) return result, changes def typed(rows): result = [] for row in rows: normalized = list(row) for col in (3, 4, 5): value = str(row[col]).replace(",", "") require(value.isascii() and value.isdigit(), "Expected nonnegative integer") normalized[col] = int(value) result.append(normalized) return result def aggregate(rows): groups = defaultdict(lambda: {"rows": 0, "quantity": 0, "amount_yen": 0}) for row in typed(rows): group = groups[row[1] or "（空欄）"] group["rows"] += 1 group["quantity"] += row[3] group["amount_yen"] += row[5] return dict(groups) def safety_checks(rows, geometry): cases = [] bad = copy.deepcopy(rows) bad[0][1] = None cases.append(("leading_blank_rejected", bad, geometry)) bad_geometry = copy.deepcopy(geometry) bad_geometry[0][1] = list(bad_geometry[0][1]) bad_geometry[0][1][3] = bad_geometry[1][0][1] cases.append(("outside_span_rejected", rows, bad_geometry)) bad = copy.deepcopy(rows) bad[1][0] = bad[0][0] cases.append(("duplicate_id_rejected", bad, geometry)) bad_geometry = copy.deepcopy(geometry) bad_geometry[1][1] = list(bad_geometry[1][0]) cases.append(("independent_blank_cell_rejected", rows, bad_geometry)) results = {} for name, candidate, boxes in cases: try: fill_categories(candidate, boxes) except ValueError: results[name] = True else: raise ValueError("Safety test failed: " + name) return results def write_json(path, data): path.write_text(json.dumps(data, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--pdf", type=Path, required=True) parser.add_argument("--truth", type=Path, required=True) parser.add_argument("--out", type=Path, required=True, help="New output directory; must not exist") args = parser.parse_args() require(not args.out.exists(), "Output directory exists; choose a new directory") extracted = [] geometry = None for _ in range(3): with pdfplumber.open(args.pdf) as pdf: require(len(pdf.pages) == 1, "T2 must have one page") tables = pdf.pages[0].find_tables() require(len(tables) == 1, "Expected exactly one table") table = tables[0].extract() require(len(table) == 10, "Expected two header rows and eight data rows") require(table[0][:3] == ["管理番号", "分類", "品名"], "Unexpected T2 header") require(table[1][3:] == ["数量", "単価（円）", "金額（円）"], "Unexpected numeric headers") extracted.append(table) geometry = [list(row.cells) for row in tables[0].rows[2:]] require(extracted[0] == extracted[1] == extracted[2], "Extraction repeats differ") rows = extracted[0][2:] require([r[0] for r in rows] == [f"{i:04d}" for i in range(1, 9)], "Unexpected IDs or order") filled, changes = fill_categories(rows, geometry) workbook = openpyxl.load_workbook(args.truth, read_only=True, data_only=True) try: truth = [list(row) for row in workbook["T2"].iter_rows(min_row=6, max_col=6, values_only=True) if row[0] is not None] finally: workbook.close() require(len(truth) == 8, "Expected eight truth rows") require(typed(filled) == truth, "Filled data differs from ground truth") checks = safety_checks(rows, geometry) checks["three_extractions_identical"] = True checks["all_48_cells_match_truth_after_numeric_conversion"] = True checks["non_category_cells_unchanged"] = all( a[:1] + a[2:] == b[:1] + b[2:] for a, b in zip(rows, filled)) totals = {"before": aggregate(rows), "after": aggregate(filled), "truth": aggregate(truth)} grand_totals = {key: sum(g["amount_yen"] for g in groups.values()) for key, groups in totals.items()} checks["grand_total_preserved"] = len(set(grand_totals.values())) == 1 changed_amount = copy.deepcopy(filled) changed_amount[0][5] = str(int(changed_amount[0][5]) + 1) checks["altered_amount_detected_by_truth_comparison"] = typed(changed_amount) != truth require(all(checks.values()), "Invariant check failed") report = { "experiment": "T2 merged categories and category totals", "executed_at_utc": datetime.now(timezone.utc).isoformat(), "scope": "One recovered self-made fictional T2 PDF; fresh extraction, not Adobe output", "environment": {"python": platform.python_version(), "os": platform.system(), "packages": {p: importlib.metadata.version(p) for p in ("pdfplumber", "openpyxl")}}, "input_sha256": {"T2.pdf": sha256(args.pdf), "ground_truth.xlsx": sha256(args.truth)}, "script_sha256": sha256(__file__), "rows": len(rows), "columns": 6, "extraction_repeats": 3, "category_blanks_before": sum(not r[1] for r in rows), "category_blanks_after": sum(not r[1] for r in filled), "changed_cells": changes, "totals": totals, "grand_totals_yen": grand_totals, "named_categories_only_before_yen": sum(v["amount_yen"] for k, v in totals["before"].items() if k != "（空欄）"), "checks": checks, } args.out.mkdir(parents=True) inputs = args.out / "inputs" inputs.mkdir() shutil.copyfile(args.pdf, inputs / "T2.pdf") shutil.copyfile(args.truth, inputs / "ground_truth.xlsx") for filename in ("run_experiment.py", "requirements.txt", "README.md"): shutil.copyfile(Path(__file__).with_name(filename), args.out / filename) write_json(args.out / "raw.json", {"table": extracted[0], "data_row_geometry": geometry}) write_json(args.out / "filled.json", filled) write_json(args.out / "report.json", report) headers = ["管理番号", "分類", "品名", "数量", "単価（円）", "金額（円）"] for filename, data in (("before.csv", rows), ("after.csv", filled)): with (args.out / filename).open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.writer(handle) writer.writerow(headers) writer.writerows(data) with (args.out / "category_totals.csv").open("w", encoding="utf-8-sig", newline="") as handle: writer = csv.writer(handle) writer.writerow(["段階", "分類", "行数", "数量合計", "金額合計（円）"]) for stage, groups in totals.items(): for category, values in groups.items(): writer.writerow([stage, category, values["rows"], values["quantity"], values["amount_yen"]]) write_json(args.out / "manifest.json", { "algorithm": "SHA-256", "files": { path.relative_to(args.out).as_posix(): sha256(path) for path in sorted(args.out.rglob("*")) if path.is_file()}}) print(json.dumps({"blanks_before": report["category_blanks_before"], "grand_total_yen": grand_totals["after"], "totals": totals, "checks": checks}, ensure_ascii=True, indent=2)) if __name__ == "__main__": main()