#!/usr/bin/env python3
"""Read-only preflight of synthetic source rows; fault cases exist only in memory."""
import argparse,csv,hashlib,json,platform,time
from collections import Counter
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

def normalized(row):
    key=row['DocID'].strip()
    if not key:raise ValueError('Empty document identifier')
    try: amount=Decimal(row['Amount'].strip().replace('$','').replace(',',''))
    except InvalidOperation as e: raise ValueError('Invalid amount for '+key) from e
    if not amount.is_finite():raise ValueError('Non-finite amount for '+key)
    payload=(datetime.strptime(row['PostingDate'].strip(),'%m/%d/%Y').date().isoformat(),row['CostCenter'].strip(),row['Account'].strip().title(),amount,row['Vendor'].strip())
    return key,payload

def guard(rows):
    seen={};duplicates=0
    for row in rows:
        key,payload=normalized(row)
        if key in seen:
            if payload!=seen[key]:raise ValueError('Conflicting duplicate '+key)
            duplicates+=1
        else:seen[key]=payload
    october=sum((Decimal(v[3]) for v in seen.values() if v[0].startswith('2026-10')),Decimal(0))
    return dict(raw_rows=len(rows),unique_documents=len(seen),identical_duplicates=duplicates,october_actual=str(october))

if __name__=='__main__':
    p=argparse.ArgumentParser();p.add_argument('csv_file',type=Path);p.add_argument('--report',type=Path,required=True);a=p.parse_args()
    original=a.csv_file.read_bytes();rows=list(csv.DictReader(original.decode().splitlines()));first=dict(rows[0])
    conflict=dict(first);conflict['Amount']=' $999,999.99 '
    empty=dict(first);empty['DocID']=' '
    padded=dict(first);padded['DocID']=' '+padded['DocID']+' '
    formatted=dict(first);formatted['Amount']=str(Decimal(first['Amount'].strip().replace('$','').replace(',','')))+'0'
    scenarios=[('original',rows),('one_more_identical_duplicate',rows+[first]),('same_id_different_amount',rows+[conflict]),('whitespace_in_same_id',rows+[padded]),('empty_id',rows+[empty]),('same_numeric_amount_different_format',rows+[formatted])]
    results=[]
    for name,case in scenarios:
        t=time.perf_counter()
        try:result=dict(status='accepted',**guard(case))
        except ValueError as e:result=dict(status='rejected',error=str(e),raw_rows=len(case))
        result.update(scenario=name,local_seconds=round(time.perf_counter()-t,6));results.append(result)
    assert results[0]['unique_documents']==480 and results[0]['october_actual']=='686500.00'
    assert [r['status'] for r in results]==['accepted','accepted','rejected','accepted','rejected','accepted']
    assert a.csv_file.read_bytes()==original
    report=dict(experiment='duplicate-preflight',python=platform.python_version(),source_sha256=hashlib.sha256(original).hexdigest(),source_rows=len(rows),results=results,
                limits='Local Python source-data preflight and deliberate in-memory fault injections. Does not run Power Query. Native workflow and the original source CSV are unchanged. Timings are single local guard executions, not Panes or complete task timings. No bill measured.')
    a.report.write_text(json.dumps(report,indent=2)+'\n');print(json.dumps(report,indent=2))