#!/usr/bin/env python3 """Consume a pinned frozen bundle; preserve attempt-scoped output evidence.""" from __future__ import annotations import argparse import importlib.util import io import json import logging import os import pathlib import re from contextlib import closing, redirect_stdout from billing_inputs import canonical, load_frozen_inputs, sha256 MONTH_PATTERN = re.compile(r"^\d{4}-(0[1-9]|1[0-2])$") RUN_PATTERN = re.compile(r"^[a-zA-Z0-9_-]{8,96}$") def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument("--bill-month", required=True) parser.add_argument("--mode", required=True, choices=("predicted", "actual")) parser.add_argument("--run-id", required=True) parser.add_argument("--attempt-id", required=True) parser.add_argument("--definition-hash", required=True) parser.add_argument("--input-manifest-uri", required=True) parser.add_argument("--input-manifest-generation", required=True) parser.add_argument("--input-manifest-sha256", required=True) parser.add_argument("--reconcile-only", action="store_true", help="Read original jobs only; never submit missing work.") return parser.parse_args() def load_allocator(): path = pathlib.Path(__file__).parents[1] / "bigquery" / "monthly_cloud_billing_revrec.py" spec = importlib.util.spec_from_file_location("monthly_cloud_billing_revrec", path) if spec is None or spec.loader is None: raise RuntimeError(f"cannot load allocator from {path}") module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def validate_scope(args: argparse.Namespace): if not MONTH_PATTERN.fullmatch(args.bill_month): raise ValueError("bill_month must be YYYY-MM") if not RUN_PATTERN.fullmatch(args.run_id) or not RUN_PATTERN.fullmatch(args.attempt_id): raise ValueError("close/attempt identity invalid") if args.mode not in ("actual", "predicted") or not re.fullmatch(r"[a-f0-9]{16,64}", args.definition_hash): raise ValueError("execution scope invalid") return {"close_run_id": args.run_id, "bill_month": args.bill_month, "mode": args.mode, "definition_hash": args.definition_hash} def guarded_load(client, frame, destination, job_id, binding, location, reconcile_only=False): from google.api_core.exceptions import Conflict, NotFound from google.cloud import bigquery labels = {"portal_binding_a": binding[:32], "portal_binding_b": binding[32:]} try: job = client.get_job(job_id, location=location, timeout=20) except NotFound: if reconcile_only: raise ValueError("original output job not found; no work submitted") # A lost/expired Job is not permission to overwrite an existing output, # including an empty table. Preserve it for explicit reconciliation. try: client.get_table(destination, timeout=20) except NotFound: config = bigquery.LoadJobConfig(write_disposition="WRITE_EMPTY", labels=labels) try: job = client.load_table_from_dataframe(frame, destination, job_id=job_id, job_config=config, location=location, timeout=30) except Conflict: job = client.get_job(job_id, location=location, timeout=20) else: raise ValueError("existing output without its provider job; reconcile required") if (job.job_type != "load" or str(job.destination) != destination or job.write_disposition != "WRITE_EMPTY" or job.labels != labels or job.job_id != job_id): raise ValueError("existing output job scope or input binding mismatch") job.result(timeout=600) if job.error_result or job.output_rows != len(frame): raise ValueError("output job row count mismatch") table = client.get_table(destination, timeout=20) if table.num_rows != len(frame): raise ValueError("output table row count mismatch") return job def run(args, client, storage, bucket, output_dataset, location): scope = validate_scope(args) if not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_]+", output_dataset): raise ValueError("explicit output dataset required") allocator_path = pathlib.Path(__file__).parents[1] / "bigquery" / "monthly_cloud_billing_revrec.py" allocator_digest = sha256(allocator_path.read_bytes()) ref = {"uri": args.input_manifest_uri, "generation": args.input_manifest_generation, "sha256": args.input_manifest_sha256} frames, manifest = load_frozen_inputs(storage, bucket, ref, scope, allocator_digest) allocator = load_allocator() # The inherited algorithm prints some customer/order diagnostics. Keep them # out of the machine receipt and do not retain their financial contents. class DiscardText(io.TextIOBase): def write(self, value): return len(value) with redirect_stdout(DiscardText()): collections, orders = allocator.process(df_orders=frames["orders"], df_billing=frames["billing"], is_recognised_revenue=args.mode == "actual") attempt_key = sha256(canonical([args.run_id, args.attempt_id])) outputs = {} for kind, frame in (("collections", collections), ("orders", orders)): if "bill_month" not in frame.columns or not frame["bill_month"].eq(args.bill_month).all(): raise ValueError("allocator output month mismatch") destination = f"{output_dataset}.monthly_revrec_{kind}_{attempt_key}" payload_digest = sha256(frame.to_parquet(index=False)) binding = sha256(canonical([scope, args.attempt_id, ref, allocator_digest, kind, payload_digest])) job_id = f"portal_billing_{attempt_key}_{kind}" job = guarded_load(client, frame, destination, job_id, binding, location, getattr(args, "reconcile_only", False)) singular = "collection" if kind == "collections" else "order" outputs.update({kind: destination, singular + "_job": job.job_id, singular + "_rows": len(frame), singular + "_payload_sha256": payload_digest}) return {"run_id": args.run_id, "attempt_id": args.attempt_id, "bill_month": args.bill_month, "mode": args.mode, "definition_hash": args.definition_hash, "order_snapshot_date": manifest["order_snapshot_date"], "input_manifest": ref, "allocator_sha256": allocator_digest, "outputs": outputs, "verification": "not_checked", "output_integrity": "provider_and_table_row_count_only"} def main(): from google.cloud import bigquery, storage logging.disable(logging.CRITICAL) args = parse_args() validate_scope(args) dataset = os.getenv("FINANCE_PORTAL_OUTPUT_DATASET", "") bucket = os.getenv("FINANCE_PORTAL_ARTIFACT_BUCKET", "") location = os.getenv("FINANCE_PORTAL_BIGQUERY_LOCATION", "us-west2") if not bucket or not re.fullmatch(r"[A-Za-z0-9_-]+\.[A-Za-z0-9_]+", dataset): raise ValueError("artifact bucket and output dataset must be explicitly configured") with closing(bigquery.Client(project=dataset.split('.')[0], location=location)) as client, closing(storage.Client(project=dataset.split('.')[0])) as objects: print(json.dumps(run(args, client, objects, bucket, dataset, location), sort_keys=True)) if __name__ == "__main__": try: main() except Exception as exc: # No provider response or financial row is emitted on failure. print(json.dumps({"status": "failed", "error_type": type(exc).__name__})) raise SystemExit(1)