"""Pure bounded reporting helpers; no Airflow import, no production extraction."""
from datetime import datetime
import hashlib
import json
import os
import uuid
from pathlib import Path

def digest(path):
    return hashlib.sha256(Path(path).read_bytes()).hexdigest()

def selected(path,start,end):
    start=datetime.fromisoformat(start);end=datetime.fromisoformat(end)
    if start>=end:raise ValueError("invalid interval")
    items={}
    for line in Path(path).read_text(encoding="utf-8").splitlines():
        event=json.loads(line)
        if type(event["amount_minor"]) is not int:raise ValueError("money must be integer minor units")
        time=datetime.fromisoformat(event["occurred_at"].replace("Z","+00:00"))
        if start<=time<end:
            key=(event["tenant_id"],event["event_id"])
            if key in items and items[key]!=event:raise ValueError("identity conflict")
            items[key]=event
    return list(items.values())

def discover(start,end,run_id,data):
    generation=hashlib.sha256(run_id.encode()).hexdigest()[:20]
    events=selected(data,start,end)
    return [{"tenant":tenant,"start":start,"end":end,"generation":generation,
             "input":str(data),"input_sha256":digest(data)}
            for tenant in sorted({e["tenant_id"] for e in events})]

def atomic_json(path,value):
    path=Path(path);path.parent.mkdir(parents=True,exist_ok=True)
    temporary=path.with_name(path.name+"."+uuid.uuid4().hex+".tmp")
    temporary.write_text(json.dumps(value,sort_keys=True),encoding="utf-8")
    os.replace(temporary,path)

def extract(spec,output):
    if digest(spec["input"])!=spec["input_sha256"]:raise ValueError("source manifest changed")
    events=[e for e in selected(spec["input"],spec["start"],spec["end"]) if e["tenant_id"]==spec["tenant"]]
    filename=hashlib.sha256(spec["tenant"].encode()).hexdigest()+".json"
    path=Path(output)/spec["generation"]/filename
    atomic_json(path,events)
    return {"uri":str(path),"sha256":digest(path),"count":len(events),"spec":spec}

def reconcile(manifests,output):
    manifests=list(manifests)
    if not manifests:raise ValueError("no source partitions: explicit empty-report policy needed")
    first=manifests[0]["spec"]
    fields=["start","end","generation","input_sha256","input"]
    for m in manifests:
        if any(m["spec"][k]!=first[k] for k in fields):raise ValueError("mixed generation/source cut")
        if digest(m["uri"])!=m["sha256"]:raise ValueError("output checksum mismatch")
        if len(json.loads(Path(m["uri"]).read_text(encoding="utf-8")))!=m["count"]:raise ValueError("manifest count mismatch")
    if len({m["spec"]["tenant"] for m in manifests})!=len(manifests):raise ValueError("duplicate manifest")
    if digest(first["input"])!=first["input_sha256"]:raise ValueError("source cut changed")
    expected=selected(first["input"],first["start"],first["end"])
    actual=[e for m in manifests for e in json.loads(Path(m["uri"]).read_text(encoding="utf-8"))]
    key=lambda e:(e["tenant_id"],e["event_id"])
    if sorted(actual,key=key)!=sorted(expected,key=key):raise ValueError("report reconciliation mismatch")
    totals={}
    for e in actual:
        k=e["tenant_id"]+"#"+e["currency"]
        totals[k]=totals.get(k,0)+e["amount_minor"]
    result={"generation":first["generation"],"interval":[first["start"],first["end"]],
            "source_sha256":first["input_sha256"],"count":len(actual),"totals":totals}
    target=Path(output)/first["generation"]/"validated-manifest.json"
    atomic_json(target,result)
    return {"uri":str(target),"sha256":digest(target),"count":len(actual)}
# Local shared volume demo; no S3 publication or cross-database snapshot claim.