import json,re,hashlib,copy,sys
from pathlib import Path
from collections import Counter,defaultdict
from datetime import datetime,timedelta,timezone
from email.utils import parsedate_to_datetime
from decimal import Decimal
P=Path(__file__).resolve().parent; BASE=P.parent
COUNT=0
def check(ok,why):
    global COUNT
    COUNT+=1
    if not ok:raise AssertionError(why)
def load(p):return json.loads(p.read_text(),parse_float=Decimal)
def typed(a,b,path=''):
    check(type(a) is type(b),(path,'type',type(a).__name__,type(b).__name__))
    if isinstance(a,dict):
        check(set(a)==set(b),(path,'keys',set(a)^set(b)))
        for k in a:typed(a[k],b[k],path+'/'+k)
    elif isinstance(a,list):
        check(len(a)==len(b),(path,'length',len(a),len(b)))
        for i,(x,y) in enumerate(zip(a,b)):typed(x,y,path+'/'+str(i))
    else:check(a==b,(path,'value',a,b))
def tree(dictionary,kind,selected):
    fields=['enum_key','enum_pid','enum_name']+(['set_value'] if kind==10 else [])
    parents={};score=0
    for original in dictionary:
        if int(original['enum_type'])!=kind:continue
        row={k:original[k] for k in fields};key=str(row['enum_key']);pid=str(row['enum_pid']);row['checked']=int(key in selected)
        if kind==10:
            row['recommend']=selected.get(key,'')
            if key in selected:score+=int(row['set_value'])
        if pid=='0':parents[key]=row
        else:parents[pid].setdefault('children',[]).append(row)
    for parent in parents.values():
        if kind==11:parent['checked']=[]
        for child in parent.get('children',[]):
            if child['checked']:
                if kind==11:parent['checked'].append(child['enum_key'])
                else:parent.update(checked=child['enum_key'],recommend=child['recommend'])
    return list(parents.values()),score
def detail(dictionary,group,body):
    if group=='FOOD':
        t,_=tree(dictionary,11,dict.fromkeys(body['enum_data'],''))
        return dict(food=t,food_allergy=body['food_allergy'],other_text=body['other_text'],food_Intolerance=body['food_Intolerance'],food_disable=body['food_disable'])
    if group=='MEALS':
        t,score=tree(dictionary,10,{e['enum_id']:e['recommend'] for e in body['enum_data']})
        return dict(meals=t,score=score,other_text=body['other_text'])
    score=dict.fromkeys(['vb','vc','va','composite_v','antioxidant','protein','dietary_fiber','calcium','iron','zinc'],0)
    out={'problem':[],'score':score,'other_text':body['other_text']}
    selected={e['enum_key']:e for e in body['enum_data'] if e['checked']!=3}
    for original in dictionary:
        kind=int(original['enum_type'])
        if kind not in (6,7,8,9):continue
        row={k:original[k] for k in ['enum_key','enum_type','enum_name','set_value']};key=row['enum_key']
        if kind==6:
            row.update(set_value=json.loads(row['set_value']),checked=3,v1=0,v2=0)
            if key in selected:
                e=selected[key];row['checked']=e['checked']
                if key=='11':row.update(v1=e['v1'],v2=e['v2'])
                if (key in ('9','25') and e['checked']==0) or (key not in ('9','25') and e['checked']==1):
                    for k,v in row['set_value'].items():
                        if k in score:score[k]+=v
            out['problem'].append(row)
        else:
            name,data={7:('lack','lack'),8:('meal','meal_type'),9:('nutrition','nutrition_type')}[kind]
            row['checked']=int(key in body[data]);out.setdefault(name,[]).append(row)
    return out
def without_trace(raw):return re.sub(r',"trace_id":"[^"\\]*"','',raw)
def rowmap(rows):return {int(r['id']):r for r in rows}
def new_answers(group,body,before,master):
    rows=[]
    if group=='PROBLEM':
        rows += [dict(type=6,reference_id=int(e['enum_key']),checkout_status=e['checked'],v1=e['v1'],v2=e['v2']) for e in body['enum_data'] if e['checked']!=3]
        sets=[('lack',7),('meal_type',8),('nutrition_type',9)]
    elif group=='MEALS':
        rows += [dict(type=10,reference_id=int(e['enum_id']),recommend=e['recommend']) for e in body['enum_data']];sets=[]
    else:sets=[('enum_data',11)]
    for key,kind in sets:
        existing={int(r['reference_id']) for r in before if int(r['type'])==kind and not int(r['is_delete'])}
        rows += [dict(type=kind,reference_id=int(x)) for x in body[key] if int(x) not in existing]
    return rows
def physical(wire,client,lane):
    case=wire['case'];body=case['body'];group=case['group'];typ={'PROBLEM':6,'MEALS':10,'FOOD':11}[group];date=body['set_date']
    before=wire['before'];after=wire['after'];bm=rowmap(before['nutrition_meals']);am=rowmap(after['nutrition_meals']);bl=rowmap(before['nutrition_lack']);al=rowmap(after['nutrition_lack'])
    scope=dict(store_id=2147001111,staff_id=body['staff_id'],set_date=date,user_id=10074 if lane=='B' else 10010)
    prior=[r for r in bm.values() if r['type']==typ and r['set_date']==date];masters=[r for r in am.values() if r['type']==typ and r['set_date']==date]
    check(len(masters)==1,'One actual scoped master');master=masters[0];mid=master['id'];create=not prior
    check(set(am)==set(bm)|({mid} if create else set()),'Only one generated master or stable pk')
    expected=copy.deepcopy(prior[0]) if prior else dict(scope,type=typ,food_allergy='',food_intolerance='',food_disable='',is_delete=0,create_by=case['create_by'],id=mid)
    expected['other_text']=body['other_text']
    if prior:expected['update_time']=master['update_time']
    if typ==11:expected.update(food_allergy=body['food_allergy'],food_intolerance=body['food_Intolerance'],food_disable=body['food_disable'])
    check(set(master)==set(expected)|{'create_time','update_time'},'Full physical master field set')
    for k,v in expected.items():typed(master[k],v,'master/'+k)
    for ident,row in bm.items():
        if ident!=mid:typed(am[ident],row,'unrelated_master')
    related={i:r for i,r in bl.items() if r['set_date']==date and (r['type'] in (6,7,8,9) if typ==6 else r['type']==typ)}
    deleted=set();soft=set();retained=set()
    if not create:
        for i,row in related.items():
            kind=row['type'];ref=str(row['reference_id'])
            if kind in (6,10):soft.add(i)
            elif kind in (7,8,9,11):
                key={7:'lack',8:'meal_type',9:'nutrition_type',11:'enum_data'}[kind]
                (retained if ref in body[key] else deleted).add(i)
    for i,row in bl.items():
        if i in deleted:check(i not in al,'Exact physically deleted answer')
        elif i in soft:
            check(al[i]['is_delete']==1,'Exact soft-delete marker');expected_row=dict(row,is_delete=1)
            # MySQL has native ON UPDATE CURRENT_TIMESTAMP; it is part of the frozen schema.
            expected_row['update_time']=al[i]['update_time'];typed(al[i],expected_row,'soft_preserve_all_other_columns')
        else:typed(al[i],row,'retained_or_unrelated_answer')
    new_ids=set(al)-set(bl);wanted=new_answers(group,body,list(related.values()),mid)
    check(len(new_ids)==len(wanted),'Exact newly generated answer count');newrows=[al[i] for i in sorted(new_ids)]
    check(set(al)==(set(bl)-deleted)|new_ids,'No other physical answer changes')
    for row,extra in zip(newrows,wanted):
        expected=dict(scope,master_id=mid,checkout_status=1,v1=0,v2=0,recommend='',is_delete=0,create_by=case['create_by'],id=row['id']);expected.update(extra)
        check(set(row)==set(expected)|{'create_time','update_time'},'Full physical answer field set')
        for k,v in expected.items():typed(row[k],v,'new_answer/'+k)
    utc=parsedate_to_datetime(next(x[6:] for x in client['response_headers'] if x.startswith('Date: ')))
    def valid_time(value):
        dt=datetime.fromisoformat(value).replace(tzinfo=timezone(timedelta(hours=8)))
        check(abs((dt-utc).total_seconds())<10,'Actual datetime within request wire time')
    if create:valid_time(master['create_time'])
    else:typed(master['create_time'],prior[0]['create_time'],'master_create_time_retained')
    valid_time(master['update_time'])
    for row in newrows:valid_time(row['create_time']);valid_time(row['update_time'])
    for i in soft:
        if al[i]['update_time']!=bl[i]['update_time']:valid_time(al[i]['update_time'])
    check(wire['successful_master_INSERT']==int(create),'Native master successful INSERT counter')
    check(wire['successful_answer_INSERT']==len(new_ids),'Native answer INSERT counter')
    return dict(master_id=mid,create=create,inserted_answer_ids=sorted(new_ids),soft_deleted_ids=sorted(soft),hard_deleted_ids=sorted(deleted),retained_ids=sorted(retained))
def sqlmaps(e):
    sql=e['queryString'];bind=e['bindValue']
    def value(tok):
        tok=tok.strip();check(tok in bind,'All DML tokens bound');b=bind[tok];check(b['ok'] is True,'Native bindValue succeeded');check(b['pdo_type'] in (0,1,2,5),'Approved real PDO type');return b['value']
    if e['operation']=='INSERT':
        if ' SET ' in sql:
            cols=re.findall(r'`?([A-Za-z_][A-Za-z0-9_]*)`?\s*=\s*(:\w+)',sql.split(' SET ',1)[1]);values={k:value(t) for k,t in cols}
        else:
            m=re.match(r'INSERT INTO \w+\s*\(([^)]+)\)\s*VALUES\s*\(([^)]+)\)',sql);check(bool(m),'Exact original DM single INSERT syntax');values={k.strip():value(t) for k,t in zip(m[1].split(','),m[2].split(','))}
        return values,{}
    where=sql.split(' WHERE ',1)[1];scope={}
    for part in re.split(r'\s+AND\s+',where.strip()):
        part=re.sub(r'(?:`?yoshop_(?:nutrition_lack|nutrition_meals)`?)\.','',part).replace('`','').strip()
        m=re.match(r'^(\w+)\s*=\s*(:\w+)$',part);check(bool(m),'Original singleton EQ/AND exact scope');scope[m[1]]=value(m[2])
    values={k:value(t) for k,t in re.findall(r'`?(\w+)`?\s*=\s*(:\w+)',sql.split(' WHERE ',1)[0].split(' SET ',1)[1])} if e['operation']=='UPDATE' else {}
    return values,scope
def expected_dml(w,lane):
    case=w['case'];body=case['body'];group=case['group'];kind={'PROBLEM':6,'MEALS':10,'FOOD':11}[group];date=body['set_date'];scope=dict(store_id=2147001111,staff_id=body['staff_id'],user_id=10074 if lane=='B' else 10010,set_date=date)
    masters=[r for r in w['before']['nutrition_meals'] if r['type']==kind and r['set_date']==date];create=not masters;mid=masters[0]['id'] if masters else w['master_generated_native_ID'];answers=[r for r in w['before']['nutrition_lack'] if r['set_date']==date]
    stamp=dict(create_by=case['create_by'],create_time='@time',update_time='@time');ops=[]
    def op(operation,table,values,where):ops.append((operation,table,values,where))
    if create:
        values=dict(scope,type=kind,other_text=body['other_text'],**stamp)
        if lane=='C' and kind!=11:values['food_allergy']=''
        where={}
    else:values=dict(other_text=body['other_text'],update_time='@time');where=dict(store_id=2147001111,id=mid)
    if kind==11:values.update(food_allergy=body['food_allergy'],food_disable=body['food_disable'],**{('food_intolerance' if lane=='C' else 'food_Intolerance'):body['food_Intolerance']})
    op('INSERT' if create else 'UPDATE','nutrition_meals',values,where)
    def insert(extra):op('INSERT','nutrition_lack',dict(scope,master_id=mid,**stamp,**extra),{})
    if kind in (6,10):
        wh=dict(scope,type=kind)
        if kind==6:wh['is_delete']=0
        op('UPDATE','nutrition_lack',{'is_delete':1},wh)
        if kind==6:
            for e in body['enum_data']:
                if e['checked']!=3:insert(dict(type=6,reference_id=e['enum_key'],checkout_status=e['checked'],v1=e['v1'],v2=e['v2']))
        else:
            for e in body['enum_data']:insert(dict(type=10,reference_id=e['enum_id'],recommend=e['recommend']))
    sets=[('lack',7),('meal_type',8),('nutrition_type',9)] if kind==6 else [('enum_data',11)] if kind==11 else []
    for key,typ in sets:
        assigned=[r['reference_id'] for r in answers if r['type']==typ and r['is_delete']==0];want=list(map(int,body[key]));gone=[ref for ref in assigned if ref not in want]
        if gone:
            check(len(gone)==1,'This approved batch deletes singleton only');op('DELETE','nutrition_lack',{},dict(scope,type=typ,reference_id=gone[0]))
        for ref in body[key]:
            if int(ref) not in assigned:insert(dict(type=typ,reference_id=ref))
    return ops
def run():
    oldseal=load(BASE/'first-phase/seal.json')
    check(len(oldseal['files'])==74,'Immutable old first-phase74 assets')
    for name,meta in oldseal['files'].items():
        f=BASE/name;check(f.stat().st_size==meta['bytes'] and hashlib.sha256(f.read_bytes()).hexdigest()==meta['sha256'],'Immutable '+name)
    out={};final={};allbodies={}
    original_source=load(BASE/'source-frozen-sha.json')['files'];final_source=load(P/'source-final-sha.json')['files'];final_c_source=load(P/'source-final-C-sha.json')['files']
    check(final_source==final_c_source and len(final_source)==6168,'Identical final B/C6168 source hashes')
    changed=[p for p in original_source if original_source[p]!=final_source[p]]
    check(changed==['app/store/model/nutrition/NutritionLack.php'] and final_source[changed[0]]=='a81a7a4eff91aec6745517a26a8794f8ccda4edf633c993406d5c557525d6afd','Exactly Root sole approved Model delta')
    for lane in ('B','C'):
        prior=BASE/'private/first-phase'/lane;folder=P/'private' if lane=='B' else P/'private/C/private'
        pre=load(folder/'preflight.json');dictionary=pre['protected']['dictionary'];cases=load(folder/'cases.json');cases.update(load(folder/'cases-extra.json'))
        expected_finish=18 if lane=='B' else 20;files=sorted(folder.glob('wire-*.json'));check(len(files)==expected_finish,'All expected finish wire receipts');check(len(list(folder.glob('client-*.json')))==expected_finish,'Actual finish clients count');check(not list(folder.glob('harness-failure-*')),'No harness failed capture')
        native=[json.loads(x,parse_float=Decimal) for x in (folder/'pdo-native.jsonl').read_text().splitlines()];prior_native=[json.loads(x,parse_float=Decimal) for x in (prior/'pdo-native.jsonl').read_text().splitlines()]
        histories=defaultdict(list);dml=[]
        for e in prior_native+native:
            if e['layer']=='actual PDOStatement.execute' and e['operation'] in ('INSERT','UPDATE','DELETE'):dml.append(e)
            histories[e['scenario']].append(e)
        verified={};bodies={};rollbacks=[]
        for f in files:
            w=load(f);n=w['scenario'];c=load(folder/('client-'+n+'.json'));check(c['raw']==w['raw'],'Actual client and server wire bytes');check(c['case']==w['case']==cases[n],'Exact frozen case receipt')
            raw=json.loads(w['raw'],parse_float=Decimal);bodies[n]=raw;allbodies[lane,n]=raw
            check(int(c['response_headers'][0].split()[1])==w['http_status'],'Client actual HTTP status equals server receipt')
            for path,sha in w['included_source_sha256'].items():check(sha==(final_source if w['source_stage']=='final' else original_source)[path],'Actual included file matches selected source stage')
            check(w['protected_before']==w['protected_after']==pre['protected'],'Whole old rows/actors/users/dictionary per request');check(raw['status']==(500 if '_CROSS' in n or '_MISSING' in n or (lane=='B' and n=='PROBLEM_UPDATE') or (lane=='C' and n=='PROBLEM_CREATE_RETRY') else 200),'Actual original status contract')
            events=histories[n];ops=[e for e in events if e['layer']=='actual PDOStatement.execute' and e['operation'] in ('INSERT','UPDATE','DELETE')]
            expected_ops=expected_dml(w,lane) if w['case']['kind']=='write' else []
            check(len(ops)<=len(expected_ops),'Native DML executed only approved prefix')
            for event,expected_op in zip(ops,expected_ops):
                values,where=sqlmaps(event);operation,table,ev,ew=expected_op
                check(event['operation']==operation and re.search(r'\byoshop_'+table+r'\b',event['queryString']) is not None,'Exact independently expected native operation/table/order')
                check(set(values)==set(ev) and set(where)==set(ew),'Exact independently expected SET/INSERT/WHERE field sets')
                for observed,expected_values in ((values,ev),(where,ew)):
                    for key,value in expected_values.items():
                        if value=='@time':check(isinstance(observed[key],str) and datetime.fromisoformat(observed[key]).year==2026,'Original bound datetime')
                        else:check(str(observed[key])==str(value),'Exact independently expected original native binding '+key)
            if raw['status']==500:
                check(w['before']==w['after'],'Rejected/failed exact rollback effect zero')
                if '人员不存在'==raw['message']:check(not ops,'Cross/missing DML0')
                else:
                    check(any(e['layer']=='actual PDO.rollBack' and e['success'] for e in events),'Actual native rollback');rollbacks.append(n)
                verified[n]={'status':500,'message':raw['message'],'effect_zero':True};continue
            if w['case']['kind']=='write':
                verified[n]=physical(w,c,lane);check(any(e['layer']=='actual PDO.beginTransaction' and e['success'] for e in events),'Real transaction begin');check(sum(e['layer']=='actual PDO.commit' for e in events)==1,'Exactly one actual commit')
                ids=[e for e in events if e['layer']=='actual PDO.lastInsertId'];check(len(ids)==w['successful_master_INSERT']+w['successful_answer_INSERT'],'Every real insert has native generated ID');check(all(e['php_type']=='string' and str(e['value']).isdigit() for e in ids),'Original PDO generated ID string types')
            else:
                check(w['before']==w['after'] and not ops,'Read DTO no DML')
                if n.startswith('DATES_'):
                    kind=int(n[6:]);expected=sorted(r['set_date'] for r in w['before']['nutrition_meals']+pre['protected']['nutrition_meals'] if int(r['staff_id'])==w['case']['body']['staff_id'] and int(r['user_id'])==(10074 if lane=='B' else 10010) and int(r['type'])==kind)
                else:
                    group=w['case']['group'];write_case=cases['FOOD_FRESH_CREATE'] if 'FRESH_DETAIL' in n else cases[group+('_CREATE' if 'DETAIL1' in n else '_UPDATE')];expected=detail(dictionary,group,write_case['body'])
                typed(raw['data']['detail'],expected,lane+'/'+n);verified[n]={'status':200,'complete_DTO_and_types':True}
            for e in ops:
                values,where=sqlmaps(e);check(not re.search(r';|--|/\*',e['queryString']),'Single safe original DML');check('yoshop_nutrition_meals' in e['queryString'] or 'yoshop_nutrition_lack' in e['queryString'],'Only two approved write tables')
                plans=[x for x in events if x['layer']=='framework original bind plan' and x['queryString']==e['queryString']];check(len(plans)==1,'Exact original framework bind plan before native DML')
                for name,pair in plans[0]['bind_plan'].items():
                    b=e['bindValue'][':'+name];check(b['value']==pair[0] and b['pdo_type']==pair[1],'Exact original framework/native binding and type')
                for k,v in where.items():
                    if k in ('staff_id','user_id','store_id','set_date'):check(str(v)==str(w['case']['body'].get(k,10074 if lane=='B' else 10010) if k=='user_id' else w['case']['body'].get(k,2147001111)),'Exact original scoped where '+k)
                if where and 'id' in where:check(int(where['id']) in [r['id'] for r in w['before']['nutrition_meals']],'Actual leased pk exists in independently full scope')
                for k in ('create_time','update_time'):values.pop(k,None)
                if e['operation']=='INSERT':
                    check(int(values['staff_id'])==w['case']['body']['staff_id'] and int(values['user_id'])==(10074 if lane=='B' else 10010) and int(values['store_id'])==2147001111 and values['set_date']==w['case']['body']['set_date'],'Every insert actual scoped binds')
        counts=Counter(e['operation'] for e in dml);success=Counter(e['operation'] for e in dml if e.get('success'));affected=Counter()
        for e in dml:
            if e.get('success'):affected[e['operation']]+=e['rowCount']
        check(counts==({'INSERT':23,'UPDATE':10,'DELETE':3} if lane=='B' else {'INSERT':25,'UPDATE':7,'DELETE':3}),('All attempt budgets',lane,counts))
        check(affected['UPDATE']==(12 if lane=='B' else 8) and affected['DELETE']==3,'Real affected includes all failed rollback attempts')
        # Failed attempts are recorded separately from committed final effects.
        check(len(rollbacks)==1,'One new exact tool rollback per side')
        for group in ('PROBLEM','MEALS','FOOD'):
            final[lane,group]=bodies[group+'_FINAL_CONTROL'] if lane=='B' else bodies[group+'_DETAIL2']
            if lane=='B':
                a=load(folder/('wire-'+group+'_DETAIL2.json'));b=load(folder/('wire-'+group+'_FINAL_CONTROL.json'));check(without_trace(a['raw'])==without_trace(b['raw']),'B complete original/final raw body exact outside trace')
        end=load(folder/'end-inspection.json');check(end['protected']==pre['protected'],'Final protected snapshot');check(end['source_files_exact']==6168,'Final full source pin before cleanup');check(len(end['own']['nutrition_meals'])==(3 if lane=='B' else 4) and len(end['own']['nutrition_lack'])==(13 if lane=='B' else 15),'Final retained marked rows')
        cleanup=load(folder/'cleanup.json');check(cleanup['own_workers_stopped'] and cleanup['own_port_closed'] and cleanup['token_files_erased'] and cleanup['source6168_pin_before_delete'],'Own resource cleanup receipt');gate=load(folder/('gate-test-result-'+lane+'.json'));check(gate['assertions']==216 and gate['negative_cases']==(197 if lane=='B' else 196),'Pure gate negative regression result')
        out[lane]={'HTTP_original':6 if lane=='B' else 4,'HTTP_finish':len(files),'HTTP_total':24,'native_DML_attempt_statements':dict(counts),'native_DML_success_statements_including_rolledback':dict(success),'native_affected_including_rolledback':dict(affected),'new_tool_rollbacks':rollbacks,'verified_finish_cases':verified,'original_protected_exact':True,'full_source_files':6168,'final_retained_masters':len(end['own']['nutrition_meals']),'final_retained_answers':len(end['own']['nutrition_lack']),'index_limit':'Initial phase did not snapshot indexes; end index catalog retained, no DDL authorized or executed'}
    for group in ('PROBLEM','MEALS','FOOD'):
        a=copy.deepcopy(final['B',group]);b=copy.deepcopy(final['C',group]);a.pop('trace_id');b.pop('trace_id');typed(a,b,'final_dual/'+group)
    out['assertions']=COUNT;out['HTTP_total']=48;out['scope']='Three original store survey writers, tested branches only; not all system compatibility';return out
if __name__=='__main__':
    result=run();target=P/'verification.json';serialized=json.dumps(result,ensure_ascii=False,indent=2,default=str)+'\n'
    if '--verify' in sys.argv:check(json.loads(target.read_text())==json.loads(serialized),'Sealed verification bytes content stable')
    else:target.write_text(serialized)
    print(json.dumps({'assertions':COUNT,'HTTP_total':48,'B_attempts':result['B']['native_DML_attempt_statements'],'C_attempts':result['C']['native_DML_attempt_statements'],'full_final_DTO_dual_equal':True},ensure_ascii=False))