"""Independent reconciliation of annual facts, lifecycle, plan and downloads.""" import csv import hashlib import json import sys from collections import Counter, defaultdict from decimal import Decimal from html.parser import HTMLParser from pathlib import Path from release_config import OUTPUT, EVIDENCE, VERSION ROOT = Path(__file__).resolve().parents[1] EXPECTED_MONTHS = ["2025-09", "2025-10", "2025-11", "2025-12", "2026-01", "2026-02", "2026-03", "2026-04", "2026-05", "2026-06", "2026-07", "2026-08"] FIELDS = {"amount": "兑换金额", "quantity": "销量", "gross": "兑换毛利额（含C端运费）"} def read(path, **kwargs): return json.loads((ROOT / path).read_text(encoding="utf-8"), **kwargs) def close(actual, expected, tolerance="0.0001"): assert actual is not None assert abs(Decimal(str(actual)) - Decimal(str(expected))) <= Decimal(tolerance), (actual, expected) def totals(rows): return {key: sum((Decimal(str(r.get(field, 0) or 0)) for r in rows), Decimal(0)) for key, field in FIELDS.items()} def match(rows, result): values = totals(rows) for key in FIELDS: close(result[key], values[key]) if values["amount"]: close(result["margin"], values["gross"] / values["amount"], "0.000000001") class Document(HTMLParser): def __init__(self): super().__init__() self.ids, self.tags, self.data, self.active = [], Counter(), "", False def handle_starttag(self, tag, attrs): values = dict(attrs) self.tags[tag] += 1 if "id" in values: self.ids.append(values["id"]) if tag == "script": self.active = values.get("id") == "dashboard-data" assert "src" not in values def handle_endtag(self, tag): if tag == "script": self.active = False def handle_data(self, value): if self.active: self.data += value def main(): a = read(EVIDENCE / "analysis.json") d = read(EVIDENCE / "dashboard_data.json") facts = read("research/private/P02_年度饮水用具明细.json", parse_float=Decimal) kitchen = read("research/private/P02_年度厨房用具明细.json", parse_float=Decimal) raw = (OUTPUT / "饮水用具行业情报看板.html").read_bytes() html = raw.decode("utf-8") doc = Document() doc.feed(html) assert not raw.startswith(b"\xef\xbb\xbf") assert len(doc.ids) == len(set(doc.ids)) assert all(doc.tags[t] == 1 for t in ("html", "head", "body")) assert json.loads(doc.data) == d assert d["business"] == a assert d["version"] == a["version"] == VERSION snapshot = read(EVIDENCE / "在售快照.json") latest = snapshot["rows"] assert {r["spu"] for r in a["spus"]} == {str(r["spu编码"]) for r in latest} assert len({str(r["sku编码"]) for r in latest}) == 290 assert a["listingSource"] == "P06" assert a["period"] == "2025-09至2026-08" assert a["factSource"] == "P02" assert len(facts) == 18314 assert {r["年月-订单创建"] for r in kitchen} == set(EXPECTED_MONTHS) assert all(r["后台二级分类"] == "厨房用具" for r in kitchen) assert len({r["后台三级分类"] for r in kitchen}) > 1 assert a["quality"]["months"] == EXPECTED_MONTHS match(facts, a["overall"]) match(kitchen, a["kitchenOverall"]) for row in a["margins"]: match([r for r in kitchen if r["频道"] == row["channel"]], row) assert row["margin"] >= row["floor"] for row in a["channels"]: match([r for r in facts if r["频道"] == row["channel"]], row) for row in a["monthly"]: match([r for r in facts if r["年月-订单创建"] == row["month"]], row) for rows in [a["monthly"]] + [[r for r in a["monthlyByChannel"] if r["channel"] == channel] for channel in ("吉拉福", "自营实物")]: assert len(rows) == 12 rows = sorted(rows, key=lambda r: r["month"]) assert rows[0]["mom"]["amountRate"] is None for prior, row in zip(rows, rows[1:]): for key in FIELDS: close(row["mom"][key + "Delta"], row[key] - prior[key]) if prior[key]: close(row["mom"][key + "Rate"], (row[key] - prior[key]) / prior[key]) match([r for r in facts if r["年月-订单创建"] in EXPECTED_MONTHS[:6]], a["half"]["first"]) match([r for r in facts if r["年月-订单创建"] in EXPECTED_MONTHS[6:]], a["half"]["second"]) close(a["bridge"]["volumeEffect"] + a["bridge"]["unitValueEffect"], a["half"]["amountDelta"]) close(a["bridge"]["amountEffectOnGross"] + a["bridge"]["marginEffectOnGross"], a["half"]["grossDelta"]) for rows in (a["categoryChanges"], a["brandChanges"]): close(sum(r["amountDelta"] for r in rows), a["half"]["amountDelta"]) close(sum(r["grossDelta"] for r in rows), a["half"]["grossDelta"]) sku_facts = defaultdict(list) for row in facts: sku_facts[row["sku_code"]].append(row) for row in a["skuLifecycle"]: source = sku_facts[row["sku"]] match(source, row["annual"]) active = [m for m in EXPECTED_MONTHS if any(r["年月-订单创建"] == m and (r.get("兑换金额", 0) > 0 or r.get("销量", 0) > 0) for r in source)] assert row["life"]["months"] == active assert row["life"]["activeMonths"] == len(active) close(row["life"]["coverage"], len(active) / 12) assert len(a["skuLifecycle"]) == 635 assert len(a["spus"]) == len(a["finalAssortment"]) == 122 assert len({r["spu"] for r in a["spus"]}) == 122 assert len({r["id"] for r in a["finalAssortment"]}) == 122 assert sum(r["currentSlots"] for r in a["allocations"]) == 122 assert sum(r["recommendedSlots"] for r in a["allocations"]) == 122 assert sum(r["delta"] for r in a["allocations"]) == 0 assigned = [sku for r in a["candidatePool"] for sku in r["evidenceSkus"]] assert len(assigned) == len(set(assigned)) == 635 for row in a["candidatePool"]: related = [fact for sku in row["evidenceSkus"] for fact in sku_facts[sku]] match(related, row["annual"]) for row in a["finalAssortment"]: assert row["action"] and row["life"]["stage"] and row["reason"] assert row["finalCategory"] in {r["category"] for r in a["allocations"] if r["recommendedSlots"]} if row["current"]: actual_skus = {str(r["sku编码"]) for r in latest if str(r["spu编码"]) == row["spu"]} assert set(row["skus"]) <= actual_skus assert set(row["listedSkus"]) == actual_skus match([fact for sku in actual_skus for fact in sku_facts[sku]], row["listedSkuAnnual"]) elif row["spu"] is None: assert row["identity"] == "历史品款组" else: assert row["identity"] == "历史系统SPU" new_jug = next(r for r in a["finalAssortment"] if r["spu"] == "210113131300") assert new_jug["finalCategory"] == "滤水壶" and new_jug["newlyObserved"] assert new_jug["action"] == "新增在售观察" assert new_jug["annual"]["amount"] == 0 and new_jug["annual"]["margin"] is None assert new_jug["firstListed"] is None assert new_jug["skus"] == ["21011313130001", "21011313130002"] mixed = next(r for r in a["finalAssortment"] if r["spu"] == "210113111487") assert mixed["skus"] == ["21011311148701"] assert "手冲" in mixed["name"] current_ids = {r["id"] for r in a["spus"]} selected_ids = {r["id"] for r in a["finalAssortment"]} assert current_ids - selected_ids == {r["outId"] for r in a["replacements"]} assert selected_ids - current_ids == {r["inId"] for r in a["replacements"]} for key, rows in (("current", a["spus"]), ("selected", a["finalAssortment"])): for metric in FIELDS: close(sum(r["annual"][metric] for r in rows), a["assortmentComparison"][key][metric]) assert all(r["marginHeadroom"] >= 0 for r in a["basketMargins"]) assert "kitchenCapacity" not in a assert a["plan"]["capacity"] == 122 assert a["plan"]["kept"] + a["plan"]["historicalGroups"] == 122 assert a["plan"]["currentCount"] - a["plan"]["replacements"] + a["plan"]["historicalGroups"] == 122 for source in d["sources"]: if source.get("file"): assert (OUTPUT / source["file"]).is_file(), source for path in source.get("images", {}).values(): assert (OUTPUT / path).is_file(), path manifest = read(EVIDENCE / "版本清单.json") for file, expected_hash in manifest["previousDeliverableHashes"].items(): assert hashlib.sha256((ROOT / file).read_bytes()).hexdigest() == expected_hash, file assert hashlib.sha256((ROOT / a["sourceFile"]).read_bytes()).hexdigest() == manifest["factSha256"] exports = {} for key, item in a["exports"].items(): with (OUTPUT / item["file"]).open(encoding="utf-8-sig", newline="") as handle: values = list(csv.reader(handle)) assert values[0] == item["headers"] expected = [["" if x is None else str(x) for x in row] for row in item["rows"]] assert values[1:] == expected, item["file"] exports[key] = len(expected) if "--exports" in sys.argv: with (EVIDENCE / "playwright" / ("download-" + key + ".csv")).open( encoding="utf-8-sig", newline="") as handle: downloaded = list(csv.reader(handle)) assert downloaded[0] == values[0] assert len(downloaded) == len(values) for actual_row, typed_row in zip(downloaded[1:], item["rows"]): assert len(actual_row) == len(typed_row) for actual, expected_cell in zip(actual_row, typed_row): if isinstance(expected_cell, (float, int)): close(actual, expected_cell, "0.000000001") else: assert actual == ("" if expected_cell is None else expected_cell), key if "--exports" in sys.argv: assert (EVIDENCE / "playwright/download-review.md").read_text(encoding="utf-8") == d["reviewMarkdown"] with (EVIDENCE / "playwright/download-margin.csv").open(encoding="utf-8-sig", newline="") as handle: rows = list(csv.DictReader(handle)) assert len(rows) == 2 for row, expected in zip(rows, a["margins"]): close(row["年度兑换金额"], expected["amount"]) forbidden = ["最终货盘待" + "关键补数", "完整年度" + "同比", "建议配额待" + "商品", "需要补充的" + "数据", "待" + "计算", "2024年9月" + "至2025年8月", "2025.4-9& " + "2026.4-9月销售数据汇总.xlsx", "厨房用具总量释放清单", "其他三级分类释放", "kitchenCapacity"] for path in (OUTPUT / "饮水用具行业情报看板.html", OUTPUT / "饮水用具全年经营与品类规划.md", EVIDENCE / "analysis.json", EVIDENCE / "dashboard_data.json"): text = path.read_text(encoding="utf-8") for value in forbidden: assert value not in text, (value, path) report = { "status": "passed", "period": a["period"], "rows": len(facts), "kitchenRows": len(kitchen), "annualAmount": a["overall"]["amount"], "annualGross": a["overall"]["gross"], "halfAmountRate": a["half"]["amountRate"], "months": 12, "skuLifecycle": 635, "currentSpus": 122, "proposedSlots": 122, "keptSystemSpus": a["plan"]["kept"], "historicalProductGroups": a["plan"]["historicalGroups"], "exports": exports, "checks": ["Decimal facts", "whole kitchen margin scope", "month-on-month", "half-year bridges", "classification and brand contributions", "SKU lifecycles", "unique SKU ownership", "fixed 122 slots and replacements", "9.21 SPU and SKU identity", "filter jug classification", "source links", "previous release hashes", "CSV equality", "obsolete wording cleanup"], "htmlSha256": hashlib.sha256(raw).hexdigest(), } (EVIDENCE / "static_validation.json").write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") print(json.dumps(report, ensure_ascii=False, indent=2)) if __name__ == "__main__": main()