"""Offline independent full DTO, real SQL, original row and resource verification."""
import base64,copy,hashlib,json,math,re
from datetime import datetime
from pathlib import Path
from zoneinfo import ZoneInfo
from oracle import oracle
HERE=Path(__file__).resolve().parent
PLAN=json.loads((HERE/'log-http-plan.json').read_text());COUNT=0
def check(ok,why):
    global COUNT
    COUNT+=1
    if not ok:raise AssertionError(why)
def exact(a,e,where='',ordered=True):
    check(type(a) is type(e),where+' type')
    if isinstance(e,dict):
        check(list(a)==list(e) if ordered else set(a)==set(e),where+' keys/order')
        for k,v in e.items():exact(a[k],v,where+'/'+str(k),ordered)
    elif isinstance(e,list):
        check(len(a)==len(e),where+' length')
        for i,v in enumerate(e):exact(a[i],v,where+'/'+str(i),ordered)
    else:check(a==e,where+' value')
def load(lane,name,phase=''):return json.loads((HERE/phase/'private'/lane/name).read_text())
def expected(lane):
    s=load(lane,'batch-before10.json');d=load(lane,'dictionary.json');out=oracle(s,d)
    frozen=load(lane,'expected10-before-http.json');exact(out,frozen,'immutable preHTTP oracle')
    # Source foreach inserts order during the first category iteration, after zhushi.
    # Correct only the original oracle's object insertion order; no returned values drive it.
    f=out['DietaryFiber']['data'];out['DietaryFiber']['data']={k:f[k] for k in ['zhushi','order','sucai','huncai','shuiguo','dannaijianguo','tangyin','sum','recomend','need_evaluate']}
    return out
def native_logs(w,case,lane,prior):
    rows=w['new_logs'];fields=PLAN['log_SQL_contract']['insert_fields_order'];check(len(rows)==case['log_rows'],'log row budget')
    native=[e for e in w['native'] if e['layer']=='actual PDOStatement.execute' and re.match(r'^INSERT\b',e['queryString'])]
    check(len(native)==1 and native[0]['success'] and native[0]['rowCount']==len(rows),'actual original insertAll')
    ev=native[0];check(ev['bound']==[] and ev['execute_params'] in [None,[]] and ev['errorInfo'][0]=='00000','literal/no binds/native no error')
    sql=ev['queryString'];prefix='INSERT INTO `ydy_initial_menu_log` ('+' , '.join('`'+f+'`' for f in fields)+') VALUES '
    check(sql.startswith(prefix),'11 exact SQL fields and order');tail=sql[len(prefix):].strip();token=r"'[^'\\]*'|-?[0-9]+";row_re=re.compile(r'\(\s*((?:'+token+r')(?:\s*,\s*(?:'+token+r')){10})\s*\)');offset=0;parsed=[]
    while offset<len(tail):
        m=row_re.match(tail,offset);check(m is not None,'independent literal tuple');values=[v[1:-1] if v.startswith("'") else int(v) for v in re.findall(token,m.group(1))];parsed.append(dict(zip(fields,values)));offset=m.end()
        if offset<len(tail):
            sep=re.compile(r'\s*,\s*').match(tail,offset);check(sep is not None,'tuple separator');offset=sep.end();check(offset<len(tail),'no trailing separator')
    exact(parsed,w['parsed_native_log_rows'],'independent parsed native')
    prep=[e for e in w['native'] if e['layer']=='actual PDO.prepare' and e.get('operation')=='approved-log-insert'];check(len(prep)==1 and prep[0]['prepared'] and prep[0]['queryString']==sql,'actual native prepare')
    for ev in w['native']:
        if ev['layer']=='actual PDOStatement.execute':check(ev['success'],'all native executions pass')
        if re.match(r'^\s*(INSERT|UPDATE|DELETE|CREATE|ALTER|DROP|TRUNCATE|MERGE)\b',ev.get('queryString',''),re.I):check(ev['queryString'].startswith(prefix),'no other native writes')
    guids={r['uuid'] for r in prior};ids={r['id'] for r in prior}
    for i,r in enumerate(rows):
        check(set(r)==set(fields)|{'id','customer_uuid'},'13 physical log columns')
        check(r['customer_uuid']==PLAN['log_SQL_contract']['default_customer_uuid'],'default customer uuid')
        check(type(r['id']) is int and r['id']>0 and r['id'] not in ids,'generated new id');ids.add(r['id'])
        check(type(r['uuid']) is str and re.fullmatch(r'[A-F0-9]{8}(-[A-F0-9]{4}){3}-[A-F0-9]{12}',r['uuid']) and r['uuid'] not in guids,'unique original GUID');guids.add(r['uuid'])
        t=datetime.fromisoformat(r['create_time']).replace(tzinfo=ZoneInfo('Asia/Shanghai')).timestamp();check(math.floor(w['started_epoch'])-1<=t<=math.ceil(w['completed_epoch'])+1,'request bounded timestamp')
        for f,v in case['log_values'][i].items():
            if f in ['uuid','create_time']:continue
            if isinstance(v,dict):v=v['lane_'+lane]
            check(str(r[f])==str(v) and str(parsed[i][f])==str(v),'exact log value '+f)
        for f,v in parsed[i].items():
            if f=='create_time':check(datetime.fromisoformat(r[f])==datetime.fromisoformat(v),'native/readback full timestamp')
            else:check(str(r[f])==str(v),'native/readback '+f)
def run():
    report=dict(HTTP=0,log_counts={},full_DTO={},original_DietaryFiber={},sentinel_deltas={},negative_cases=0,oracle_order_correction='Source foreach inserts order after zhushi; preHTTP value oracle retained immutable')
    all_wires={};expecteds={}
    for lane in ['B','C']:
        exp=expected(lane);expecteds[lane]=exp;prior=load(lane,'logs-before10.json');check(len(prior)==97,'original97');original=copy.deepcopy(prior);all_wires[lane]={}
        for phase,attemptfile in [('', 'http-attempts10.json'),('clientheaders','http-attempts-clientheaders.json')]:
            attempts=load(lane,attemptfile,phase);check(len(attempts)==(10 if phase=='' else 1),'actual HTTP budget');report['HTTP']+=len(attempts)
            for a in attempts:
                w=load(lane,'wire-'+a['label']+'.json',phase);all_wires[lane][w['label']]=w
                check(a['complete'] and base64.b64decode(a['body_base64']).decode()==w['body'],'exact client/wire bytes');check(hashlib.sha256(w['body'].encode()).hexdigest()==w['body_sha256'],'bodySHA')
                check(w['status']==200 and w['source_unchanged'] and w['source_before']==w['source_after'],'native200/original8 protected')
                check(w['old_logs_unchanged'] and w['logs_before']==prior and w['logs_after']==prior+w['new_logs'],'every prior log row unchanged')
                case=next(c for c in PLAN['cases'] if c['path']==w['path']);native_logs(w,case,lane,prior);prior=w['logs_after']
                action=w['path'].split('/')[-1]
                if action=='DietaryFiber' and phase=='':
                    decoded,end=json.JSONDecoder().raw_decode(w['body']);exact(decoded,exp[action],lane+'/'+action+' original typed prefix')
                    check(w['body'][end:].startswith('<div id="think_page_trace"'),'original complete raw debug suffix preserved')
                    report['original_DietaryFiber'][lane]=dict(status=200,full_body_is_JSON=False,raw_bytes=len(w['body'].encode()),json_prefix_bytes=len(w['body'][:end].encode()),suffix_bytes=len(w['body'][end:].encode()),source_return_type='json_encode string / HTML response')
                else:
                    exact(json.loads(w['body']),exp[action],lane+'/'+action+' fullbody')
                    if phase:
                        headers={'Accept':'application/json, text/plain, */*','Cache-Control':'max-age=28880'};exact(w['actual_client_headers'],headers,'actual original axios headers');exact(a['headers_sent'],headers,'client headers sent')
                        check('think_page_trace' not in w['body'],'no trace with original Accept')
                        encoded=json.dumps(exp[action],ensure_ascii=True,separators=(',',':'));check(w['body']==encoded,'original string bytes/order/unicode escapes retained')
            fin=load(lane,'final10.json',phase);check(fin['own_rows_unchanged'] and fin['logs']==prior,'final exact log/source set')
            resource=load(lane,'resource-check.json',phase);check(all(resource[k] for k in ['own_worker_absent','own_port_closed','original_parent_present','request_absent','cache_absent']) and resource['AppleDouble']==[],'exact own cleanup')
            prov=load(lane,'source-provenance.json',phase);check(prov['source_php_files']==958 and prov['source_lock_sha256']=='57d57588410be71682c8597d092b125cb26cf2b3d489b6ce7f1a136c56bad96b','APP958 pin')
            if phase:exact(fin,load(lane,'final10-toolfirst.json',phase),'toolfirst saved final unchanged')
        check(len(prior)==148 and prior[:97]==original,'97+48+3 full conservation')
        af=load(lane,'batch-after10.json');bef=load(lane,'batch-before10.json');report['sentinel_deltas'][lane]={k:dict(before=v,after=af['sentinel'][k]) for k,v in bef['sentinel'].items() if v!=af['sentinel'][k]}
        report['log_counts'][lane]=dict(old=97,original10_new=48,clientheaders_new=3,final=148);report['full_DTO'][lane]=list(exp)
        for f in [HERE/'private'/lane/'typed-expected10.json',HERE/'clientheaders'/'private'/lane/'typed-expected-DietaryFiber.json']:
            value=exp if 'expected10' in f.name else exp['DietaryFiber'];f.write_text(json.dumps(value,ensure_ascii=False,indent=2)+'\n');f.chmod(0o600)
    exact(expecteds['B'],expecteds['C'],'independent source DTO parity')
    for action in expecteds['B']:
        suffix='-clientheaders' if action=='DietaryFiber' else '-positive';check(all_wires['B'][action+suffix]['body']==all_wires['C'][action+suffix]['body'],'full native body parity '+action)
    dto=expecteds['B']['Protein']
    for bad in [None,{},[],{**dto,'code':'0'},{**dto,'unexpected':1},{**dto,'data':{}},{**dto,'data':{**dto['data'],'all':{}}},{**dto,'data':{**dto['data'],'morning':{**dto['data']['morning'],'all_sum':'10'}}},{**dto,'data':{**dto['data'],'morning':{**dto['data']['morning'],'all_sum':10.0}}},{'message':dto['message'],'code':0,'data':dto['data']}]:
        try:exact(bad,dto,'counterexample')
        except AssertionError:report['negative_cases']+=1
        else:raise AssertionError('Verifier accepted counterexample')
    report['assertions']=COUNT;report['status']='10 methods selected full DTO pass; raw original DietaryFiber trace preserved; client Accept supplemental verified'
    (HERE/'verification10.json').write_text(json.dumps(report,ensure_ascii=False,indent=2)+'\n');print(json.dumps({k:v for k,v in report.items() if k not in ['full_DTO','sentinel_deltas']},ensure_ascii=False))
if __name__=='__main__':run()