#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""钟馗夜馒 V2 <-> V3 迭代对比页生成器（idempotent, 按 ab-compare-html 蓝图）"""
import json, os, glob, re, shutil, subprocess, datetime, html
BASE = "/data/workspace/kais-hermes-skills/skills/kais-movie-pipeline/episodes"
V = {"V2": os.path.join(BASE, "ep-zhongkui-v2-ep01"),
"V3": os.path.join(BASE, "ep-zhongkui-v3-ep01")}
OUT = "/home/kai/shared/2026-08-27/zhongkui-v2v3-comparison-tmp"
OUTPAGE = "/home/kai/shared/2026-08-27/zhongkui-v2v3-compare"
ISO = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}")
for d in (OUTPAGE,):
os.makedirs(os.path.join(OUTPAGE, "m"), exist_ok=True)
os.makedirs(OUT, exist_ok=True)
# ---------- state extraction ----------
def load(ep):
return json.load(open(os.path.join(V[ep], ".pipeline-state.json")))
def collect_ts(entry):
found = []
def walk(o):
if isinstance(o, dict):
for x in o.values(): walk(x)
elif isinstance(o, list):
for x in o: walk(x)
elif isinstance(o, str) and ISO.match(o): found.append(o)
walk(entry); return sorted(found)
def durations(ep):
s = load(ep); seq = []
for k, pv in s.get("phases", {}).items():
if isinstance(pv, dict):
tss = collect_ts(pv)
if tss: seq.append((datetime.datetime.fromisoformat(min(tss)), k))
seq.sort()
out, prev = {}, None
for t, k in seq:
out[k] = None if prev is None else round((t - prev).total_seconds() / 60, 1)
prev = t
return out
D2, D3 = durations("V2"), durations("V3")
PHASES = [p for p in D3.keys()]
allp = list(dict.fromkeys(list(D2.keys()) + list(D3.keys())))
STATE = {e: load(e) for e in V}
def anchor(ep, probe):
best = None
for pk, pv in STATE[ep].get("phases", {}).items():
if isinstance(pv, dict):
ms = re.findall(r'"%s":\s*([0-9.]+)' % probe, json.dumps(pv))
if ms: best = (pk, float(ms[-1]))
return best
A = {}
for probe in ("total_score", "fit_score", "consistency_score", "overall_score"):
A[probe] = {"V2": anchor("V2", probe), "V3": anchor("V3", probe)}
# ---------- media ----------
from PIL import Image
def thumb(src, dst, maxw=900, q=82):
try:
im = Image.open(src);
if im.mode != "RGB": im = im.convert("RGB")
w, h = im.size
if w > maxw: im = im.resize((maxw, int(h * maxw / w)))
im.save(dst, "JPEG", quality=q)
return True
except Exception:
return False
def fcp(a, b):
try: shutil.copy2(a, b); os.chmod(b, 0o644); return True
except Exception: return False
M = lambda name: "m/" + name
med_manifest = []
# masters
mast = {}
for e in V:
g = glob.glob(V[e] + "/assets/P13/master.mp4")
if g:
dst = f"{e.lower()}_master.mp4"
fcp(g[0], os.path.join(OUTPAGE, "m", dst))
r = subprocess.run(["ffprobe","-v","quiet","-show_entries","format=duration","-of","csv=p=0",g[0]],capture_output=True,text=True)
mast[e] = {"src": M(dst), "dur": float(r.stdout.strip() or 0)}
med_manifest.append(dst)
# per-shot pairing: V2 final-render channel vs V3 segment channel
v2shots = {os.path.basename(p): p for p in glob.glob(V["V2"] + "/assets/P11/S*_B*.mp4")}
v3shots = {os.path.basename(p): p for p in glob.glob(V["V3"] + "/assets/P11a5/segments/S*_B*.mp4")}
pairs = sorted(set(v2shots) & set(v3shots))
v2_only = sorted(set(v2shots) - set(v3shots))
v3_only = sorted(set(v3shots) - set(v2shots))
PAIRS = []
for sid in pairs:
d2 = f"v2_{sid}"; d3 = f"v3_{sid}"
ok1 = fcp(v2shots[sid], os.path.join(OUTPAGE,"m",d2))
ok2 = fcp(v3shots[sid], os.path.join(OUTPAGE,"m",d3))
if ok1 and ok2:
PAIRS.append({"sid": sid[:-4], "a": M(d2), "b": M(d3)})
med_manifest += [d2, d3]
UNIQ = []
for tag, dd, pfx in (("V2", v2_only, "v2u"), ("V3", v3_only, "v3u")):
srcmap = v2shots if tag == "V2" else v3shots
for sid in dd[:8]:
dst = f"{pfx}_{sid}"
if fcp(srcmap[sid], os.path.join(OUTPAGE,"m",dst)):
UNIQ.append({"tag": tag, "sid": sid[:-4], "src": M(dst)})
med_manifest.append(dst)
# scene images S07
def collect_scene(ep):
root = os.path.join(V[ep], "assets", "S07")
out = {}
for p in glob.glob(root + "/**/*.png", recursive=True) + glob.glob(root + "/**/*.jp*g", recursive=True):
rel = os.path.relpath(p, root)
key = rel.replace("\\", "/")
if key not in out: out[key] = p
return out
S2, S3 = collect_scene("V2"), collect_scene("V3")
scene_pairs, scene_miss, uniq_imgs = [], [], []
common_keys = sorted(set(S2) & set(S3))
for key in common_keys:
if key.endswith(".jpg") or os.path.getsize(S2[key]) < 30000: continue
dst = "v2_" + key.replace("/", "_"); dst2 = "v3_" + key.replace("/", "_")
if os.path.splitext(dst)[1] == ".png": dst = os.path.splitext(dst)[0]+".jpg"; dst2 = os.path.splitext(dst2)[0]+".jpg"
ok1 = thumb(S2[key], os.path.join(OUTPAGE,"m",dst)); ok2 = thumb(S3[key], os.path.join(OUTPAGE,"m",dst2))
if ok1 and ok2:
scene_pairs.append({"key": key, "a": M(dst), "b": M(dst2)})
med_manifest += [dst, dst2]
for key in sorted(set(S2)-set(S3)):
dst = "v2only_" + key.replace("/","_")
if thumb(S2[key], os.path.join(OUTPAGE,"m",dst)): uniq_imgs.append({"tag":"V2","key":key,"src":M(dst)})
for key in sorted(set(S3)-set(S2))[:12]:
dst = "v3only_" + key.replace("/","_")
if thumb(S3[key], os.path.join(OUTPAGE,"m",dst)): uniq_imgs.append({"tag":"V3","key":key,"src":M(dst)})
# turnarounds + storyboard
TA2 = {os.path.basename(p): p for p in glob.glob(V["V2"]+"/assets/turnaround_sheets/*.png")+glob.glob(V["V2"]+"/assets/turnaround_final/*.png")}
TA3 = {os.path.basename(p): p for p in glob.glob(V["V3"]+"/assets/turnaround_final/*.png")+glob.glob(V["V3"]+"/assets/Tier0/**/*.png", recursive=True)[:6]}
ta_pairs = []
for k in sorted(set(TA2)&set(TA3))[:4]:
a="tav2_"+k; b="tav3_"+k
if thumb(TA2[k],os.path.join(OUTPAGE,"m",a),700) and thumb(TA3[k],os.path.join(OUTPAGE,"m",b),700):
ta_pairs.append({"key":k,"a":M(a),"b":M(b)})
SB2 = glob.glob(V["V2"]+"/assets/storyboard_sheets/*.png")
SB3 = glob.glob(V["V3"]+"/assets/storyboard_sheets/*.png")
def csize(pat_map):
tot=0
for p in pat_map.values(): tot+=os.path.getsize(p)
return tot
vid_mb = sum(os.path.getsize(os.path.join(OUTPAGE,"m",f)) for f in med_manifest if f.endswith(".mp4"))/1e6
tot2 = sum(v for v in D2.values() if v); tot3 = sum(v for v in D3.values() if v)
NOW = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")
def fmt(v, unit="min"):
return "—" if v is None else (f"{v:.0f}{unit}" if unit=="min" else str(v))
rows_dur = []
mx = 1
for p in allp:
a,b = D2.get(p), D3.get(p)
mx = max(mx, a or 0, b or 0)
for p in allp:
a,b = D2.get(p), D3.get(p)
if a is None and b is None: continue
dlt = ""
if a is not None and b is not None:
d = b - a
cls = "up" if d < -1 else ("down" if d > 1 else "flat")
sign = "+" if d>0 else ""
dlt = f'
{sign}{d:.0f}m
' if abs(d)>=1 else '
~
'
else:
na = "新增" if a is None else ""
nb = "移除" if b is None else ""
dlt = f"
"
ca = f'{(a or 0)/mx*100:.0f}' ; cb = f'{(b or 0)/mx*100:.0f}'
note=""
if (a or 0)>240: note="含夜间挂机/人工等待窗"
wa = a if a is not None else "—"; wb = b if b is not None else "—"
rows_dur.append(f"""
{html.escape(p)}
{wa}
{wb}
{dlt}
{note}
""")
def anchor_row(name, probe, better_up=True, unit=""):
v2r, v3r = A[probe]["V2"], A[probe]["V3"]
a = f"{v2r[1]:g}{unit}" if v2r else "—"
b = f"{v3r[1]:g}{unit}" if v3r else "—"
cls = "flat"; arrow="→"
if v2r and v3r:
dv = v3r[1]-v2r[1]
good = (dv>0.005) if better_up else (dv<-0.005)
bad = (dv<-0.005) if better_up else (dv>0.005)
if abs(dv)<=0.005: cls="flat"; arrow=f"≈ ({dv:+.2f})"
elif good: cls="up"; arrow=f"↑ (+{dv:.2f})" if dv>0 else f"↑ ({dv:.2f})"
elif bad: cls="down"; arrow=f"↓ ({dv:+.2f})"
src2 = v2r[0] if v2r else ""; src3 = v3r[0] if v3r else ""
nt = (f"V2@{src2}" + (f" · V3@{src3}" if src3 and src3!=src2 else ""))
return f"
" for n,a,b in counts)
video_pair_html = ""
for pr in PAIRS:
video_pair_html += f"""
{pr['sid']}V2 成片
{pr['sid']}V3 终版
"""
uniq_vid_html = "".join(f"""
{u['tag']} 独有{u['sid']}
""" for u in UNIQ)
scene_group = {}
for sp in scene_pairs:
g = sp["key"].split("/")[0]
scene_group.setdefault(g, []).append(sp)
scene_html = ""
for g, items in sorted(scene_group.items()):
cells = []
for sp in items[:6]:
nm = os.path.splitext(os.path.basename(sp["key"]))[0]
cells.append(f"""
{nm}
V3
""")
scene_html += f"
🏞 {g}
{''.join(cells)}
"
uniq_img_html = "".join(f"{u['tag']}{u['key'][:40]}" for u in uniq_imgs[:16])
ta_html = "".join(f"""
V2{t['key']}
V3{t['key']}
""" for t in ta_pairs)
sb_thumbs = ""
for i,p in enumerate(SB2[:2]):
n=f"sb2_{i}.jpg"; thumb(p,os.path.join(OUTPAGE,"m",n),800); sb_thumbs+=f"
V2
"
for i,p in enumerate(SB3[:2]):
n=f"sb3_{i}.jpg"; thumb(p,os.path.join(OUTPAGE,"m",n),800); sb_thumbs+=f"
