import json,re
from decimal import Decimal
from compare import ROOT,equal
def audit():
 out={}
 for lane in ('B','C'):
  p=ROOT/'private-evidence'/lane;f=json.loads((p/'fixture-journal.json').read_text());assert f['committed'] and f['in_transaction_old23_stable'] and len(f['dml'])==6
  for item in f['dml']:
   t=item['table'];pk={'order':'order_id','order_goods':'order_goods_id','order_export':'id'}[t];row=next(r for r in f['after'][t] if int(r[pk])==item['generated_PK']);assert set(row)==set(item['values'])|{pk}
   for k,v in item['values'].items():
    if type(v) is int:assert int(row[k])==v
    elif k in ('total_price','order_price','coupon_money','points_money','pay_price','update_price','express_price','goods_price','line_price','grade_goods_price','grade_total_money','total_pay_price'):assert Decimal(str(row[k]))==Decimal(v)==0
    else:equal(row[k],v,lane+'.fixture_full_column.'+t+'.'+k)
   if t=='order':assert int(row['pay_status'])==20 and int(row['order_status'])==30 and int(row['is_settled'])==int(row['is_comment'])==1 and int(row['trade_id'])==0 and row['pay_method']=='cash'
   if t=='order_goods':assert int(row['total_num'])==int(row['delivery_num'])==1
  for t,old in f['before'].items():
   if t in ('order','order_goods','order_export'):
    assert len(f['after'][t])==len(old)+2
    for r in old:assert r in f['after'][t]
   else:equal(old,f['after'][t],lane+'.old_unchanged.'+t)
  equal(f['before_protect'],f['after_protect'],lane+'.fixture_natural')
  counts={'HTTP':0,'HTTP_DML':0,'DDL':0,'preparation_INSERT':6,'preparation_UPDATE':0,'preparation_DELETE':0,'business_SELECT':0,'metadata_session':0,'independent_after_original_table_sets':0,'independent_before_original_table_sets':0}
  for line in (p/'private-sql.jsonl').read_text().splitlines():
   r=json.loads(line);sql=r['sql'];assert r['layer'].startswith('ThinkPHP framework listener debug interpolation');assert not re.match(r'^\s*(insert|update|delete|alter|drop|create|truncate|merge|replace|grant|revoke)\b',sql,re.I)
   if re.match(r'^\s*SELECT\b',sql,re.I) and not re.search(r'\bSYSOBJECTS\b|\bSYSCOLUMNS\b|USER_TAB_COLUMNS|USER_CONS_COLUMNS|USER_CONSTRAINTS|\bDUAL\b',sql,re.I):counts['business_SELECT']+=1
   else:assert re.match(r'^\s*(SELECT|SHOW|SET SCHEMA|CALL SP_TABLEDEF)\b',sql,re.I);counts['metadata_session']+=1
  for line in (p/'private-wire.jsonl').read_text().splitlines():
   w=json.loads(line);counts['HTTP']+=1;counts['independent_after_original_table_sets']+=23;counts['independent_before_original_table_sets']+=23;equal(w['before'],f['after'],w['scenario']+'.before23');equal(w['original'],f['after'],w['scenario']+'.after23');equal(w['protect_before'],w['protect_after'],'natural/read')
  for path in p.glob('run-*.json'):assert json.loads(path.read_text())['all_own_tokens_absent']
  final=json.loads((p/'final-original.json').read_text());equal(final['original'],f['after'],lane+'.final23');equal(final['protect'],f['before_protect'],lane+'.final_natural')
  out[lane]={'counts':counts,'SQL_evidence_layer':'framework Db::listen getRealSql/getLastSql debug interpolation, NOT prepared queryString or server trace','original_failed_prepare_unobserved_in_listener':2 if lane=='C' else 0}
 return out
if __name__=='__main__':
 r=audit();(ROOT/'sql-audit.json').write_text(json.dumps(r,indent=2)+'\n');print(json.dumps(r))