"""Read-only local CSV audit. Output aggregates/row numbers, no contact values."""
import csv,json,sys,re
from pathlib import Path
from datetime import date
from collections import Counter
AS_OF=date(2026,10,10)
ALLOWED={'active','pending','unsubscribed','inactive','paused','needs_approval'}
REQUIRED={'email','status','consent_state','joined_on'}
def invalid(code,record=None):
    result={'error':code,'scope':'CSV inspection stopped; no contact values emitted','providerOperations':0}
    if record is not None:result['record']=record
    return result
def inspect(path):
    # Validate structure before reading any contact fields. DictReader uses a
    # None key for extra columns and None values for missing columns.
    try:
        with Path(path).open(encoding='utf-8-sig',newline='') as f:
            reader=csv.DictReader(f,strict=True)
            header=reader.fieldnames
            if not header or any(not name.strip() for name in header) or len(header)!=len(set(header)) or not REQUIRED.issubset(header):
                return invalid('invalid_header')
            rows=[]
            for n,row in enumerate(reader,2):
                if None in row or any(value is None for value in row.values()):
                    return invalid('invalid_row_width',n)
                rows.append(row)
    except csv.Error:
        return invalid('malformed_csv')
    except (OSError,UnicodeError):
        return invalid('csv_unreadable')
    issues=[];seen=set();valid=set();active=set();eligible=set();counts=Counter()
    for n,row in enumerate(rows,2):
        email=(row.get('email') or '').strip()
        key=email # exact-address matching; no provider-specific normalization
        status=(row.get('status') or '').strip()
        counts[status if status in ALLOWED else 'unmapped']+=1
        flags=[]
        if not re.fullmatch(r'[^\s@]+@[^\s@]+\.[^\s@]+',email):flags.append('email_shape')
        else:valid.add(key)
        if key and key in seen:flags.append('duplicate_exact_address')
        seen.add(key)
        if status not in ALLOWED:flags.append('unmapped_status')
        try:
            if date.fromisoformat(row.get('joined_on',''))>AS_OF:flags.append('future_join_date')
        except ValueError:flags.append('invalid_join_date')
        if any((v or '').lstrip().startswith(('=','+','-','@')) for v in row.values()):flags.append('spreadsheet_formula_prefix')
        if status=='active' and key in valid:active.add(key)
        if status=='active' and row.get('consent_state')=='confirmed' and not flags:eligible.add(key)
        if flags:issues.append({'row':n,'flags':flags})
    return {'scope':'Synthetic local inspection; no provider import, billing count or consent verification','asOf':AS_OF.isoformat(),'rows':len(rows),'uniqueShapeValidAddresses':len(valid),'exactDuplicateRows':sum('duplicate_exact_address' in x['flags'] for x in issues),'statusRowCounts':dict(sorted(counts.items())),'activeLabelUniqueShapeValid':len(active),'localReviewReadyUnique':len(eligible),'issueRows':issues,'providerOperations':0}
if __name__=='__main__':
    result=inspect(sys.argv[1] if len(sys.argv)>1 else 'subscribers.synthetic.csv')
    print(json.dumps(result,sort_keys=True,indent=2))
    sys.exit(1 if 'error' in result else 0)