#!/usr/bin/env python3
"""Aggregate and publish every eligible city pair and winning-pair trace evidence."""
import collections as co
import functools
import gzip
import html
import json
import sys
from pathlib import Path
import orjson as J
sys.dont_write_bytecode=True
import run_experiment as R
R.init()
HERE=R.HERE;PAPER=R.PAPER
sys.path.insert(0,str(PAPER/'city_group_rebuild_20261108'))
import prepare_current as C

def records():
    seen=set()
    for p in sorted((HERE/'daily').glob('*.evidence.jsonl.gz')):
        for line in gzip.open(p,'rb'):
            q=J.loads(line);tid=q['context']['trace_id']
            if tid in seen:continue
            seen.add(tid);yield q

def detail(o,ctx):
    l=R.ANCHORS[R.ANCHOR_IPS[o[5]]];rr=R.ANCHORS[R.ANCHOR_IPS[o[7]]]
    def anchor(z,idx,ttl):
        return {k:v for k,v in dict(z,ip=R.ANCHOR_IPS[idx],ttl=ttl).items() if k not in ('upstream_round1_representatives',)}
    return {'ip':R.TARGETS[o[0]],'mode':R.MODES[o[1]],'target_ttl':o[4],
      'left':anchor(l,o[5],o[6]),'right':anchor(rr,o[7],o[8]),
      'left_hops':o[4]-o[6],'right_hops':o[8]-o[4],
      'distance_km':R.CDIST[o[2],o[3]],'trace':ctx}

def sample_rank(o,ctx):
    return (*R.obs_rank(o)[:2],not ctx['selected_cohort'],ctx['path_grade']!='high',ctx['trace_id'],o[5],o[7],o[4])

def select_rank(g):
    return (g['distance_km'],*g['_rank'][:2],-g['trace_count'],g['left_city'],g['right_city'])

@functools.lru_cache(None)
def resolve_ref(rdns,asn,ipinfo):
    d=C.rdns_check.rdns_city_details(rdns,asn)
    if d.get('city'):
        confirmed=(d.get('source') or '').startswith('confirmed_library:')
        return {'city':d['city'],'country':d.get('country',''),'source':'rDNS确认库' if confirmed else 'rDNS脚本',
            'needs_review':not confirmed,'source_detail':d.get('source',''),'parser_confidence':d.get('confidence',''),
            'confirmed_hit':confirmed,'evidence':d.get('evidence',{})}
    return {'city':ipinfo,'country':'','source':'IPinfo' if ipinfo else '无位置','needs_review':False,
            'source_detail':'ipinfo_fallback' if ipinfo else 'missing','confirmed_hit':False}

def recover_selected(found):
    missing=set(R.SELECTED)-found
    if not missing:return []
    # The broad AS14593 archive requires at least one AS14593 internal response.
    # Recover page rows that fail that archive filter from the page's own cache.
    sys.path.insert(0,str(R.ROOT/'interconnect_db/scripts'))
    import build_routes_map as M
    recovered=[]
    folder=R.ROOT/'interconnect_db/2026-10-31/_work/active_ip_maps'
    for p in sorted(folder.glob('continent_*.jsonl.gz')):
        for n,line in enumerate(gzip.open(p,'rt'),1):
            tid,cat,js=line.rstrip('\n').split('\t',2)
            if tid not in missing:continue
            x=J.loads(js);hh=M.trace_hops(x);hops=[];meta={}
            for t,h in hh.items():
                if h['ip'] in ('*',None,''):continue
                hops.append(dict(h,ttl=t,ttl_uncertain=False));meta[h['ip']]={k:h.get(k) for k in ('asn','rdns','city')}
            r={'trace_id':tid,'day':'2026-10-31','start_utc':x.get('timestamp_utc'),
               'src':x.get('remote_ip'),'raw_src':x.get('remote_ip'),'node':x.get('remote_id'),
               'dst':x.get('ip'),'source_kind':'cloud','direction':'ingress','hops':hops,'meta':meta,
               'candidates':[],'unknown_candidates':[],'source_file':str(p),'source_index':n,
               '_partition_file':str(p),'_partition_line':n,'same_id_content_conflict':False,
               'record_origin':'page_trace_cache_reconstruction','original_page_cache_row':x}
            recovered.append(r);missing.remove(tid)
    assert not missing,('selected page cache missing',sorted(missing)[:10])
    with gzip.open(HERE/'selected_recovered_from_page.jsonl.gz','wb',compresslevel=1) as out:
        for r in recovered:out.write(J.dumps(r)+b'\n')
    return recovered

def build():
    scope=R.read(HERE/'scope_manifest.json');scan=R.read(HERE/'scan_manifest.json')
    groups={};eligible=0;selected_eligible=set();actual_utc_dates=set()
    for q in records():
        ctx=q['context'];eligible+=1;actual_utc_dates.add(ctx['start_utc'][:10])
        if ctx['selected_cohort']:selected_eligible.add(ctx['trace_id'])
        keys=set()
        for o in q['observations']:
            key=tuple(o[:4]);assert key not in keys;keys.add(key)
            assert o[6]<o[4]<o[8]
            if key not in groups:
                lc=R.CITY_KEYS[o[2]].rsplit('|',1);rc=R.CITY_KEYS[o[3]].rsplit('|',1)
                groups[key]={'ip':R.TARGETS[o[0]],'mode':R.MODES[o[1]],'left_city':lc[0],'left_country':lc[1],
                  'right_city':rc[0],'right_country':rc[1],'distance_km':R.CDIST[o[2],o[3]],'trace_count':0,
                  'partition_day_counts':co.Counter(),'measurement_utc_day_counts':co.Counter(),'source_counts':co.Counter(),
                  'confidence_patterns':co.Counter(),'path_grades':co.Counter(),'selected_trace_count':0,
                  '_rank':None,'example':None}
            g=groups[key];g['trace_count']+=1;g['partition_day_counts'][ctx['day']]+=1
            g['measurement_utc_day_counts'][ctx['start_utc'][:10]]+=1
            g['source_counts'][ctx['node'] or ctx['src']]+=1;g['selected_trace_count']+=ctx['selected_cohort']
            conf='/'.join(R.ANCHORS[R.ANCHOR_IPS[i]]['confidence'] for i in (o[5],o[7]));g['confidence_patterns'][conf]+=1;g['path_grades'][ctx['path_grade']]+=1
            rank=sample_rank(o,ctx)
            if g['_rank'] is None or rank<g['_rank']:g['_rank']=rank;g['example']=detail(o,ctx)
    byip=co.defaultdict(lambda:co.defaultdict(list));winners={}
    for k,g in groups.items():byip[k[0]][k[1]].append((k,g))
    for ti,modes in byip.items():
        for mode,gs in modes.items():
            gs.sort(key=lambda kv:select_rank(kv[1]));winners[ti,mode]=gs[0][0]
    candidates={R.TARGETS[ti]:{R.MODES[m]:[{k:v for k,v in g.items() if not k.startswith('_')} for key,g in gs] for m,gs in modes.items()} for ti,modes in byip.items()}
    R.write(HERE/'all_city_pair_candidates.json.gz',candidates)
    print(J.dumps({'stage':'ranked','eligible_unique_traces':eligible,'candidate_city_pairs_modes':len(groups),'targets_with_pair':len(byip)}).decode(),flush=True)
    # Resolve the original observed PTRs, preserving every metadata variant.
    metadata={}
    for p in sorted((HERE/'daily').glob('*.metadata.json.gz')):
        for x in R.read(p):
            key=tuple(x[k] for k in ('ip','rdns','asn','ipinfo_city'))
            if key not in metadata:metadata[key]=dict(x,partition_days=[])
            else:metadata[key]['trace_count']+=x['trace_count']
            metadata[key]['partition_days'].append(x['day'])
    refs=co.defaultdict(list);script_refs=[]
    for x in metadata.values():
        x['resolved']=resolve_ref(x['rdns'],x['asn'],x['ipinfo_city']);refs[x['ip']].append(x)
        if x['resolved']['needs_review']:script_refs.append(x)
    for vals in refs.values():vals.sort(key=lambda x:-x['trace_count'])
    R.write(HERE/'rdns_and_ipinfo_references.json.gz',dict(refs))
    R.write(HERE/'rdns_script_review.json',script_refs,True)
    # Winning-pair supports remain individually inspectable with their own hops.
    evidence=co.defaultdict(list);hop_stats={};selected_global_support=co.Counter()
    for q in records():
        ctx=q['context']
        for o in q['observations']:
            k=tuple(o[:4]);tm=(o[0],o[1])
            if winners.get(tm)!=k:continue
            evidence[o[0]].append({'observation':o,'trace':ctx})
            if tm not in hop_stats:
                hop_stats[tm]={'joint':co.Counter(),'ip_pairs':{},'pop_counts':co.Counter(),'directions':co.Counter(),
                    'repeat_target_traces':0,'known_other_city_inside':0}
            s=hop_stats[tm];s['joint'][o[4]-o[6],o[8]-o[4]]+=1
            ik=(o[5],o[7])
            if ik not in s['ip_pairs']:s['ip_pairs'][ik]={'trace_count':0,'day_counts':co.Counter(),'joint':co.Counter(),'example':detail(o,ctx)}
            pp=s['ip_pairs'][ik];pp['trace_count']+=1;pp['day_counts'][ctx['day']]+=1;pp['joint'][o[4]-o[6],o[8]-o[4]]+=1
            dp=ctx['destination_pop'];s['pop_counts'][dp.get('pop') or '['+dp['status']+']']+=1;s['directions'][ctx['direction']]+=1
            s['repeat_target_traces']+=R.TARGETS[o[0]] in ctx['repeated_target_ips']
            # A positive flag is a real observed other-city anchor; absence is not proof of no hidden detour.
            other=any(o[6]<oo[pos+1]<o[8] and oo[cpos] not in (o[2],o[3]) for oo in q['observations'] for pos,cpos in ((5,2),(7,3)))
            s['known_other_city_inside']+=other
            if ctx['selected_cohort'] and o[1]==0:selected_global_support[o[0]]+=1
    for ti,es in evidence.items():
        es.sort(key=lambda e:(e['observation'][1],e['trace']['day'],e['trace']['trace_id']))
    def joint(c):return [{'left_hops':k[0],'right_hops':k[1],'trace_count':v} for k,v in sorted(c.items(),key=lambda kv:(-kv[1],kv[0]))]
    rows=[]
    for ti,ip in enumerate(R.TARGETS):
        row=dict(scope['target_rows'][ti],index=ti,reference_locations=refs.get(ip,[]),nearest=None,nearest_high_only=None,
                 candidate_city_pair_count=len(byip.get(ti,{}).get(0,[])))
        for mode,col in ((0,'nearest'),(1,'nearest_high_only')):
            if (ti,mode) not in winners:continue
            g={k:v for k,v in groups[winners[ti,mode]].items() if not k.startswith('_')};s=hop_stats[ti,mode]
            g['joint_hop_distribution']=joint(s['joint']);g['joint_hop_mode']=g['joint_hop_distribution'][0]
            g['left_hop_range']=[min(k[0] for k in s['joint']),max(k[0] for k in s['joint'])]
            g['right_hop_range']=[min(k[1] for k in s['joint']),max(k[1] for k in s['joint'])]
            g['destination_pop_counts']=dict(s['pop_counts']);g['directions']=dict(s['directions'])
            g['repeated_target_trace_count']=s['repeat_target_traces'];g['observed_other_city_inside_trace_count']=s['known_other_city_inside']
            g['anchor_ip_pairs']=[]
            for (li,ri),p in sorted(s['ip_pairs'].items(),key=lambda kv:-kv[1]['trace_count']):
                g['anchor_ip_pairs'].append(dict(p,left_ip=R.ANCHOR_IPS[li],right_ip=R.ANCHOR_IPS[ri],joint=joint(p['joint'])))
            g['representative_ip_pair_trace_count']=next(z['trace_count'] for z in g['anchor_ip_pairs'] if z['left_ip']==g['example']['left']['ip'] and z['right_ip']==g['example']['right']['ip'])
            assert sum(s['joint'].values())==g['trace_count']==sum(p['trace_count'] for p in g['anchor_ip_pairs'])
            g['flags']=[]
            if g['trace_count']==1:g['flags'].append('仅 1 条 trace')
            if len(g['partition_day_counts'])==1:g['flags'].append('仅 1 个分区日')
            if len(g['source_counts'])==1:g['flags'].append('仅 1 个源测点')
            if not g['selected_trace_count']:g['flags'].append('该城市对仅见于原始样本之外')
            if g['distance_km']==0:g['flags'].append('同城锚点不等于目标已定位同城')
            if s['known_other_city_inside']:g['flags'].append('部分区间内另有已知城市')
            if s['repeat_target_traces']:g['flags'].append('部分 trace 中目标 IP 重复出现')
            row[col]=g
        rows.append(row)
    # Cover every selected trace, including the 615 rows outside the archive's initial filter.
    raws={}
    for p in (HERE/'daily').glob('*.selected_raw.jsonl.gz'):
        for ln in gzip.open(p,'rb'):
            r=J.loads(ln);raws.setdefault(r['trace_id'],r)
    recovered=recover_selected(set(raws))
    for r in recovered:raws[r['trace_id']]=r
    assert set(raws)==set(R.SELECTED)
    selected_results=[];selected_with_local=0;local_by_ip=co.defaultdict(list)
    for tid,s in R.SELECTED.items():
        r=raws[tid];local,hs,found,grade,outside=R.observations(r)
        members=sorted({h[1] for h in s['internal'] if h[1] in R.TARGET_SET})
        target_results=[]
        for ip in members:
            ti=R.TI[ip];opts=[o for k,o in local.items() if k[0]==ti and k[1]==0]
            best=min(opts,key=lambda o:(R.CDIST[o[2],o[3]],*R.obs_rank(o))) if opts else None
            global_key=winners.get((ti,0));same=global_key in local if global_key else False
            z={'ip':ip,'original_ttls':[h[0] for h in s['internal'] if h[1]==ip],
               'local_nearest_observation':best,'global_pair_seen_in_this_trace':same,
               'global_pair_observation':local.get(global_key),'path_grade':grade,
               'status':'本 trace 有有效夹击' if best else '本 trace 未形成合格两侧锚点'}
            target_results.append(z);local_by_ip[ti].append({'trace_id':tid,'node':r.get('node'),'src':r.get('src'),
                'dst':s['dst'],'page_pop':s['pop'],'page_pop_city':s['pop_city'],'start_utc':r.get('start_utc'),
                'review_category':s['cat'],'primary_label':s['label'],**z})
        selected_with_local+=any(z['local_nearest_observation'] for z in target_results)
        selected_results.append({'trace_id':tid,'review_category':s['cat'],'primary_label':s['label'],'city_pair':s['pair'],
          'node':r.get('node'),'src':r.get('src'),'dst':s['dst'],'page_pop':s['pop'],'page_pop_city':s['pop_city'],
          'start_utc':r.get('start_utc'),'partition_day':r['day'],'record_origin':r.get('record_origin','normalized_raw_partition'),
          'source_file':r.get('source_file'),'source_index':r.get('source_index'),'target_results':target_results})
    R.write(HERE/'selected_path_results.json.gz',selected_results)
    # Independent raw-path retrieval for all main and high-only representatives.
    reps={(x[col]['example']['trace']['partition_file'],x[col]['example']['trace']['partition_line']) for x in rows for col in ('nearest','nearest_high_only') if x[col]}
    rawreps={};want=co.defaultdict(set)
    for p,n in reps:want[p].add(n)
    for p,ns in sorted(want.items()):
        with gzip.open(p,'rb') as f:
            last=max(ns)
            for n,line in enumerate(f,1):
                if n in ns:rawreps[p,n]=J.loads(line)
                if n>=last:break
        print('verified raw representatives',Path(p).name,len(ns),flush=True)
    checks=[]
    for row in rows:
        for col in ('nearest','nearest_high_only'):
            if not row[col]:continue
            e=row[col]['example'];ctx=e['trace'];raw=rawreps[ctx['partition_file'],ctx['partition_line']]
            assert raw['trace_id']==ctx['trace_id']
            hs=co.defaultdict(set)
            for h in raw['hops']:
                if h.get('ip') not in (None,'*') and not h.get('ttl_uncertain'):hs[h['ttl']].add(h['ip'])
            for ip,t in ((row['ip'],e['target_ttl']),(e['left']['ip'],e['left']['ttl']),(e['right']['ip'],e['right']['ttl'])):assert hs[t]=={ip}
            assert e['left']['ttl']<e['target_ttl']<e['right']['ttl']
            e['path']=[{'ttl':t,'ips':sorted(hs.get(t,set()))} for t in range(e['left']['ttl'],e['right']['ttl']+1)]
            e['other_known_cities_in_interval']=sorted({R.ANCHORS[ip]['city'] for t,ips in hs.items() if e['left']['ttl']<t<e['right']['ttl'] for ip in ips if ip in R.ANCHORS and R.ANCHORS[ip]['city'] not in (e['left']['city'],e['right']['city'])})
            e['target_meta_reference']=raw.get('meta',{}).get(row['ip'],{})
            for side in ('left','right'):
                a=e[side];m=raw.get('meta',{}).get(a['ip'],{});a['observed_rdns']=m.get('rdns') or ''
                a['observed_city_reference']=resolve_ref(a['observed_rdns'],str(m.get('asn') or ''),m.get('ipinfo_city',m.get('city')) or '')
                a['upstream_round1_representatives']=R.ANCHORS[a['ip']]['upstream_round1_representatives']
            checks.append({'ip':row['ip'],'selection':col,'trace_id':raw['trace_id'],'ordered_single_reply_ttls_verified':True})
    R.write(HERE/'representative_raw_traces.json.gz',[dict(r,_partition_file=p,_partition_line=n) for (p,n),r in rawreps.items()])
    stats={'selected_traces':len(R.SELECTED),'selected_traces_with_targets':scope['selected_traces_with_targets'],
       'selected_traces_with_local_bracket':selected_with_local,'target_count':len(rows),
       'targets_with_bracket':sum(bool(r['nearest']) for r in rows),'targets_without_bracket':sum(not r['nearest'] for r in rows),
       'targets_with_high_only_bracket':sum(bool(r['nearest_high_only']) for r in rows),
       'targets_whose_main_pair_has_selected_support':sum(bool(r['nearest'] and r['nearest']['selected_trace_count']) for r in rows),
       'raw_partition_traces':scope['raw_trace_count'],'matching_partition_records':sum(x['stats']['matching_traces'] for x in scan),
       'eligible_unique_traces':eligible,'candidate_city_pairs_and_modes':len(groups),'partition_days':[x['day'] for x in scan],
       'eligible_measurement_utc_days':sorted(actual_utc_dates),'archive_selected_found':len(raws)-len(recovered),
       'selected_recovered_from_page_cache':len(recovered),'recovered_with_targets':sum(any(h['ip'] in R.TARGET_SET for h in r['hops']) for r in recovered),
       'rdns_script_pending_variants':len(script_refs),'anchor_count':len(R.ANCHORS),
       'main_pair_single_trace_targets':sum(bool(r['nearest'] and r['nearest']['trace_count']==1) for r in rows),
       'main_pair_same_city_targets':sum(bool(r['nearest'] and r['nearest']['distance_km']==0) for r in rows)}
    result={'stats':stats,'rules':{
      'scope':'绕路洲对 + 支持早入 / 第一跳未定位 / ic 无位置；保留原始 primary label，不把全部样本重新判为早入。',
      'selection':'先筛同 trace 中严格夹住目标的有效锚点；最小化城市中心大圆距离；同距离优先双高/较短 TTL，次数仅参与后续平局。',
      'anchors':'冻结严格一二轮单城市中高置信、第三轮新清理单城市中高置信、唯一城市 Starlink IXP；不递归增加本轮目标。',
      'rdns':R.INV['rdns_policy'],
      'eligibility':'cloud 外侧公网回应 TTL：0–1 排除、2–3 路径上限中、4+ 高；非 cloud 沿用原项目高上限。区间同 TTL 多 IP 排除；静默 TTL 最多 2；内容冲突排除；不使用源/目的 IP 充当内部锚点。',
      'counts':'城市对支持数按 trace_id 去重；每条 trace 每城市对选择一组代表锚点/目标 TTL。精确锚点 IP 对统计为被选为代表的支持数，不能当作该 IP 对所有出现次数。',
      'hops':'TTL 差，邻接为 1，包含静默 TTL；不代表 MPLS 隐藏物理跳。众数来自同一条观测的左右组合，不拼接左右独立最小值。',
      'dates':'保留分区日、UTC 时间及 UTC 日；目的 PoP 取 UTC 测量时刻之前最近的同日快照，没有更早同日快照则用该日首个快照。',
      'interpretation':'仅给出路径顺序的城市对候选；不是地理包围、地理线段证明或城市定位置信概率。不同 trace 的 hop 数与路径不能移植。',
      'coverage':'37,880,706 条是本地已筛选的“至少有一个 AS14593 内部回应”归档，不是整个项目未过滤的全部测量。615 条选中样本从原网页 trace 缓存补回；其中 614 条无本轮目标。',
      'provenance':'冻结轮次锚点可能依赖上游 IPinfo；所附首轮来源为代表证据，不宣称对所有上游证明完成审计。'},'rows':rows}
    R.write(HERE/'results.json',result,True)
    validation={'passed':True,'selected_trace_manifest_complete':len(selected_results)==7204,'all_targets_seen_in_archive':all(any(ip in z['per_ip'] for z in scan) for ip in R.TARGETS),
       'target_rows':len(rows),'target_anchor_overlap':sorted(R.TARGET_SET&set(R.ANCHORS)),
       'representative_checks':checks,'distance_minima_checked':all(not r['nearest'] or r['nearest']['distance_km']==min(g['distance_km'] for k,g in byip[r['index']][0]) for r in rows),
       'scope_sha256_matches':R.sha(R.PAGE)==scope['data_sha256'],'selected_cache_recovered':len(recovered),
       'limitations':result['rules']['coverage']}
    assert validation['passed'] and not validation['target_anchor_overlap'] and validation['distance_minima_checked'] and validation['scope_sha256_matches']
    R.write(HERE/'validation.json',validation,True)
    (HERE/'evidence').mkdir(exist_ok=True);(HERE/'details').mkdir(exist_ok=True)
    for row in rows:
        ti=row['index'];es=evidence.get(ti,[])
        with gzip.open(HERE/'evidence'/f'{ti}.jsonl.gz','wb',compresslevel=1) as out:
            for e in es:out.write(J.dumps(detail(e['observation'],e['trace']))+b'\n')
        payload={'row':row,'evidence':es,'local_paths':local_by_ip.get(ti,[]),'candidates':candidates.get(row['ip'],{}),
                 'anchors':{str(i):dict(R.ANCHORS[R.ANCHOR_IPS[i]],ip=R.ANCHOR_IPS[i]) for e in es for i in (e['observation'][5],e['observation'][7])},
                 'rules':result['rules']}
        # Include local-only anchors as well.
        for e in local_by_ip.get(ti,[]):
            for k in ('local_nearest_observation','global_pair_observation'):
                o=e.get(k)
                if o:
                    for i in (o[5],o[7]):payload['anchors'][str(i)]=dict(R.ANCHORS[R.ANCHOR_IPS[i]],ip=R.ANCHOR_IPS[i])
        (HERE/'details'/f'{ti}.js').write_bytes(b'window.DATA='+J.dumps(payload)+b';')
    from render_report import render
    render(result)
    print(J.dumps(stats).decode(),flush=True)

if __name__=='__main__':build()