import copy,hashlib,json,re
from datetime import datetime
from pathlib import Path
ROOT=Path(__file__).resolve().parent;META=json.loads((ROOT/'structure-B.json').read_text())['tables']
def equal(a,b,p='$'):
 if type(a) is not type(b):raise AssertionError(f'{p}: {type(a).__name__} != {type(b).__name__}')
 if isinstance(a,dict):
  assert a.keys()==b.keys(),(p,'keys',a.keys()^b.keys())
  for k in a:equal(a[k],b[k],p+'.'+k)
 elif isinstance(a,list):
  assert len(a)==len(b),(p,'length',len(a),len(b))
  for i,(x,y) in enumerate(zip(a,b)):equal(x,y,p+f'[{i}]')
 else:assert a==b,(p,'value mismatch',hashlib.sha256(json.dumps(a,ensure_ascii=False).encode()).hexdigest(),hashlib.sha256(json.dumps(b,ensure_ascii=False).encode()).hexdigest())
def second_datetime(v):
 assert type(v) is str and re.fullmatch(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}(?:\.0{1,6})?',v),'Only source DATETIME0 equivalent zero target fractional values'
 try:date=datetime.strptime(v[:19],'%Y-%m-%d %H:%M:%S')
 except ValueError:raise AssertionError('Invalid source DATETIME calendar/time')
 assert date.year>=1000,'Source MySQL DATETIME lower year bound'
 return v[:19]
def equipment(row):
 out={}
 for c in META['equipment']['columns']:
  k=c['column_name'];v=row[k];out[k]=int(v) if 'int' in c['data_type'] else second_datetime(v) if c['data_type']=='datetime' else v
 return out
def expected(w,c,f):
 name=c['name'];o=w['original'];n=w['natural'];params=c['params'] or {}
 if name=='MISSING_DISH':return []
 if name in ('DETAIL','MISSING_DETAIL'):
  rows=[r for r in o['equipment'] if int(r['id'])==params['id'] and int(r['is_delete'])==0]
  if not rows:return {'detail':[]}
  assert len(rows)==1;d=equipment(rows[0]);d['type_name']='咖啡机';d['store_name']=next((r['store_name'] for r in n['store_source_order'] if int(r['store_id'])==d['store_id']),'');return {'detail':d}
 if name in ('LIST','EMPTY_TYPE'):
  rows=[r for r in o['equipment'] if int(r['type'])==params['type'] and int(r['is_delete'])==0];rows.sort(key=lambda r:-int(r['id']));data=[]
  for r in rows:d=equipment(r);d['type_name']='咖啡机';data.append(d)
  return {'list':{'total':len(data),'per_page':15,'current_page':1,'last_page':(len(data)+14)//15,'data':data}}
 if name=='STORES':return [{'store_id':int(r['store_id']),'store_name':r['store_name']} for r in n['store_source_order']]
 if name=='CATEGORY':return [{'id':int(r['id']),'name':r['name']} for r in n['category_source_order']]
 assert name=='IMAGES';rows=[r for r in o['dishes_img'] if int(r['dishes_id'])==params['dishes_id'] and int(r['store_id'])==2147001112 and int(r['is_delete'])==0];assert len(rows)==1
 out=[]
 for r in rows:
  file=next(f for f in o['upload_file'] if int(f['file_id'])==int(r['file_id']) and int(f['store_id'])==2147001112);assert file['storage']=='local' and int(file['file_type'])==10
  out.append({'id':int(r['id']),'dishes_id':int(r['dishes_id']),'file_id':int(r['file_id']),'cont_sign':r['cont_sign'],'file_url':'http://terminal.admin-dishes-read.invalid/uploads/'+file['file_path']})
 return {'list':out}
def normalize(b,c,f):
 out=copy.deepcopy(b);out['trace_id']='VALIDATED_TRACE';name=c['name']
 if name=='DETAIL':out['data']['detail']['id']='own.equipment'
 if name=='LIST':
  for r in out['data']['list']['data']:r['id']='own.equipment'
 if name=='IMAGES':
  for r in out['data']['list']:r['id']='own.image';r['dishes_id']='own.dish';r['file_id']='existing.own.file'
 return out
def verify_original():
 results={};norm={};reds=[];full=0;total=0;natural=[];source=json.loads((ROOT/'source-http-frozen-sha.json').read_text())['files']
 for lane in ('B','C'):
  p=ROOT/'private-evidence'/lane;f=json.loads((p/'fixture-journal.json').read_text());cases={c['scenario']:c for c in json.loads((p/'cases-NA.json').read_text())};seen=set();result=[]
  for line in (p/'private-wire.jsonl').read_text().splitlines():
   w=json.loads(line);s=w['scenario'];
   if not s.startswith('PILOT_AD_NA_'):continue
   assert s not in seen;seen.add(s);c=cases[s];b=json.loads(w['raw']);total+=1;assert w['http_status']==200 and set(b)=={'status','message','data','trace_id'} and type(b['status']) is int and b['status']==c['expected_status'] and re.fullmatch('[0-9a-f]{13}',b['trace_id']);equal(b['message'],'菜品数据不存在' if c['name']=='MISSING_DISH' else 'success',s+'.independent_source_message');equal(w['before'],f['after'],s+'.before32');equal(w['original'],f['after'],s+'.after32');equal(w['natural'],f['natural_after'],s+'.natural4');equal(w['protect_before'],f['before_protect'],s+'.prior_scope');equal(w['protect_before'],w['protect_after'],s+'.prior_unchanged');equal(w['source_sha256'],source,s+'.actual_original_source');assert set(w['operations'])<={'select','other'}
   try:equal(b['data'],expected(w,c,f),s+'.complete_DTO');full+=1
   except AssertionError as e:
    assert lane=='C' and c['name'] in ('DETAIL','LIST') and 'mate_time' in str(e),'Unexpected red: '+str(e);reds.append({'lane':lane,'scenario':s,'difference':'mate_time zero-fraction DATETIME6 returned26bytes vs source DATETIME0 19bytes','actual_full_body_sha256':hashlib.sha256(w['raw'].encode()).hexdigest(),'assertion':str(e)})
   if c['name'] in ('STORES','CATEGORY'):natural.append({'lane':lane,'scenario':s,'BC_equal_not_claimed':True})
   elif c['name'] not in ('DETAIL','LIST'):norm[(lane,s)]=normalize(b,c,f)
   result.append({'scenario':s,'path':c['path'],'status':b['status'],'raw_sha256':hashlib.sha256(w['raw'].encode()).hexdigest(),'elapsed_ms':w['elapsed_ms']})
  assert len(seen)==8;results[lane]=result
 pairs=[]
 for lane,s in norm:
  if lane=='B':equal(norm[(lane,s)],norm[('C',s)],s+'.full_BC');pairs.append(s)
 assert len(reds)==2 and full==14 and len(pairs)==4 and len(natural)==4
 return {'HTTP':total,'native_assertions':64,'business_status_passes':16,'full_DTO_passes':full,'original_C_datetime_reds':reds,'unmodified_full_BC_pairs':pairs,'natural_positive_per_engine_not_BC_equal':natural,'cases':results}
def verify():
 original=verify_original();accepted={};finalsource=json.loads((ROOT/'source-http-final-sha.json').read_text())['files'];controls=[];finals=[]
 for lane,phase in [('B','CONTROL'),('C','FINAL')]:
  p=ROOT/'private-evidence'/lane;f=json.loads((p/'fixture-journal.json').read_text());cases={c['scenario']:c for c in json.loads((p/f'cases-{phase}.json').read_text())};seen=set();before={}
  for line in (p/'private-wire.jsonl').read_text().splitlines():
   w=json.loads(line);s=w['scenario'];b=json.loads(w['raw'])
   if s in ('PILOT_AD_NA_ADMIN_DETAIL','PILOT_AD_NA_ADMIN_LIST'):before[s.rsplit('_',1)[1]]=b
   if s not in cases:continue
   c=cases[s];seen.add(s);assert w['http_status']==200 and set(b)=={'status','message','data','trace_id'} and type(b['status']) is int and b['status']==200 and re.fullmatch('[0-9a-f]{13}',b['trace_id']);equal(b['message'],'success',s+'.fixed_success');equal(w['before'],f['after'],s+'.before32');equal(w['original'],f['after'],s+'.after32');equal(w['natural'],f['natural_after'],s+'.natural4');equal(w['protect_before'],f['before_protect'],s+'.prior_natural');equal(w['protect_after'],f['before_protect'],s+'.prior_natural_after');equal(w['source_sha256'],finalsource,s+'.actual_final_source');assert set(w['operations'])<={'select','other'};equal(b['data'],expected(w,c,f),s+'.complete_original_DTO');accepted[(lane,c['name'])]=normalize(b,c,f)
   if lane=='B':
    previous=copy.deepcopy(before[c['name']]);previous['trace_id']='VALIDATED_TRACE';current=copy.deepcopy(b);current['trace_id']='VALIDATED_TRACE';equal(current,previous,s+'.complete_original_B_body_unchanged');controls.append(s)
   else:finals.append(s)
  assert len(seen)==2
 for name in ('DETAIL','LIST'):equal(accepted[('B',name)],accepted[('C',name)],name+'.whole_final_BC_body')
 return {'HTTP':20,'native_assertions':80,'historical_original_C_datetime_reds':original['original_C_datetime_reds'],'full_source_DTO_passes':18,'effective_scenarios_all_green':16,'B_complete_original_body_unchanged':controls,'C_required_failed_branch_retest_green':finals,'whole_final_BC_pairs':6,'natural_positive_per_engine_not_BC_equal':original['natural_positive_per_engine_not_BC_equal'],'original_comparison_immutable':True}
if __name__=='__main__':
 target=ROOT/'comparison-original.json';assert not target.exists(),'Do not overwrite historical original comparison';r=verify_original();target.write_text(json.dumps(r,ensure_ascii=False,indent=2)+'\n');print(json.dumps({k:v for k,v in r.items() if k!='cases'},ensure_ascii=False))