xsl
4b39ae2e49
四泳道并行全流水线（资产→开发→测试→玩家迭代闭环），双双 R1-R2 连续达标裁定可交付： - cipher《暗码》223 断言：摩尔斯解码嵌叙事/四结局矩阵/暴露度逐NIGHT验算/三层兜底；R2 三修（链首标记音/预警震/结局计数句+4句补录）4/4 全满分档 - teahouse《深夜茶馆》205 断言：泡茶时机窗/经济逐晚对账/多晚弧线；连续 2 轮 4/4 愿意玩 P0=0 - keeper《守塔人》257 断言：时空签名教学-兑现/插叙触发回返/trust×truth 档位/4结局；R2 F1/F2 坐实（震型豁免/首播跳句锁）4/4 - bridge《星环舰长》336 断言：巡检节奏逐毫秒/危机×系统态矩阵/仲裁7场景/忠诚差分/评级验算；R1 当场修F1-F4、R2 F5 菜单顺延保护 4/4 上行 公共：index.html 四卡片+profiles.js 四桶+check_all_assets 登记（预接线）、00-program 八款进度账收账、 gen_voice_dual 四游戏段、README/HANDOVER 阵容更新（11 款）、voice_text.json 四游戏键、mureka 台账（累计$1.26/$2.00） 验证：四游戏 playtest 223/205/257/336 全绿 + guide 122/122 + 全项目 15 游戏资产终检绿 遗留：keeper 三代人声滤波人耳试听（最高优先，ASR 对滤波句失明）、bridge 音频 32MB 超软指引 2MB（压缩候选）
208 lines
7.2 KiB
Python
208 lines
7.2 KiB
Python
#!/usr/bin/env python3
|
||
"""《深夜茶馆》live 试玩驱动：真 Chromium + 真实混音 + SenseVoice ASR。
|
||
通过 control.mjs HTTP 接口逐步玩游戏，/listen 落判定日志。
|
||
用法: python3 tools/../research/new8/teahouse/live_drive.py [A|B]
|
||
"""
|
||
import json
|
||
import sys
|
||
import time
|
||
import urllib.request
|
||
|
||
BASE = "http://localhost:8157"
|
||
|
||
|
||
def post(path, body):
|
||
req = urllib.request.Request(BASE + path, data=json.dumps(body).encode(),
|
||
headers={"content-type": "application/json"}, method="POST")
|
||
with urllib.request.urlopen(req, timeout=30) as r:
|
||
return json.loads(r.read().decode())
|
||
|
||
|
||
def get(path):
|
||
with urllib.request.urlopen(BASE + path, timeout=30) as r:
|
||
return json.loads(r.read().decode())
|
||
|
||
|
||
LOG = []
|
||
|
||
|
||
def log(msg):
|
||
line = f"[{time.strftime('%H:%M:%S')}] {msg}"
|
||
print(line, flush=True)
|
||
LOG.append(line)
|
||
|
||
|
||
def gesture(g):
|
||
post("/gesture", {"type": g})
|
||
log(f"→ gesture {g}")
|
||
|
||
|
||
def drain_listen(tag):
|
||
try:
|
||
rows = get("/listen")
|
||
except Exception as e:
|
||
log(f" (listen err {e})")
|
||
return
|
||
for r in rows.get("rows", []) if isinstance(rows, dict) else rows:
|
||
v = r.get("verdict", "?")
|
||
hid = r.get("voiceId") or ""
|
||
sim = r.get("similarity")
|
||
sims = f" sim={sim:.2f}" if isinstance(sim, (int, float)) else ""
|
||
log(f" ♪ {r.get('t0', 0):.1f}-{r.get('t1', 0):.1f}s {hid} {v}{sims}"
|
||
+ (f" 「{str(r.get('heard', ''))[:40]}」" if v != "OK" else ""))
|
||
|
||
|
||
def sleep_listen(sec, tag=""):
|
||
t0 = time.time()
|
||
while time.time() - t0 < sec:
|
||
time.sleep(min(4, sec - (time.time() - t0)))
|
||
drain_listen(tag)
|
||
|
||
|
||
EVT_SINCE = [0]
|
||
|
||
|
||
def wait_evt(name, timeout=60, poll=0.8, ev=None, nth=1):
|
||
"""轮询 /events 直到 name 子串第 nth 次出现。
|
||
ev='vend' 锚「语音句播完」（等待态开放）；ev='start' 锚「sfx 起播」（时间线零点）。"""
|
||
seen = 0
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
time.sleep(poll)
|
||
try:
|
||
rows = get(f"/events?since={EVT_SINCE[0]}")
|
||
except Exception:
|
||
continue
|
||
evts = rows.get("events", []) if isinstance(rows, dict) else rows
|
||
if isinstance(rows, dict) and "next" in rows:
|
||
EVT_SINCE[0] = rows["next"]
|
||
for e in evts:
|
||
if e.get("ev") in ("fetch", "asset404", "console.error", "console.warn"):
|
||
continue # 只认真实播放/语音事件（prewarm fetch 会误锚）
|
||
if ev and e.get("ev") != ev:
|
||
continue
|
||
nm = str(e.get("name", "")) + str(e.get("voiceId", "")) + str(e.get("url", ""))
|
||
if name in nm:
|
||
seen += 1
|
||
if seen >= nth:
|
||
return True
|
||
log(f" (等待事件 {name}#{nth} 超时 {timeout}s)")
|
||
return False
|
||
|
||
|
||
def wait_any(conds, timeout=60, poll=0.8):
|
||
"""conds=[(ev, name), ...]：单窗口内任一条件命中即返回命中索引（-1=超时）。
|
||
游标只在本窗口结束时前进一次——前置等待不会吞掉后续锚点。"""
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
time.sleep(poll)
|
||
try:
|
||
rows = get(f"/events?since={EVT_SINCE[0]}")
|
||
except Exception:
|
||
continue
|
||
evts = rows.get("events", []) if isinstance(rows, dict) else rows
|
||
if isinstance(rows, dict) and "next" in rows:
|
||
EVT_SINCE[0] = rows["next"]
|
||
for e in evts:
|
||
if e.get("ev") in ("fetch", "asset404", "console.error", "console.warn"):
|
||
continue
|
||
nm = str(e.get("url", "")) + str(e.get("name", ""))
|
||
for i, (ev, name) in enumerate(conds):
|
||
if (not ev or e.get("ev") == ev) and name in nm:
|
||
return i
|
||
log(f" (等待事件 {conds} 超时 {timeout}s)")
|
||
return -1
|
||
|
||
|
||
def wait_cue(voice_id, timeout=60, poll=1.2):
|
||
"""轮询 /listen 直到目标 voiceId 被判定（OK/MISMATCH 均可，TRUNCATED 也算落声）。"""
|
||
t0 = time.time()
|
||
while time.time() - t0 < timeout:
|
||
time.sleep(poll)
|
||
rows = None
|
||
try:
|
||
rows = get("/listen")
|
||
except Exception:
|
||
continue
|
||
for r in (rows.get("rows", []) if isinstance(rows, dict) else rows):
|
||
v = r.get("verdict", "?")
|
||
hid = r.get("voiceId") or ""
|
||
sim = r.get("similarity")
|
||
sims = f" sim={sim:.2f}" if isinstance(sim, (int, float)) else ""
|
||
log(f" ♪ {r.get('t0', 0):.1f}-{r.get('t1', 0):.1f}s {hid} {v}{sims}"
|
||
+ (f" 「{str(r.get('heard', ''))[:40]}」" if v not in ("OK",) else ""))
|
||
if hid == voice_id:
|
||
return True
|
||
log(f" (等待 {voice_id} 超时 {timeout}s)")
|
||
return False
|
||
|
||
|
||
def session_a():
|
||
post("/session", {"game": "teahouse",
|
||
"settings": {"voice": "female", "speed": "normal", "bgm": "normal", "guide": "full"}})
|
||
EVT_SINCE[0] = 0
|
||
log("== R2 局A′：female/normal/full（F1/F3 教学新句验证） ==")
|
||
wait_evt("probe_ready", 150, poll=1.0)
|
||
gesture("tap")
|
||
wait_evt("bgm/menu", 60, poll=1.0)
|
||
time.sleep(1.5)
|
||
gesture("tap")
|
||
log("-- D0 --")
|
||
wait_evt("tut_d0", 120, ev="vend")
|
||
gesture("swiperight")
|
||
wait_evt("tut_tea", 60, ev="vend")
|
||
gesture("tap")
|
||
log("-- D2：F1 新 tut_water（对照口径） --")
|
||
wait_evt("water_pour", 40, ev="start")
|
||
sleep_listen(2.2)
|
||
gesture("doubletap")
|
||
r = wait_any([("vend", "water_ok"), ("vend", "water_early"), ("vend", "water_late")], 40)
|
||
log(f" 候汤归因档 = {['ok','early','late'][r] if r >= 0 else '??'}")
|
||
log("-- D3：F1 新 tut_steep --")
|
||
wait_evt("steep_rustle", 60, ev="start")
|
||
sleep_listen(4.5)
|
||
gesture("doubletap")
|
||
wait_evt("tut_serve", 60, ev="vend")
|
||
gesture("tap")
|
||
log("-- D5：F3 新 tut_status --")
|
||
wait_evt("tut_status", 90, ev="vend")
|
||
gesture("swipeup")
|
||
wait_evt("tut_done", 90, ev="vend")
|
||
log("-- F2 复验需新档；本局收尾 --")
|
||
drain_listen("end")
|
||
out = post("/stop", {})
|
||
log(f"== R2 局A′ 结束 == {json.dumps(out)[:200]}")
|
||
|
||
|
||
def session_b():
|
||
post("/session", {"game": "teahouse",
|
||
"settings": {"voice": "male", "speed": "normal", "bgm": "normal", "guide": "full"}})
|
||
EVT_SINCE[0] = 0
|
||
log("== R2 局B′：male/normal/full（F2 跳过兜底验证） ==")
|
||
wait_evt("probe_ready", 150, poll=1.0)
|
||
gesture("tap")
|
||
wait_evt("bgm/menu", 60, poll=1.0)
|
||
time.sleep(1.5)
|
||
gesture("tap")
|
||
wait_evt("tut_d0", 120, ev="vend")
|
||
gesture("swipedown") # 跳过教学
|
||
ok = wait_any([("vend", "tut_skip"), ("vend", "sys_help")], 40)
|
||
log(f" 跳过链: tut_skip={'已播' if ok >= 0 else '??'} sys_help={'已播(F2 ✓)' if ok == 1 else '未播'}")
|
||
drain_listen("end")
|
||
out = post("/stop", {})
|
||
log(f"== R2 局B′ 结束 == {json.dumps(out)[:200]}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
which = sys.argv[1] if len(sys.argv) > 1 else "A"
|
||
try:
|
||
session_a() if which == "A" else session_b()
|
||
except Exception as e:
|
||
log(f"!! 异常: {e}")
|
||
try:
|
||
post("/stop", {})
|
||
except Exception:
|
||
pass
|
||
with open(f"/tmp/tea_live_{which}.log.full", "w") as f:
|
||
f.write("\n".join(LOG))
|