"""Offline exact JSON/independent SQL CAST consumer verification. No DB or HTTP."""
from pathlib import Path
from decimal import Decimal
import argparse
import copy
import json
import struct
import hashlib

HERE = Path(__file__).resolve().parent
PRIVATE = HERE / 'private-evidence'

def read(name):
    return json.loads((PRIVATE / name).read_text(), parse_float=Decimal)

def manifest_check():
    path=HERE/'FINAL_EXECUTION_MANIFEST.json'
    if not path.exists(): return
    manifest=json.loads(path.read_text())
    for group in ['public_files','private_files']:
        for item in manifest[group]:
            file=HERE/item['path']
            assert file.is_file()
            assert hashlib.sha256(file.read_bytes()).hexdigest()==item['sha256'],item['path']
            assert file.stat().st_size==item['bytes']
            if group=='private_files': assert file.stat().st_mode & 0o777 == 0o600
    assert PRIVATE.stat().st_mode & 0o777 == 0o700

def same(a, b):
    if type(a) is not type(b):
        return False
    if isinstance(a, dict):
        return a.keys() == b.keys() 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 amount(value):
    return f'{Decimal(str(value)):.2f}'

def php_hash(value):
    text = json.dumps(value, ensure_ascii=True, separators=(',', ':')).replace('/', '\\/')
    return hashlib.sha256(text.encode()).hexdigest()

def preparation(lane):
    journal = read(f'fixture-journal-{lane}.json')
    proposal = read('FIXTURE_PROPOSAL.json')['rows'][lane]
    before = read(f'all-before-{lane}.json')
    after = read(f'all-after-{lane}.json')
    assert journal['status'] == 'committed' and journal['DML_attempts'] == 5
    assert journal['rows'] == proposal
    assert php_hash(before) == journal['before_sha256']
    own = journal['generated_rows']
    ids = [row['log_id'] for row in own]
    assert len(ids) == len(set(ids)) == 5 and all(int(x) > 0 for x in ids)
    assert [row for row in after if row['log_id'] not in ids] == before
    assert [row for row in after if row['log_id'] in ids] == own
    assert len(after) == len(before) + 5
    assert 'LOG_ID' not in journal['last_sql'].split('VALUES')[0].upper()
    for row, explicit in zip(own, proposal):
        assert set(row) == set(explicit) | {'log_id'}
        for key, value in explicit.items():
            if key in ['money', 'balance_before', 'balance_after']:
                assert Decimal(row[key]) == Decimal(str(value))
            elif lane == 'C' and key in ['create_time', 'update_time']:
                assert row[key] == value + '.000000'
            else:
                assert row[key] == str(value)

def source_and_sql(lane, phase):
    proof=read(f'query-layer-{lane}-{phase}.json')
    assert proof['assertions']==210 and proof['passed'] is True
    assert proof['DML']==proof['DDL']==proof['HTTP']==0
    for kind in ['account','wallet']:
        assert proof['result'][kind+'_page1']['total']==5
        assert len(proof['result'][kind+'_page1']['data'])==5
        assert proof['result'][kind+'_page2']['data']==[]
        assert proof['result'][kind+'_empty']['total']==0
        assert proof['result'][kind+'_empty']['data']==[]
    sqls=read(f'query-layer-sql-{lane}-{phase}.json')
    count=total=precise=0
    for event in sqls:
        # Legacy listener interpolation is explicitly not the SQL oracle:
        # empty binds can leave metadata values that corrupt its SELECT text.
        for key in ['store_queryString','ai_queryString']:
            if key not in event: continue
            sql=event[key].lstrip().upper()
            assert sql.startswith(('SELECT','CALL','SHOW'))
            if 'USER_GIFT_LOG' not in sql: continue
            if sql.startswith('SELECT COUNT('):
                count+=1;assert 'GIFT_BIZ_ORDER_ID_EXACT' not in sql
            if sql.startswith('SELECT SUM('):
                total+=1;assert 'GIFT_BIZ_ORDER_ID_EXACT' not in sql
            if 'GIFT_BIZ_ORDER_ID_EXACT' in sql: precise+=1
    assert count>=1 and total>=1
    assert (precise>0) is (lane=='C' and phase=='green')
    audit=read(f'source-audit-{lane}.json')
    assert len(audit['classes'])==18 and audit['PDO']==audit['SQL']==0
    targets={'app\\p\\logic\\AccountSetting':'AccountSetting.php',
             'app\\p\\logic\\WalletFlow':'WalletFlow.php',
             'app\\common\\service\\account\\GiftFlowRead':'GiftFlowRead.php'}
    for row in audit['classes']:
        if row['class'] in targets:
            assert hashlib.sha256((PRIVATE/'source-final'/targets[row['class']]).read_bytes()).hexdigest()==row['sha256']

def expected(lane, kind):
    journal = read(f'fixture-journal-{lane}.json')
    assert journal['status'] == 'committed' and journal['DML_attempts'] == 5
    profile = read(f'profile-{lane}.json')
    source = list(reversed(journal['generated_rows']))
    rows = []
    for row in source:
        if kind == 'account':
            item = {key: row[key] for key in ['log_id','user_id','scene','money',
                'balance_before','balance_after','biz_type','biz_order_id',
                'biz_order_no','describe','remark','operator_name','create_time']}
            for key in ['log_id','user_id','scene']:
                item[key] = int(item[key])
            value = int(item['biz_order_id'])
            item['biz_order_id'] = value if value <= 9223372036854775807 else str(value)
            for key in ['money','balance_before','balance_after']:
                item[key] = amount(item[key])
            item.update(create_time='2001-01-01 00:00:00',
                        nickname=profile['nickname'], mobile=profile['mobile'])
        else:
            item = dict(flow_no='GIFT--'+row['log_id'],occur_time='2001-01-01 00:00:00',
                name=profile['nickname'],current_department=profile['department'],
                account_type='gift',account_type_name='赠送',change_type='人工调整',
                direction='入账',balance_before=amount(row['balance_before']),
                change_amount='0.00',balance_after=amount(row['balance_after']),
                business_no=row['biz_order_no'],source_channel='管理后台',
                operator_name=row['operator_name'])
        rows.append(item)
    data = dict(total=5,per_page=100,current_page=1,data=rows)
    if kind == 'account':
        data.update(last_page=1,income_amount='0.00',expense_amount='0.00')
    else:
        data['summary'] = dict(total_inflow='0.00',total_outflow='0.00',net_change='0.00')
    return dict(code=0,message='查询成功',data=data)

def run(phase):
    manifest_check()
    report = {'phase':phase,'HTTP_replayed':0,'DB_access':0,'checks':[]}
    for lane in ['B','C']:
        preparation(lane)
        source_and_sql(lane,phase)
        use_phase = 'red' if lane == 'B' else phase
        for kind in ['account','wallet']:
            body = read(f'body-{lane}-{use_phase}-{kind}-positive.json')
            oracle = expected(lane,kind)
            matches = same(body,oracle)
            if lane == 'C' and phase == 'red' and kind == 'account':
                wrong = [r['biz_order_no'] for r,e in zip(body['data']['data'],oracle['data']['data']) if not same(r['biz_order_id'],e['biz_order_id'])]
                assert len(wrong) == 3 and not matches
                # The other existing DM discrepancy is the 6 zero microseconds
                # leaked by AccountSetting::formatTimestamp, not a wide-id loss.
                aligned = copy.deepcopy(body)
                for r,e in zip(aligned['data']['data'],oracle['data']['data']):
                    r['biz_order_id'] = e['biz_order_id']
                    assert r['create_time'] == '2001-01-01 00:00:00.000000'
                    r['create_time'] = e['create_time']
                assert same(aligned,oracle)
                report['checks'].append({'lane':lane,'kind':kind,'expected_red_wide_rows':3,'expected_red_zero_microsecond_rows':5})
            else:
                assert matches,(lane,kind,'full body value/type mismatch')
                report['checks'].append({'lane':lane,'kind':kind,'full_body_exact':True})
    # Exact binary64 reconstruction demonstrates native driver loss independently
    # of PHP's integer cast, scientific decimal rendering, or JavaScript Number.
    native = read('internal-C.json')['rows']
    losses = 0
    for row in native:
        value = struct.unpack('>d',bytes.fromhex(row['native_double_be_hex']))[0]
        num,den = value.as_integer_ratio()
        losses += num != int(row['exact_sql_text']) * den
    assert losses == 3
    report['native_binary64_exact_losses'] = losses
    logic = read(f'logic-C-{phase}.json')
    source_map={row['log_id']:row for row in read('fixture-journal-C.json')['generated_rows']}
    failures=0
    assert len(logic['rows']) == 5
    for row in logic['rows']:
        exact=source_map[row['log_id']]['biz_order_id']
        assert row['source_exact_text'] == exact
        source_int=min(int(exact),9223372036854775807)
        assert row['source_contract_order_id'] == source_int
        matches=same(row['normalized_order_id'],source_int)
        assert row['matches_source_contract'] is matches
        failures += not matches
    assert failures == (3 if phase == 'red' else 0)
    report['real_Wallet_logic_internal_failures'] = failures
    for lane in ['B','C']:
        assert read(f'readback-{lane}.json')['all_old_rows_exact_unchanged']
        assert read(f'readback-{lane}.json')['all_owner_staff_actor_hash_unchanged']
    return report

if __name__ == '__main__':
    parser=argparse.ArgumentParser();parser.add_argument('--phase',choices=['red','green'],default='red')
    result=run(parser.parse_args().phase)
    print(json.dumps(result,ensure_ascii=False,indent=2))