#!/usr/bin/env python3
"""fixbatch-004 平价验收：local qwen-eye vs hosted glm-5.3-flash 双臂 28 对判定对比。

用法: python3 parity_run.py <arm>   # arm ∈ {local, hosted}
每个 arm 在独立进程跑一次(模块级 sha256 缓存进程间隔离)。
样本 = ep-cat-worker-001 标准 28 对(20 pos + 8 交叉负),每镜头抽 f0/f2/f4 三帧,
verify_image_content 逐帧判 → 多数票聚合成镜头 verdict。
输出 results/parity_<arm>.jsonl
"""
import asyncio, json, sys, os
from pathlib import Path

# 绕代理直连智谱(hosted 臂需要; local 不受影响)
for k in list(os.environ):
    if 'proxy' in k.lower():
        del os.environ[k]

WT = Path('/data/workspace/kmc-fix-vvadapter-wt')
sys.path.insert(0, str(WT))
from plugins.kais_aigc import vision_verify as vv  # noqa: E402

EP = Path('/home/kai/runs/ep-cat-worker-001')
FR = Path('/home/kai/shared/qwen-eye-exp/frames')
OUT = Path('/home/kai/shared/qwen-eye-exp/results')
ARM = sys.argv[1]
ORDER = ['qwen-eye'] if ARM == 'local' else ['glm-5.3-flash']

shots = json.load(open(EP / '.pipeline-assets/shot-list.json'))['value']
spec_by_id = {s['shot_id']: s for s in shots}
CONTROLS = [
    ("S1-01", "S4-01"), ("S5-03", "S8-01"), ("S7-03", "S6-02"),
    ("S4-02", "S2-02"), ("S8-01", "S1-03"), ("S3-01", "S1-02"),
    ("S1-02", "S2-01"), ("S7-02", "S5-03"),
]

def spec_text(s):
    st = s.get('shot_type') or ''
    cm = s.get('camera_movement') or ''
    parts = [f"景别{st}，{cm}运镜，构图上主体{s.get('composition','居中')}"]
    em = s.get('emotion') or ''
    if em:
        parts.append(f"情绪{em}")
    return '，'.join(parts)

tasks = []
for s in shots:
    tasks.append(('pos', s['shot_id'], s['shot_id'], s))
for vid, sid in CONTROLS:
    tasks.append(('neg', vid, sid, spec_by_id[sid]))

FRAMES = ['f0.jpg', 'f2.jpg', 'f4.jpg']
THRESH = 0.7  # 与管线默认门限一致

async def run_case(tag, vid, sid, spec):
    q = spec_text(spec)
    votes = []
    per_frame = []
    for fn in FRAMES:
        img = FR / vid / fn
        if not img.exists():
            continue
        try:
            r = await vv.verify_image_content(
                str(img), q, None, None, engine_order=ORDER)
            votes.append(bool(r.get('pass')))
            per_frame.append({'frame': fn, 'pass': r.get('pass'),
                              'conf': round(float(r.get('confidence') or 0), 2),
                              'engine': r.get('engine')})
        except Exception as e:
            per_frame.append({'frame': fn, 'error': repr(e)[:80]})
    n_pass = sum(votes)
    if not votes:
        verdict = 'error'   # 引擎不可用≠判负,聚合时剔除
    else:
        verdict = 'pass' if n_pass * 2 > len(votes) else 'fail'
    return {'tag': tag, 'video': vid, 'spec_of': sid, 'verdict': verdict,
            'votes': f"{n_pass}/{len(votes)}", 'frames': per_frame}

async def main():
    log = open(OUT / f'parity_{ARM}.jsonl', 'w')
    for i, (tag, vid, sid, spec) in enumerate(tasks):
        rec = await run_case(tag, vid, sid, spec)
        log.write(json.dumps(rec, ensure_ascii=False) + '\n')
        log.flush()
        print(f"[{i+1}/{len(tasks)}] {tag} {vid}->{sid}: {rec['verdict']} ({rec['votes']})",
              flush=True)

asyncio.run(main())
print("PARITY_RUN_DONE")