#!/usr/bin/env python3
import copy,hashlib,json,re
from datetime import datetime,timedelta
from decimal import Decimal
from pathlib import Path
from zoneinfo import ZoneInfo
ROOT=Path(__file__).resolve().parent;TZ=ZoneInfo('Asia/Shanghai');META=json.loads((ROOT/'structure-B.json').read_text())['tables'];DAY='2026-10-11'
def equal(a,b,p='$'):
 if type(a) is not type(b):raise AssertionError(f'{p}: {type(a).__name__} != {type(b).__name__}')
 if isinstance(a,dict):
  assert a.keys()==b.keys(),(p,'keys',a.keys()^b.keys())
  for k in a:equal(a[k],b[k],p+'.'+k)
 elif isinstance(a,list):
  assert len(a)==len(b),(p,'length')
  for i,(x,y) in enumerate(zip(a,b)):equal(x,y,p+f'[{i}]')
 else:assert a==b,(p,a,b)
def diff(a,b,p='$'):
 if type(a) is not type(b):return [p]
 if isinstance(a,dict):
  if a.keys()!=b.keys():return [p+'.keys']
  return [x for k in a for x in diff(a[k],b[k],p+'.'+k)]
 if isinstance(a,list):
  if len(a)!=len(b):return [p+'.length']
  return [x for i,(aa,bb) in enumerate(zip(a,b)) for x in diff(aa,bb,p+f'[{i}]')]
 return [] if a==b else [p]
def date(v,day=False):return datetime.fromtimestamp(int(v),TZ).strftime('%Y-%m-%d' if day else '%Y-%m-%d %H:%M:%S')
def money(v):return format(Decimal(str(v)),'.2f')
def count(n):return format(n,',')
def rows(o,t,tenant):return [r for r in o[t] if int(r['store_id'])==tenant]
def during(v,day):return date(v,True)==day
def paid_orders(o,tenant,day=None,status_filter=True):return [r for r in rows(o,'order',tenant) if int(r['pay_status'])==20 and int(r['is_delete'])==0 and (not status_filter or int(r['order_status'])!=20) and (day is None or during(r['pay_time'],day))]
def survey(o,tenant,day=None):
 orders=paid_orders(o,tenant,day);consumers=paid_orders(o,tenant,day,False);users=[r for r in rows(o,'user',tenant) if int(r['is_delete'])==0 and (day is None or during(r['create_time'],day))];goods=[r for r in rows(o,'goods',tenant) if int(r['is_delete'])==0 and (day is None or during(r['create_time'],day))];recharge=[r for r in rows(o,'recharge_order',tenant) if int(r['pay_status'])==20 and (day is None or during(r['pay_time'],day))]
 return {'userTotal':count(len(users)),'consumeUsers':count(len({r['user_id'] for r in consumers})),'orderTotal':count(len(orders)),'orderTotalPrice':money(sum((Decimal(str(r['pay_price'])) for r in orders),Decimal(0))),'goodsTotal':count(len(goods)),'rechargeTotalMoney':money(sum((Decimal(str(r['actual_money'])) for r in recharge),Decimal(0)))}
def trend(o,tenant):
 days=[(datetime.fromisoformat(DAY)-timedelta(days=i)).date().isoformat() for i in range(6,-1,-1)];return {'date':days,'orderTotal':[count(len(paid_orders(o,tenant,d))) for d in days],'orderTotalPrice':[money(sum((Decimal(str(r['pay_price'])) for r in paid_orders(o,tenant,d)),Decimal(0))) for d in days]}
def ranking(o,tenant):
 orders={int(r['order_id']):r for r in o['order']};groups={}
 for r in rows(o,'order_goods',tenant):
  order=orders[int(r['order_id'])];assert int(order['store_id'])==tenant
  if int(order['pay_status'])!=20 or int(order['order_status'])==20:continue
  gid=int(r['goods_id']);g=groups.setdefault(gid,{'goods_id':gid,'names':[],'money':Decimal(0),'num':0});g['names'].append(r['goods_name']);g['money']+=Decimal(str(r['total_pay_price']));g['num']+=int(r['total_num'])
 values=sorted(groups.values(),key=lambda r:-r['money'])[:10];return [{'goods_id':r['goods_id'],'goods_name':max(r['names']),'sales_volume':money(r['money']),'total_sales_num':str(r['num'])} for r in values]
def expected_data(w,c):
 o=w['original'];tenant=2147001111 if c['module']=='store' else 2147001112;name=c['name']
 if name=='SURVEY':assert c['params']=={'startDate':DAY,'endDate':DAY};return survey(o,tenant,DAY)
 if name=='STATISTICS':
  users=[r for r in rows(o,'user',tenant) if int(r['is_delete'])==0];assert not users,'Nonempty active User ranking not covered';return {'data':{'overview':survey(o,tenant),'tradeTrend':trend(o,tenant),'goodsRanking':ranking(o,tenant),'userExpendRanking':[]}}
 if name=='HOME':
  ytd=(datetime.fromisoformat(DAY)-timedelta(days=1)).date().isoformat();ov={}
  for key in ('orderTotalPrice','orderTotal','newUserTotal','consumeUserTotal'):
   values={}
   for label,day in [('tday',DAY),('ytd',ytd)]:
    orders=paid_orders(o,tenant,day);users=[r for r in rows(o,'user',tenant) if int(r['is_delete'])==0 and during(r['create_time'],day)]
    values[label]=money(sum((Decimal(str(r['pay_price'])) for r in orders),Decimal(0))) if key=='orderTotalPrice' else count(len(orders)) if key=='orderTotal' else count(len(users)) if key=='newUserTotal' else count(len({r['user_id'] for r in paid_orders(o,tenant,day,False)}))
   ov[key]=values
  allorders=rows(o,'order',tenant);allgoods=rows(o,'goods',tenant);allusers=rows(o,'user',tenant)
  st={'goodsTotal':count(sum(int(r['is_delete'])==0 for r in allgoods)),'userTotal':count(sum(int(r['is_delete'])==0 for r in allusers)),'orderTotal':count(len(paid_orders(o,tenant))),'consumeUserTotal':count(sum(int(r['is_delete'])==0 and Decimal(str(r['pay_money']))>0 for r in allusers))}
  pending={'deliverOrderTotal':count(sum(int(r['is_delete'])==0 and int(r['pay_status'])==20 and int(r['delivery_status'])!=20 and int(r['order_status']) in (10,21) for r in allorders)),'refundTotal':count(sum(int(r['status'])==0 for r in rows(o,'order_refund',tenant))),'paidOrderTotal':count(sum(int(r['is_delete'])==0 and int(r['pay_status'])==10 and int(r['order_status'])==10 for r in allorders)),'soldoutGoodsTotal':count(sum(int(r['is_delete'])==0 and int(r['stock_total'])==0 and int(r['status'])==10 for r in allgoods))};return {'data':{'overview':ov,'statistics':st,'pending':pending,'tradeTrend':trend(o,tenant)}}
 assert name=='EXPORT_LIST';data=[]
 for r in sorted(rows(o,'order_export',tenant),key=lambda r:-int(r['create_time'])):
  d={col['column_name']:(date(r[col['column_name']],True) if col['column_name'] in ('start_time','end_time') else date(r[col['column_name']]) if col['column_name']=='create_time' else int(r[col['column_name']]) if 'int' in col['data_type'] else r[col['column_name']]) for col in META['order_export']['columns']};d['download_url']='http://'+c['module']+'.dashboard.invalid/'+r['file_path'];data.append(d)
 n=len(data);assert n==1;return {'list':{'total':n,'per_page':10,'current_page':1,'last_page':1,'data':data}}
def normalize(body,w,fixture,c):
 d=copy.deepcopy(body);d['trace_id']='VALIDATED_TRACE';ids=fixture['owned'][c['module']]
 if c['name']=='STATISTICS':
  for r in d['data']['data']['goodsRanking']:assert r['goods_id']==ids['goods'];r['goods_id']=c['module']+'.own_goods'
 if c['name']=='EXPORT_LIST':
  for r in d['data']['list']['data']:assert r['id']==ids['stat_order_export'];r['id']=c['module']+'.own_export'
 return d
def verify():
 normal={};bodymap={};casesout={};source=None;strict_reds=[];sql_reds=[];strict_pass=0;total=0
 alias={'store':'9b9b9ed24a630d2e2314648d5585b2a8bc1ffec6ffde5002b65db240bb2750ee','terminal':'78285345d0925a94eb4631741796ea9b3e8449082d6957ce4c91c1af91343cba'}
 final={m:hashlib.sha256((ROOT/(m+'GoodsRankingTextCandidate.php')).read_bytes()).hexdigest() for m in ('store','terminal')}
 for lane in ('B','C'):
  p=ROOT/'private-evidence'/lane;f=json.loads((p/'fixture-journal.json').read_text());cases={c['scenario']:c for path in p.glob('cases-*.json') for c in json.loads(path.read_text())};out=[]
  for line in (p/'private-wire.jsonl').read_text().splitlines():
   w=json.loads(line);s=w['scenario'];c=cases[s];phase=s.split('_')[2];b=json.loads(w['raw']);total+=1;assert w['http_status']==200 and set(b)=={'status','message','data','trace_id'} and type(b['status']) is int and type(b['message']) is str and re.fullmatch('[0-9a-f]{13}',b['trace_id']);equal(w['before'],f['after'],s+'.before23');equal(w['original'],f['after'],s+'.after23');equal(w['protect_before'],f['before_protect'],s+'.natural');equal(w['protect_before'],w['protect_after'],s+'.natural_read');assert not set(w['operations'])&{'insert','update','delete','merge','replace'}
   if source is None:source=w['source_sha256']
   wanted=copy.deepcopy(source)
   if phase!='NA':
    for m in ('store','terminal'):wanted['app/'+m+'/service/statistics/data/GoodsRanking.php']=(alias if phase=='GREEN' else final)[m]
   equal(w['source_sha256'],wanted,s+'.actual_phase_source')
   sqlred=lane=='C' and phase=='NA' and c['name']=='STATISTICS';typered=lane=='C' and phase=='GREEN'
   if sqlred:assert b['status']==500 and '-2007' in b['message'] and '[order]' in b['message'];equal(b['data'],[],s+'.sql_error_data');sql_reds.append(s)
   else:
    assert b['status']==200 and b['message']=='success';e=expected_data(w,c)
    if typered:
     paths=diff(b['data'],e);assert set(paths)=={'$.data.goodsRanking[0].sales_volume','$.data.goodsRanking[0].total_sales_num'},(s,paths);strict_reds.append({'scenario':s,'paths':paths})
    else:equal(b['data'],e,s+'.complete_source_DTO');strict_pass+=1;normal[(lane,s)]=normalize(b,w,f,c)
   bodymap[(lane,s)]=b;out.append({'scenario':s,'status':b['status'],'historical_SQL_red':sqlred,'historical_SUM_DTO_type_red':typered,'full_original_DTO_equal':not(sqlred or typered),'raw_sha256':hashlib.sha256(w['raw'].encode()).hexdigest(),'elapsed_ms':w['elapsed_ms']})
  casesout[lane]=out
 pairs=[];unchanged=[]
 for module in ('STORE','TERMINAL'):
  for name in ('HOME','STATISTICS','SURVEY','EXPORT_LIST'):
   bkey=f'PILOT_DS_'+('CONTROL' if name=='STATISTICS' else 'NA')+f'_{module}_{name}';ckey=f'PILOT_DS_'+('FINAL' if name=='STATISTICS' else 'NA')+f'_{module}_{name}';equal(normal[('B',bkey)],normal[('C',ckey)],ckey+'.full_BC_body');pairs.append({'B':bkey,'C':ckey})
   if name=='STATISTICS':
    old=copy.deepcopy(bodymap[('B',f'PILOT_DS_NA_{module}_{name}')]);new=copy.deepcopy(bodymap[('B',bkey)]);old['trace_id']=new['trace_id']='VALIDATED_TRACE';equal(old,new,bkey+'.MySQL_original_body_exact');unchanged.append(bkey)
 assert len(sql_reds)==2 and len(strict_reds)==2 and len(pairs)==8
 return {'HTTP':total,'historical_C_SQL_reds':sql_reds,'historical_C_SUM_type_reds':strict_reds,'full_source_DTO_passes':strict_pass,'complete_BC_positive_pairs':pairs,'MySQL_final_body_unchanged':unchanged,'original_source_sha256':source,'cases':casesout}
if __name__=='__main__':
 r=verify();(ROOT/'comparison.json').write_text(json.dumps(r,ensure_ascii=False,indent=2)+'\n');print(json.dumps({k:v for k,v in r.items() if k not in ('cases','original_source_sha256')}))