"""Explicit-schema, local single-writer example. Review before replacing real data."""
import argparse,csv,json,os,tempfile
from pathlib import Path
def unique(pairs):
    result={}
    for k,v in pairs:
        if k in result: raise ValueError('duplicate JSON key')
        result[k]=v
    return result
def constant(value): raise ValueError('nonstandard JSON constant')
def export(source,target,replace=False):
    source=Path(source).resolve(strict=True); target=Path(target).absolute()
    if target.is_symlink(): raise ValueError('target symlink is outside demo contract')
    if source==target.resolve(): raise ValueError('source and target must differ')
    if target.exists() and (not replace or not target.is_file()): raise ValueError('existing file requires --replace; directory is not a target')
    with source.open('rb') as f: raw=f.read(1000001)
    if len(raw)>1000000: raise ValueError('demo input exceeds 1000000 bytes')
    records=json.loads(raw.decode('utf-8'),object_pairs_hook=unique,parse_constant=constant)
    if not isinstance(records,list): raise ValueError('top-level array required')
    stage=None;matrix=[]
    try:
        with tempfile.NamedTemporaryFile(mode='w',encoding='utf-8',newline='',dir=target.parent,prefix='.csv-stage-',suffix='.tmp',delete=False) as f:
            stage=Path(f.name); w=csv.writer(f); w.writerow(['id','text'])
            for r in records:
                if not isinstance(r,dict) or set(r)!={'id','text'}: raise ValueError('exact id/text schema required')
                values=[r['id'],r['text']]
                if not all(isinstance(v,str) for v in values): raise ValueError('string fields required')
                for v in values:v.encode('utf-8',errors='strict')
                w.writerow(values);matrix.append(values)
            f.flush()
        # Close the writer before reopening or replacing the stage, including on Windows.
        with stage.open(encoding='utf-8',newline='') as f: restored=list(csv.reader(f,strict=True))
        if restored!=[['id','text']]+matrix:raise ValueError('staged field read-back differs')
        os.replace(stage,target)
        return {'data_records':len(matrix),'exact_staged_readback':True,'committed':True}
    except Exception as e:
        if stage is not None:e.add_note('Staging path retained if still present: '+str(stage))
        raise
if __name__=='__main__':
    p=argparse.ArgumentParser(description=__doc__);p.add_argument('source');p.add_argument('target');p.add_argument('--replace',action='store_true',help='intentionally replace an existing CSV after staging and verification');a=p.parse_args()
    try: print(json.dumps(export(a.source,a.target,a.replace)))
    except (ValueError,OSError,csv.Error,RecursionError) as e:
        p.exit(1,type(e).__name__+': '+str(e)+'\n'+'\n'.join(getattr(e,'__notes__',[]))+'\n')