"""Offline independent audit of the Food phase; no DB/HTTP and no normalization of dates."""
import base64,copy,hashlib,json,re
from pathlib import Path
ROOT=Path(__file__).parent; checks=0
def same(a,b):
 if type(a) is not type(b): return False
 if isinstance(a,dict): return list(a)==list(b) and all(same(a[k],b[k]) for k in a)
 if isinstance(a,list): return len(a)==len(b) and all(same(x,y) for x,y in zip(a,b))
 return a==b
def eq(a,b,label):
 global checks
 if not same(a,b): raise AssertionError(label)
 checks+=1
def read(p):return json.loads(p.read_text())
def partial(s):return {k:v for k,v in s.items() if k not in ('all_primary_users','all_config','all_dish')}
def protected(s):return {k:v for k,v in partial(s).items() if k!='own'}
NUTRIENTS='energy carbohydrate protein fat sfa mufa pufa cholesterol va renieratene retinol vb1 vb2 vpp vc ve folic_acid biotin choline sodium calcium magnesium iron zinc selenium purine dietary_fiber moisture ash iodine phosphorus cuprum manganese pantothenic_acid potassium gi gl'.split()
def format_row(row):
 row=dict(row)
 for k in NUTRIENTS:
  if k in row and row[k] is not None: row[k]=format(float(row[k]),'.2f')
 return row
def dto(own,categories,method,stage):
 if stage==4:return {'code':1,'message':'可食部分取值范围是1-100','data':''}
 if method=='readById':
  row=format_row(own[0]) if own else None
  if row is not None:
   names={c['id']:c['name'] for c in categories}
   for key,out in [('category','category_name'),('minute_sort','minute_sort_name'),('reclassify','reclassify_name')]:row[out]=names.get(row[key],'') if row[key] else ''
  return {'code':0,'message':'请求成功','data':row}
 if method=='lists':
  fields='id name edible_part energy carbohydrate protein fat dietary_fiber cooking_food category minute_sort sub_name create_by create_time'.split()
  return {'code':0,'message':'请求成功','data':{'rows':[format_row({k:r[k] for k in fields}) for r in own],'total':len(own)}}
 return {'code':0,'message':{'add':'添加成功','update':'修改成功','delete':'删除成功'}[method],'data':{'id':str(own[0]['id'])} if method=='add' else []}
summary={}; bodies={}; dates={}
for lane in ('B','C'):
 d=ROOT/'private'/(lane+'-main'); base=read(d/'baseline.json'); owned=read(d/'owned.json'); prev=partial(base); bodies[lane]={}
 eq(len(owned['full_created_row']),62,'Complete original Food physical row62')
 eq(read(d/'tap-proof.json')['before_HTTP'],True,'Actual dual connection proof before first add')
 for role,p in read(d/'tap-proof.json')['proof'].items():
  eq(p['pdo'],'FoodPDO','Actual PDO tap');eq(p['driver'],'think\\db\\connector\\Mysql' if lane=='B' else 'app\\cores\\database\\Dm','Original actual driver')
 native_reads=len(read(d/'tap-proof.json')['native']); obs=len(read(d/'preflight.json')['observer'])+len(read(d/'auth-proof.json')['observer'])+len(read(d/'tap-proof.json')['observer']); writes=[]; spy=[]; notifications=[]; isolates=[]
 if lane=='B':
  bad=read(d/'wire-01.json');eq(bad['body'],'','Original failed metadata empty response');eq(bad['DML_attempts'],0,'Original failed metadata no target DML');eq(bad['before'],bad['after'],'Original failed metadata protected whole state exact');eq(bad['before']['own'],[],'Original failed metadata no own row');eq(bad['native'][0]['sql'],'SHOW COLUMNS FROM `ydy_config`','Original exact rejected metadata');eq(bad['native'][0]['success'],False,'Original pre-PDO execution reject');eq(len(bad['native']),1,'Only one failed native metadata attempt');assert not (d/'response-01.json').exists();obs+=len(bad['observer'])+len(read(d/'before-01.json')['observer'])
 for stage in range(1,11):
  label='01-fixed' if lane=='B' and stage==1 else f'{stage:02d}'
  w=read(d/f'wire-{label}.json');r=read(d/f'response-{label}.json');c=read(d/f'client-{label}.json');case=w['case'];after=w['after'];method=case['method']
  eq(w['before'],prev,'Actual scoped chronological fresh before');eq(protected(after),protected(base),'Full protected natural hash/categories/actor/nutrition/critical config unchanged')
  eq(c['status'],200,'Actual client HTTP200');eq(r['status'],200,'Original Response HTTP200');eq(c['body'],r['body'],'Exact client response bytes');eq(w['body'],r['body'],'Exact server response bytes');eq(hashlib.sha256(r['body'].encode()).hexdigest(),r['body_sha256'],'Original body SHA');eq(json.loads(base64.b64decode(w['raw_request_base64'])),case['params'],'Original typed raw JSON request')
  eq(w['warnings'],[],'Request warnings');eq(w.get('business_error'),None,'No source exception');eq(w.get('observer_error'),None,'No observer failure')
  expected=dto(after['own'],after['categories'],method,stage);actual=json.loads(r['body']);eq(actual,expected,'Complete independent DTO including raw time/null/data string');bodies[lane][stage]=actual
  sql=[x for x in w['native'] if 'sql' in x];dd=[x for x in sql if re.match(r'^\s*(INSERT|UPDATE|DELETE)\b',x['sql'])];native_reads+=len(sql)-len(dd)
  eq(len(dd),1 if stage in (1,6,8) else 0,'Only prescribed native writes');eq(w['DML_attempts'],len(dd),'Native DML counter exact');writes+=dd
  for x in dd:
   eq(x['schema'],'store','Only primary Food write');eq(x['success'],True,'Original native write success');eq(x['affected'],1,'All original writes affected1');eq(x['execute_parameters'],None,'Explicit original typed binds')
   assert 'ydy_foods_composition' in x['sql']
   if stage!=1:assert str(owned['id']) in [str(b['value']) for b in x['binds'].values()]
   for key,b in x['binds'].items():
    eq(b['ok'],True,'Native bind succeeded')
    if key.startswith(':data__'):
     field=key[7:]
     if field in case['params']:eq(b['value'],case['params'][field].strip() if field in ('name','sub_name') else case['params'][field],'Original exact input PHP value')
     if field in ('energy','protein','fat','carbohydrate','dietary_fiber'):
      eq(b['type'],1,'Original decimal PDO PARAM_INT preserved');eq(b['php_type'],'string' if stage==1 else 'double','Original string vs JSON double distinction')
  if stage==1:
   ids=[x for x in w['native'] if x.get('layer')=='PDO.lastInsertId'];eq(len(ids),1,'Native identity once');eq(ids[0]['value'],str(owned['id']),'Native identity vs actual PK');eq(after['own'][0],owned['full_created_row'],'Creation journal complete physical row');eq(after['own'][0]['name'],'CDXFOOD11W1','Original trim');eq(after['own'][0]['sub_name'],'CDXFOOD11W1A','Original alias trim');eq(after['own'][0]['store_id'],10001,'Actual default tenant')
  if stage==4:eq(w['before'],after,'Invalid edible0 entire row and protection unchanged')
  if stage in (1,6):
   row=after['own'][0]
   for k in ('edible_part','energy','protein','fat','carbohydrate','dietary_fiber'):eq(float(row[k]),float(case['params'][k]),'Input value retained physically, not washed by row oracle')
  if stage==6:
   before=w['before']['own'][0];changed={'name','sub_name','edible_part','energy','protein','fat','carbohydrate','dietary_fiber','en_name','create_time','update_time'}
   eq({k:v for k,v in row.items() if k not in changed},{k:v for k,v in before.items() if k not in changed},'All other physical columns exact')
   dates[lane]={'created':{k:owned['full_created_row'][k] for k in ('create_time','update_time')},'updated':{k:row[k] for k in ('create_time','update_time')},'automatic_update_time_changed':before['update_time']!=row['update_time']}
  if stage==8:
   lease=read(d/'delete-lease.json');eq(lease['full_own_row']['id'],owned['id'],'Own deletion PK lease');eq(lease['full_own_row']['name'],'CDXFOOD11W1-U','Own deletion name lease');eq(lease['all_dish_references'],[],'All tenant dish reference0');eq(after['own'],[],'Only own row gone')
  obs+=len(w['observer']);spy+=w['transport_spy'];notifications+=w['notification_spy'];isolates+=w['tchef_config_isolation_calls'];prev=after
 first='01-fixed' if lane=='B' else '01';obs+=len(read(d/f'before-{first}.json')['observer']);final=read(d/'final.json');obs+=len(final['observer']);eq(final['source'],base,'Fresh final natural+whole primary users/config/dish and all scoped source exact')
 eq(len(spy),2,'Two explicit failed AI transport spies');eq(len(notifications),3,'Three isolated notifications');eq(len(isolates),5 if lane=='B' else 0,'Only B explicit process disabled tchef5 calls')
 for x in spy:eq(x['response'],False,'Actual explicit failure, never simulated success');eq(x['api_type'],2,'Only AI food sync API2');eq(str(x['params']['form']['id']),str(owned['id']),'Only newly created own Food external payload')
 budget={p.stem.replace('budget-',''):int(p.read_text()) for p in d.glob('budget-*.json')};eq(budget['observer'],obs,'Every observer execution counted');eq(budget['business-read'],native_reads,'Every executed native read counted incl proof');eq(budget['total-read'],obs+native_reads+(12 if lane=='B' else 11)+(1 if lane=='B' else 0),'Unified reads incl prior and rejected metadata attempt');assert budget['total-read']<=800
 eq(budget['HTTP'],11 if lane=='B' else 10,'All HTTP incl original tool failure');eq(len(writes),3,'Only Food I/U/D3');eq(read(d/'resource-proof.json')['source_files'],959,'Actual APP959 pinned')
 summary[lane]={'formal_HTTP':10,'all_HTTP':budget['HTTP'],'INSERT':1,'UPDATE':1,'DELETE':1,'native_read_executed':native_reads,'observer_read':obs,'prior_readonly':12 if lane=='B' else 11,'rejected_metadata_attempt':1 if lane=='B' else 0,'all_readonly_total':budget['total-read'],'transport_false_spy':2,'notifications_spy':3,'tchef_process_disabled_calls':len(isolates),'final_all_state_exact':True,'runtime_retained_for_next_authorized_read_planning':True}
eq(dates['B']['automatic_update_time_changed'],True,'Original MySQL automatic update across seconds');eq(dates['C']['automatic_update_time_changed'],False,'Original DM automatic update missing preserved')
diff=[]
for stage in range(1,11):
 b=copy.deepcopy(bodies['B'][stage]);c=copy.deepcopy(bodies['C'][stage])
 if stage==1:c['data']['id']=b['data']['id']
 if stage in (2,3,5,7):
  br=b['data']['rows'][0] if stage==3 else b['data'];cr=c['data']['rows'][0] if stage==3 else c['data']
  for k in ('id','create_by'):cr[k]=br[k]
  for k in br:
   if not same(br[k],cr[k]):diff.append({'stage':stage,'field':k,'B':br[k],'C':cr[k]})
 else:eq(c,b,'Other full DTO exact')
eq(sorted(set(x['field'] for x in diff)),['create_time','update_time'],'Only date response differences remain after ID/creator mapping')
for a,b in [(None,[]),('',[]),(1,1.0),(0,False),('1.00',1.0)]:eq(same(a,b),False,'Typed DTO mutation rejected')
print(json.dumps({'status':'PRIMARY_CRUD_INPUT_PASS; ORIGINAL_DATE0_AND_ON_UPDATE_RED; AI_FAILURE_ISOLATED_TCHEF_ENABLED_UNTESTED','checks':checks,'lanes':summary,'dates':dates,'cross_full_DTO_differences':diff},ensure_ascii=False,indent=2))