#!/usr/bin/env python3 """Validate the lifetime aggregate snapshot and delivery artifacts.""" from __future__ import annotations import csv import json from datetime import date from pathlib import Path ROOT = Path(__file__).resolve().parent PROJECTS = {"airport", "saidi"} DIMENSION_PERIODS = {"上线以来", "近30天", "前30天"} EXPECTED_LAUNCH = {"airport": "2026-01-20", "saidi": "2026-01-18"} def read_csv(name: str) -> list[dict[str, str]]: with (ROOT / name).open(encoding="utf-8-sig", newline="") as handle: return list(csv.DictReader(handle)) def integer(row: dict[str, str], field: str) -> int: return int(float(row[field])) def number(row: dict[str, str], field: str) -> float: return float(row[field]) def main() -> None: meta = read_csv("project_meta.csv") summary = read_csv("order_summary.csv") dimensions = { "渠道": read_csv("channel_mix.csv"), "支付方式": read_csv("payment_mix.csv"), "餐段": read_csv("meal_mix.csv"), } daily = read_csv("daily_usage.csv") monthly = read_csv("monthly_usage.csv") frequency = read_csv("user_frequency.csv") weekdays = read_csv("weekday_mix.csv") restaurants = read_csv("restaurant_rank.csv") features = read_csv("feature_activity.csv") management = read_csv("management_activity.csv") quality = read_csv("data_quality.csv") meta_by_project = {row["project_key"]: row for row in meta} assert set(meta_by_project) == PROJECTS for project, expected_start in EXPECTED_LAUNCH.items(): assert meta_by_project[project]["observable_launch_date"] == expected_start assert meta_by_project[project]["observable_launch_definition"] == "首个有效订单日" summary_by_key = {(row["project_key"], row["period"]): row for row in summary} assert PROJECTS == {row["project_key"] for row in summary} for project in PROJECTS: life = summary_by_key[(project, "上线以来")] current = summary_by_key[(project, "近30天")] assert life["start_date"] == EXPECTED_LAUNCH[project] assert life["end_date"] == current["end_date"] == "2026-08-03" assert (date.fromisoformat(current["end_date"]) - date.fromisoformat(current["start_date"])).days == 29 assert integer(life, "orders") >= integer(current, "orders") > 0 assert integer(life, "active_users") >= integer(current, "active_users") > 0 for label, rows in dimensions.items(): for period in DIMENSION_PERIODS: part = [row for row in rows if row["project_key"] == project and row["period"] == period] assert part, f"{project} {period} missing {label} rows" assert sum(integer(row, "orders") for row in part) == integer(summary_by_key[(project, period)], "orders") assert abs(sum(number(row, "order_share") for row in part) - 1) <= 0.00001 current_daily = [ row for row in daily if row["project_key"] == project and current["start_date"] <= row["date"] <= current["end_date"] ] assert len(current_daily) == 30 assert sum(integer(row, "orders") for row in current_daily) == integer(current, "orders") project_months = [row for row in monthly if row["project_key"] == project] assert project_months assert sum(integer(row, "orders") for row in project_months) == integer(life, "orders") for period in ("上线以来", "近30天"): project_frequency = [ row for row in frequency if row["project_key"] == project and row["period"] == period ] assert sum(integer(row, "users") for row in project_frequency) == integer(summary_by_key[(project, period)], "active_users") assert sum(integer(row, "orders") for row in project_frequency) == integer(summary_by_key[(project, period)], "orders") project_weekdays = [ row for row in weekdays if row["project_key"] == project and row["period"] == period ] assert len(project_weekdays) == 7 assert sum(integer(row, "orders") for row in project_weekdays) == integer(summary_by_key[(project, period)], "orders") project_restaurants = [row for row in restaurants if row["project_key"] == project] assert sum(integer(row, "orders") for row in project_restaurants) == integer(life, "orders") assert [integer(row, "rank") for row in project_restaurants] == list(range(1, len(project_restaurants) + 1)) for row in features: assert integer(row, "table_total_records") >= integer(row, "lifetime_records") >= 0 assert integer(row, "lifetime_records") >= integer(row, "current_30d_records") >= 0 assert integer(row, "lifetime_records") >= integer(row, "previous_30d_records") >= 0 for project in PROJECTS: life_management = [row for row in management if row["project_key"] == project and row["period"] == "上线以来"] current_management = [row for row in management if row["project_key"] == project and row["period"] == "近30天"] assert life_management and current_management core_quality = [row for row in quality if not row.get("quality_check")] assert {row["project_key"] for row in core_quality} == PROJECTS for row in core_quality: assert integer(row, "rows") == integer(row, "distinct_ids") == integer(row, "distinct_order_numbers") assert integer(row, "missing_user_orders") == 0 assert row["first_meal_date"] == EXPECTED_LAUNCH[row["project_key"]] pickup_lifetime = next( row for row in quality if row.get("quality_check") == "取餐流程状态" and row.get("period") == "上线以来" ) assert integer(pickup_lifetime, "prepare_status") == 10 assert integer(pickup_lifetime, "records") == 257 artifact = json.loads((ROOT / "artifact.json").read_text(encoding="utf-8")) title = artifact["manifest"]["title"] assert title == "赛迪与首都机场：上线以来哪些功能真正被使用" assert len(artifact["manifest"]["charts"]) == 4 assert len(artifact["manifest"]["tables"]) == 7 datasets = artifact["snapshot"]["datasets"] assert sum(row["orders"] for row in datasets["monthly_airport"]) == 895908 assert sum(row["orders"] for row in datasets["monthly_saidi"]) == 145955 html = (ROOT / "saidi-airport-feature-usage.html").read_text(encoding="utf-8") assert title in html assert "data-analytics-portable-artifact-payload-source" in html assert "2026-01-18" in html and "2026-01-20" in html print("PASS: lifetime reconciliation, launch dates, feature evidence, artifact structure, and HTML delivery validated") if __name__ == "__main__": main()