feat: VR字幕默认MarginV改为700并统一历史字幕样式

- srt-to-dual-eye-ass: 默认顶部安全边距 margin_top 120→700(2026-09), 120落在画面最顶需抬头看,700为实测合适值、视线自然可读 - nodes/ass.py: 样式定义抽为单一事实来源(DEFAULT_MARGIN_TOP/颜色常量 +ass_header/style_row/dialogue_line),新生成字幕与历史统一脚本共用同一出口,以后调样式只改一处、两边永不漂移 - 新增scripts/unify_ass_style.py: 解析旧文件分辨率与全部Dialogue后经 ass_header/dialogue_line重建,把媒体库历史样式(底部an2实心白1200/ 半透明1020/顶部120)原地统一为an8顶部对齐+70%透明+MarginV=700; 默认dry-run预览,--apply写盘;非VR字幕自动跳过 - /mnt/fnOS/123 库401个.CN_dual_eye.ass已全部改写(0失败),269764条字幕事件无an2残留,二次运行幂等 - 测试: ass默认值断言120→700,新增invoke默认700用例;新增 tests/test_unify_ass_style.py(历史三世代收敛/与write_ass逐字节一致/ 幂等/CLI dry-run与apply)
fix: 批量进度只统计无字幕视频，SKIPPED 不计入 total/done
2026-09-06 18:02:31 +08:00 · 2026-09-06 17:25:01 +08:00
10 changed files with 723 additions and 88 deletions
@@ -61,8 +61,27 @@ vrsub/
 | `frame-extract` | `video_uri` | `frames_manifest`、`frame_count` | 按**帧间隔**抽帧（解析 fps → step=round(间隔秒×fps)，ffmpeg select 按帧号精确取帧，帧时间=帧号/fps 无累计偏差）并 crop 裁切字幕区域，参数：`interval_seconds`（默认 0.5）、`crop`（[x,y,w,h] 0~1，**默认画面底部 1/4** `[0,0.75,1,0.25]`——字幕很少出现在画面上半部分，2026-08 调整）。**帧文件必须按帧号数值排序读取**（`_sorted_frame_files`）：ffmpeg `%04d` 编号超过 9999 帧后扩为 5 位，字典序 `sorted()` 会把 5 位编号排在 4 位之前导致时间与图像错位（真实发生于 run_339ec7ee437f 的 14236 帧任务，回归测试见 `test_frame_files_read_order_matches_frame_number`） |
 | `subtitle-ocr` | `frames_manifest` | `srt_uri`、`count` | 自适应线程池并发逐帧调 vlm-ocr → 垃圾过滤（无文字帧）→ 相同字幕合并（记录最后可见帧）→ 组装 SRT，消失时间=最后可见帧+采样间隔（间隔从帧清单推导），参数：`min_chars`、`min_alnum_ratio`、`garbage_tokens`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold` |
 | `llm-filter` | `srt_uri` | `srt_uri`、`kept`、`removed` | 两级过滤：①**规则层**（不调 LLM）正则确定性删除——横线装饰、URL/邮箱/**裸网址域名**（含中文夹杂的注册地址）、**HTML/水印模式**（html code/标签/javascript 等）、overlay token（html/marketing 等）、单双 ASCII 字符；②**LLM 五类分类**（garbage/overlay/noise 删，repeat/dialogue 留，未识别回退保留）每条连同前后各 `context_size`（默认 10）条纯文本分批判断——**上下文净化**：喂给 LLM 的是**过滤后的字幕**，规则层确定性垃圾从上下文中剔除（原文不进 LLM），避免覆盖层垃圾污染场景判断误删真实对话（回归：run_011d01f19999 曾 190 条含 ≥4 汉字对话被误删）；**长文本保护**：≥`min_keep_len`（默认 12）时 noise 不构成删除依据——LLM 判定不稳定，长度是必要兜底（实测移除保护后新增误删 124 条真实长对话）。**限流自适应**：LLM 调用 429/5xx 指数退避重试（最多 3 次，1s/2s/4s），worker 捕获限流错误时调用线程池 `report_failure()` **内存中临时降低最大线程数并缩容**（连续无错误窗口后逐步回升），失败条目在收紧后的并发下**重试一轮**，二次仍失败才整体失败——20 并发一拥而上触发 429 时自动收敛到配额内而不打挂任务。**节点级断点存档**（2026-08）：每条判定成功立即追加 `filter_partial.jsonl`（`{"index","category"}`，多线程加锁串行化），失败/中断后重跑只重判未判定条目，已判定结果复用（与 OCR 存档同机制）。**按文本去重**（忽略空白/大小写，相同文本只调一次 LLM，上下文取首次出现）保证判定一致并省调用。参数：`context_size`、`min_keep_len`、`overlay_tokens`（JSON 数组）、`dedupe`（默认开）、`model`、`pool_min_workers`/`pool_max_workers`/`pool_window_seconds`/`pool_fast_threshold`/`pool_slow_threshold`。回归数据：testdata/ocr_srt_run_ac7f480a3ccb.srt（真实任务 1666 条 OCR 输出） |
-| `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数：`resolution`（如 `3840x1920`）、`margin_top`（顶部安全边距，默认 120）。左右眼各占左右半幅且水平相对位置一致（**A-1 零视差**：字幕固定在屏幕平面，不做景深偏移）；对齐 `an8` 顶部居中 + `MarginV=margin_top`（**B-1 顶部安全区**，避开画面中央人脸区，2026-09）；文字填充 `&HB3FFFFFF`（约 70% 透明）描边 `&H80000000`（半透明黑），降低遮挡感 |
+| `srt-to-dual-eye-ass` | `cn_srt_uri` | `ass_uri` | 参数：`resolution`（如 `3840x1920`）、`margin_top`（顶部安全边距，**默认 700**——2026-09 调整：120 落在画面最顶需抬头看，700 使字幕处于视线自然可读位置）。左右眼各占左右半幅且水平相对位置一致（**A-1 零视差**：字幕固定在屏幕平面，不做景深偏移）；对齐 `an8` 顶部居中 + `MarginV=margin_top`（**B-1 顶部安全区**，避开画面中央人脸区）；文字填充 `&HB3FFFFFF`（约 70% 透明）描边 `&H80000000`（半透明黑），降低遮挡感 |

+
+#### 字幕样式统一（2026-09 起）
+
+`nodes/ass.py` 顶部的 `DEFAULT_MARGIN_TOP=700` 与左右眼样式常量是**单一事实来源**：
+新生成的字幕（`write_ass`/`invoke`）与历史字幕统一脚本共用 `ass_header()`/
+`style_row()`/`dialogue_line()` 同一出口，两边永不漂移。历史媒体库里由旧版本
+批量生成的 `*.CN_dual_eye.ass` 混有多种旧样式（底部 an2 实心白 / 底部半透明 /
+顶部 120），用 `scripts/unify_ass_style.py` 统一原地改写为当前新样式：
+
+```bash
+# 先 dry-run 预览将改哪些文件（默认不改盘）
+uv run python scripts/unify_ass_style.py /mnt/fnOS/123
+# 确认无误后真正改写（原地，不产生 .bak）
+uv run python scripts/unify_ass_style.py /mnt/fnOS/123 --apply
+```
+
+脚本解析旧文件分辨率与全部 Dialogue 事件后经 `ass_header()`/`dialogue_line()`
+重建，输出与代码新产物**逐字节一致**；非 VR 字幕（无 LeftEye/RightEye 样式行）
+自动跳过。相关测试见 `tests/test_unify_ass_style.py`。
 ### 模型权重解析（本地优先）

 whisper 节点按以下顺序解析模型路径，默认避免从远端下载：
@@ -11,40 +11,103 @@ from pathlib import Path

 from wov_sdk.models import InvokeRequest, InvokeResponse

-def _ass_header(resolution: str, margin_top: int = 120) -> str:
-    """生成 ASS 文件头：脚本信息、左右眼样式和事件格式。
+# ---------------------------------------------------------------------------
+# 统一 ASS 样式常量（单一事实来源）
+# ---------------------------------------------------------------------------
+# 说明：新生成字幕（write_ass/invoke）与"历史字幕统一脚本"
+# （scripts/unify_ass_style.py）共用下面这套样式定义——要调整字幕样式
+# （位置/透明度/描边等）只改这里，两条输出路径保持一致，不会各自漂移。
+#
+# 2026-09 调整：默认顶部安全边距 DEFAULT_MARGIN_TOP 由 120 改为 700。
+# 旧值 120 顶部对齐时字幕贴近画面最顶端，VR 头盔里需抬头才看得到；
+# 700 为实测合适值，字幕落在更接近视线自然平视的高度。
+DEFAULT_MARGIN_TOP = 700

-    样式说明（B-1 顶部安全区 + 透明度）：
-    - 对齐 8（顶部居中，\an8），MarginV=margin_top：字幕渲染在画面顶部安全区，
-      避开人脸/近景主体常出现的画面中央偏下区域，降低遮挡；
-    - PrimaryColour &HB3FFFFFF：约 70% 透明文字填充，比原先 &H80FFFFFF（约 50%）
-      更通透，减少对场景的遮挡感；
-    - OutlineColour &H80000000：半透明黑描边（不再是实心纯黑），
-      在保留可读性的同时不产生生硬黑框；
-    - 左右眼水平边距一致（LeftEye/RightEye 各占左右半幅）⇒ 水平相对位置相同，
-      即零视差（A-1）：字幕被渲染在屏幕平面，不产生额外景深冲突。
-    """
-    width, height = resolution.lower().split("x", 1)
-    # 左眼样式占左半边，右眼样式占右半边，各留 50px 内边距。
-    left_margin = 50
-    right_margin = 50
-    return f"""[Script Info]
-Title: VR Dual-Eye Subtitle
-ScriptType: v4.00+
-Collisions: Normal
-PlayResX: {width}
-PlayResY: {height}
-WrapStyle: 1
-ScaledBorderAndShadow: yes
+# 左右眼样式行字段（列顺序与 ASS Style Format 一一对应）：
+# - PrimaryColour &HB3FFFFFF：约 70% 透明文字填充，弱化对画面的遮挡；
+# - OutlineColour &H80000000：半透明黑描边（取代早期实心纯黑），保留可读性
+#   又不产生生硬黑框；
+# - Alignment 8（\an8 顶部居中）：配合 MarginV 形成顶部安全区——避开画面
+#   中央人脸高发区，同时落在视线自然高度。
+_ASS_FONT = "Arial"
+_ASS_FONT_SIZE = 50
+_ASS_PRIMARY = "&HB3FFFFFF"
+_ASS_SECONDARY = "&H000000FF"
+_ASS_OUTLINE = "&H80000000"
+_ASS_BACK = "&H80000000"
+_ASS_SCALE_X = 50
+_ASS_SCALE_Y = 100
+_ASS_BORDER_STYLE = 1
+_ASS_OUTLINE_WIDTH = 4
+_ASS_SHADOW = 0
+_ASS_ALIGN = 8
+_ASS_ENCODING = 1
+_EYE_PAD = 50  # 左眼距屏幕左缘 / 右眼距右缘的水平内边距

-[V4+ Styles]
-Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding
-Style: LeftEye,Arial,50,&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,{left_margin},{int(width) // 2},{margin_top},1
-Style: RightEye,Arial,50,&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,{int(width) // 2},{right_margin},{margin_top},1
+# 样式/事件表头格式行（列顺序即上面注释的顺序，勿改动）。
+ASS_STYLE_FORMAT = (
+    "Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,"
+    "BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,"
+    "BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding"
+)
+ASS_EVENT_FORMAT = "Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text"

-[Events]
-Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text
-"""
+
+def style_row(eye: str, width: int, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
+    """生成单眼（LeftEye/RightEye）的完整 ASS 样式行。
+
+    左眼占左半幅（左缘留 _EYE_PAD 内边距、向中线收 50%），右眼占右半幅，
+    两眼的水平相对位置一致 → 零视差（A-1）：字幕渲染在屏幕平面，不产生
+    额外景深冲突。margin_top 即该眼样式的 MarginV（距画面上缘的安全边距）。"""
+    mid = width // 2
+    margin_l, margin_r = (_EYE_PAD, mid) if eye == "LeftEye" else (mid, _EYE_PAD)
+    return (
+        f"Style: {eye},{_ASS_FONT},{_ASS_FONT_SIZE},{_ASS_PRIMARY},{_ASS_SECONDARY},"
+        f"{_ASS_OUTLINE},{_ASS_BACK},0,0,0,0,{_ASS_SCALE_X},{_ASS_SCALE_Y},0,0,"
+        f"{_ASS_BORDER_STYLE},{_ASS_OUTLINE_WIDTH},{_ASS_SHADOW},{_ASS_ALIGN},"
+        f"{margin_l},{margin_r},{margin_top},{_ASS_ENCODING}"
+    )
+
+
+def ass_header(width: int, height: int, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
+    """生成标准 ASS 文件头：Script Info（含分辨率）+ 左右眼样式 + 事件格式行。
+
+    新生成字幕与历史字幕统一脚本共用此函数，保证两者输出样式完全一致。"""
+    return (
+        "[Script Info]\n"
+        "Title: VR Dual-Eye Subtitle\n"
+        "ScriptType: v4.00+\n"
+        "Collisions: Normal\n"
+        f"PlayResX: {width}\n"
+        f"PlayResY: {height}\n"
+        "WrapStyle: 1\n"
+        "ScaledBorderAndShadow: yes\n"
+        "\n"
+        "[V4+ Styles]\n"
+        f"{ASS_STYLE_FORMAT}\n"
+        f"{style_row('LeftEye', width, margin_top)}\n"
+        f"{style_row('RightEye', width, margin_top)}\n"
+        "\n"
+        "[Events]\n"
+        f"{ASS_EVENT_FORMAT}\n"
+    )
+
+
+def dialogue_line(style: str, start: str, end: str, text: str) -> str:
+    """生成一行标准 Dialogue 事件。
+
+    文本前缀固定 \\an8 顶部居中对齐（与样式 Alignment 一致），使每句字幕
+    都落到样式定义的顶部安全区位置。"""
+    return f"Dialogue: 0,{start},{end},{style},,0,0,0,,{{\\an8}}{text}"
+
+
+def _ass_header(resolution: str, margin_top: int = DEFAULT_MARGIN_TOP) -> str:
+    """兼容旧接口的 ASS 头生成：resolution 形如 "3840x1920"。
+
+    内部委托给 ass_header()（统一样式出口），仅负责把字符串分辨率解析为
+    整数宽高。"""
+    width, height = (int(part) for part in resolution.lower().split("x", 1))
+    return ass_header(width, height, margin_top=margin_top)


 def parse_srt(text: str) -> list[tuple[str, str, str]]:
@@ -79,17 +142,22 @@ def parse_srt(text: str) -> list[tuple[str, str, str]]:


 def write_ass(
-    entries: list[tuple[str, str, str]], output_path: Path, resolution: str, margin_top: int = 120
+    entries: list[tuple[str, str, str]],
+    output_path: Path,
+    resolution: str,
+    margin_top: int = DEFAULT_MARGIN_TOP,
 ) -> None:
    """把解析后的条目写入 ASS 文件，每个条目输出左右眼两行 Dialogue。

-    margin_top 控制字幕距画面顶部的安全边距（默认 120），顶部对齐（\an8）
-    使字幕整体落在顶部安全区。左右眼使用相同文本与水平相对位置（零视差，A-1）。"""
-    lines = [_ass_header(resolution, margin_top=margin_top)]
+    margin_top 控制字幕距画面顶部的安全边距（默认 700，2026-09 起），顶部对齐（\an8）
+    使字幕整体落在顶部安全区下方。左右眼使用相同文本与水平相对位置（零视差，A-1）。"""
+    width, height = (int(part) for part in resolution.lower().split("x", 1))
+    header = ass_header(width, height, margin_top=margin_top)
+    # an8 对齐到屏幕顶部，配合 MarginV 形成顶部安全区，避开中央人脸区域。
+    lines = [header.rstrip("\n")]
    for start, end, text in entries:
-        # an8 对齐到屏幕顶部，配合 MarginV 形成顶部安全区，避开中央人脸区域。
-        lines.append(f"Dialogue: 0,{start},{end},LeftEye,,0,0,0,,{{\\an8}}{text}")
-        lines.append(f"Dialogue: 0,{start},{end},RightEye,,0,0,0,,{{\\an8}}{text}")
+        lines.append(dialogue_line("LeftEye", start, end, text))
+        lines.append(dialogue_line("RightEye", start, end, text))
    output_path.write_text("\n".join(lines) + "\n", encoding="utf-8")

 def invoke(request: InvokeRequest) -> InvokeResponse:
@@ -108,8 +176,9 @@ def invoke(request: InvokeRequest) -> InvokeResponse:
    output_path = output_dir / "dual_eye.ass"
    # 分辨率默认 3840x1920，覆盖常见 VR 视频尺寸。
    resolution = str(request.params.get("resolution", "3840x1920"))
-    # margin_top 可选：顶部安全边距，不同分辨率/内容可用工作流参数微调。
-    margin_top = int(request.params.get("margin_top", 120))
+    # margin_top 可选：顶部安全边距（默认 700，2026-09 起），不同分辨率/内容
+    # 仍可用工作流参数微调（历史 120 已过时，勿再使用）。
+    margin_top = int(request.params.get("margin_top", 700))
    write_ass(entries, output_path, resolution, margin_top=margin_top)
    return InvokeResponse(status="completed", outputs={"ass_uri": str(output_path)})

@@ -0,0 +1,227 @@
+"""统一历史 VR 双目字幕（*.CN_dual_eye.ass）的样式与位置。
+
+背景
+----
+`nodes/ass.py` 生成的字幕样式在迭代中多次变化，媒体库（如 /mnt/fnOS/123）
+里由旧版本批量生成的 *.CN_dual_eye.ass 因此混有多种历史样式：
+
+- 最早期：底部对齐（an2）+ 实心白字（&H00FFFFFF），MarginV≈1200；
+- 过渡期：底部对齐 + 半透明白（&H80FFFFFF），MarginV≈1020；
+- 2026-09 新默认：顶部对齐（an8）+ 70% 透明（&HB3FFFFFF）+ MarginV=120；
+- 用户实测合适值：顶部对齐 + 70% 透明 + MarginV=700。
+
+本脚本把库内所有目标字幕**原地改写**为当前 nodes/ass.py 的统一样式
+（an8 顶部对齐 + 70% 透明 + DEFAULT_MARGIN_TOP=700，单点定义见
+nodes/ass.py 顶部常量），保证"之后生成的字幕"与"历史字幕"外观一致。
+
+实现要点
+--------
+- 不对手写字符串做手术，而是解析旧文件的 [Script Info] 分辨率与全部
+  Dialogue 事件后，用 nodes/ass.py 的 ass_header()/dialogue_line()
+  重新拼装——与代码新生成产物共用同一出口，两边永不漂移；以后要调样式
+  只需改 nodes/ass.py 一处常量。
+- 事件文本原样保留（含原有 \\N 多行分隔与事件内其它标签），只规整前导
+  \\an 对齐标签：历史 an2（底部）/an8（顶部）一律统一为 an8。
+- 仅改写样式声明与事件对齐标签，不触碰时间轴、事件文本与左右眼对称性。
+- 文件最终以 UTF-8 写回、行尾 \\n、末尾一个换行——与 write_ass() 一致。
+
+用法（在 vrsub 根目录执行）
+----------------------------
+    # 预览（默认 dry-run，不改盘）
+    uv run python scripts/unify_ass_style.py /mnt/fnOS/123
+
+    # 真正原地改写
+    uv run python scripts/unify_ass_style.py /mnt/fnOS/123 --apply
+
+    # 指定递归匹配模式 / 只处理某旧 MarginV
+    uv run python scripts/unify_ass_style.py /mnt/fnOS/123 --pattern "*.ass"
+    uv run python scripts/unify_ass_style.py /mnt/fnOS/123 --only-margin 1200
+
+安全说明
+--------
+- 默认 --dry-run 只打印不写入；--apply 才真正原地改写。
+- 改写前脚本先打印库内旧样式分布，便于核对范围。
+- 不产生 .bak 副本（用户已确认直接原地改写）；如需回滚请先自行备份。
+"""
+
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from collections import Counter
+from pathlib import Path
+
+# 复用节点单一事实来源：统一脚本必须与"新生成字幕"完全一致。
+from nodes.ass import (  # noqa: E402
+    DEFAULT_MARGIN_TOP,
+    ass_header,
+    dialogue_line,
+)
+
+_DIALOGUE_RE = re.compile(r"^Dialogue: ")
+_AN_TAG_RE = re.compile(r"\{\\an[0-9]\}")
+
+
+def parse_resolution(text: str) -> tuple[int, int] | None:
+    """从 [Script Info] 解析 PlayResX/PlayResY；解析失败返回 None。"""
+    width = height = None
+    for line in text.splitlines():
+        m = re.match(r"^\s*PlayResX:\s*(\d+)", line)
+        if m:
+            width = int(m.group(1))
+            continue
+        m = re.match(r"^\s*PlayResY:\s*(\d+)", line)
+        if m:
+            height = int(m.group(1))
+    if width is None or height is None:
+        return None
+    return width, height
+
+
+def inspect_style(text: str) -> tuple[str, str] | None:
+    """提取旧样式 (PrimaryColour, MarginV) 用于统计/过滤。
+
+    只有存在 "Style: LeftEye" 行才视为 VR 双眼字幕产物；普通 ASS 字幕
+    （无左右眼样式）跳过，避免误伤。"""
+    for line in text.splitlines():
+        if line.startswith("Style: LeftEye"):
+            fields = line.split(",")
+            # 字段：Name0,Fontname1,Fontsize2,Primary3,...,Alignment18,MarginL19,MarginR20,MarginV21
+            primary = fields[3].strip() if len(fields) > 3 else "?"
+            margin_v = fields[21].strip() if len(fields) > 21 else "?"
+            return primary, margin_v
+    return None
+
+
+def parse_dialogues(text: str) -> list[tuple[str, str, str, str]]:
+    """抽取全部 Dialogue 事件为 (开始, 结束, 样式, 文本)。
+
+    历史文件每条可能带前导 {\an2}/{\an8} 对齐标签；这里剥掉（统一由
+    dialogue_line 补 an8），文本其余内容原样保留。"""
+    entries: list[tuple[str, str, str, str]] = []
+    for line in text.splitlines():
+        if not _DIALOGUE_RE.match(line):
+            continue
+        # Dialogue: 0,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text
+        parts = line.split(",", 9)
+        if len(parts) < 10:
+            continue
+        text_field = _AN_TAG_RE.sub("", parts[9], count=1)
+        entries.append((parts[1], parts[2], parts[3], text_field))
+    return entries
+
+
+def rewrite_content(original: str) -> str | None:
+    """把整份 ASS 文本重建为统一样式；已符合目标样式时返回 None。
+
+    返回的新文本 = ass_header()（新样式行，MarginV=DEFAULT_MARGIN_TOP）
+    + dialogue_line() 重建的全部事件，即与 write_ass() 产物逐字节一致
+    （除事件文本本身）。"""
+    res = parse_resolution(original)
+    if res is None:
+        return None  # 无 PlayResX/Y，非本工具产物
+    width, height = res
+    if inspect_style(original) is None:
+        return None  # 无 LeftEye 样式行，非 VR 双目字幕
+
+    lines = [ass_header(width, height, margin_top=DEFAULT_MARGIN_TOP).rstrip("\n")]
+    for start, end, style, text in parse_dialogues(original):
+        lines.append(dialogue_line(style, start, end, text))
+    rewritten = "\n".join(lines) + "\n"
+    return None if rewritten == original else rewritten
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="统一历史 VR 双目字幕样式与位置")
+    parser.add_argument("root", help="要扫描的媒体库根目录（递归）")
+    parser.add_argument(
+        "--pattern",
+        default="*.CN_dual_eye.ass",
+        help="递归匹配的文件名模式（默认 *.CN_dual_eye.ass）",
+    )
+    parser.add_argument(
+        "--apply",
+        action="store_true",
+        help="真正原地改写文件；不带此参数等价于 dry-run 预览",
+    )
+    parser.add_argument(
+        "--only-margin",
+        type=int,
+        default=None,
+        help="只处理旧样式 MarginV 等于该值的文件（如 1200），其余跳过",
+    )
+    args = parser.parse_args()
+
+    root = Path(args.root)
+    if not root.is_dir():
+        print(f"错误：目录不存在或不可读: {root}", file=sys.stderr)
+        return 2
+
+    files = sorted(root.rglob(args.pattern))
+    if not files:
+        print(f"未找到匹配 {args.pattern} 的文件")
+        return 0
+
+    # 第一遍统计：库内旧样式分布 + 圈定目标文件。
+    distribution: Counter[tuple[str, str]] = Counter()
+    unreadable = 0
+    target_files: list[Path] = []
+    for p in files:
+        try:
+            text = p.read_text(encoding="utf-8")
+        except (UnicodeDecodeError, OSError):
+            unreadable += 1
+            continue
+        styled = inspect_style(text)
+        if styled is None:
+            unreadable += 1
+            continue
+        primary, margin = styled
+        try:
+            margin_i = int(margin)
+        except ValueError:
+            margin_i = -1
+        if args.only_margin is not None and margin_i != args.only_margin:
+            continue
+        distribution[(primary, margin)] += 1
+        target_files.append(p)
+
+    print(f"扫描 {root}：共 {len(files)} 个匹配文件，可识别 VR 字幕 {len(target_files)}，跳过 {unreadable}")
+    print("旧样式分布（PrimaryColour, MarginV）：")
+    for (primary, margin), count in distribution.most_common():
+        print(f"  {count:4d}  {primary}  MarginV={margin}")
+    if not target_files:
+        print("没有需要统一的文件。")
+        return 0
+
+    print(f"\n目标样式：an8 顶部对齐 + 70%透明(&HB3FFFFFF) + MarginV={DEFAULT_MARGIN_TOP}（与代码新生成一致）")
+
+    changed = failed = 0
+    for p in target_files:
+        try:
+            original = p.read_text(encoding="utf-8")
+            rewritten = rewrite_content(original)
+        except Exception as exc:  # 单文件失败不中断整体
+            print(f"  [错误] {p}: {exc}")
+            failed += 1
+            continue
+        if rewritten is None:
+            continue  # 已是目标样式，跳过
+        changed += 1
+        marker = "改写" if args.apply else "将改写"
+        if args.apply:
+            p.write_text(rewritten, encoding="utf-8")
+        # 摘要：旧样式 -> 新样式。
+        styled = inspect_style(original)
+        old_desc = f"{styled[0]}@{styled[1]}" if styled else "?"
+        print(f"  [{marker}] {p.relative_to(root)}  ({old_desc} -> &HB3FFFFFF@{DEFAULT_MARGIN_TOP})")
+
+    print(f"\n完成：{changed} 个文件{'已' if args.apply else '待'}统一，{failed} 个失败跳过，其余已是最新样式。")
+    if not args.apply:
+        print("（以上为预览，未改动任何文件；确认无误后加 --apply 执行真正改写）")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -239,6 +239,16 @@ def create_job(
            "created_at": now,
            "updated_at": now,
        })
+    # total = 本批真正需要处理（无字幕）的视频数；已有字幕被 SKIPPED 的
+    # 不计入总数也不计入完成数——进度条只反映"实际待处理"的这批。
+    if pending == 0:
+        # 整批都已有字幕、无任何可处理项：直接视为完成，不排队。
+        db.update_batch_job(
+            job_id, status="COMPLETED", total=0, progress=1.0,
+            updated_at=_now_iso(),
+        )
+    else:
+        db.update_batch_job(job_id, total=pending, updated_at=_now_iso())
    logger.info(
        "创建批量任务 %s: 文件夹 %s, 工作流 %s, 共 %d 个视频（%d 待处理, %d 已有字幕跳过）",
        job_id, folder, workflow_id, len(videos), pending, skipped,
@@ -340,27 +350,30 @@ class BatchWorker:
        definition.validate()

        items = self.db.list_batch_videos(job_id)
-        total = len(items)
+        # total 在创建任务时已固定为"无字幕需处理的视频数"，这里不覆盖；
+        # 老任务（历史口径 total=全部视频数）由 sync_batch_job_progress 在读取
+        # 时自我修正为不含 SKIPPED 的口径。
        self.db.update_batch_job(
-            job_id, status="RUNNING", total=total, progress=0,
+            job_id, status="RUNNING", progress=0,
            current_video=None, error=None, updated_at=_now_iso(),
        )
+        # total 用于进度条分母；无字幕项为 0 时表示整批跳过（创建即 COMPLETED，
+        # 正常不会进入本循环）。
+        total = int(job["total"] or 0)

-        for index, item in enumerate(items):
+        for item in items:
            # 暂停检查：批量任务被暂停后停止处理后续视频，等待用户继续。
            current = self.db.get_batch_job(job_id)
            if current is None or current["status"] == "PAUSED":
-                # 停下前把已完成的视频实时入账：任务可能已处理多个视频才被暂停，
-                # 若不在暂停边界同步，前端会一直看到 0/总数 0%（回归 batch_fee668175444）。
+                # 停下前把已完成/失败的视频实时入账，让暂停中的前端也能看到
+                # 真实进度（回归 batch_fee668175444）。
                self.db.sync_batch_job_progress(job_id)
                logger.info("批量任务 %s 已暂停，停止在视频 %s", job_id, item["video_path"])
                return

-            # 已完成/已跳过的视频不再处理（跳过决策在创建任务时已定）。
+            # 已完成/已跳过的视频不再处理：COMPLETED 由断点续跑逻辑跳过，
+            # SKIPPED 在创建任务时已定（不参与 total/done，故无需同步进度）。
            if item["status"] in ("COMPLETED", "SKIPPED"):
-                # 已完成的视频同样是任务进度的一部分：continue 前实时同步汇总，
-                # 避免长任务（大量 SKIPPED）中途汇总停留在 0。
-                self.db.sync_batch_job_progress(job_id)
                continue

            video = Path(item["video_path"])
@@ -371,9 +384,7 @@ class BatchWorker:

            work_dir = Path(item["work_dir"])
            self.db.update_batch_job(
-                job_id, current_video=str(video),
-                progress=index / total if total else 0,
-                updated_at=_now_iso(),
+                job_id, current_video=str(video), updated_at=_now_iso(),
            )
            try:
                self._process_video(job, item, version, definition, work_dir)
@@ -395,7 +406,8 @@ class BatchWorker:
                self.db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())
                return

-        # 全部视频处理完成：先用明细实时对齐汇总（含已跳过），再置 COMPLETED。
+        # 全部视频处理完成：先用明细实时对齐汇总（done 只计实际完成的，
+        # 不含 SKIPPED），再置 COMPLETED。
        self.db.sync_batch_job_progress(job_id)
        job = self.db.get_batch_job(job_id)
        done = int(job["done"]) if job else 0
@@ -405,7 +417,7 @@ class BatchWorker:
            current_video=None, error=None, updated_at=_now_iso(),
        )
        logger.info(
-            "批量任务 %s 完成: 共 %d 个视频, 完成/跳过 %d, 失败 %d",
+            "批量任务 %s 完成: 待处理 %d 个视频, 完成 %d, 失败 %d",
            job_id, total, done, failed,
        )

@@ -559,21 +559,25 @@ class Database:


    def sync_batch_job_progress(self, job_id: str) -> None:
-        """按视频明细实时重算任务的 done/failed 汇总并落库。
+        """按视频明细实时对齐任务的 total/done/failed 汇总并落库。
+
+        统计口径（2026-09 用户确认）：total = 本批**无字幕、需要处理**的视频数
+        （= 明细里非 SKIPPED 的数量，创建时已固定，运行中 SKIPPED 不会变化）；
+        done = 实际**处理完成**的视频数（仅 COMPLETED，SKIPPED 不计）；
+        failed = 处理失败的视频数。已有字幕直接跳过的视频不参与 total/done。

-        明细里 COMPLETED 与 SKIPPED 都计入 done（它们都不再需要处理，是任务
-        已完成的工作量）；FAILED 计入 failed；PAUSED/PENDING 等状态不计。
        引擎在暂停、视频间检查、收尾等边界调用，router 在读取前也调用，保证
-        前端看到的进度始终与明细一致——即使任务被暂停或进程被终止，汇总字段
-        也不会停留在创建时的 0（回归 batch_fee668175444：处理了 15 个仍显示
+        前端看到的进度始终与明细一致——即使任务被暂停或进程被终止（回归
+        batch_fee668175444：整批 431 个含 383 个已有字幕，应显示 15/48 而非
        0/431）。任务不存在时静默返回。
        """
        with self._connect() as conn:
            counts = conn.execute(
                """
                SELECT
-                    SUM(CASE WHEN status IN ('COMPLETED', 'SKIPPED') THEN 1 ELSE 0 END) AS done,
-                    SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed
+                    SUM(CASE WHEN status = 'COMPLETED' THEN 1 ELSE 0 END) AS done,
+                    SUM(CASE WHEN status = 'FAILED' THEN 1 ELSE 0 END) AS failed,
+                    SUM(CASE WHEN status != 'SKIPPED' THEN 1 ELSE 0 END) AS total
                FROM batch_videos WHERE job_id = ?
                """,
                (job_id,),
@@ -582,16 +586,16 @@ class Database:
                # 任务不存在或没有任何明细：无需更新。
                return
            conn.execute(
-                "UPDATE batch_jobs SET done = ?, failed = ?, updated_at = ? WHERE id = ?",
-                (int(counts["done"]), int(counts["failed"]), _now_iso(), job_id),
+                "UPDATE batch_jobs SET done = ?, failed = ?, total = ?, updated_at = ? WHERE id = ?",
+                (int(counts["done"]), int(counts["failed"]), int(counts["total"]), _now_iso(), job_id),
            )

    def refresh_batch_job(self, job_id: str) -> dict[str, Any] | None:
        """实时对齐任务汇总后返回最新记录（读取侧统一入口）。

-        先 sync_batch_job_progress 让 done/failed 与明细一致，再返回最新 job；
-        任务不存在返回 None。批量引擎与 router 共用此入口，保证各处看到的
-        进度数字一致。
+        先 sync_batch_job_progress 让 total/done/failed 与明细一致，再返回
+        最新 job；任务不存在返回 None。批量引擎与 router 共用此入口，保证
+        各处看到的进度数字一致。
        """
        self.sync_batch_job_progress(job_id)
        return self.get_batch_job(job_id)
@@ -81,8 +81,10 @@ def create_batch_job(
 def list_batch_jobs(db: Database = Depends(_get_db)) -> list[dict]:
    """返回最近的批量任务列表（不含视频明细，明细按需单独查询）。

-    返回前对每个任务实时对齐 done/failed：任务被暂停或引擎不在运行时，汇总
-    字段也能与明细一致，前端列表的进度数字不会停留在 0（回归 batch_fee668175444）。
+    返回前对每个任务实时对齐 total/done/failed：任务被暂停或引擎不在运行时，
+    汇总字段也能与明细一致，前端列表的进度数字不会停留在 0
+    （回归 batch_fee668175444：431 个含 383 个已有字幕，应显示实际处理进度
+    而非 0/431）。
    """
    jobs = db.list_batch_jobs()
    for job in jobs:
@@ -318,14 +318,19 @@ def test_create_job_skips_videos_with_existing_subtitles(tmp_path) -> None:


 def test_create_job_all_videos_skipped_still_created(tmp_path) -> None:
-    """文件夹里全部视频都已有字幕时任务仍可创建（全部 SKIPPED，不再处理）。"""
+    """文件夹里全部视频都已有字幕时任务直接视为完成（无字幕视频数为 0）。
+
+    新口径（2026-09）：total = 扫描出的无字幕视频数，全被 SKIPPED 时
+    total=0 且没有可处理项，创建即置 COMPLETED，不排队空跑。
+    """
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    (folder / "a.CN.srt").write_text("x", encoding="utf-8")
    (folder / "b.CN_dual_eye.ass").write_text("x", encoding="utf-8")
    job = _make_job(db, folder)
-    assert job["status"] == "QUEUED"
+    assert job["status"] == "COMPLETED"
+    assert job["total"] == 0 and job["done"] == 0
    assert all(v["status"] == "SKIPPED" for v in db.list_batch_videos(job["id"]))


@@ -364,31 +369,39 @@ def test_batch_worker_processes_all_videos_and_cleans_up(tmp_path) -> None:


 def test_batch_worker_skips_videos_with_existing_subtitles(tmp_path) -> None:
-    """已有字幕的视频在创建时记 SKIPPED，引擎运行时不再为它触发流水线。"""
+    """已有字幕的视频在创建时记 SKIPPED，不进入 total/done，只统计真正处理的。
+
+    新口径（2026-09）：total=无字幕视频数（不含 SKIPPED），done=实际完成数。
+    """
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    (folder / "a.CN_dual_eye.ass").write_text("已处理", encoding="utf-8")
    job = _make_job(db, folder)
+    # 创建时 total 已固定为无字幕数（b 一个），SKIPPED 不计入。
+    assert db.get_batch_job(job["id"])["total"] == 1
    BatchWorker(db, interval_seconds=0.05)._process_job(job)
    videos = {Path(v["video_path"]).name: v for v in db.list_batch_videos(job["id"])}
    assert videos["a.mp4"]["status"] == "SKIPPED"
    assert videos["a.mp4"]["run_id"] is None
    assert videos["b.mp4"]["status"] == "COMPLETED"
-    assert db.get_batch_job(job["id"])["done"] == 2
+    job = db.get_batch_job(job["id"])
+    # done 只计实际完成的 b（1 个），不再把 SKIPPED 的 a 计入。
+    assert job["done"] == 1 and job["total"] == 1


 def test_batch_worker_all_skipped_job_completes(tmp_path) -> None:
-    """任务里全部视频都是 SKIPPED 时引擎正常完成，不创建任何 run。"""
+    """全部视频都有字幕时创建即视为完成（total=0），不创建任何 run。"""
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    (folder / "a.srt").write_text("x", encoding="utf-8")
    (folder / "b.CN.srt").write_text("x", encoding="utf-8")
    job = _make_job(db, folder)
-    BatchWorker(db, interval_seconds=0.05)._process_job(job)
+    # 新口径：无字幕视频数为 0 → 创建即 COMPLETED，引擎不需要再跑。
    assert db.get_batch_job(job["id"])["status"] == "COMPLETED"
-    assert db.get_batch_job(job["id"])["done"] == 2
+    assert db.get_batch_job(job["id"])["total"] == 0
+    assert db.get_batch_job(job["id"])["done"] == 0
    assert db.list_runs() == []


@@ -917,7 +930,10 @@ def test_batch_progress_sync_recounts_done_failed(tmp_path) -> None:


 def test_batch_progress_sync_counts_skipped(tmp_path) -> None:
-    """SKIPPED 视频同样计入 done（它们不需要处理，属于已完成的工作量）。"""
+    """SKIPPED 视频不计入 total 也不计入 done（它不需要本批处理）。
+
+    新口径（2026-09）：进度只反映"本批无字幕待处理"的视频。
+    """
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4",))
@@ -925,7 +941,9 @@ def test_batch_progress_sync_counts_skipped(tmp_path) -> None:
    # 把 a 改为 SKIPPED（等同创建时旁挂字幕被跳过的语义）。
    db.update_batch_video(db.list_batch_videos(job["id"])[0]["id"], status="SKIPPED", updated_at=_now_iso())
    db.sync_batch_job_progress(job["id"])
-    assert db.get_batch_job(job["id"])["done"] == 1
+    job = db.get_batch_job(job["id"])
+    assert job["done"] == 0
+    assert job["total"] == 0


 def test_batch_progress_sync_returns_refreshed_job(tmp_path) -> None:
@@ -979,22 +997,29 @@ def test_batch_worker_paused_job_reports_real_done(tmp_path, monkeypatch) -> Non


 def test_batch_worker_skipped_continue_syncs_done(tmp_path) -> None:
-    """循环里遇到 SKIPPED/COMPLETED 的 continue 分支也会把汇总实时对齐。"""
+    """循环里遇到 SKIPPED/COMPLETED 的 continue 分支不把它们计入 done。
+
+    新口径：total=无字幕视频数（本例 a 有字幕 SKIPPED → total=1 即 b），
+    done 只计实际完成（b=COMPLETED → done=1）。SKIPPED 不占分母也不占分子。
+    """
    db = _db(tmp_path)
    _seed_echo_workflow(db)
    folder = _video_folder(tmp_path, names=("a.mp4", "b.mp4"))
    (folder / "a.CN.srt").write_text("x", encoding="utf-8")  # a 创建即 SKIPPED
    job = _make_job(db, folder)
    job_id = job["id"]
-    # b 也手工置为 COMPLETED（引擎会因 continue 分支跳过它）。
+    # 创建即固定：total=无字幕数（仅 b=1）。
+    assert db.get_batch_job(job_id)["total"] == 1
+    # b 手工置为 COMPLETED（模拟此前已完成）；a 在循环里走 continue 分支。
    db.update_batch_video(
        [v for v in db.list_batch_videos(job_id) if v["video_path"].endswith("b.mp4")][0]["id"],
        status="COMPLETED", updated_at=_now_iso(),
    )
-    # 任务先置 PAUSED：首轮循环在 a（SKIPPED）即停下，若 continue 前未对齐则 done=0。
    db.update_batch_job(job_id, status="PAUSED", updated_at=_now_iso())
    BatchWorker(db, interval_seconds=0.05)._run_job(job_id)
-    assert db.get_batch_job(job_id)["done"] == 2
+    job = db.get_batch_job(job_id)
+    # done 只计实际完成的 b（1），SKIPPED 的 a 不计；total 保持 1。
+    assert job["done"] == 1 and job["total"] == 1

 def test_batch_worker_ghost_job_id_returns(tmp_path) -> None:
    """_run_job 在任务不存在时直接返回（幽灵任务处理无副作用）。"""
@@ -1059,6 +1084,8 @@ def test_batch_api_create_list_detail(tmp_path) -> None:
        detail = client.get(f"/api/batch/jobs/{job['id']}").json()
        assert detail["workflow_id"] == "echo-app"
        assert len(detail["videos"]) == 2
+        # 新口径：total = 无字幕需处理数（b 一个），SKIPPED 的 a 不占分母。
+        assert detail["total"] == 1 and detail["done"] == 0

        missing = client.get("/api/batch/jobs/ghost")
        assert missing.status_code == 404
@@ -1066,6 +1093,24 @@ def test_batch_api_create_list_detail(tmp_path) -> None:
        client.__exit__(None, None, None)


+
+
+def test_batch_api_all_skipped_created_completed(tmp_path) -> None:
+    """全部视频已有字幕时，创建响应直接为 COMPLETED，total=0。"""
+    client, folder = _client_with_echo_workflow(tmp_path)
+    try:
+        # 两个视频都已有旁挂字幕 → 无任何无字幕项。
+        (folder / "a.CN.srt").write_text("x", encoding="utf-8")
+        (folder / "b.CN_dual_eye.ass").write_text("x", encoding="utf-8")
+        job = client.post(
+            "/api/batch/jobs",
+            json={"folder": str(folder), "workflow_id": "echo-app"},
+        ).json()
+        assert job["status"] == "COMPLETED"
+        assert job["total"] == 0 and job["done"] == 0
+        assert all(v["status"] == "SKIPPED" for v in job["videos"])
+    finally:
+        client.__exit__(None, None, None)
 def test_batch_api_creation_errors(tmp_path) -> None:
    """批量 API 拒绝：文件夹不存在、无视频、未发布工作流。"""
    client, folder = _client_with_echo_workflow(tmp_path)
@@ -678,20 +678,21 @@ def test_ass_parse_and_write(tmp_path) -> None:
 def test_ass_top_aligned_and_translucent(tmp_path) -> None:
    """验证字幕默认渲染到顶部安全区且文字/描边带透明度。

-    B-1：对齐 an8（顶部居中），MarginV 取顶部安全边距默认 120；
+    B-1：对齐 an8（顶部居中），MarginV 取顶部安全边距默认 700（2026-09 起，
+    字幕置于视线自然可读位置；旧默认 120 落在画面最顶需抬头观看）；
    透明度：文字填充 &HB3FFFFFF（约 70% 透明），描边 &H80000000（半透明黑），
    而左右眼水平相对位置一致保持零视差（A-1，字幕在屏幕平面）。"""
    entries = parse_srt(SAMPLE_SRT)
    output = tmp_path / "out.ass"
    write_ass(entries, output, "3840x1920")
    content = output.read_text(encoding="utf-8")
-    # 顶部对齐 an8、顶部安全边距默认 120。
+    # 顶部对齐 an8、顶部安全边距默认 700。
    assert r"{\an8}" in content
    assert r"{\an2}" not in content
-    # 样式行：MarginV=120（顶部安全区），填充 &HB3FFFFFF，描边 &H80000000。
+    # 样式行：MarginV=700（顶部安全区），填充 &HB3FFFFFF，描边 &H80000000。
    assert "&HB3FFFFFF" in content
    assert "&H80000000" in content
-    assert ",0,0,0,0,50,100,0,0,1,4,0,8,50,1920,120,1" in content
+    assert ",0,0,0,0,50,100,0,0,1,4,0,8,50,1920,700,1" in content
    # 左右眼两行文本一致（水平相对位置相同 → 零视差/A-1）。
    assert content.count(r"第一行\N第二行") == 2

@@ -725,6 +726,28 @@ def test_ass_invoke_reads_margin_top(tmp_path) -> None:
    assert ",0,0,0,0,50,100,0,0,1,4,0,8,50,960,90,1" in content


+def test_ass_invoke_default_margin_top_is_700(tmp_path) -> None:
+    """验证 invoke 未显式传 margin_top 时默认生成 MarginV=700 的样式。
+
+    保证之后生成的字幕默认都落在顶部安全区下方（2026-09 起），
+    不再回退到旧的 120（需抬头观看）。"""
+    source = tmp_path / "in.srt"
+    source.write_text(SAMPLE_SRT, encoding="utf-8")
+    response = ass_invoke(
+        InvokeRequest(
+            run_id="run_def700",
+            node_instance_id="ni_def700",
+            inputs={"cn_srt_uri": str(source)},
+            params={"resolution": "1920x1080"},
+            output_dir=str(tmp_path / "out"),
+        )
+    )
+    assert response.status == "completed"
+    content = Path(response.outputs["ass_uri"]).read_text(encoding="utf-8")
+    assert "an8" in content
+    assert ",0,0,0,0,50,100,0,0,1,4,0,8,50,960,700,1" in content
+
+
 def test_ass_parse_malformed(tmp_path) -> None:
    """验证畸形 SRT 不会抛出异常且返回空条目或忽略坏行。"""
    source = tmp_path / "bad.srt"
@@ -0,0 +1,231 @@
+"""历史 VR 双目字幕统一脚本（scripts/unify_ass_style.py）测试。
+
+背景
+----
+`scripts/unify_ass_style.py` 负责把媒体库中由旧版本生成、混有多种历史样式
+的 *.CN_dual_eye.ass 原地改写为 nodes/ass.py 当前统一样式
+（an8 顶部对齐 + 70% 透明 + DEFAULT_MARGIN_TOP=700）。
+
+本测试验证三条关键路径：
+1. 各历史样式（底部实心白 an2+MarginV1200 / 底部半透明 an2+MarginV1020 /
+   顶部 120 / 已是 700）都能正确重建为统一的 700 样式；
+2. 重建结果与 nodes/ass.py 的 ass_header()/dialogue_line()（新生成字幕的
+   唯一出口）**逐字节一致**，防止"新字幕"与"历史改写"两套逻辑漂移；
+3. 脚本文件能被直接加载执行（仓库根运行 uv run python scripts/...），
+   且 rewrite_content 对非 VR 字幕返回 None（不误伤普通 ASS 文件）。
+
+测试使用真实 ASS 内容构造（Script Info + 样式 + 事件），不 mock 文件系统
+之外的任何逻辑。
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+# 仓库根（tests/ 的上一级），用于导入 scripts/unify_ass_style.py。
+WORKSPACE = Path(__file__).resolve().parent.parent
+SCRIPT_PATH = WORKSPACE / "scripts" / "unify_ass_style.py"
+
+# 用真实可执行的统一脚本构造各历史样式样本：以下文本代表旧版实际输出格式。
+HEADER_3840_OLD = """[Script Info]
+Title: VR Dual-Eye Subtitle
+ScriptType: v4.00+
+Collisions: Normal
+PlayResX: 3840
+PlayResY: 1920
+WrapStyle: 1
+ScaledBorderAndShadow: yes
+
+[V4+ Styles]
+Format: Name,Fontname,Fontsize,PrimaryColour,SecondaryColour,OutlineColour,BackColour,Bold,Italic,Underline,StrikeOut,ScaleX,ScaleY,Spacing,Angle,BorderStyle,Outline,Shadow,Alignment,MarginL,MarginR,MarginV,Encoding
+"""
+
+
+def _load_module():
+    """按文件路径加载 scripts/unify_ass_style.py（scripts 不是包，无法 import）。
+
+    与仓库运行方式一致：uv run python scripts/unify_ass_style.py 时 pythonpath
+    含仓库根，模块内 `from nodes.ass import ...` 可正常解析。"""
+    spec = importlib.util.spec_from_file_location("unify_ass_style", SCRIPT_PATH)
+    module = importlib.util.module_from_spec(spec)
+    assert spec is not None and spec.loader is not None
+    spec.loader.exec_module(module)
+    return module
+
+
+U = _load_module()
+
+
+@pytest.fixture
+def sandbox(tmp_path):
+    """构造含三种历史样式 + 一个非 VR 文件的临时媒体库。
+
+    每个测试独立临时目录（function scope），CLI 写盘测试不会污染其他测试。"""
+    root = tmp_path
+    # 1) 最早期：底部 an2 + 实心白 &H00FFFFFF + MarginV=1200。
+    (root / "old_solid.CN_dual_eye.ass").write_text(
+        HEADER_3840_OLD
+        + "Style: LeftEye,Arial,50,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,50,1970,1200,1\n"
+        + "Style: RightEye,Arial,50,&H00FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,1970,50,1200,1\n"
+        + "\n[Events]\n"
+        + "Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text\n"
+        + 'Dialogue: 0,00:00:00.000,00:00:03.000,LeftEye,,0,0,0,,{\\an2}第一行\\N第二行\n'
+        + 'Dialogue: 0,00:00:00.000,00:00:03.000,RightEye,,0,0,0,,{\\an2}第一行\\N第二行\n',
+        encoding="utf-8",
+    )
+    # 2) 过渡期：底部 an2 + 半透明 &H80FFFFFF + MarginV=1020。
+    (root / "mid_translucent.CN_dual_eye.ass").write_text(
+        HEADER_3840_OLD
+        + "Style: LeftEye,Arial,50,&H80FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,50,1970,1020,1\n"
+        + "Style: RightEye,Arial,50,&H80FFFFFF,&H000000FF,&H00000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,2,1970,50,1020,1\n"
+        + "\n[Events]\n"
+        + "Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text\n"
+        + 'Dialogue: 0,00:00:00.000,00:00:03.000,LeftEye,,0,0,0,,{\\an2}你好\n',
+        encoding="utf-8",
+    )
+    # 3) 已是顶部 120 样式（旧代码默认），仍应规整为 700。
+    (root / "top120.CN_dual_eye.ass").write_text(
+        HEADER_3840_OLD
+        + "Style: LeftEye,Arial,50,&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,50,1920,120,1\n"
+        + "Style: RightEye,Arial,50,&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,1920,50,120,1\n"
+        + "\n[Events]\n"
+        + "Format: Layer,Start,End,Style,Name,MarginL,MarginR,MarginV,Effect,Text\n"
+        + 'Dialogue: 0,00:00:00.000,00:00:03.000,LeftEye,,0,0,0,,{\\an8}第三行\n',
+        encoding="utf-8",
+    )
+    # 4) 非 VR 字幕（无 LeftEye/RightEye 样式），脚本不应改动。
+    (root / "not_vr.CN_dual_eye.ass").write_text(
+        "[Script Info]\nPlayResX: 1920\nPlayResY: 1080\n\n[V4+ Styles]\n"
+        "Format: Name,Fontname,Fontsize,PrimaryColour,...\nStyle: Default,Arial,20,&H00FFFFFF,...\n",
+        encoding="utf-8",
+    )
+    return root
+
+
+def _read(path: Path) -> str:
+    """以 UTF-8 读取测试文件内容。"""
+    return path.read_text(encoding="utf-8")
+
+
+def test_rewrite_converges_all_generations(sandbox) -> None:
+    """三种历史样式统一后都含 an8 顶部对齐 + 70% 透明 + MarginV=700。"""
+    # (文件名, 该文件期望保留的事件文本)
+    expected_text = {
+        "old_solid": r"第一行\N第二行",  # 最早期多行字幕（an2 底部）
+        "mid_translucent": "你好",  # 过渡期
+        "top120": "第三行",  # 旧顶部 120
+    }
+    for name, text in expected_text.items():
+        original = _read(sandbox / f"{name}.CN_dual_eye.ass")
+        rewritten = U.rewrite_content(original)
+        assert rewritten is not None, f"{name} 应被改写"
+        # 左/右眼样式都落到统一值：70%透明 + an8 + MarginV=700。
+        assert "&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,50,1920,700,1" in rewritten
+        assert "&HB3FFFFFF,&H000000FF,&H80000000,&H80000000,0,0,0,0,50,100,0,0,1,4,0,8,1920,50,700,1" in rewritten
+        # 对齐标签全部为 an8，不再残留 an2。
+        assert r"{\an8}" in rewritten
+        assert r"{\an2}" not in rewritten
+        # 事件文本原样保留。
+        assert text in rewritten
+
+
+def test_rewrite_byte_identical_to_write_ass(sandbox) -> None:
+    """历史字幕统一后的输出与 write_ass() 生成的新字幕逐字节一致。
+
+    防止"统一脚本"与"节点生成"两套样式逻辑漂移：两边都经由 ass_header()
+    /dialogue_line()（nodes/ass.py 单一出口）。"""
+    from nodes.ass import DEFAULT_MARGIN_TOP, ass_header, dialogue_line
+
+    # 取 old_solid 样本的事件，用 write_ass 语义重建期望文本。
+    original = _read(sandbox / "old_solid.CN_dual_eye.ass")
+    rewritten = U.rewrite_content(original)
+    assert rewritten is not None
+    hdr = ass_header(3840, 1920, margin_top=DEFAULT_MARGIN_TOP).rstrip("\n")
+    expected = "\n".join(
+        [
+            hdr,
+            dialogue_line("LeftEye", "00:00:00.000", "00:00:03.000", "第一行\\N第二行"),
+            dialogue_line("RightEye", "00:00:00.000", "00:00:03.000", "第一行\\N第二行"),
+        ]
+    ) + "\n"
+    assert rewritten == expected
+
+
+def test_rewrite_skips_non_vr_and_already_canonical(sandbox) -> None:
+    """非 VR 字幕返回 None；已是目标样式（MarginV=700）也返回 None（幂等）。"""
+    not_vr = _read(sandbox / "not_vr.CN_dual_eye.ass")
+    assert U.rewrite_content(not_vr) is None
+
+    # 构造一份已是 700 的新样式文本，改写应返回 None（内容不变）。
+    from nodes.ass import ass_header, dialogue_line
+
+    canonical = (
+        ass_header(3840, 1920, margin_top=U.DEFAULT_MARGIN_TOP).rstrip("\n")
+        + "\n"
+        + dialogue_line("LeftEye", "00:00:00.000", "00:00:03.000", "嗨")
+        + "\n"
+    )
+    assert U.rewrite_content(canonical) is None
+
+
+def test_module_runs_as_cli_dry_run(sandbox) -> None:
+    """脚本可作为 CLI 以 dry-run 方式运行（uv run python scripts/...），不写盘。
+
+    复现仓库根运行方式；断言 dry-run 输出含"旧样式分布"与"将改写"，
+    且运行后沙盒内文件一个都没被改动（dry-run 语义）。"""
+    before = {p.name: _read(p) for p in sandbox.glob("*.ass")}
+    # 不带 --apply = dry-run：应只打印不写盘。
+    proc = subprocess.run(
+        [sys.executable, str(SCRIPT_PATH), str(sandbox)],
+        capture_output=True,
+        text=True,
+        cwd=WORKSPACE,  # 与 uv run python 一致：仓库根在 sys.path
+    )
+    assert proc.returncode == 0, proc.stderr
+    out = proc.stdout
+    assert "[将改写]" in out  # 预览行标记
+    assert "[改写]" not in out  # 未真正写盘
+    after = {p.name: _read(p) for p in sandbox.glob("*.ass")}
+    assert before == after  # dry-run 不改动任何文件
+
+
+
+def test_module_cli_apply_writes_files(sandbox) -> None:
+    """脚本加 --apply 后真正原地改写，写盘结果与 rewrite_content 一致。
+
+    覆盖 main() 的写盘分支：统计输出显示 [改写]，且文件内容已变为
+    MarginV=700 的统一新样式（第二次运行幂等，不再改写）。"""
+    # 备份一个将改写的样本，确保 CLI 走的是与单元层相同的重建逻辑。
+    old_solid = sandbox / "old_solid.CN_dual_eye.ass"
+    expected = U.rewrite_content(old_solid.read_text(encoding="utf-8"))
+    assert expected is not None
+
+    proc = subprocess.run(
+        [sys.executable, str(SCRIPT_PATH), str(sandbox), "--apply"],
+        capture_output=True,
+        text=True,
+        cwd=WORKSPACE,
+    )
+    assert proc.returncode == 0, proc.stderr
+    assert "[改写]" in proc.stdout
+    assert old_solid.read_text(encoding="utf-8") == expected
+
+    # 幂等：第二次 --apply 不再有任何 [改写]（文件已是目标样式）。
+    proc2 = subprocess.run(
+        [sys.executable, str(SCRIPT_PATH), str(sandbox), "--apply"],
+        capture_output=True,
+        text=True,
+        cwd=WORKSPACE,
+    )
+    assert proc2.returncode == 0, proc2.stderr
+    assert "[改写]" not in proc2.stdout
+
+def test_module_loads_and_exports() -> None:
+    """模块可加载且暴露 rewrite_content / DEFAULT_MARGIN_TOP。"""
+    assert U.DEFAULT_MARGIN_TOP == 700
+    assert callable(U.rewrite_content)
@@ -54,12 +54,15 @@ async function createBatchJob() {
  await loadBatchJobs();
 }

-// 渲染单个任务的进度：已完成/跳过与失败计入不同颜色，总数为任务视频数。
+// 渲染单个任务的进度：total=本批无字幕（需处理）的视频数，done=实际处理
+// 完成的视频数。SKIPPED（已有字幕）不参与 total/done——进度只反映真正
+// 待处理的这批（2026-09 口径调整）。失败视频单独红字提示。
 function batchProgress(job) {
  const total = job.total || 0;
  const done = job.done || 0;
  const failed = job.failed || 0;
-  if (!total) return "待扫描";
+  // total=0：本批没有需要处理的视频（全部已有字幕被跳过），视为无处理量。
+  if (!total) return "无待处理视频";
  const percent = Math.round(((done + failed) / total) * 100);
  // 有失败视频时追加红色失败计数，避免把"部分失败"误看成全部完成。
  const failedText = failed > 0 ? ` <span class="danger">，失败 ${failed}</span>` : "";