新增：字幕审核暂停与异步烧录流水线

DEVELOPMENT_LOG.md

@@ -1,5 +1,17 @@
 # Development Log
 
+## 2026-08-24 字幕审核、异步渲染与自动流水线整合（PR 4）
+
+- 全自动流水线在切片后创建原片/切片字幕草稿，并停在 `pending_subtitle_review`；不会继续生成文案或发送任务。
+- 用户必须选择“审核并批量烧录”或“跳过字幕，使用原片继续”。决定写入任务配置，恢复流水线后不会静默切换视频来源。
+- 字幕批量烧录接入 `workflow_jobs` 持久化队列，固定引用已审核 revision，逐条保存 checkpoint；重启只复用文件存在且验证通过的结果。
+- FFmpeg 渲染支持真实进度、取消、无进展超时、临时文件和原子替换；优先尝试可用的 NVENC，失败回退 `libx264`，输出必须通过 H.264、`yuv420p`、音轨和时长校验。
+- 旧 active 字幕成片只在新结果验证成功后切换；失败、取消或进程异常不会覆盖旧文件，精确临时文件会被清理。
+- AI 纠错按所选 cue 分批创建 `ai_suggestion` revision，仅接受文字和断句建议；差异需人工选择，接受后再生成新的人工 revision，不自动覆盖 active 版本。
+- 发送中心新增字幕来源证据门禁：自动使用字幕时必须同时满足 revision 已审核、渲染已验证、文件存在；明确跳过时只使用原片。
+- 新增第四阶段专项测试，覆盖暂停/恢复、显式跳过、审核门禁、checkpoint、取消清理、NVENC 回退、验证失败、AI 建议与发布就绪证据。
+- 验证通过：Ruff、Python 编译、两个 JavaScript 语法检查、定向测试 `143 passed`、完整测试 `500 passed`；本机 9:16、16:9、1:1 三种画幅真实 FFmpeg 烧录与 FFprobe 校验通过。
+
 ## 2026-08-23 字幕数据层与专业编辑器重构（PR 3）
 
 - 新增 `subtitle_tracks / subtitle_revisions / subtitle_cues`，以原片主时间轴为事实源；每次人工编辑、导入或同步都创建不可变 revision，渲染任务固定引用 revision。

NEXT_STEPS.md

@@ -5,17 +5,17 @@
 - [x] PR 1：模式必选、已有文件入口、媒体/磁盘预检、持久化重型 Job、转写断点与词级时间戳。
 - [x] PR 2：`long_live_talk` 5 分钟重叠窗口、每小时覆盖、全局配额、去重、90% 覆盖门禁和窗口级恢复。
 - [x] PR 3：统一字幕 track/revision/cue、pysubs2 导入导出、wavesurfer 波形与专业编辑器。
-- [ ] PR 4：字幕审核暂停、AI 建议 revision、异步批量烧录、NVENC 回退和发送中心门禁。
+- [x] PR 4：字幕审核暂停、AI 建议 revision、异步批量烧录、NVENC 回退和发送中心门禁。
 
-当前验证重点：PR 3 完成专项、页面和全量回归后进入自动流水线整合；异步批量烧录、AI 建议 revision 和发送中心审核门禁仍属于 PR 4，不能把 PR 3 的旧同步烧录兼容入口当成最终自动流程。
+当前验证重点：四个开发阶段均已落地。合并时必须按 PR 1 → PR 2 → PR 3 → PR 4 的顺序；每个 PR 只合并到它标明的前序分支，最终再进入 `master`，不要跳过中间数据库与接口变更。
 
-### PR 3 人工检查
+### PR 4 人工检查
 
-1. 打开任一已有转写和切片的任务字幕工作台，先选“原片主字幕”，确认视频、波形和数千行虚拟列表可滚动。
-2. 修改文字或毫秒时间，等待状态显示新 revision 已保存；刷新页面后修改仍在，旧 revision 仍可从 API 查询。
-3. 切换切片字幕轨，确认时间从 0 开始；人工修改后更新原片，切片应显示待同步而不是被覆盖。
-4. 分别导入、导出 SRT/VTT/ASS；ASS 按规范以 10ms 为单位，SRT/VTT 与数据库保持毫秒精度。
-5. PR 4 完成前不要用本页做自动批量烧录或真实发送；当前“自动加字幕”仍是旧同步兼容入口。
+1. 新建一个低风险全自动测试任务并完成切片；任务应停在“字幕待审核”，发送中心不能提前出现该任务。
+2. 打开字幕工作台，修改并审核字幕后点击“审核并批量烧录”；观察队列进度，完成后确认带字幕视频可播放，任务随后恢复生成文案。
+3. 另建测试任务，在暂停页明确点击“跳过字幕，使用原片继续”；发送中心应显示原片来源，不应误用旧字幕文件。
+4. 选中少量 cue 请求 AI 建议时会调用已配置的文本 AI 并产生额度；先检查 Provider，再确认差异只包含文字，人工接受前 active revision 不变。
+5. 取消一条正在烧录的测试任务，确认状态进入已取消、旧 active 字幕仍可播放、刷新后可以重试。不要用真实待发布内容做这一项。
 
 ## 2026-08-23 v2.1.0 主线同步后检查
 

app/db/database.py

@@ -44,6 +44,12 @@ def init_db() -> None:
             settings.data_dir / "backups",
             "subtitle-editor-rebuild",
         )
+    if _requires_subtitle_auto_schema_migration(settings.database_path):
+        create_schema_migration_backup(
+            settings.database_path,
+            settings.data_dir / "backups",
+            "subtitle-auto-workflow",
+        )
 
     with get_connection() as connection:
         connection.executescript(
@@ -181,17 +187,23 @@ def init_db() -> None:
                 task_id TEXT NOT NULL,
                 output_clip_id TEXT NOT NULL,
                 revision_id TEXT,
+                workflow_job_id TEXT,
                 style_preset_id TEXT,
                 status TEXT NOT NULL DEFAULT 'pending',
                 subtitle_file_path TEXT,
                 output_file_path TEXT,
                 error_message TEXT,
+                validation_status TEXT NOT NULL DEFAULT 'legacy_unverified',
+                validation_json TEXT NOT NULL DEFAULT '{}',
+                encoder TEXT NOT NULL DEFAULT '',
+                verified_at TEXT,
                 is_active INTEGER NOT NULL DEFAULT 1,
                 created_at TEXT NOT NULL,
                 updated_at TEXT NOT NULL,
                 FOREIGN KEY(task_id) REFERENCES tasks(id),
                 FOREIGN KEY(output_clip_id) REFERENCES output_clip(id),
                 FOREIGN KEY(revision_id) REFERENCES subtitle_revisions(id),
+                FOREIGN KEY(workflow_job_id) REFERENCES workflow_jobs(id),
                 FOREIGN KEY(style_preset_id) REFERENCES subtitle_style_presets(id)
             );
 
@@ -561,6 +573,26 @@ def _requires_subtitle_editor_schema_migration(database_path) -> bool:
     )
 
 
+def _requires_subtitle_auto_schema_migration(database_path) -> bool:
+    """已有字幕任务缺少异步渲染验证字段时，迁移前先备份。"""
+    if not database_path.exists() or database_path.stat().st_size == 0:
+        return False
+    connection = None
+    try:
+        connection = sqlite3.connect(f"{database_path.resolve().as_uri()}?mode=ro", uri=True, timeout=10)
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(subtitle_jobs)").fetchall()}
+    finally:
+        if connection:
+            connection.close()
+    return bool(columns) and not {
+        "workflow_job_id",
+        "validation_status",
+        "validation_json",
+        "encoder",
+        "verified_at",
+    } <= columns
+
+
 def _get_table_columns(connection: sqlite3.Connection, table_name: str) -> set[str]:
     rows = connection.execute(f"PRAGMA table_info({table_name})").fetchall()
     return {row["name"] for row in rows}
@@ -584,6 +616,7 @@ def _create_indexes(connection: sqlite3.Connection) -> None:
         "CREATE INDEX IF NOT EXISTS idx_clip_feedback_task_clip ON clip_feedback(task_id, clip_candidate_id)",
         # 字幕任务（按任务、输出切片、状态）
         "CREATE INDEX IF NOT EXISTS idx_subtitle_jobs_task_output_status ON subtitle_jobs(task_id, output_clip_id, status)",
+        "CREATE INDEX IF NOT EXISTS idx_subtitle_jobs_workflow_job ON subtitle_jobs(workflow_job_id)",
         "CREATE INDEX IF NOT EXISTS idx_subtitle_tracks_task_type ON subtitle_tracks(task_id, track_type, is_active)",
         "CREATE UNIQUE INDEX IF NOT EXISTS uq_subtitle_tracks_active_source ON subtitle_tracks(task_id) WHERE track_type = 'source' AND is_active = 1",
         "CREATE INDEX IF NOT EXISTS idx_subtitle_revisions_track_created ON subtitle_revisions(track_id, created_at)",
@@ -724,10 +757,12 @@ def _migrate_tasks_table(connection: sqlite3.Connection) -> None:
         UPDATE tasks SET status = 'failed' WHERE status IN ('失败', 'failed');
         UPDATE tasks SET status = 'pending_video' WHERE status NOT IN (
             'CREATED', 'PREPARING_SOURCE', 'TRANSCRIBING', 'AI_ANALYZING',
-            'CLIP_SELECTING', 'VIDEO_CUTTING', 'METADATA_GENERATING',
+            'CLIP_SELECTING', 'VIDEO_CUTTING', 'SUBTITLE_DRAFTING',
+            'PENDING_SUBTITLE_REVIEW', 'METADATA_GENERATING',
             'SCHEDULE_CREATING', 'PUBLISH_JOB_CREATING', 'READY_TO_PUBLISH',
             'COMPLETED', 'FAILED_PREPARING_SOURCE', 'FAILED_TRANSCRIBING',
             'FAILED_AI_ANALYZING', 'FAILED_CLIP_SELECTING', 'FAILED_VIDEO_CUTTING',
+            'FAILED_SUBTITLE_DRAFTING',
             'FAILED_METADATA_GENERATING', 'FAILED_SCHEDULE_CREATING',
             'FAILED_PUBLISH_JOB_CREATING',
             'pending_video', 'pending_processing', 'audio_extracting', 'transcribing',
@@ -901,11 +936,16 @@ def _migrate_subtitle_jobs_table(connection: sqlite3.Connection) -> None:
         "task_id": "ALTER TABLE subtitle_jobs ADD COLUMN task_id TEXT",
         "output_clip_id": "ALTER TABLE subtitle_jobs ADD COLUMN output_clip_id TEXT",
         "revision_id": "ALTER TABLE subtitle_jobs ADD COLUMN revision_id TEXT",
+        "workflow_job_id": "ALTER TABLE subtitle_jobs ADD COLUMN workflow_job_id TEXT",
         "style_preset_id": "ALTER TABLE subtitle_jobs ADD COLUMN style_preset_id TEXT",
         "status": "ALTER TABLE subtitle_jobs ADD COLUMN status TEXT NOT NULL DEFAULT 'pending'",
         "subtitle_file_path": "ALTER TABLE subtitle_jobs ADD COLUMN subtitle_file_path TEXT",
         "output_file_path": "ALTER TABLE subtitle_jobs ADD COLUMN output_file_path TEXT",
         "error_message": "ALTER TABLE subtitle_jobs ADD COLUMN error_message TEXT",
+        "validation_status": "ALTER TABLE subtitle_jobs ADD COLUMN validation_status TEXT NOT NULL DEFAULT 'legacy_unverified'",
+        "validation_json": "ALTER TABLE subtitle_jobs ADD COLUMN validation_json TEXT NOT NULL DEFAULT '{}'",
+        "encoder": "ALTER TABLE subtitle_jobs ADD COLUMN encoder TEXT NOT NULL DEFAULT ''",
+        "verified_at": "ALTER TABLE subtitle_jobs ADD COLUMN verified_at TEXT",
         "created_at": "ALTER TABLE subtitle_jobs ADD COLUMN created_at TEXT NOT NULL DEFAULT ''",
         "updated_at": "ALTER TABLE subtitle_jobs ADD COLUMN updated_at TEXT NOT NULL DEFAULT ''",
         "is_active": "ALTER TABLE subtitle_jobs ADD COLUMN is_active INTEGER NOT NULL DEFAULT 1",

app/models/subtitle.py

@@ -64,6 +64,15 @@ class SubtitleAIRevisionRequest(BaseModel):
     instructions: str = Field(default="", max_length=2000)
 
 
+class SubtitleAISuggestionAcceptRequest(BaseModel):
+    base_revision_id: str = Field(min_length=1, max_length=64)
+    cue_ids: list[str] = Field(min_length=1, max_length=500)
+
+
+class SubtitleTaskRenderRequest(BaseModel):
+    approve_active_revisions: bool = True
+
+
 class SubtitleStyleExtendedUpdate(BaseModel):
     font_family: str = Field(default="Microsoft YaHei", min_length=1, max_length=120)
     font_size: int = Field(default=42, ge=12, le=160)

app/models/task.py

@@ -11,6 +11,8 @@ class TaskStatus(str, Enum):
     AI_ANALYZING = "AI_ANALYZING"
     CLIP_SELECTING = "CLIP_SELECTING"
     VIDEO_CUTTING = "VIDEO_CUTTING"
+    SUBTITLE_DRAFTING = "SUBTITLE_DRAFTING"
+    PENDING_SUBTITLE_REVIEW = "PENDING_SUBTITLE_REVIEW"
     METADATA_GENERATING = "METADATA_GENERATING"
     SCHEDULE_CREATING = "SCHEDULE_CREATING"
     PUBLISH_JOB_CREATING = "PUBLISH_JOB_CREATING"
@@ -21,6 +23,7 @@ class TaskStatus(str, Enum):
     FAILED_AI_ANALYZING = "FAILED_AI_ANALYZING"
     FAILED_CLIP_SELECTING = "FAILED_CLIP_SELECTING"
     FAILED_VIDEO_CUTTING = "FAILED_VIDEO_CUTTING"
+    FAILED_SUBTITLE_DRAFTING = "FAILED_SUBTITLE_DRAFTING"
     FAILED_METADATA_GENERATING = "FAILED_METADATA_GENERATING"
     FAILED_SCHEDULE_CREATING = "FAILED_SCHEDULE_CREATING"
     FAILED_PUBLISH_JOB_CREATING = "FAILED_PUBLISH_JOB_CREATING"

app/routers/subtitles.py

@@ -6,12 +6,23 @@
 
 from app.models.subtitle import (
     SubtitleApproveRequest,
+    SubtitleAISuggestionAcceptRequest,
+    SubtitleAIRevisionRequest,
     SubtitleOperationsRequest,
     SubtitleRevisionCreate,
+    SubtitleTaskRenderRequest,
     SubtitleSyncRequest,
 )
+from app.services import job_service
+from app.services.subtitle_ai_service import generate_subtitle_suggestions
+from app.services.subtitle_auto_workflow_service import (
+    enqueue_task_subtitle_render,
+    prepare_task_subtitle_review,
+    skip_task_subtitles_and_resume,
+)
 from app.services.subtitle_data_service import (
     SubtitleRevisionConflict,
+    accept_suggestion_revision,
     apply_revision_operations,
     approve_revision,
     create_manual_revision,
@@ -40,6 +51,37 @@ def generate_source_track(task_id: str, payload: SubtitleSyncRequest) -> dict:
     return _call(lambda: {"track": ensure_source_track(task_id, force=payload.force)})
 
 
+@router.post("/tasks/{task_id}/prepare-review")
+def prepare_review(task_id: str) -> dict:
+    return _call(lambda: prepare_task_subtitle_review(task_id))
+
+
+@router.post("/tasks/{task_id}/approve-and-render")
+def approve_and_render(task_id: str, payload: SubtitleTaskRenderRequest) -> dict:
+    return _call(
+        lambda: enqueue_task_subtitle_render(
+            task_id,
+            approve_active_revisions=payload.approve_active_revisions,
+            continue_pipeline=True,
+        )
+    )
+
+
+@router.post("/tasks/{task_id}/skip-and-resume")
+def skip_and_resume(task_id: str) -> dict:
+    return _call(lambda: skip_task_subtitles_and_resume(task_id))
+
+
+@router.get("/tasks/{task_id}/jobs")
+def list_subtitle_jobs(task_id: str) -> dict:
+    jobs = [
+        job
+        for job in job_service.list_jobs(task_id=task_id)
+        if job.get("job_type") == job_service.JOB_TYPE_SUBTITLE
+    ]
+    return {"task_id": task_id, "jobs": jobs, "count": len(jobs)}
+
+
 @router.get("/tracks/{track_id}")
 def read_track(track_id: str) -> dict:
     return _call(lambda: {"track": get_track(track_id)})
@@ -109,6 +151,36 @@ def sync_source(track_id: str, payload: SubtitleSyncRequest) -> dict:
     return _call(lambda: {"track": sync_clip_track(track_id, force=payload.force)})
 
 
+@router.post("/tracks/{track_id}/ai-suggestions")
+def ai_suggestions(track_id: str, payload: SubtitleAIRevisionRequest) -> dict:
+    return _call(
+        lambda: generate_subtitle_suggestions(
+            track_id,
+            revision_id=payload.revision_id,
+            cue_ids=payload.cue_ids,
+            instructions=payload.instructions,
+        )
+    )
+
+
+@router.post("/tracks/{track_id}/ai-suggestions/{suggestion_revision_id}/accept")
+def accept_ai_suggestions(
+    track_id: str,
+    suggestion_revision_id: str,
+    payload: SubtitleAISuggestionAcceptRequest,
+) -> dict:
+    return _call(
+        lambda: {
+            "revision": accept_suggestion_revision(
+                track_id,
+                suggestion_revision_id=suggestion_revision_id,
+                base_revision_id=payload.base_revision_id,
+                cue_ids=payload.cue_ids,
+            )
+        }
+    )
+
+
 @router.post("/tracks/{track_id}/import")
 async def import_subtitle(
     track_id: str,

app/routers/tasks.py

@@ -441,7 +441,14 @@ async def resume_auto_pipeline(task_id: str, background_tasks: BackgroundTasks)
 @router.post("/{task_id}/output-clips/{output_clip_id}/subtitles")
 async def render_output_clip_subtitles(task_id: str, output_clip_id: str) -> dict:
     try:
-        return task_service.render_subtitles_for_output_clip(task_id, output_clip_id)
+        from app.services.subtitle_auto_workflow_service import enqueue_task_subtitle_render
+
+        return enqueue_task_subtitle_render(
+            task_id,
+            output_clip_ids=[output_clip_id],
+            approve_active_revisions=False,
+            continue_pipeline=False,
+        )
     except ValueError as exc:
         raise HTTPException(status_code=400, detail=str(exc)) from exc
     except RuntimeError as exc:

app/services/ai/ai_clip_analyzer.py

@@ -624,6 +624,11 @@ def _loads_ai_json(raw_text: str) -> Any:
     raise AIAnalysisError("JSON 解析失败：AI 没有返回可识别的 JSON 内容")
 
 
+def loads_ai_json(raw_text: str) -> Any:
+    """供不同 AI 工作流复用同一套容错 JSON 解析。"""
+    return _loads_ai_json(raw_text)
+
+
 def _iter_json_candidates(raw_text: str) -> list[str]:
     cleaned = (raw_text or "").strip().lstrip("\ufeff")
     candidates: list[str] = []
@@ -886,5 +891,6 @@ def result_to_jsonable(result: AIClipAnalysisResult) -> dict[str, Any]:
     "build_provider",
     "build_remote_provider",
     "inspect_local_analysis_plan",
+    "loads_ai_json",
     "result_to_jsonable",
 ]