"""Local-only reporting pipeline: validate, reconcile, stage, then retain a run bundle."""
from __future__ import annotations
import argparse
import csv
import hashlib
import io
import json
import os
import platform
import re
import sys
import tempfile
from datetime import datetime, timezone
from pathlib import Path

FIELDS=['event_id','occurred_at','received_at','region','amount_paise','status']
REGIONS={'North','South','Unknown'}


class ReportError(ValueError):
    pass


def digest(data):return hashlib.sha256(data).hexdigest()


def canonical(value):return (json.dumps(value,sort_keys=True,indent=2)+'\n').encode('utf-8')


def timestamp(value):
    if not isinstance(value,str) or not re.fullmatch(r'\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}Z',value):
        raise ReportError('invalid_utc_timestamp')
    try:return datetime.strptime(value,'%Y-%m-%dT%H:%M:%SZ').replace(tzinfo=timezone.utc)
    except ValueError as error:raise ReportError('invalid_utc_timestamp') from error


def validate_config(config):
    if set(config)!={'period_start','period_end','region','max_extract_lag_seconds'}:
        raise ReportError('invalid_config_fields')
    start,end=timestamp(config['period_start']),timestamp(config['period_end'])
    if start>=end:raise ReportError('invalid_period')
    if config['region'] not in REGIONS|{'all'}:raise ReportError('invalid_region')
    lag=config['max_extract_lag_seconds']
    if type(lag) is not int or lag<0:raise ReportError('invalid_extract_lag')
    return start,end


def load_source(data_path,manifest_path,config):
    start,end=validate_config(config)
    raw=Path(data_path).read_bytes();manifest_raw=Path(manifest_path).read_bytes()
    manifest=json.loads(manifest_raw)
    if type(manifest.get('source_schema_version')) is not int or manifest['source_schema_version']!=1:raise ReportError('source_schema_version')
    if digest(raw)!=manifest.get('data_sha256'):raise ReportError('source_hash_mismatch')
    if type(manifest.get('raw_rows')) is not int or manifest['raw_rows']<0:raise ReportError('source_row_count_invalid')
    watermark=timestamp(manifest['event_watermark']);extracted=timestamp(manifest['extracted_at'])
    if watermark<end:raise ReportError('source_watermark_before_period_end')
    if extracted<watermark or (extracted-end).total_seconds()>config['max_extract_lag_seconds']:
        raise ReportError('source_extract_outside_allowed_window')
    reader=csv.DictReader(io.StringIO(raw.decode('utf-8'),newline=''),strict=True)
    if reader.fieldnames!=FIELDS:raise ReportError('source_header_mismatch')
    rows=list(reader)
    if len(rows)!=manifest['raw_rows']:raise ReportError('source_row_count_mismatch')
    unique={};replays=0
    for row in rows:
        if set(row)!=set(FIELDS) or any(v is None for v in row.values()):raise ReportError('source_row_shape')
        if not re.fullmatch(r'E[1-9][0-9]*',row['event_id']):raise ReportError('invalid_event_id')
        occurred,received=timestamp(row['occurred_at']),timestamp(row['received_at'])
        if occurred>received or received>extracted:raise ReportError('invalid_event_chronology')
        if row['region'] not in REGIONS or row['status'] not in {'paid','cancelled'}:raise ReportError('invalid_event_enum')
        if not re.fullmatch(r'0|[1-9][0-9]*',row['amount_paise']):raise ReportError('invalid_amount')
        key=row['event_id']
        if key in unique:
            if unique[key]!=row:raise ReportError('conflicting_event_id')
            replays+=1
        else:unique[key]=row
    selected=[row for row in unique.values() if start<=timestamp(row['occurred_at'])<end
              and row['status']=='paid' and (config['region']=='all' or row['region']==config['region'])]
    selected.sort(key=lambda row:row['event_id'])
    groups={}
    for row in selected:
        group=groups.setdefault(row['region'],{'events':0,'amount_paise':0})
        group['events']+=1;group['amount_paise']+=int(row['amount_paise'])
    total=sum(int(row['amount_paise']) for row in selected)
    if sum(group['amount_paise'] for group in groups.values())!=total:raise ReportError('reconciliation_failed')
    metrics={'period_start':config['period_start'],'period_end':config['period_end'],'region_filter':config['region'],
             'currency':'INR','amount_unit':'paise','raw_rows':len(rows),'unique_events':len(unique),
             'identical_replays':replays,'selected_events':len(selected),'amount_paise':total,
             'by_region':groups,'event_watermark':manifest['event_watermark'],'extracted_at':manifest['extracted_at']}
    evidence={'source_sha256':digest(raw),'source_manifest_sha256':digest(manifest_raw),
              'config_sha256':digest(canonical(config)),'code_sha256':digest(Path(__file__).read_bytes()),
              'python_version':platform.python_version()}
    return selected,metrics,evidence


def verify_bundle(folder,expected_evidence=None):
    folder=Path(folder)
    manifest=json.loads((folder/'manifest.json').read_text(encoding='utf-8'))
    if expected_evidence is not None and manifest['evidence']!=expected_evidence:raise ReportError('run_evidence_mismatch')
    if manifest['run_id']!=digest(canonical(manifest['evidence'])):raise ReportError('run_id_mismatch')
    if set(manifest['outputs'])!={'metrics.json','selected-events.csv'}:raise ReportError('unexpected_bundle_outputs')
    for name,expected in manifest['outputs'].items():
        if digest((folder/name).read_bytes())!=expected:raise ReportError('bundle_checksum_mismatch')
    return manifest


def run(data_path,manifest_path,config,output_root,fail_after=None):
    selected,metrics,evidence=load_source(data_path,manifest_path,config)
    root=Path(output_root).resolve();root.mkdir(parents=True,exist_ok=True)
    run_id=digest(canonical(evidence));target=root/run_id
    if target.exists():
        verify_bundle(target,evidence)
        return target,False
    # Staging and destination share one output root/filesystem. No existing bundle is overwritten.
    with tempfile.TemporaryDirectory(prefix='.report-stage-',dir=root) as temporary:
        staging_parent=Path(temporary).resolve()
        if staging_parent.parent!=root:raise ReportError('staging_outside_output_root')
        stage=staging_parent/'bundle';stage.mkdir()
        (stage/'metrics.json').write_bytes(canonical(metrics))
        if fail_after=='metrics':raise ReportError('injected_failure_after_metrics')
        with (stage/'selected-events.csv').open('w',encoding='utf-8',newline='') as handle:
            writer=csv.DictWriter(handle,fieldnames=FIELDS);writer.writeheader();writer.writerows(selected)
        outputs={name:digest((stage/name).read_bytes()) for name in ('metrics.json','selected-events.csv')}
        manifest={'run_id':run_id,'evidence':evidence,'outputs':outputs,
                  'created_at':datetime.now(timezone.utc).strftime('%Y-%m-%dT%H:%M:%SZ'),
                  'status':'prepared_for_review','distribution':'not_sent'}
        (stage/'manifest.json').write_bytes(canonical(manifest))
        verify_bundle(stage,evidence)
        if fail_after=='before_commit':raise ReportError('injected_failure_before_commit')
        try:os.rename(stage,target)
        except OSError:
            if not target.exists():raise
            verify_bundle(target,evidence)
            return target,False
    return target,True


def check_freshness(period_end,now,max_age_seconds):
    if type(max_age_seconds) is not int or max_age_seconds<0:raise ReportError('invalid_freshness_limit')
    age=(timestamp(now)-timestamp(period_end)).total_seconds()
    if age<0 or age>max_age_seconds:raise ReportError('report_outside_freshness_window')
    return age


def main(argv=None):
    parser=argparse.ArgumentParser(description='Prepare a local report bundle for review; never distribute it.')
    parser.add_argument('--data',type=Path,required=True)
    parser.add_argument('--source-manifest',type=Path,required=True)
    parser.add_argument('--config',type=Path,required=True)
    parser.add_argument('--output',type=Path,required=True)
    args=parser.parse_args(argv)
    try:
        config=json.loads(args.config.read_text(encoding='utf-8'))
        folder,created=run(args.data,args.source_manifest,config,args.output)
        print(json.dumps({'event':'report_prepared','run_id':folder.name,'created':created,'distribution':'not_sent'}))
        return 0
    except (OSError,ValueError,KeyError,TypeError,csv.Error) as error:
        # Only the fixed diagnostic code/class is logged; no row payloads or credentials.
        code=str(error) if isinstance(error,ReportError) else type(error).__name__
        print(json.dumps({'event':'report_failed','code':code}),file=sys.stderr)
        return 1


if __name__=='__main__':raise SystemExit(main())