"""Local fictional learning reference. No website/database/cloud connection."""
import csv,json,sqlite3,hashlib,math,re,statistics,collections
from pathlib import Path

def rows(source,name):
 with (Path(source)/name).open(encoding='utf-8',newline='') as f:return list(csv.DictReader(f))

def de_pipeline(source,out):
 # Full snapshot with staged output; no append or inferred incremental semantics.
 from datetime import date
 order_rows=rows(source,'orders.csv');line_rows=rows(source,'order_lines.csv');products=rows(source,'products.csv')
 if len({r['order_id'] for r in order_rows})!=len(order_rows):raise ValueError('Duplicate order keys')
 if len({r['product_id'] for r in products})!=len(products):raise ValueError('Duplicate product keys')
 for r in order_rows:
  if not r['order_id'] or r['status'] not in ['completed','cancelled']:raise ValueError('Invalid order key/status')
  if not re.fullmatch(r'\d{4}-\d{2}-\d{2}',r['order_date']):raise ValueError('ISO calendar date required')
  date.fromisoformat(r['order_date'])
 orders={r['order_id']:r for r in order_rows};product_keys={r['product_id'] for r in products};accepted=[];rejected=[]
 for n,r in enumerate(line_rows,2):
  reasons=[]
  try:
   quantity=int(r['quantity']);price=int(r['unit_price_cents'])
   if quantity<=0 or price<0:reasons.append('invalid_domain')
  except ValueError:reasons.append('invalid_number');quantity=price=0
  if r['order_id'] not in orders:reasons.append('unmatched_order')
  if r['product_id'] not in product_keys:reasons.append('unmatched_product')
  if not r['line_id']:reasons.append('missing_line_id')
  if reasons:rejected.append({'row':n,'line_id':r['line_id'],'reasons':reasons});continue
  accepted.append(dict(r,value_cents=quantity*price,status=orders[r['order_id']]['status']))
 if len({r['line_id'] for r in accepted})!=len(accepted):raise ValueError('Duplicate line keys')
 completed=[r for r in accepted if r['status']=='completed']
 manifest={'source_lines':len(line_rows),'accepted_lines_all_statuses':len(accepted),'quarantined_lines':len(rejected),'completed_lines':len(completed),'completed_value_cents':sum(r['value_cents'] for r in completed),'source_hashes':{n:hashlib.sha256((Path(source)/n).read_bytes()).hexdigest() for n in ['orders.csv','order_lines.csv','products.csv']}}
 assert len(line_rows)==len(accepted)+len(rejected)
 (Path(out)/'prepared.json').write_text(json.dumps(accepted,indent=2)+'\n');(Path(out)/'rejects.json').write_text(json.dumps(rejected,indent=2)+'\n')
 return manifest

if __name__=='__main__':
    import argparse,platform
    parser=argparse.ArgumentParser();parser.add_argument('--source',default='.');parser.add_argument('--output',required=True);args=parser.parse_args()
    source=Path(args.source).resolve();out=Path(args.output).resolve()
    if source==out:raise ValueError('Source and output must differ')
    out.mkdir(parents=False,exist_ok=False)
    # Partial folders after an error are NOT successful; result.json is written last.
    result=de_pipeline(source,out)
    with (out/'result.json').open('x',encoding='utf-8') as handle:json.dump(result,handle,indent=2,allow_nan=False);handle.write(chr(10))
    print(json.dumps(result,indent=2,allow_nan=False))