@@ -164,6 +242,30 @@ 字幕样式模板
启用阴影和描边，适合短视频平台观看
+
+
+
+
+
+
+
+
+
样式会保存到 SQLite，后续自动加字幕会使用这套模板。
@@ -304,4 +406,10 @@ 还没有进入字幕流程的切片
{% endblock %}
{% block extra_scripts %}
+{% if subtitle_task_mode %}
+
+
+
+
+{% endif %}
{% endblock %}
diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md
index 7a80917..dc8dbb6 100644
--- a/docs/ARCHITECTURE.md
+++ b/docs/ARCHITECTURE.md
@@ -1,5 +1,21 @@
# 系统架构
+## 2026-08-23：统一字幕架构
+
+```text
+转写 checkpoint / 旧 transcript.md
+ ↓
+source subtitle_track → immutable subtitle_revision → subtitle_cues(ms)
+ ↓ 按 output_clip 原片边界快照截取
+clip subtitle_track → immutable subtitle_revision → SRT/VTT/ASS/编辑器
+```
+
+- `subtitle_data_service.py` 是字幕数据事实入口，负责轨、版本、cue、同步、导入导出、质量检查和服务端波形 peaks。
+- 原片轨只保存一条 active track；内容变化产生新 revision。切片轨记录来源 track/revision，未人工编辑时可同步，人工编辑后只进入 `pending_sync`。
+- revision 内容不可原地更新；审核只改变 revision 状态，渲染必须固定引用 revision id，避免编辑过程中改变已排队输出。
+- `pysubs2` 负责字幕格式与 ASS，不再手写固定分辨率 ASS。`wavesurfer.js` 仅消费服务端 peaks 与媒体流，不读取六小时完整音频到浏览器内存。
+- 当前同步 FFmpeg 烧录仅作为旧 API 兼容；持久化渲染队列、取消、重试、NVENC 回退与发送中心门禁在 PR 4 接入。
+
## 2026-08-23：长直播选片层
`long_live_talk` 使用独立的 `long_live_talk_analyzer`：
diff --git a/docs/DATABASE_SCHEMA.md b/docs/DATABASE_SCHEMA.md
index 542c346..f878e93 100644
--- a/docs/DATABASE_SCHEMA.md
+++ b/docs/DATABASE_SCHEMA.md
@@ -1,5 +1,14 @@
# 数据库结构说明
+## 2026-08-23：字幕 revision 数据结构
+
+- `output_clip` 新增 `source_start_ms / source_end_ms / source_duration_ms / source_fingerprint / snapshot_source`。新切片使用 `cut_commit`，旧切片第一次使用时可从候选边界生成 `legacy_inferred`；之后候选修改不会漂移已保存边界。
+- `subtitle_tracks`：任务级原片轨或切片轨，记录 `source_track_id / source_revision_id / active_revision_id / sync_status / has_manual_edits`。
+- `subtitle_revisions`：不可变内容版本，记录来源 `asr / markdown / source_sync / manual / import / ai_suggestion`、父版本、状态、cue 数和 checksum。
+- `subtitle_cues`：毫秒级 `start_ms / end_ms`、文字、置信度、手工说话人和 `source_cue_id`；按 revision 和时间查询。
+- `subtitle_jobs.revision_id` 固定渲染使用的版本；样式表新增描边宽度、阴影深度、安全区百分比与说话人样式 JSON。
+- 启动迁移保持幂等；已有数据库缺少上述结构时先通过 SQLite 在线 backup 创建 `subtitle-editor-rebuild` 迁移前快照，不删除旧字幕 job 或历史切片。
+
## 2026-08-23：长直播 AI 窗口 checkpoint
新增 `ai_analysis_windows`：
diff --git a/docs/SUBTITLE_AND_PUBLISH_PLAN.md b/docs/SUBTITLE_AND_PUBLISH_PLAN.md
index 8cbd3a2..82a4c0c 100644
--- a/docs/SUBTITLE_AND_PUBLISH_PLAN.md
+++ b/docs/SUBTITLE_AND_PUBLISH_PLAN.md
@@ -1,5 +1,11 @@
# 字幕与一键推送功能计划书
+## 2026-08-23 PR 3 实现状态
+
+- 已完成统一原片/切片 track、不可变 revision、毫秒 cue、切片边界快照、人工编辑保护。
+- 已完成 pysubs2 SRT/VTT/ASS、动态 ASS 样式、本地 wavesurfer 波形、虚拟列表和专业编辑操作。
+- 尚未完成自动流水线审核暂停、AI 纠错建议、持久化批量渲染、NVENC 回退和发送中心审核门禁；这些仅在 PR 4 实现。
+
## 1. 最终目标
把当前“上传视频 → 转写 → AI 分析 → 片段审核 → 自动切片”的流程继续向后延伸，形成：
diff --git a/docs/TASK_FLOW.md b/docs/TASK_FLOW.md
index 9cc6727..1fe2b59 100644
--- a/docs/TASK_FLOW.md
+++ b/docs/TASK_FLOW.md
@@ -1,5 +1,13 @@
# 任务状态流转
+## 2026-08-23：字幕编辑数据流（PR 3）
+
+1. 完成转写后，从成功的结构化 checkpoint 创建原片主字幕 revision；旧任务没有 checkpoint 时兼容读取完整 `transcript.md`。
+2. 切片完成时固化原片起止毫秒快照；首次进入字幕工作台时按快照截取主字幕并换算为切片局部时间。
+3. 编辑、导入、拆分、合并、位移和替换均创建新 revision；自动保存使用当前 active revision 做乐观并发校验，冲突时返回 409，不覆盖他人或新版本。
+4. 未人工编辑的切片可跟随原片新 revision；人工切片只标记 `pending_sync`，用户明确强制同步前保留其 active revision。
+5. 审核将指定 revision 标记为 `approved`。PR 3 仍保留单条同步烧录兼容入口；PR 4 才会让自动流水线暂停审核并异步批量烧录。
+
## 2026-08-23：长直播高光流程
1. 逐句时间戳转写按约 5 分钟、60 秒重叠生成窗口。
diff --git a/docs/UI_REFERENCE.md b/docs/UI_REFERENCE.md
index 4e7cc6a..7c7bf9d 100644
--- a/docs/UI_REFERENCE.md
+++ b/docs/UI_REFERENCE.md
@@ -1,5 +1,15 @@
# UI 参考说明
+## 2026-08-23 更新：专业字幕编辑器
+
+- 字幕任务页顶部新增“原片主时间轴 · 切片继承编辑器”，轨下拉明确区分原片和每条切片，并显示人工版、待同步、revision 与审核状态。
+- 视频与字幕双向联动；视频上覆盖当前字幕，波形使用本地 wavesurfer Regions/Timeline，拖动蓝色区间直接修改毫秒起止。
+- 字幕列表固定高度并虚拟渲染，只创建可见行 DOM；适合数千 cue。每行提供勾选、跳转、开始/结束毫秒、文字、说话人和质量提醒。
+- 工具栏包含搜索替换、新增、拆分、合并、删除、批量位移、撤销、重做和立即保存；普通编辑 1.8 秒后自动保存为新 revision。
+- 质量卡将重叠显示为错误，时长、间隔、行长、行数和阅读速度显示为提醒；页面不会自动改字。
+- 样式侧栏增加描边宽度、阴影深度和安全区；实际 ASS 画布按视频 9:16、16:9 或 1:1 尺寸生成。
+- 小屏将视频与波形改为单列，字幕行横向保留关键编辑字段；不引入 React/Vue。
+
## 2026-08-23 更新：长直播分析完整性状态
- 长直播 AI 窗口失败时，任务详情的 AI 状态显示“分析不完整”和当前覆盖率，不把部分结果冒充为已完整分析。
diff --git a/docs/agent_tasks/2026-08-23-subtitle-editor-rebuild.md b/docs/agent_tasks/2026-08-23-subtitle-editor-rebuild.md
new file mode 100644
index 0000000..ad971f2
--- /dev/null
+++ b/docs/agent_tasks/2026-08-23-subtitle-editor-rebuild.md
@@ -0,0 +1,65 @@
+# PR 3 执行任务：字幕数据层与专业编辑器重构
+
+## 背景
+
+现有字幕工作台从 `transcript.md` 最多读取 120 行，按整秒时间生成固定 1080×1920 ASS，并在 HTTP 请求内同步烧录。长直播需要以原片毫秒时间轴为事实来源，切片字幕必须继承且不能覆盖人工编辑。
+
+## 目标
+
+1. 建立 `subtitle_tracks / subtitle_revisions / subtitle_cues` 统一字幕模型。
+2. 保存 output clip 的不可变原片起止毫秒快照，完成原片→切片本地时间换算。
+3. 使用 `pysubs2==1.9.0` 实现 SRT/VTT/ASS 导入导出与 ASS 序列化。
+4. 本地固定 `wavesurfer.js@7.12.11`，使用 Regions/Timeline 和服务端 peaks 构建长视频编辑器。
+5. 提供文本、毫秒时间、说话人、增删、拆分、合并、批量位移、搜索替换、撤销重做、自动保存和质量告警。
+
+## 允许修改范围
+
+- 字幕数据库表、output_clip 快照和 subtitle_jobs revision 引用。
+- 新字幕服务、模型、API 路由、媒体 peaks 接口。
+- 字幕工作台模板、专用 JS/CSS 和固定版本 vendor 文件。
+- requirements、第三方许可证、测试和项目文档。
+
+## 禁止修改范围
+
+- 不实现自动说话人分离、翻译、卡拉 OK 或 AI 自动覆盖。
+- 不复制 GPL-3.0 VideoCaptioner 源码；仅借鉴流程。
+- PR 3 不接入自动流水线审核暂停和异步批量烧录，这属于 PR 4。
+- 不删除旧 `subtitle_jobs` 或旧带字幕成片；旧同步烧录入口暂时兼容。
+- 不读取或写入 secrets，不合并 PR，不改写 Git 历史。
+
+## 已确定实现要求
+
+- revision 内容创建后不可原地改写；每次人工保存创建子 revision 并切换 active。
+- 渲染 Job 增加 `revision_id`，后续只能引用固定 revision。
+- source track 来自结构化转写 checkpoint；无结构化结果时完整读取 Markdown，不再截断 120 行。
+- clip track 使用 `source_start_ms/source_end_ms` 快照截取 source cues 并换算本地毫秒。
+- 未人工编辑 clip 自动跟随 source active revision；已有人工 revision 只标记 `pending_sync`。
+- 质量规则只返回 warning/error，不自动修改文字。
+- 波形 peaks 服务端缓存并限制点数，浏览器不解码完整 6 小时音频。
+- 样式序列化使用实际视频宽高，并提供安全区与说话人样式。
+
+## 验收标准
+
+- SRT/VTT/ASS 往返保留毫秒时间，中文不乱码。
+- 超过 120 cues 不截断；重叠、时长、间隔、行长和阅读速度规则正确。
+- 原片→切片边界截取准确；人工 clip revision 不被 source 更新覆盖。
+- 数千 cues 范围查询分页；前端实现虚拟滚动及编辑操作。
+- 六小时 peaks 输出点数受控且有缓存。
+- 旧字幕任务、通用/综艺/长直播流程不回归。
+
+## 测试命令
+
+```powershell
+.venv\Scripts\ruff.exe check app tests
+.venv\Scripts\python.exe -m compileall app tests
+node --check app/static/js/subtitle-editor.js
+.venv\Scripts\python.exe -m pytest tests/test_subtitle_editor.py tests/test_split_services.py tests/test_versioning_rollback.py -q
+.venv\Scripts\python.exe -m pytest tests/ -q
+```
+
+## 返回格式
+
+- 数据模型、API 和编辑器能力摘要
+- 第三方版本与许可证
+- 专项/全量测试证据
+- 中文 commit、推送分支、堆叠 PR 链接
diff --git a/requirements.in b/requirements.in
index c84c629..a50c48c 100644
--- a/requirements.in
+++ b/requirements.in
@@ -10,3 +10,4 @@ tzdata>=2026.1
playwright>=1.58,<1.63
aiofiles>=25.1,<26
faster-whisper>=1.2,<1.3
+pysubs2==1.9.0
diff --git a/requirements.txt b/requirements.txt
index d44dc29..0b340de 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -13,3 +13,4 @@ tzdata==2026.3
playwright==1.62.0
aiofiles==25.1.0
faster-whisper==1.2.1
+pysubs2==1.9.0
diff --git a/tests/test_subtitle_editor.py b/tests/test_subtitle_editor.py
new file mode 100644
index 0000000..8468c9f
--- /dev/null
+++ b/tests/test_subtitle_editor.py
@@ -0,0 +1,466 @@
+from __future__ import annotations
+
+from array import array
+import hashlib
+import json
+from uuid import uuid4
+
+import pysubs2
+import pytest
+from fastapi.testclient import TestClient
+
+from app.db.database import get_connection, init_db
+from app.main import app
+from app.services.subtitle_data_service import (
+ apply_revision_operations,
+ create_manual_revision,
+ ensure_clip_track,
+ ensure_source_track,
+ evaluate_subtitle_quality,
+ export_subtitle_text,
+ get_revision,
+ get_track,
+ get_waveform_peaks,
+ import_subtitle_text,
+ inherit_cues_for_clip,
+ serialize_revision_to_ass,
+)
+from app.services.video_cut_service import CutResult
+from app.services.video_cut_workflow_service import _insert_output_clip_record
+from app.services.subtitle_workflow_service import _write_ass_file
+
+
+PREFIX = "test-subtitle-editor-"
+
+
+@pytest.fixture(autouse=True)
+def subtitle_editor_database():
+ init_db()
+ _cleanup()
+ yield
+ _cleanup()
+
+
+def _cleanup() -> None:
+ with get_connection() as connection:
+ connection.execute(
+ "DELETE FROM subtitle_cues WHERE revision_id IN (SELECT id FROM subtitle_revisions WHERE track_id IN (SELECT id FROM subtitle_tracks WHERE task_id LIKE ?))",
+ (f"{PREFIX}%",),
+ )
+ connection.execute(
+ "DELETE FROM subtitle_revisions WHERE track_id IN (SELECT id FROM subtitle_tracks WHERE task_id LIKE ?)",
+ (f"{PREFIX}%",),
+ )
+ connection.execute("DELETE FROM subtitle_tracks WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM subtitle_jobs WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM output_clip WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM cut_runs WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM clip_candidates WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM transcription_chunks WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM transcription_runs WHERE task_id LIKE ?", (f"{PREFIX}%",))
+ connection.execute("DELETE FROM tasks WHERE id LIKE ?", (f"{PREFIX}%",))
+ connection.commit()
+
+
+def _create_task(*, segments: list[dict] | None = None, with_clip: bool = True) -> tuple[str, str | None]:
+ task_id = f"{PREFIX}{uuid4().hex[:10]}"
+ now = "2026-08-23T12:00:00+00:00"
+ with get_connection() as connection:
+ connection.execute(
+ "INSERT INTO tasks (id, task_name, platform, status, created_at, updated_at) VALUES (?, ?, 'general', 'completed', ?, ?)",
+ (task_id, "字幕编辑器测试", now, now),
+ )
+ if segments is not None:
+ _insert_transcription(connection, task_id, segments, now)
+ output_id = None
+ if with_clip:
+ output_id = f"out-{uuid4().hex[:10]}"
+ connection.execute(
+ """
+ INSERT INTO output_clip (
+ id, task_id, output_file_path, output_file_name, status, is_active,
+ source_start_ms, source_end_ms, source_duration_ms,
+ source_fingerprint, snapshot_source, created_at, updated_at
+ ) VALUES (?, ?, ?, 'clip.mp4', 'completed', 1, 1000, 5000, 4000,
+ 'source-v1', 'cut_commit', ?, ?)
+ """,
+ (output_id, task_id, "C:/missing/clip.mp4", now, now),
+ )
+ connection.commit()
+ return task_id, output_id
+
+
+def _insert_transcription(connection, task_id: str, segments: list[dict], now: str) -> None:
+ run_id = f"run-{uuid4().hex[:10]}"
+ raw = json.dumps(segments, ensure_ascii=False, separators=(",", ":"))
+ checksum = hashlib.sha256(raw.encode("utf-8")).hexdigest()
+ connection.execute(
+ """
+ INSERT INTO transcription_runs (
+ id, task_id, source_fingerprint, provider, model, device, compute_type,
+ chunk_seconds, overlap_seconds, status, total_chunks, completed_chunks,
+ is_active, created_at, updated_at, completed_at
+ ) VALUES (?, ?, 'source-v1', 'local', 'small', 'cpu', 'int8',
+ 120, 5, 'completed', 1, 1, 1, ?, ?, ?)
+ """,
+ (run_id, task_id, now, now, now),
+ )
+ connection.execute(
+ """
+ INSERT INTO transcription_chunks (
+ id, run_id, task_id, chunk_index, start_ms, end_ms, status,
+ attempt_count, result_json, result_checksum, created_at, updated_at
+ ) VALUES (?, ?, ?, 1, 0, 120000, 'completed', 1, ?, ?, ?, ?)
+ """,
+ (f"chunk-{uuid4().hex[:10]}", run_id, task_id, raw, checksum, now, now),
+ )
+
+
+def _segments(count: int = 4) -> list[dict]:
+ return [
+ {
+ "start_seconds": index * 1.5 + 0.123,
+ "end_seconds": index * 1.5 + 1.345,
+ "text": f"第{index + 1}条中文字幕",
+ "confidence": 0.91,
+ "words": [
+ {
+ "start_ms": round((index * 1.5 + 0.123) * 1000),
+ "end_ms": round((index * 1.5 + 0.5) * 1000),
+ "text": "第",
+ "confidence": 0.9,
+ }
+ ],
+ }
+ for index in range(count)
+ ]
+
+
+def test_schema_migration_is_idempotent_and_contains_revision_tables():
+ init_db()
+ init_db()
+ with get_connection() as connection:
+ names = {
+ row[0]
+ for row in connection.execute(
+ "SELECT name FROM sqlite_master WHERE type = 'table'"
+ ).fetchall()
+ }
+ output_columns = {row[1] for row in connection.execute("PRAGMA table_info(output_clip)")}
+ assert {"subtitle_tracks", "subtitle_revisions", "subtitle_cues"} <= names
+ assert {"source_start_ms", "source_end_ms", "source_duration_ms", "source_fingerprint"} <= output_columns
+
+
+def test_source_track_uses_structured_checkpoint_with_millisecond_precision():
+ task_id, _ = _create_task(segments=_segments(), with_clip=False)
+ track = ensure_source_track(task_id)
+ revision = get_revision(track["active_revision_id"], include_cues=True)
+ assert revision["cue_count"] == 4
+ assert revision["cues"][0]["start_ms"] == 123
+ assert revision["cues"][0]["end_ms"] == 1345
+ assert revision["cues"][0]["confidence"] == pytest.approx(0.91)
+
+
+def test_more_than_120_cues_are_not_truncated():
+ task_id, _ = _create_task(segments=_segments(150), with_clip=False)
+ track = ensure_source_track(task_id)
+ revision = get_revision(track["active_revision_id"], include_cues=True)
+ assert revision["cue_count"] == 150
+ assert len(revision["cues"]) == 150
+ assert revision["cues"][-1]["text"] == "第150条中文字幕"
+
+
+def test_source_to_clip_boundary_conversion_is_exact():
+ source = [
+ {"id": "a", "start_ms": 0, "end_ms": 1500, "text": "开头"},
+ {"id": "b", "start_ms": 1500, "end_ms": 3000, "text": "中间"},
+ {"id": "c", "start_ms": 4500, "end_ms": 6000, "text": "结尾"},
+ {"id": "d", "start_ms": 7000, "end_ms": 8000, "text": "范围外"},
+ ]
+ inherited = inherit_cues_for_clip(source, 1000, 5000)
+ assert [(cue["start_ms"], cue["end_ms"], cue["text"]) for cue in inherited] == [
+ (0, 500, "开头"),
+ (500, 2000, "中间"),
+ (3500, 4000, "结尾"),
+ ]
+ assert [cue["source_cue_id"] for cue in inherited] == ["a", "b", "c"]
+
+
+def test_clip_track_inherits_snapshot_and_manual_revision_is_not_overwritten():
+ task_id, output_id = _create_task(segments=_segments(), with_clip=True)
+ clip_track = ensure_clip_track(task_id, output_id)
+ active = get_revision(clip_track["active_revision_id"], include_cues=True)
+ edited = [{**cue, "text": f"人工：{cue['text']}"} for cue in active["cues"]]
+ manual = create_manual_revision(
+ clip_track["id"],
+ base_revision_id=active["id"],
+ cues=edited,
+ note="人工精修",
+ )
+
+ with get_connection() as connection:
+ chunk = connection.execute(
+ "SELECT * FROM transcription_chunks WHERE task_id = ?", (task_id,)
+ ).fetchone()
+ changed = _segments()
+ changed[1]["text"] = "原片字幕已经变化"
+ raw = json.dumps(changed, ensure_ascii=False, separators=(",", ":"))
+ connection.execute(
+ "UPDATE transcription_chunks SET result_json = ?, result_checksum = ? WHERE id = ?",
+ (raw, hashlib.sha256(raw.encode("utf-8")).hexdigest(), chunk["id"]),
+ )
+ connection.commit()
+
+ ensure_source_track(task_id, force=True)
+ protected = get_track(clip_track["id"])
+ assert protected["active_revision_id"] == manual["id"]
+ assert protected["sync_status"] == "pending_sync"
+ assert protected["has_manual_edits"] is True
+
+
+def test_source_manual_revision_syncs_unedited_clip_but_only_flags_edited_clip():
+ task_id, output_id = _create_task(segments=_segments(), with_clip=True)
+ source_track = ensure_source_track(task_id)
+ clip_track = ensure_clip_track(task_id, output_id)
+ source_revision = get_revision(source_track["active_revision_id"], include_cues=True)
+ source_cues = [{**cue, "text": f"原片修改：{cue['text']}"} for cue in source_revision["cues"]]
+ new_source = create_manual_revision(
+ source_track["id"],
+ base_revision_id=source_revision["id"],
+ cues=source_cues,
+ )
+ followed = get_track(clip_track["id"])
+ assert followed["source_revision_id"] == new_source["id"]
+ assert followed["sync_status"] == "up_to_date"
+
+ clip_revision = get_revision(followed["active_revision_id"], include_cues=True)
+ manual_clip = create_manual_revision(
+ clip_track["id"],
+ base_revision_id=clip_revision["id"],
+ cues=[{**cue, "text": f"切片精修：{cue['text']}"} for cue in clip_revision["cues"]],
+ )
+ unchanged_source = ensure_clip_track(task_id, output_id)
+ assert unchanged_source["active_revision_id"] == manual_clip["id"]
+ assert unchanged_source["sync_status"] == "manual"
+ source_after = get_revision(new_source["id"], include_cues=True)
+ create_manual_revision(
+ source_track["id"],
+ base_revision_id=new_source["id"],
+ cues=[{**cue, "text": f"再次修改：{cue['text']}"} for cue in source_after["cues"]],
+ )
+ protected = get_track(clip_track["id"])
+ assert protected["active_revision_id"] == manual_clip["id"]
+ assert protected["sync_status"] == "pending_sync"
+
+
+def test_multiple_operations_keep_cue_ids_until_new_revision_is_committed():
+ task_id, _ = _create_task(segments=_segments(), with_clip=False)
+ track = ensure_source_track(task_id)
+ revision = get_revision(track["active_revision_id"], include_cues=True)
+ cue_id = revision["cues"][0]["id"]
+ updated = apply_revision_operations(
+ track["id"],
+ base_revision_id=revision["id"],
+ operations=[
+ {"type": "update", "cue_id": cue_id, "text": "连续操作"},
+ {"type": "shift", "cue_ids": [cue_id], "delta_ms": 250},
+ ],
+ )
+ assert updated["cues"][0]["text"] == "连续操作"
+ assert updated["cues"][0]["start_ms"] == 373
+
+
+@pytest.mark.parametrize("format_name", ["srt", "vtt", "ass"])
+def test_pysubs2_round_trip_preserves_chinese_and_milliseconds(format_name: str):
+ task_id, _ = _create_task(segments=_segments(), with_clip=False)
+ track = ensure_source_track(task_id)
+ content, _media_type, _filename = export_subtitle_text(track["id"], format_name=format_name)
+ parsed = pysubs2.SSAFile.from_string(content, format_=format_name)
+ # ASS 规范使用厘秒；SRT/VTT 保留 1ms，ASS 最多产生 5ms 的量化误差。
+ tolerance_ms = 5 if format_name == "ass" else 0
+ assert abs(parsed.events[0].start - 123) <= tolerance_ms
+ assert abs(parsed.events[0].end - 1345) <= tolerance_ms
+ assert "中文字幕" in parsed.events[0].plaintext
+
+ imported = import_subtitle_text(track["id"], content=content, format_name=format_name)
+ assert abs(imported["cues"][0]["start_ms"] - 123) <= tolerance_ms
+ assert abs(imported["cues"][0]["end_ms"] - 1345) <= tolerance_ms
+
+
+@pytest.mark.parametrize("dimensions", [(1080, 1920), (1920, 1080), (1080, 1080)])
+def test_ass_resolution_follows_real_media_dimensions(monkeypatch, dimensions):
+ task_id, output_id = _create_task(segments=_segments(), with_clip=True)
+ track = ensure_clip_track(task_id, output_id)
+ monkeypatch.setattr("app.services.subtitle_data_service._probe_media_dimensions", lambda _path: dimensions)
+ ass = serialize_revision_to_ass(track["id"], track["active_revision_id"])
+ document = pysubs2.SSAFile.from_string(ass, format_="ass")
+ assert int(document.info["PlayResX"]) == dimensions[0]
+ assert int(document.info["PlayResY"]) == dimensions[1]
+ assert document.styles["Default"].marginv == round(dimensions[1] * 0.05)
+
+
+def test_ass_applies_default_host_and_guest_speaker_styles(monkeypatch):
+ task_id, output_id = _create_task(segments=_segments(), with_clip=True)
+ track = ensure_clip_track(task_id, output_id)
+ revision = get_revision(track["active_revision_id"], include_cues=True)
+ cues = []
+ for index, cue in enumerate(revision["cues"]):
+ cues.append({**cue, "speaker": "主播" if index == 0 else "嘉宾"})
+ manual = create_manual_revision(
+ track["id"],
+ base_revision_id=revision["id"],
+ cues=cues,
+ )
+ monkeypatch.setattr(
+ "app.services.subtitle_data_service._probe_media_dimensions",
+ lambda _path: (1920, 1080),
+ )
+ document = pysubs2.SSAFile.from_string(
+ serialize_revision_to_ass(track["id"], manual["id"]),
+ format_="ass",
+ )
+ assert document.events[0].style != "Default"
+ assert document.events[1].style != "Default"
+ assert document.styles[document.events[0].style].primarycolor == pysubs2.Color(255, 255, 255)
+ assert document.styles[document.events[1].style].primarycolor == pysubs2.Color(255, 214, 10)
+
+
+def test_ass_render_uses_explicit_immutable_revision_not_latest_active(monkeypatch, tmp_path):
+ task_id, output_id = _create_task(segments=_segments(), with_clip=True)
+ track = ensure_clip_track(task_id, output_id)
+ original = get_revision(track["active_revision_id"], include_cues=True)
+ create_manual_revision(
+ track["id"],
+ base_revision_id=original["id"],
+ cues=[{**cue, "text": "最新人工版本"} for cue in original["cues"]],
+ )
+ monkeypatch.setattr(
+ "app.services.subtitle_workflow_service.get_artifact_paths",
+ lambda _task_id: {"subtitled_dir": tmp_path},
+ )
+ path = _write_ass_file(
+ task_id,
+ {"id": output_id, "output_file_name": "fixed.mp4"},
+ {},
+ revision_id=original["id"],
+ )
+ document = pysubs2.load(str(path), encoding="utf-8")
+ assert "最新人工版本" not in document.events[0].plaintext
+ assert "中文字幕" in document.events[0].plaintext
+
+
+def test_quality_rules_only_report_and_do_not_change_text():
+ cues = [
+ {"id": "a", "start_ms": 0, "end_ms": 500, "text": "这是一行非常非常非常非常非常长的中文字幕"},
+ {"id": "b", "start_ms": 400, "end_ms": 9000, "text": "发生重叠\n第二行\n第三行"},
+ ]
+ original = json.loads(json.dumps(cues, ensure_ascii=False))
+ quality = evaluate_subtitle_quality(cues)
+ codes = {issue["code"] for issue in quality["issues"]}
+ assert {"too_short", "line_too_long", "reading_speed", "overlap", "too_long", "too_many_lines"} <= codes
+ assert quality["error_count"] == 1
+ assert cues == original
+
+
+def test_cut_commit_saves_immutable_source_bounds():
+ task_id, _ = _create_task(segments=None, with_clip=False)
+ candidate_id = f"candidate-{uuid4().hex[:8]}"
+ run_id = f"cut-{uuid4().hex[:8]}"
+ now = "2026-08-23T12:00:00+00:00"
+ with get_connection() as connection:
+ connection.execute(
+ """
+ INSERT INTO clip_candidates (
+ id, task_id, title, start_time, end_time, duration_seconds,
+ created_at, updated_at
+ ) VALUES (?, ?, '候选', '00:00:01.250', '00:00:05.750', 5, ?, ?)
+ """,
+ (candidate_id, task_id, now, now),
+ )
+ connection.execute(
+ "INSERT INTO cut_runs (id, task_id, run_number, status, is_active, created_at, updated_at) VALUES (?, ?, 1, 'processing', 0, ?, ?)",
+ (run_id, task_id, now, now),
+ )
+ connection.commit()
+ _insert_output_clip_record(
+ task_id,
+ run_id,
+ CutResult(candidate_id, "C:/missing/output.mp4", "output.mp4", "completed"),
+ source_fingerprint="fingerprint-v1",
+ )
+ with get_connection() as connection:
+ connection.execute(
+ "UPDATE clip_candidates SET start_time = '00:01:00', end_time = '00:02:00' WHERE id = ?",
+ (candidate_id,),
+ )
+ row = connection.execute(
+ "SELECT * FROM output_clip WHERE task_id = ? AND clip_candidate_id = ?",
+ (task_id, candidate_id),
+ ).fetchone()
+ assert row["source_start_ms"] == 1250
+ assert row["source_end_ms"] == 5750
+ assert row["source_duration_ms"] == 4500
+ assert row["source_fingerprint"] == "fingerprint-v1"
+ assert row["snapshot_source"] == "cut_commit"
+
+
+def test_cue_api_supports_time_range_and_pagination():
+ task_id, _ = _create_task(segments=_segments(20), with_clip=False)
+ track = ensure_source_track(task_id)
+ response = TestClient(app).get(
+ f"/api/subtitles/tracks/{track['id']}/cues",
+ params={"start_ms": 3000, "end_ms": 9000, "offset": 1, "limit": 2},
+ )
+ assert response.status_code == 200
+ payload = response.json()
+ assert payload["total"] >= 4
+ assert len(payload["cues"]) == 2
+ assert all(cue["end_ms"] > 3000 and cue["start_ms"] < 9000 for cue in payload["cues"])
+
+
+def test_subtitle_page_loads_local_editor_and_vendored_wavesurfer():
+ task_id, _ = _create_task(segments=_segments(), with_clip=True)
+ response = TestClient(app).get(f"/subtitles/{task_id}")
+ assert response.status_code == 200
+ assert 'id="subtitle-editor"' in response.text
+ assert "vendor/wavesurfer/wavesurfer.min.js" in response.text
+ assert "vendor/wavesurfer/regions.min.js" in response.text
+ assert "js/subtitle-editor.js" in response.text
+
+
+def test_waveform_peaks_are_precomputed_at_low_sample_rate_and_cached(monkeypatch, tmp_path):
+ media_path = tmp_path / "source.mp4"
+ media_path.write_bytes(b"fake-video-source")
+ task_id, _ = _create_task(segments=_segments(), with_clip=False)
+ with get_connection() as connection:
+ connection.execute(
+ "UPDATE tasks SET original_video_path = ? WHERE id = ?",
+ (str(media_path), task_id),
+ )
+ connection.commit()
+ track = ensure_source_track(task_id)
+ transcript_path = tmp_path / "artifacts" / "transcript.md"
+ calls = []
+
+ class Result:
+ returncode = 0
+ stderr = b""
+ stdout = array("h", [0, 1000, -2000, 32000, -12000] * 400).tobytes()
+
+ def fake_run(command, **_kwargs):
+ calls.append(command)
+ return Result()
+
+ monkeypatch.setattr("app.services.subtitle_data_service.shutil.which", lambda _name: "ffmpeg")
+ monkeypatch.setattr("app.services.subtitle_data_service.subprocess.run", fake_run)
+ monkeypatch.setattr(
+ "app.services.subtitle_data_service.get_artifact_paths",
+ lambda _task_id: {"transcript_path": transcript_path},
+ )
+ first = get_waveform_peaks(track["id"], max_points=1000)
+ second = get_waveform_peaks(track["id"], max_points=1000)
+ assert calls and calls[0][calls[0].index("-ar") + 1] == "100"
+ assert first["point_count"] <= 1000
+ assert first["cached"] is False
+ assert second["cached"] is True
+ assert len(calls) == 1
diff --git a/third_party_licenses/pysubs2-LICENSE.txt b/third_party_licenses/pysubs2-LICENSE.txt
new file mode 100644
index 0000000..dc62e89
--- /dev/null
+++ b/third_party_licenses/pysubs2-LICENSE.txt
@@ -0,0 +1,19 @@
+Copyright (c) 2014-2026 Tomas Karabela
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.