"""Integrity-checked frozen billing inputs. No warehouse queries or writes."""
from __future__ import annotations

import hashlib
import io
import json
import re
from datetime import datetime, timezone
from urllib.parse import urlsplit

MAX_MANIFEST_BYTES = 1 << 20
MAX_PARQUET_BYTES = 64 << 20
MAX_UNCOMPRESSED_BYTES = 512 << 20
MAX_ROWS = 500_000
HEX = re.compile(r"[a-f0-9]{64}")


class InputContractError(ValueError):
    pass


def sha256(body):
    return hashlib.sha256(body).hexdigest()


def canonical(value):
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()


def object_name(uri, bucket):
    u = urlsplit(uri)
    if (not bucket or u.scheme != "gs" or u.netloc != bucket or u.query or u.fragment
            or not re.fullmatch(r"/finance-portal/frozen/[a-f0-9]{64}/(?:manifest.json|billing.parquet|orders.parquet)", u.path)):
        raise InputContractError("frozen artifact namespace mismatch")
    return u.path.lstrip("/")


def pinned_read(storage, bucket, ref, limit):
    if not isinstance(ref, dict) or not HEX.fullmatch(str(ref.get("sha256", ""))):
        raise InputContractError("artifact digest required")
    generation = str(ref.get("generation", ""))
    if not re.fullmatch(r"[1-9][0-9]{0,19}", generation):
        raise InputContractError("artifact generation required")
    name = object_name(ref.get("uri", ""), bucket)
    blob = storage.bucket(bucket).blob(name, generation=int(generation))
    blob.reload(timeout=20)
    if not blob.size or blob.size > limit or str(blob.generation) != generation:
        raise InputContractError("artifact size or generation mismatch")
    body = blob.download_as_bytes(if_generation_match=int(generation), timeout=30)
    if len(body) != blob.size or sha256(body) != ref["sha256"]:
        raise InputContractError("artifact content mismatch")
    return body


def validate_manifest(manifest, ref, scope, allocator_sha256, bucket):
    if not isinstance(manifest, dict) or manifest.get("contract") != "finance-billing-frozen-inputs/v1":
        raise InputContractError("unsupported input bundle contract")
    for key in ("close_run_id", "bill_month", "mode", "definition_hash"):
        if manifest.get(key) != scope[key]:
            raise InputContractError("frozen input scope mismatch")
    if manifest.get("allocator_sha256") != allocator_sha256:
        raise InputContractError("allocator version differs from frozen inputs")
    try:
        snapshot_date = datetime.strptime(manifest["order_snapshot_date"], "%Y-%m-%d").date()
        cutoff = datetime.fromisoformat(manifest["source_cutoff"])
    except (KeyError, TypeError, ValueError):
        raise InputContractError("frozen input time binding invalid") from None
    if snapshot_date.isoformat() != manifest["order_snapshot_date"] or cutoff.tzinfo is None or cutoff > datetime.now(timezone.utc):
        raise InputContractError("frozen input time binding invalid")
    freeze_attempt = manifest.get("freeze_attempt_id")
    if not isinstance(freeze_attempt, str) or not re.fullmatch(r"[A-Za-z0-9_-]{8,96}", freeze_attempt):
        raise InputContractError("freeze attempt identity required")
    expected_folder = "finance-portal/frozen/" + sha256(canonical([scope["close_run_id"], freeze_attempt]))
    if object_name(ref["uri"], bucket) != expected_folder + "/manifest.json":
        raise InputContractError("manifest attempt namespace mismatch")
    upstream = manifest.get("upstream_receipts")
    if not isinstance(upstream, list) or not upstream or any(not isinstance(r, dict) or not r.get("run_id") or not HEX.fullmatch(str(r.get("sha256", ""))) for r in upstream):
        raise InputContractError("upstream receipt bindings required")
    inputs = manifest.get("inputs")
    if not isinstance(inputs, dict) or set(inputs) != {"billing", "orders"}:
        raise InputContractError("billing and computed order inputs required together")
    for kind, item in inputs.items():
        if not isinstance(item, dict) or object_name(item.get("uri", ""), bucket) != expected_folder + "/" + kind + ".parquet":
            raise InputContractError("input artifact attempt mismatch")
        if type(item.get("rows")) is not int or not 0 <= item["rows"] <= MAX_ROWS or not HEX.fullmatch(str(item.get("schema_sha256", ""))):
            raise InputContractError("input row or schema binding invalid")
    business_scope = manifest.get("business_scope")
    if not isinstance(business_scope, dict) or business_scope.get("kind") != "billing_tenant_set" or not HEX.fullmatch(str(business_scope.get("sha256", ""))):
        raise InputContractError("business scope binding required")
    return snapshot_date


def decode_frame(body, ref):
    import pyarrow.parquet as pq

    parquet = pq.ParquetFile(io.BytesIO(body))
    if parquet.metadata.num_rows != ref["rows"] or parquet.metadata.num_rows > MAX_ROWS:
        raise InputContractError("frozen input row count mismatch")
    if sum(parquet.metadata.row_group(i).total_byte_size for i in range(parquet.metadata.num_row_groups)) > MAX_UNCOMPRESSED_BYTES:
        raise InputContractError("frozen input decompression bound exceeded")
    schema = parquet.schema_arrow.remove_metadata()
    if len(schema.names) != len(set(schema.names)) or sha256(schema.serialize().to_pybytes()) != ref["schema_sha256"]:
        raise InputContractError("frozen input schema mismatch")
    return parquet.read().to_pandas()


def validate_frames(frames, manifest, snapshot_date):
    import pandas as pd

    billing, orders = frames["billing"], frames["orders"]
    required_billing = {"billing_month", "tenant_id", "accumulate_revenue_to_org_tenant_id", "actual_revenue_recognition_date"}
    required_orders = {"date_day", "order_service_start_date", "accumulate_revenue_to_org_tenant_id",
                       "accumulated_actual_revenue_before_billing", "accumulated_actual_adjusted_revenue_before_billing",
                       "accumulated_actual_credit_amount_before_billing"}
    if not required_billing <= set(billing.columns) or not required_orders <= set(orders.columns):
        raise InputContractError("frozen input consumer columns missing")
    if billing.empty or not billing["billing_month"].eq(manifest["bill_month"]).all():
        raise InputContractError("billing month empty or mixed")
    recognition = billing["actual_revenue_recognition_date"]
    recognized = recognition.notna() & recognition.ne("")
    if not (recognized if manifest["mode"] == "actual" else ~recognized).all():
        raise InputContractError("billing mode mixed in frozen inputs")
    tenants = billing["accumulate_revenue_to_org_tenant_id"]
    if tenants.isna().any() or tenants.map(lambda t: not isinstance(t, str) or not t.strip() or t != t.strip()).any():
        raise InputContractError("billing tenant identity missing")
    tenant_set = sorted(set(tenants))
    if sha256(canonical(tenant_set)) != manifest["business_scope"]["sha256"]:
        raise InputContractError("billing business scope changed")
    if not orders.empty:
        # No filtering here: a producer must freeze the exact rows consumed,
        # including prior-period actual balances computed in its snapshot.
        days = pd.to_datetime(orders["date_day"], errors="coerce").dt.date
        starts = pd.to_datetime(orders["order_service_start_date"], errors="coerce").dt.date
        if days.isna().any() or starts.isna().any() or not days.eq(snapshot_date).all() or not starts.le(snapshot_date).all():
            raise InputContractError("order snapshot scope mixed")
        if not orders["accumulate_revenue_to_org_tenant_id"].isin(tenant_set).all():
            raise InputContractError("orders outside billing tenant scope")
        orders["date_day"], orders["order_service_start_date"] = days, starts


def load_frozen_inputs(storage, bucket, ref, scope, allocator_sha256):
    body = pinned_read(storage, bucket, ref, MAX_MANIFEST_BYTES)
    manifest = json.loads(body)
    snapshot_date = validate_manifest(manifest, ref, scope, allocator_sha256, bucket)
    frames = {kind: decode_frame(pinned_read(storage, bucket, item, MAX_PARQUET_BYTES), item)
              for kind, item in manifest["inputs"].items()}
    validate_frames(frames, manifest, snapshot_date)
    return frames, manifest