"""Audit and extract the current private business sources. The annual redemption detail is the sole sales fact source for the final dashboard. The historical comparison workbook is intentionally outside this ingestion boundary. """ import argparse import hashlib import json import posixpath import zipfile from pathlib import Path from xml.etree import ElementTree as ET ROOT = Path(__file__).resolve().parents[1] PRIVATE = ROOT / "新补充私有数据" OUT = ROOT / "research" / "private" NS = {"s": "http://schemas.openxmlformats.org/spreadsheetml/2006/main"} W = {"w": "http://schemas.openxmlformats.org/wordprocessingml/2006/main"} SOURCE_FILES = [ ("P01", "2026.9.18坑位 - 副本.xlsx"), ("P02", "兑换明细_业绩-2025.9.1-2026.8.31.xlsx"), ("P03", "饮水用具品类规划.xlsx"), ("P04", "饮水用具品类规划汇报.docx"), ("P05", "4、【商城】品类管理规划-模版V2.xlsx"), ] CURRENT_GENERATED_FILES = { "P01_坑位规划.json", "P01_上架清单9.18.json", "P01_去重上架清单9.18.json", "P02_年度饮水用具明细.json", "P02_年度厨房用具明细.json", "inventory.json", "extraction_summary.json", } def write_json(path, value): path.parent.mkdir(parents=True, exist_ok=True) path.write_text( json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False), encoding="utf-8", newline="\n", ) class WorkbookReader: def __init__(self, path): self.archive = zipfile.ZipFile(path) self.strings = [] if "xl/sharedStrings.xml" in self.archive.namelist(): tree = ET.fromstring(self.archive.read("xl/sharedStrings.xml")) self.strings = ["".join(node.itertext()) for node in tree.findall("s:si", NS)] workbook = ET.fromstring(self.archive.read("xl/workbook.xml")) relations = ET.fromstring(self.archive.read("xl/_rels/workbook.xml.rels")) rels = {node.attrib["Id"]: node.attrib["Target"] for node in relations} self.sheets = [] for sheet in workbook.findall("s:sheets/s:sheet", NS): rid = sheet.attrib["{http://schemas.openxmlformats.org/officeDocument/2006/relationships}id"] target = rels[rid] member = target.lstrip("/") if target.startswith("/") else posixpath.normpath("xl/" + target) self.sheets.append({"name": sheet.attrib["name"], "member": member, "state": sheet.attrib.get("state", "visible")}) def rows(self, sheet): with self.archive.open(sheet["member"]) as stream: for _, node in ET.iterparse(stream, events=("end",)): if node.tag != f"{{{NS['s']}}}row": continue cells, formulas, errors = {}, {}, {} for cell in node.findall("s:c", NS): address = cell.attrib["r"] value_node = cell.find("s:v", NS) value = value_node.text if value_node is not None else None kind = cell.attrib.get("t") if kind == "s" and value is not None: value = self.strings[int(value)] elif kind == "inlineStr": inline = cell.find("s:is", NS) value = "".join(inline.itertext()) if inline is not None else "" elif kind == "e": errors[address] = value elif value is not None and kind not in ("str", "d"): value = float(value) if value.is_integer(): value = int(value) if value is not None and str(value).strip(): cells[address] = value formula = cell.find("s:f", NS) if formula is not None: formulas[address] = {"text": formula.text, **formula.attrib} if cells or formulas: yield {"row": int(node.attrib["r"]), "cells": cells, "formulas": formulas, "errors": errors} node.clear() def close(self): self.archive.close() def col_letters(address): return "".join(char for char in address if char.isalpha()) def sheet_records(book, sheet): iterator = book.rows(sheet) first = next(iterator, None) if not first: return [], {} headers = {col_letters(cell): str(value).strip() for cell, value in first["cells"].items()} rows = [] for raw in iterator: record = {headers.get(col_letters(cell), cell): value for cell, value in raw["cells"].items()} record["_row"] = raw["row"] rows.append(record) return rows, headers def extract_workbook(source_id, path): book = WorkbookReader(path) sheets = [] for sheet in book.sheets: rows, headers = sheet_records(book, sheet) sheets.append({ "name": sheet["name"], "member": sheet["member"], "headers": headers, "rows": len(rows), "preview": rows[:12], }) if source_id == "P01": safe = { "坑位规划": "P01_坑位规划.json", "上架清单9.18": "P01_上架清单9.18.json", "去重上架清单9.18": "P01_去重上架清单9.18.json", }.get(sheet["name"]) if safe: write_json(OUT / safe, rows) elif source_id == "P02": kitchen = [ row for row in rows if str(row.get("后台二级分类", "")).strip() == "厨房用具" ] write_json(OUT / "P02_年度厨房用具明细.json", kitchen) kept = [ row for row in rows if str(row.get("后台三级分类", "")).strip() == "饮水用具" ] write_json(OUT / "P02_年度饮水用具明细.json", kept) book.close() return {"id": source_id, "file": path.relative_to(ROOT).as_posix(), "sheets": sheets} def extract_docx(source_id, path): with zipfile.ZipFile(path) as archive: document_xml = archive.read("word/document.xml") return { "id": source_id, "file": path.relative_to(ROOT).as_posix(), "documentBytes": len(document_xml), "role": "既有规划参考资料；最终销售事实采用P02全年兑换明细", } def inspect(): OUT.mkdir(parents=True, exist_ok=True) for path in OUT.iterdir(): if path.is_file() and path.name.startswith(("P01_", "P02_", "P03_", "P04_", "P05_")) and path.name not in CURRENT_GENERATED_FILES: path.unlink() inventory = [] for source_id, filename in SOURCE_FILES: path = PRIVATE / filename if path.suffix.lower() == ".xlsx": entry = extract_workbook(source_id, path) else: entry = extract_docx(source_id, path) entry["sha256"] = hashlib.sha256(path.read_bytes()).hexdigest() entry["bytes"] = path.stat().st_size inventory.append(entry) print(source_id, filename, flush=True) write_json(OUT / "inventory.json", inventory) def extract(): inspect() write_json( OUT / "extraction_summary.json", json.loads((OUT / "inventory.json").read_text(encoding="utf-8")), ) if __name__ == "__main__": parser = argparse.ArgumentParser() parser.add_argument("--inspect", action="store_true") parser.add_argument("--extract", action="store_true") parser.parse_args() extract()