#!/usr/bin/env python3 import argparse import hashlib import json from collections import defaultdict from datetime import datetime, timezone from decimal import Decimal from pathlib import Path MONEY = Decimal("0.01") COMPARE_FIELDS = ( "tenant_id", "status", "invoice_type", "invoice_date", "created_at", "sent_date", "cost_price", "raw_price", ) def read_jsonl(path: Path) -> list[dict]: with path.open(encoding="utf-8") as handle: return [json.loads(line) for line in handle if line.strip()] def write_json(path: Path, value: object) -> None: path.write_text(json.dumps(value, indent=2, ensure_ascii=False) + "\n", encoding="utf-8") def sha256(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def money(value: Decimal) -> str: return format(value.quantize(MONEY), "f") def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--primary", type=Path, required=True) parser.add_argument("--comparison", type=Path, required=True) parser.add_argument("--output-dir", type=Path, required=True) parser.add_argument("--run-id", required=True) parser.add_argument("--tenant-id", required=True) parser.add_argument("--period-start", required=True) parser.add_argument("--period-end", required=True) parser.add_argument("--source-snapshot", type=Path, required=True) parser.add_argument("--bigquery-job", type=Path, required=True) args = parser.parse_args() primary = read_jsonl(args.primary) comparison = read_jsonl(args.comparison) primary_by_id = {record["invoice_id"]: record for record in primary} comparison_by_id = {record["invoice_id"]: record for record in comparison} monthly = defaultdict(lambda: {"invoices": [], "cost": Decimal("0"), "raw": Decimal("0")}) for record in primary: issue_month = record["created_at"][:7] monthly[issue_month]["invoices"].append(record) monthly[issue_month]["cost"] += Decimal(record["cost_price"]) monthly[issue_month]["raw"] += Decimal(record["raw_price"]) monthly_results = [] for issue_month in sorted(monthly): values = monthly[issue_month] monthly_results.append( { "issue_month": issue_month, "invoice_count": len(values["invoices"]), "cost_price_total": money(values["cost"]), "raw_price_total": money(values["raw"]), "invoices": values["invoices"], } ) total_cost = sum((Decimal(record["cost_price"]) for record in primary), Decimal("0")) total_raw = sum((Decimal(record["raw_price"]) for record in primary), Decimal("0")) result = { "run_id": args.run_id, "tenant_id": args.tenant_id, "currency": "USD", "period_start_inclusive": args.period_start, "period_end_exclusive": args.period_end, "grouping": "created_at UTC calendar month", "invoice_count": len(primary), "cost_price_total": money(total_cost), "raw_price_total": money(total_raw), "months": monthly_results, } mismatches = [] for invoice_id in sorted(primary_by_id.keys() & comparison_by_id.keys()): differences = { field: {"primary": primary_by_id[invoice_id].get(field), "comparison": comparison_by_id[invoice_id].get(field)} for field in COMPARE_FIELDS if primary_by_id[invoice_id].get(field) != comparison_by_id[invoice_id].get(field) } if differences: mismatches.append({"invoice_id": invoice_id, "differences": differences}) primary_only = sorted(primary_by_id.keys() - comparison_by_id.keys()) comparison_only = sorted(comparison_by_id.keys() - primary_by_id.keys()) validation = { "status": "passed_with_source_freshness_warning" if primary_only else "passed", "primary_row_count": len(primary), "comparison_row_count": len(comparison), "overlap_count": len(primary_by_id.keys() & comparison_by_id.keys()), "overlap_value_mismatches": mismatches, "primary_only_invoice_ids": primary_only, "comparison_only_invoice_ids": comparison_only, "duplicate_primary_invoice_ids": len(primary_by_id) != len(primary), "monthly_total_reconciliation": money(sum(Decimal(month["cost_price_total"]) for month in monthly_results)) == money(total_cost), } lineage = { "run_id": args.run_id, "generated_at": datetime.now(timezone.utc).isoformat(), "rule": {"id": "tenant-invoice-history", "version": 1, "status": "draft"}, "query_bundle_version": 1, "input_schema_version": 1, "calculation_version": 1, "primary_source": { "kind": "production_database_csv_export", "snapshot": str(args.source_snapshot), "sha256": sha256(args.source_snapshot), "normalized": str(args.primary), "normalized_sha256": sha256(args.primary), }, "comparison_source": { "kind": "bigquery", "job_metadata": str(args.bigquery_job), "normalized": str(args.comparison), "normalized_sha256": sha256(args.comparison), }, "calculation_script_sha256": sha256(Path(__file__)), } args.output_dir.mkdir(parents=True, exist_ok=True) write_json(args.output_dir / "result.json", result) write_json(args.output_dir / "validation.json", validation) write_json(args.output_dir / "lineage.json", lineage) lines = [ "# Manual invoice history", "", "## Billing summary", "", f"- Customer ID: `{args.tenant_id}`", f"- Invoice creation period: `{args.period_start}` to `{args.period_end}` (end exclusive)", "- Currency: `USD`", f"- Total cost price: `{result['cost_price_total']}`", f"- Total raw price: `{result['raw_price_total']}`", f"- Run ID: `{args.run_id}`", "", "## Monthly invoices", "", "| Issue month | Invoice ID | Type | Invoice date | Status | Cost price | Raw price |", "|---|---:|---|---|---|---:|---:|", ] for month in monthly_results: for invoice in month["invoices"]: lines.append( f"| {month['issue_month']} | {invoice['invoice_id']} | {invoice['invoice_type']} | " f"{invoice['invoice_date'][:10]} | {invoice['status']} | {money(Decimal(invoice['cost_price']))} | " f"{money(Decimal(invoice['raw_price']))} |" ) lines.append( f"| **{month['issue_month']} subtotal** | | | | | **{month['cost_price_total']}** | " f"**{month['raw_price_total']}** |" ) lines.extend( [ "", "## Validation and reconciliation", "", f"- Production export rows: `{len(primary)}`", f"- BigQuery comparison rows: `{len(comparison)}`", f"- Exact overlapping invoice IDs: `{len(primary_by_id.keys() & comparison_by_id.keys())}`", f"- Overlap value mismatches: `{len(mismatches)}`", f"- Production-only invoice IDs: `{', '.join(primary_only) if primary_only else 'none'}`", "- Status: `DRAFT`", "", ] ) (args.output_dir / "manual-invoice.md").write_text("\n".join(lines), encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())