diff --git a/.codemap/modules.json b/.codemap/modules.json index da5c3d5..19644c9 100644 --- a/.codemap/modules.json +++ b/.codemap/modules.json @@ -631,86 +631,51 @@ ], "loc": 2409, "contentHash": "fb31a5d9516731fe561cbfdeaae24f92b052ca23194294b4972664716f93d32e", - "score": 70, + "score": 69, "grade": "C", "tags": [ "fallback", "silent-except", "legacy", + "duplication", "bloat", - "god-component", - "monkeypatch" + "god-component" ], "findings": [ { "sev": "MED", - "loc": "app/services/subtitle_data_service.py:352", - "text": "手工 revision 的 active/base 检查在事务外，并发编辑可覆盖 active 选择。" + "loc": "app/services/subtitle_auto_workflow_service.py:42", + "text": "prepare_task_subtitle_review 先提交 ensure_source_track/ensure_clip_track 的数据库副作用，之后才在 BEGIN IMMEDIATE 中检查活动 Job、lease 和 Task 状态。并发状态变化或冲突 Job 会使入口失败，但已生成的字幕 revision/track 变更无法回滚。" }, { "sev": "MED", - "loc": "app/services/subtitle_data_service.py:489", - "text": "批准 revision 在事务外校验，并发或重放旧请求可回退 active 版本。" + "loc": "app/services/subtitle_data_service.py:1222", + "text": "_sync_dependent_clip_tracks 逐条调用 sync_clip_track、逐条提交。源轨 revision 已提交后，任一切片同步失败会留下部分 up_to_date、部分 pending_sync 的状态。" }, { "sev": "MED", - "loc": "app/services/subtitle_data_service.py:135", - "text": "source/clip track 先查后插，NULL output_clip_id 约束不足，并发可能重复源轨。" + "loc": "app/services/subtitle_data_service.py:1017", + "text": "_load_source_cues 对 checksum 不匹配、JSON 损坏或非法 segment 直接跳过；只要仍有有效 cue 就返回部分 ASR 字幕，可能生成静默缺句的 revision。" }, { "sev": "MED", - "loc": "app/services/subtitle_auto_workflow_service.py:87", - "text": "批量批准逐 clip 独立提交，Job 后创建；中途失败会留下部分批准。" - }, - { - "sev": "MED", - "loc": "app/services/subtitle_data_service.py:168", - "text": "source revision 提交后逐 clip 独立同步，异常会形成混合版本。" - }, - { - "sev": "MED", - "loc": "app/services/subtitle_workflow_service.py:475", - "text": "字幕 job 完成与激活分两次写入，迟到 worker 仍可能激活旧成片。" - }, - { - "sev": "LOW", - "loc": "app/services/subtitle_auto_workflow_service.py:273", - "text": "auto_config_json 损坏时静默降为空并写回，可能丢弃其他配置。" - }, - { - "sev": "MED", - "loc": "app/services/subtitle_data_service.py:1045", - "text": "字幕数据服务聚合 track、revision、cue、导入导出、波形和渲染辅助，回归半径大。" - }, - { - "sev": "LOW", - "loc": "app/services/subtitle_workflow_service.py:312", - "text": "统一 revision 外仍保留旧调用方适配，存在历史口径漂移成本。" + "loc": "app/services/subtitle_data_service.py:1341", + "text": "_probe_media_dimensions 在 ffprobe 不存在、超时、输出损坏或尺寸缺失时静默回退 1080x1920，真实媒体故障不可见。" }, { "sev": "MED", - "loc": "tests/test_subtitle_editor.py:189", - "text": "缺并发 save/approve/ensure track 与批处理中途失败测试。" - }, - { - "sev": "LOW", - "loc": "tests/test_subtitle_auto_workflow.py:172", - "text": "核心真实渲染在 FFmpeg/FFprobe 缺失时会跳过，异常 Provider 证据不足。" + "loc": "app/services/subtitle_data_service.py:1322", + "text": "字幕数据层仍集中承担 revision/cue 持久化、导入导出、ASS、波形与 ffprobe，约 1434 行；默认样式加载存在重复实现，修改影响面偏大。" }, { "sev": "LOW", - "loc": "app/services/subtitle_data_service.py:761", - "text": "波形处理把完整 PCM 捕获到内存，超长媒体存在时长线性内存峰值。" - }, - { - "sev": "LOW", - "loc": "app/services/subtitle_data_service.py:1024", - "text": "单条字幕文本可接近文件上限，放大渲染、导出和 Prompt 资源消耗。" + "loc": "app/services/subtitle_workflow_service.py:237", + "text": "旧版 _activate_subtitle_job 仍可无 lease/revision/status 校验激活记录；正常生产渲染路径已不再调用，但遗留兼容入口仍有风险。" } ], "auditedHash": "fb31a5d9516731fe561cbfdeaae24f92b052ca23194294b4972664716f93d32e", - "auditedAt": "2026-08-24", - "auditedRev": "9ae8440d80d0527c7dd296fe1331b63aac2818ea" + "auditedAt": "2026-08-25", + "auditedRev": "2a2774dd248cc80377dd7edf8e1f98e8ec2f4a20" }, { "id": "pipeline_queue", @@ -742,30 +707,29 @@ ], "loc": 3350, "contentHash": "79bcc0b7de14b28318a7578e71d5281909254180ff351895b72fb10cefb87c68", - "score": 88, + "score": 87, "grade": "B", "tags": [ "god-component", "bloat", - "legacy", "glue", - "over-fit" + "legacy" ], "findings": [ { "sev": "LOW", - "loc": "app/services/pipeline_engine.py:977-1073", - "text": "PUBLISH_JOB_CREATING 尚无专用 reconcile 分支，恢复主要依赖重复创建的幂等去重。" + "loc": "app/services/pipeline_engine.py:1135", + "text": "PUBLISH_JOB_CREATING 重启恢复已具备 source、workflow_job_id、active output 和文件证据校验，但 provisional 仍把全部 recovered_ids 归入 created，skipped 恒为空；混合批次恢复后的计数和取消语义可能与原批次不同。" }, { - "sev": "LOW", - "loc": "tests/test_pipeline_checkpoint.py:358-625; tests/test_pipeline_state_stability.py:275-313", - "text": "仍缺真实进程重启、lease 过期边界及发布内容证据变更的进程级回归测试。" + "sev": "MED", + "loc": "app/services/pipeline_engine.py:87", + "text": "PipelineEngine 约1810行、46个方法，继续集中9步执行、checkpoint/reconcile、发布证据、取消清理和失败收口；lease、旧checkpoint、切片与发布协议高度耦合，维护及回归面仍偏大。" } ], "auditedHash": "79bcc0b7de14b28318a7578e71d5281909254180ff351895b72fb10cefb87c68", - "auditedAt": "2026-08-24", - "auditedRev": "" + "auditedAt": "2026-08-25", + "auditedRev": "2a2774dd248cc80377dd7edf8e1f98e8ec2f4a20" }, { "id": "publish_center", diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index 28d7a6e..7168294 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1225,3 +1225,15 @@ - Windows 子进程树终止现在校验 `taskkill` 退出码并等待进程退出；超时、启动失败、非零退出和无法确认退出都会显式报错，不再让上层误以为已经安全停止。 - 移除路径测试中的全局模块 reload 泄漏，并让边界测试始终引用当前运行时 `storage_service`，消除全量测试顺序依赖。 - 独立验收：边界专项 `52 passed`，全量 `533 passed`；Ruff、Python Compileall 和 `git diff --check` 全部通过。未访问真实 NAS 内容，未触发 AI 或真实投稿。 + +## 2026-08-25 稳定 V1 P1.3d 字幕批次原子性与恢复 + +- 字幕“审核并批量烧录”改为一个 `BEGIN IMMEDIATE` 事务：所有当前 active revision 在事务内重新校验，批准、创建/复用 Workflow Job 和 `subtitle_delivery_mode` 同时提交；任一 revision、配置或 Job 插入失败时整批回滚。 +- 单条批准也在写锁内重新核对 active revision，过期页面请求不能把字幕轨回退到旧版本；损坏的 `auto_config_json` 改为明确失败，不再静默覆盖为 `{}`。 +- 字幕成片的临时文件切换、completed/verified 写回和 active 版本切换合并为一次短事务，并绑定当前 Workflow Job 的 owner、lease token、未过期时间和取消状态；旧 Worker 或旧 revision 不能激活迟到结果。 +- 新 Worker 接管时会收口同一 Workflow Job 遗留的 processing 子任务，只清理本执行标记的 `.part.mp4` 和没有数据库引用的中断最终文件；其他 Job、已验证 active 文件、外部文件和历史 revision 均保留。 +- 若进程在“字幕 DB 已提交、checkpoint 尚未写入”的窄窗口退出，重启会从同一 Workflow Job 的 active + verified 结果恢复 checkpoint，不重复运行 FFmpeg。 +- 字幕完成与后续自动流水线 Job 改为同事务提交；取消、过期 lease 或不兼容的续跑 payload 会整笔回滚。发布草稿批次在提交前再次核验 lease，并可从同一 Workflow Job 的持久化草稿证据恢复 checkpoint。 +- 源轨生成、切片轨同步、手工保存和字幕导入均在写锁内重读 active revision 并使用条件更新；迟到的旧读取不能覆盖较新的人工版本。“跳过字幕”也会拒绝已经排队的后续自动流水线。 +- Worker 子进程启动失败会明确写入 failed 而不是让队列线程退出；重启接管同时收口 processing/queued 字幕子记录，按 revision 复用的 ASS 缓存不会被误删。 +- 最终独立验收 Ruff、Compileall 通过，字幕/Job fencing/队列/自动流水线/checkpoint/状态机/发布关联/版本回滚共 `167 passed`。Pytest 明确使用临时 `test_workflow.sqlite3`；活动库前后 size 与 SHA256 完全一致。未调用 AI、Chrome 或真实平台投稿。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index 6932184..51277e9 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -20,7 +20,7 @@ - [x] P1.3b：自动流水线已增加版本化步骤 checkpoint、输入/产物证据、旧 Job 原位重试和重领恢复；已覆盖 AI、转写、切片、文案、排期与发布草稿的失效边界。 - [x] P1.3c：已增加 `schema_migrations` 版本/checksum 账本、迁移前 Online Backup 和 fail-closed v2 活动发布唯一索引；重复数据、索引漂移或账本异常都会拒绝启动。本轮未直接迁移活动库。 - [x] 任务 20 专项：完成状态恢复、重复 AI 409 保护、上传单入口、最近 5 个任务名、长直播参数显隐和“跳过字幕后进入片段审核”闭环。 -- [ ] P1.3d：处理字幕批准批次原子性、字幕/切片中断目录清理和剩余跨进程恢复边界。 +- [x] P1.3d：字幕批准、Job 创建和交付模式已原子提交；成片激活绑定 lease/revision；接管会清理本执行残留并恢复“DB 已提交但 checkpoint 未写”的结果。 - [ ] P1.4：统一第三方 AI/FFmpeg 超时、错误 JSON、429/5xx 与重试幂等边界，并避免重复计费。 - [ ] P1.5：在不扩大个人本地项目范围的前提下处理密钥日志、输入校验和本地管理员接口门禁。 - [ ] P2：拆分 God Service、去除查询重复、补核心集成/故障测试和可观测性；不做全面重构。 @@ -34,6 +34,8 @@ 5. P1B.1/P1B.2、P1.3a 与 P1.3b 代码及隔离测试均已完成；活动库尚未增加 `workflow_jobs.lease_token`，本轮也未重启正式服务。下一独立轮先确认没有活动 Workflow Job，再做在线备份、幂等迁移和正式重启烟测。 6. P1.3b 不增加数据库列。合并并完成正式迁移/重启后，可用一条低风险测试任务在 AI 或切片完成后重启 Worker，确认同一 Job 从 checkpoint 续跑；不要点击真实发布，也不要用重要待发布素材做故障演练。 7. P1.3c 对正式库只做了只读预检：当前无活动发布重复组，旧唯一索引仍在，尚未写入 `schema_migrations`。合并后的首次正式重启会先创建迁移备份再切换 v2 索引；应与尚未部署的 Workflow fencing Schema 一起安排受控迁移和重启验收，不要在服务运行中手工改索引。 +8. P1.3d 不增加数据库列，也不需要手工清理 `06_subtitled`。合并并正式重启后，可用一条低风险测试任务执行“审核并批量烧录”；不要通过强杀正式服务制造故障，跨进程接管已由隔离 lease 回归验证。 +9. 下一轮进入 P1.4，优先统一 AI/FFmpeg/FFprobe 的超时、429/5xx、坏 JSON 和持久化调用幂等；在该轮完成前，不要把 Provider 自动重试次数调高。 ## 2026-08-24 工程体检确认的原始整改顺序 diff --git a/PROJECT_AUDIT.md b/PROJECT_AUDIT.md index cd19321..4804154 100644 --- a/PROJECT_AUDIT.md +++ b/PROJECT_AUDIT.md @@ -109,6 +109,23 @@ Codemap 独立复评结果：`Publish Scheduler 62→71（C）`、`Publishers & Codemap 独立复评将 `SQLite Persistence` 从 **64/C 提升到 73/C**，原 HIGH“先删索引、重建失败静默吞掉”和 MED“没有迁移账本”均已移除。模块仍为 C：历史列探测和多处 `executescript` 继续属于 pre-ledger compatibility，不能保证整个旧库升级原子；备份恢复工具还缺 FK/ledger/关键索引验证与运行中服务闸门。P1.3c 已完成有限范围目标，但项目整体仍维持 **可用 V1**，下一轮按路线处理 P1.3d，而不是借机全面重写数据库层。 +## 0.7 P1.3d 字幕批次与跨进程恢复状态（2026-08-25） + +| P1.3d 项目 | 结果 | 验证证据 | +| --- | --- | --- | +| 批量批准原子性 | 已封口 | active revision、cue/质量、批准、Workflow Job 创建/复用和字幕交付模式在同一个 `BEGIN IMMEDIATE` 中提交；任一条失败整批回滚。 | +| Revision 并发覆盖 | 已封口 | 手工保存、单条批准、原片轨生成、切片轨同步和字幕导入均在写锁内重读 current active revision 并使用条件更新；旧页面或迟到执行不能把 active 指针回退。 | +| 成片激活与旧 Worker | 已封口 | 临时文件切换、completed/verified 和 active 版本切换绑定当前 workflow owner/token/expiry/cancel 与当前已批准 revision；迟到结果不能覆盖新版本。 | +| 中断目录与子记录 | 已封口 | 接管会收口同一 Workflow Job 的 `processing/queued` 子记录，只清理本次标记的 `.part.mp4` 和无数据库引用的孤儿最终视频；其他 Job、active 文件、历史结果和可复用 ASS 不删除。 | +| 跨进程 checkpoint | 已封口 | “字幕 DB 已提交但 checkpoint 未写”可恢复；字幕完成与后续 auto-pipeline Job 同事务提交；发布草稿已提交但 checkpoint 未写时按 schedule、任务目录和 workflow 证据恢复。 | +| Lease 与进程启动 | 已封口 | 所有通用 Job 写回拒绝过期 lease；发布批次提交前二次验 lease；子进程启动失败明确标记 failed，不再让 Worker 线程无声退出。 | +| 路径和证据 | 已封口（自动流水线） | Workflow 创建发布草稿前，视频/封面必须位于当前任务受控目录、扩展名合法且非空，并持久化 size/fingerprint，坏草稿不会先提交再被去重掩盖。 | +| 隔离回归 | 通过 | Ruff、Compileall 通过；10 组字幕/Job/自动流水线/checkpoint/状态机/发布关联/版本回滚测试 `167 passed`。活动库前后 size 与 SHA256 一致。 | + +Codemap 最终独立复评未再发现 P1.3d 新的 HIGH。`Subtitle` 为 **69/C**：本轮的批准、激活、接管和 revision 覆盖风险已封口，剩余中风险主要是源轨向多个切片逐条同步的部分完成、损坏 ASR chunk 静默跳过、FFprobe 画布 fallback 以及 1400 行 God Service；前两项中的输入/探测故障归入 P1.4，结构拆分归入 P2。`Pipeline & Job Queue` 为 **87/B**，核心 lease/follow-up/发布恢复与创建时媒体/文案证据边界已经收口；剩余问题主要是恢复计数语义和 God Component 拆分，均不值得在本轮扩大重构。 + +本轮没有增加数据库列、没有迁移活动库、没有调用 AI Provider、Chrome 或真实投稿。项目成熟度继续维持 **可用 V1**：P1.3d 已解决“字幕正常时能跑、进程切换时可能重复或覆盖”的问题，但 P1.4 的第三方超时/错误响应/计费幂等和 P1.5 的密钥/输入/XSS/管理员门禁仍是改判“稳定 V1”的必要条件。 + ## 1. Executive Summary ### 结论 diff --git a/app/services/auto_publish_service.py b/app/services/auto_publish_service.py index 841dd88..44d10f3 100644 --- a/app/services/auto_publish_service.py +++ b/app/services/auto_publish_service.py @@ -3,15 +3,22 @@ from __future__ import annotations import json +from pathlib import Path from uuid import uuid4 from app.core.config import settings from app.db.database import get_connection +from app.services import job_service from app.services.publish_copy_rules import PUBLISH_COPY_RULE_VERSION from app.services.publish_service import DEFAULT_BILIBILI_TID, USER_REMOVED_ERROR_CODE, get_publish_job from app.services.publish_domain import AUTO_PUBLISH_PLATFORMS, validate_publish_mode, validate_target_platform -from app.services.storage_service import resolve_video_file_path +from app.services.storage_service import ( + IMAGE_EXTENSIONS, + resolve_task_media_file_path, + resolve_video_file_path, +) from app.services.task_service import _now_iso +from app.services.transcription_checkpoint_service import fingerprint_file def platforms_for_task(task: dict) -> list[str]: @@ -35,18 +42,66 @@ def create_auto_publish_jobs( raise ValueError("字幕交付模式必须是 subtitled 或 original") created_ids: list[str] = [] skipped_ids: list[str] = [] - now = _now_iso() + lease = job_service.current_job_lease() if workflow_job_id else None + if workflow_job_id and (not lease or lease[0] != workflow_job_id): + raise job_service.JobLeaseLostError(f"发布草稿创建缺少当前 Workflow Job 租约：{workflow_job_id}") with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + if workflow_job_id: + active_lease = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE id = ? AND status = 'running' AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') + AND cancel_requested = 0 + """, + (workflow_job_id, lease[1], lease[2]), + ).fetchone() + if not active_lease: + connection.rollback() + raise job_service.JobLeaseLostError( + f"发布草稿创建前 Workflow Job 租约已失效：{workflow_job_id}" + ) for item in scheduled_items: output_clip = item["output_clip"] - video_source, video_file_path, subtitle_evidence = _resolve_auto_video_source( + video_source, video_file_path, resolved_video, subtitle_evidence = _resolve_auto_video_source( + task, output_clip, subtitle_delivery_mode, + require_managed_path=bool(workflow_job_id), ) metadata = item["metadata"] cover = item.get("cover") or {} platform = validate_target_platform(metadata["platform"]) publish_mode = validate_publish_mode(settings.publish_default_mode) + cover_file_path = str(cover.get("cover_file_path") or "").strip() + if not cover_file_path: + raise ValueError(f"{output_clip.get('id') or '未知切片'} 没有生成封面，已停止创建不完整的发布任务") + resolved_cover = ( + resolve_task_media_file_path( + cover_file_path, + task_id=str(task["id"]), + task_dir_name=str(task.get("task_dir_name") or "") or None, + allowed_subdirectories=("07_covers",), + allowed_extensions=IMAGE_EXTENSIONS, + ) + if workflow_job_id + else resolve_video_file_path(cover_file_path) + ) + if ( + resolved_cover is None + or not resolved_cover.exists() + or not resolved_cover.is_file() + or resolved_cover.stat().st_size <= 0 + ): + raise ValueError(f"{output_clip.get('id') or '未知切片'} 的封面文件无效或不在受控目录") + cover_file_path = str(resolved_cover) + scheduled_at = str(item.get("scheduled_at") or "").strip() + if scheduled_at: + from app.services.publish_time import to_utc_iso + + scheduled_at = to_utc_iso(scheduled_at, settings.app_timezone) latest = connection.execute( """ SELECT id, status, error_code @@ -64,9 +119,14 @@ def create_auto_publish_jobs( ): skipped_ids.append(latest["id"]) continue + if latest and str(latest["status"] or "").upper() in {"PUBLISHED", "EXPORTED"}: + skipped_ids.append(latest["id"]) + continue existing = connection.execute( """ - SELECT id + SELECT id, video_file_path, cover_file_path, provider_response, + title, description, caption, tags, hashtags, cover_text, + scheduled_at, cover_time_seconds FROM publish_jobs WHERE output_clip_id = ? AND platform = ? AND publish_mode = ? AND status IN ('DRAFT', 'WAITING', 'SCHEDULED', 'PUBLISHING', 'NEED_REVIEW') @@ -76,12 +136,63 @@ def create_auto_publish_jobs( (output_clip["id"], platform, publish_mode), ).fetchone() if existing: + if workflow_job_id: + try: + evidence = json.loads(str(existing["provider_response"] or "{}")) + except json.JSONDecodeError as exc: + raise ValueError("已有发布草稿证据已损坏，拒绝复用") from exc + expected_caption = str(metadata.get("caption") or "") + expected_hashtags = ", ".join(metadata.get("hashtags") or []) + existing_video = resolve_task_media_file_path( + str(existing["video_file_path"] or ""), + task_id=str(task["id"]), + task_dir_name=str(task.get("task_dir_name") or "") or None, + allowed_subdirectories=( + ("06_subtitled",) + if video_source == "subtitled" + else ("05_clips", "clips") + ), + ) + existing_cover = resolve_task_media_file_path( + str(existing["cover_file_path"] or ""), + task_id=str(task["id"]), + task_dir_name=str(task.get("task_dir_name") or "") or None, + allowed_subdirectories=("07_covers",), + allowed_extensions=IMAGE_EXTENSIONS, + ) + if ( + not isinstance(evidence, dict) + or existing_video is None + or existing_cover is None + or not existing_video.is_file() + or not existing_cover.is_file() + or existing_video.resolve() != resolved_video.resolve() + or existing_cover.resolve() != resolved_cover.resolve() + or int(evidence.get("video_file_size") or -1) != existing_video.stat().st_size + or str(evidence.get("video_file_fingerprint") or "") + != fingerprint_file(existing_video) + or int(evidence.get("cover_file_size") or -1) != existing_cover.stat().st_size + or str(evidence.get("cover_file_fingerprint") or "") + != fingerprint_file(existing_cover) + or str(evidence.get("metadata_policy_version") or "") + != PUBLISH_COPY_RULE_VERSION + or str(evidence.get("subtitle_delivery_mode") or "") + != subtitle_delivery_mode + or str(existing["title"] or "") + != str(metadata.get("title") or "精彩片段") + or str(existing["description"] or "") != expected_caption + or str(existing["caption"] or "") != expected_caption + or str(existing["tags"] or "") != expected_hashtags + or str(existing["hashtags"] or "") != expected_hashtags + or str(existing["cover_text"] or "") + != str(metadata.get("cover_text") or "") + or str(existing["scheduled_at"] or "") != scheduled_at + or float(existing["cover_time_seconds"] or 0) + != float(cover.get("cover_time_seconds") or 0) + ): + raise ValueError("已有发布草稿的媒体或文案证据已失效，拒绝复用或重复创建") skipped_ids.append(existing["id"]) continue - - cover_file_path = str(cover.get("cover_file_path") or "").strip() - if not cover_file_path: - raise ValueError(f"{output_clip.get('id') or '未知切片'} 没有生成封面，已停止创建不完整的发布任务") cover_time_seconds = float(cover.get("cover_time_seconds") or 0) account_rows = connection.execute( """ @@ -96,11 +207,6 @@ def create_auto_publish_jobs( if len(account_rows) == 1 and str(account_rows[0]["login_status"] or "") == "normal" else None ) - scheduled_at = str(item.get("scheduled_at") or "").strip() - if scheduled_at: - from app.services.publish_time import to_utc_iso - - scheduled_at = to_utc_iso(scheduled_at, settings.app_timezone) status = "NEED_REVIEW" if metadata.get("risk_flags") else ( "SCHEDULED" if scheduled_at and (publish_mode != "local_browser" or account_id) else "WAITING" ) @@ -119,6 +225,10 @@ def create_auto_publish_jobs( "risk_flags": metadata.get("risk_flags") or [], "publish_mode": publish_mode, "subtitle_delivery_mode": subtitle_delivery_mode, + "video_file_size": int(resolved_video.stat().st_size), + "video_file_fingerprint": fingerprint_file(resolved_video), + "cover_file_size": int(resolved_cover.stat().st_size), + "cover_file_fingerprint": fingerprint_file(resolved_cover), **subtitle_evidence, "note": "全自动流水线已直接创建最终发布任务，可在发送中心设置排期。", } @@ -168,6 +278,21 @@ def create_auto_publish_jobs( ), ) created_ids.append(job_id) + if workflow_job_id: + active_lease = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE id = ? AND status = 'running' AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') + AND cancel_requested = 0 + """, + (workflow_job_id, lease[1], lease[2]), + ).fetchone() + if not active_lease: + connection.rollback() + raise job_service.JobLeaseLostError( + f"发布草稿提交前 Workflow Job 租约已失效：{workflow_job_id}" + ) connection.commit() return { @@ -178,16 +303,40 @@ def create_auto_publish_jobs( } -def _resolve_auto_video_source(output_clip: dict, delivery_mode: str) -> tuple[str, str, dict]: +def _resolve_auto_video_source( + task: dict, + output_clip: dict, + delivery_mode: str, + *, + require_managed_path: bool, +) -> tuple[str, str, Path, dict]: if delivery_mode == "original": raw_path = str(output_clip.get("output_file_path") or "") - path = resolve_video_file_path(raw_path) if raw_path else None - if not path or not path.exists() or not path.is_file(): + path = ( + resolve_task_media_file_path( + raw_path, + task_id=str(task["id"]), + task_dir_name=str(task.get("task_dir_name") or "") or None, + allowed_subdirectories=("05_clips", "clips"), + ) + if require_managed_path + else (resolve_video_file_path(raw_path) if raw_path else None) + ) + if not path or not path.exists() or not path.is_file() or path.stat().st_size <= 0: raise ValueError("原片切片文件不存在，不能创建发布任务") - return "original", raw_path, {"subtitle_skip_confirmed": True} + return "original", str(path), path, {"subtitle_skip_confirmed": True} raw_path = str(output_clip.get("subtitled_output_file_path") or "") - path = resolve_video_file_path(raw_path) if raw_path else None + path = ( + resolve_task_media_file_path( + raw_path, + task_id=str(task["id"]), + task_dir_name=str(task.get("task_dir_name") or "") or None, + allowed_subdirectories=("06_subtitled",), + ) + if require_managed_path + else (resolve_video_file_path(raw_path) if raw_path else None) + ) if ( output_clip.get("subtitle_status") != "completed" or output_clip.get("subtitle_validation_status") != "verified" @@ -195,11 +344,13 @@ def _resolve_auto_video_source(output_clip: dict, delivery_mode: str) -> tuple[s or not path or not path.exists() or not path.is_file() + or path.stat().st_size <= 0 ): raise ValueError("字幕成片尚未同时通过 revision 审核和 FFprobe 验证，不能进入发送中心") return ( "subtitled", - raw_path, + str(path), + path, { "subtitle_revision_id": output_clip.get("subtitle_revision_id") or "", "subtitle_revision_status": output_clip.get("subtitle_revision_status") or "", diff --git a/app/services/job_service.py b/app/services/job_service.py index f345b52..152b47d 100644 --- a/app/services/job_service.py +++ b/app/services/job_service.py @@ -104,7 +104,8 @@ def _lease_write_condition( lease = _resolve_job_lease(lease_owner, lease_token) if lease: return ( - "id = ? AND status = ? AND lease_owner = ? AND lease_token = ?", + "id = ? AND status = ? AND lease_owner = ? AND lease_token = ? " + "AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now')", (JOB_STATUS_RUNNING, lease[0], lease[1]), True, ) @@ -188,56 +189,69 @@ def create_or_get_active_job( 返回值中的布尔值表示是否新建。`BEGIN IMMEDIATE` 将“查询 + 新建” 串行化，避免连续点击或并发请求为同一任务创建重复的切片作业。 """ - resolved_job_id = uuid4().hex[:12] - now = _now_iso() - payload_json = json.dumps(payload or {}, ensure_ascii=False) - with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") - existing = connection.execute( - """ - SELECT id - FROM workflow_jobs - WHERE task_id = ? - AND job_type = ? - AND status IN (?, ?) - ORDER BY created_at DESC - LIMIT 1 - """, - ( - task_id, - job_type, - JOB_STATUS_QUEUED, - JOB_STATUS_RUNNING, - ), - ).fetchone() - if existing: - connection.commit() - return get_job(existing["id"]), False - - connection.execute( - """ - INSERT INTO workflow_jobs ( - id, task_id, job_type, status, progress, message, - payload_json, result_json, error_message, - created_at, updated_at, started_at, finished_at - ) - VALUES (?, ?, ?, ?, 0, ?, ?, '{}', NULL, ?, ?, NULL, NULL) - """, - ( - resolved_job_id, - task_id, - job_type, - JOB_STATUS_QUEUED, - f"{JOB_TYPE_LABELS.get(job_type, job_type)}任务已加入队列", - payload_json, - now, - now, - ), + resolved_job_id, created = create_or_get_active_job_with_connection( + connection, + task_id=task_id, + job_type=job_type, + payload=payload, ) connection.commit() + return get_job(resolved_job_id), created - return get_job(resolved_job_id), True + +def create_or_get_active_job_with_connection( + connection, + *, + task_id: str, + job_type: str, + payload: Optional[dict] = None, +) -> tuple[str, bool]: + """在调用方事务中原子复用或创建活动 Job，不自行 commit。""" + existing = connection.execute( + """ + SELECT id + FROM workflow_jobs + WHERE task_id = ? + AND job_type = ? + AND status IN (?, ?) + ORDER BY created_at DESC + LIMIT 1 + """, + ( + task_id, + job_type, + JOB_STATUS_QUEUED, + JOB_STATUS_RUNNING, + ), + ).fetchone() + if existing: + return str(existing["id"]), False + + resolved_job_id = uuid4().hex[:12] + now = _now_iso() + connection.execute( + """ + INSERT INTO workflow_jobs ( + id, task_id, job_type, status, progress, message, + payload_json, result_json, error_message, + created_at, updated_at, started_at, finished_at + ) + VALUES (?, ?, ?, ?, 0, ?, ?, '{}', NULL, ?, ?, NULL, NULL) + """, + ( + resolved_job_id, + task_id, + job_type, + JOB_STATUS_QUEUED, + f"{JOB_TYPE_LABELS.get(job_type, job_type)}任务已加入队列", + json.dumps(payload or {}, ensure_ascii=False), + now, + now, + ), + ) + return resolved_job_id, True # ── 查询 job ───────────────────────────────────────────────────── @@ -306,12 +320,12 @@ def mark_job_running(job_id: str) -> dict | None: def claim_job(job_id: str, lease_owner: str, lease_seconds: int = 120) -> dict | None: """原子领取一个排队任务，或接管 lease 已过期的运行任务。""" - now = datetime.now(timezone.utc) - now_iso = now.isoformat(timespec="seconds") - lease_expires_at = (now + timedelta(seconds=max(30, lease_seconds))).isoformat(timespec="seconds") lease_token = uuid4().hex with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") + now = datetime.now(timezone.utc) + now_iso = now.isoformat(timespec="seconds") + lease_expires_at = (now + timedelta(seconds=max(30, lease_seconds))).isoformat(timespec="seconds") row = connection.execute( "SELECT status, lease_expires_at, cancel_requested, attempt_count, max_attempts FROM workflow_jobs WHERE id = ?", (job_id,), @@ -352,12 +366,12 @@ def claim_job(job_id: str, lease_owner: str, lease_seconds: int = 120) -> dict | def claim_next_job(lease_owner: str, lease_seconds: int = 120) -> dict | None: """按创建时间领取一个重型任务，保证本地默认串行。""" - now_value = datetime.now(timezone.utc) - now = now_value.isoformat(timespec="seconds") - lease_expires_at = (now_value + timedelta(seconds=max(30, lease_seconds))).isoformat(timespec="seconds") lease_token = uuid4().hex with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") + now_value = datetime.now(timezone.utc) + now = now_value.isoformat(timespec="seconds") + lease_expires_at = (now_value + timedelta(seconds=max(30, lease_seconds))).isoformat(timespec="seconds") connection.execute( """ UPDATE workflow_jobs @@ -437,14 +451,16 @@ def validate_job_lease( *, require_unexpired: bool = True, ) -> dict | None: - now = _now_iso() - expiry_clause = "AND lease_expires_at > ?" if require_unexpired else "" + expiry_clause = ( + "AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now')" + if require_unexpired + else "" + ) params: tuple[str, ...] = ( job_id, JOB_STATUS_RUNNING, lease_owner, lease_token, - *((now,) if require_unexpired else ()), ) with get_connection() as connection: row = connection.execute( @@ -475,9 +491,9 @@ def heartbeat_job( """ UPDATE workflow_jobs SET heartbeat_at = ?, lease_expires_at = ?, updated_at = ? WHERE id = ? AND status = ? AND lease_owner = ? AND lease_token = ? - AND lease_expires_at > ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') """, - (now_iso, expires, now_iso, job_id, JOB_STATUS_RUNNING, lease[0], lease[1], now_iso), + (now_iso, expires, now_iso, job_id, JOB_STATUS_RUNNING, lease[0], lease[1]), ) connection.commit() if cursor.rowcount == 0: @@ -815,6 +831,87 @@ def mark_job_completed( return get_job(job_id) +def mark_job_completed_with_followup( + job_id: str, + result: Optional[dict], + *, + followup_task_id: str, + followup_job_type: str, + followup_payload: Optional[dict] = None, + result_followup_key: str = "followup_job_id", + lease_owner: str | None = None, + lease_token: str | None = None, +) -> tuple[dict, dict, bool]: + """在同一事务内完成当前 Job 并创建/复用后续 Job。""" + lease = _resolve_job_lease(lease_owner, lease_token) + if lease is None: + raise ValueError("完成并创建后续 Job 需要有效的 Workflow Job 租约") + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + active = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE id = ? AND status = ? AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') + AND cancel_requested = 0 + """, + (job_id, JOB_STATUS_RUNNING, lease[0], lease[1]), + ).fetchone() + if not active: + connection.rollback() + raise JobLeaseLostError(f"Workflow Job 完成前租约已失效或已取消：{job_id}") + followup_job_id, created = create_or_get_active_job_with_connection( + connection, + task_id=followup_task_id, + job_type=followup_job_type, + payload=followup_payload, + ) + followup_row = connection.execute( + "SELECT payload_json FROM workflow_jobs WHERE id = ?", + (followup_job_id,), + ).fetchone() + try: + existing_followup_payload = json.loads(followup_row["payload_json"] or "{}") + except (TypeError, json.JSONDecodeError) as exc: + connection.rollback() + raise ValueError("后续 Workflow Job payload 已损坏，拒绝复用") from exc + if existing_followup_payload != (followup_payload or {}): + connection.rollback() + raise ValueError("已有后续 Workflow Job 的执行参数不同，拒绝错误复用") + result_payload = {**(result or {}), result_followup_key: followup_job_id} + cursor = connection.execute( + """ + UPDATE workflow_jobs + SET status = ?, progress = 100, message = '任务已完成', result_json = ?, + finished_at = ?, updated_at = ?, lease_owner = NULL, + lease_token = NULL, lease_expires_at = NULL, heartbeat_at = NULL + WHERE id = ? AND status = ? AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') + AND cancel_requested = 0 + """, + ( + JOB_STATUS_COMPLETED, + json.dumps(result_payload, ensure_ascii=False), + now, + now, + job_id, + JOB_STATUS_RUNNING, + lease[0], + lease[1], + ), + ) + if cursor.rowcount != 1: + connection.rollback() + raise JobLeaseLostError(f"Workflow Job 完成提交冲突：{job_id}") + connection.commit() + completed_job = get_job(job_id) + followup_job = get_job(followup_job_id) + if not completed_job or not followup_job: + raise RuntimeError("Workflow Job 原子提交后无法读取结果") + return completed_job, followup_job, created + + def mark_job_failed( job_id: str, error_message: str, diff --git a/app/services/job_worker.py b/app/services/job_worker.py index 3570099..3a7db0a 100644 --- a/app/services/job_worker.py +++ b/app/services/job_worker.py @@ -70,6 +70,16 @@ def execute_job( except job_service.JobLeaseLostError: raise except Exception as exc: + if job_type == job_service.JOB_TYPE_SUBTITLE: + from app.services.subtitle_auto_workflow_service import cleanup_interrupted_subtitle_job + + cleanup_interrupted_subtitle_job( + job_id, + lease_owner=owner, + lease_token=lease_token, + status="failed", + message=str(exc), + ) job_service.mark_job_failed(job_id, str(exc)) raise @@ -159,6 +169,26 @@ def _execute_subtitle(job_id: str, task_id: str, payload: dict) -> None: if job_service.is_cancel_requested(job_id): job_service.mark_job_cancelled(job_id, "字幕批量烧录已取消") return + if result.get("resume_requested"): + from app.models.task import TaskStatus + + _completed_job, resume_job, created = job_service.mark_job_completed_with_followup( + job_id, + result, + followup_task_id=task_id, + followup_job_type=job_service.JOB_TYPE_AUTO_PIPELINE, + followup_payload={"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + result_followup_key="resume_job_id", + ) + from app.services.task_log_service import append_task_log + + append_task_log( + task_id, + "字幕成片全部验证通过，已排队恢复自动文案与发送中心流程" + if created + else f"字幕成片全部验证通过，已复用自动流水线恢复 Job：{resume_job['id']}", + ) + return job_service.mark_job_completed(job_id, result) @@ -195,12 +225,24 @@ def _run_job_subprocess(self, job_id: str) -> None: lease_token = str(job_before_start.get("lease_token") or "") if job_before_start.get("lease_owner") != self.owner or not lease_token: return - process = popen_process_group( - [sys.executable, "-m", "app.services.job_worker_process", job_id, self.owner, lease_token], - cwd=str(settings.project_root), - stdout=subprocess.DEVNULL, - stderr=subprocess.DEVNULL, - ) + try: + process = popen_process_group( + [sys.executable, "-m", "app.services.job_worker_process", job_id, self.owner, lease_token], + cwd=str(settings.project_root), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + except Exception as exc: + try: + job_service.mark_job_failed( + job_id, + f"无法启动 Job 子进程：{exc}", + lease_owner=self.owner, + lease_token=lease_token, + ) + except job_service.JobLeaseLostError: + pass + return last_heartbeat = 0.0 last_progress_at = time.monotonic() previous_progress: tuple[int, str] | None = None @@ -307,13 +349,8 @@ def _run_job_subprocess(self, job_id: str) -> None: terminate_process_tree(process) return last_heartbeat = time.monotonic() - final_job = job_service.get_job(job_id) - owns_final_lease = bool( - final_job - and final_job.get("status") == job_service.JOB_STATUS_RUNNING - and final_job.get("lease_owner") == self.owner - and final_job.get("lease_token") == lease_token - ) + final_job = job_service.validate_job_lease(job_id, self.owner, lease_token) + owns_final_lease = bool(final_job) if owns_final_lease and int(final_job.get("cancel_requested") or 0): job_service.mark_job_cancelled( job_id, @@ -345,6 +382,22 @@ def _run_job_subprocess(self, job_id: str) -> None: lease_token=lease_token, ) elif process.returncode == 0 and owns_final_lease: + if final_job.get("job_type") == job_service.JOB_TYPE_SUBTITLE: + from app.services.subtitle_auto_workflow_service import cleanup_interrupted_subtitle_job + + try: + job_service.heartbeat_job(job_id, self.owner, lease_token) + except job_service.JobLeaseLostError: + return + cleaned = cleanup_interrupted_subtitle_job( + job_id, + lease_owner=self.owner, + lease_token=lease_token, + status="failed", + message="字幕 Job 子进程正常退出但没有写入终态", + ) + if not cleaned: + return job_service.mark_job_failed( job_id, "Job 子进程已退出但没有写入终态", diff --git a/app/services/pipeline_engine.py b/app/services/pipeline_engine.py index 4b326c9..42d73d2 100644 --- a/app/services/pipeline_engine.py +++ b/app/services/pipeline_engine.py @@ -491,13 +491,17 @@ def _publish_job_evidence(self, task_id: str, job_ids: list[str]) -> list[dict]: with get_connection() as connection: rows = connection.execute( f""" - SELECT id, output_clip_id, platform, publish_mode, video_source, - video_file_path, cover_file_path, scheduled_at, status, - title, description, caption, tags, hashtags, cover_text, - error_code, provider_response - FROM publish_jobs - WHERE task_id = ? AND id IN ({placeholders}) - ORDER BY id + SELECT pj.id, pj.output_clip_id, pj.platform, pj.publish_mode, pj.video_source, + pj.video_file_path, pj.cover_file_path, pj.scheduled_at, pj.status, + pj.title, pj.description, pj.caption, pj.tags, pj.hashtags, pj.cover_text, + pj.error_code, pj.provider_response, + oc.status AS output_status, oc.is_active AS output_is_active, + oc.cut_run_id, cr.status AS cut_run_status, cr.is_active AS cut_run_is_active + FROM publish_jobs pj + LEFT JOIN output_clip oc ON oc.id = pj.output_clip_id AND oc.task_id = pj.task_id + LEFT JOIN cut_runs cr ON cr.id = oc.cut_run_id AND cr.task_id = oc.task_id + WHERE pj.task_id = ? AND pj.id IN ({placeholders}) + ORDER BY pj.id """, (task_id, *normalized_ids), ).fetchall() @@ -514,6 +518,13 @@ def _publish_job_evidence(self, task_id: str, job_ids: list[str]) -> list[dict]: raise PipelineCheckpointError("发布任务 provider_response 已损坏") from exc if not isinstance(provider_response, dict): raise PipelineCheckpointError("发布任务 provider_response 不是对象") + if str(row["output_status"] or "") != "completed" or not bool(row["output_is_active"]): + raise PipelineCheckpointError("发布任务对应的切片已失活或未完成") + if row["cut_run_id"] and ( + str(row["cut_run_status"] or "") not in {"completed", "completed_with_errors"} + or not bool(row["cut_run_is_active"]) + ): + raise PipelineCheckpointError("发布任务对应的切片批次已失活或未完成") status = str(row["status"] or "").upper() if status not in allowed_statuses and not ( status == "CANCELLED" and str(row["error_code"] or "") == "user_removed" @@ -544,6 +555,16 @@ def _publish_job_evidence(self, task_id: str, job_ids: list[str]) -> list[dict]: raise PipelineCheckpointError("发布草稿的封面文件无效") from app.services.transcription_checkpoint_service import fingerprint_file + video_fingerprint = fingerprint_file(video_path) + cover_fingerprint = fingerprint_file(cover_path) + if ( + int(provider_response.get("video_file_size") or -1) != video_path.stat().st_size + or str(provider_response.get("video_file_fingerprint") or "") != video_fingerprint + or int(provider_response.get("cover_file_size") or -1) != cover_path.stat().st_size + or str(provider_response.get("cover_file_fingerprint") or "") != cover_fingerprint + ): + raise PipelineCheckpointError("发布草稿媒体与创建时指纹不一致") + evidence.append( { "id": str(row["id"]), @@ -553,7 +574,7 @@ def _publish_job_evidence(self, task_id: str, job_ids: list[str]) -> list[dict]: "video_source": video_source, "video_file_path": str(video_path.resolve()), "video_file_size": int(video_path.stat().st_size), - "video_file_fingerprint": fingerprint_file(video_path), + "video_file_fingerprint": video_fingerprint, "cover_file_path": str(cover_path.resolve()), "cover_file": self._file_evidence(cover_path), "subtitle_delivery_mode": str( @@ -1075,6 +1096,50 @@ def _reconcile_interrupted_step( outputs = self._checkpoint_outputs(task_id, step, provisional) return self._restore_checkpoint_step(task_id, step, outputs) return None + if step == TaskStatus.PUBLISH_JOB_CREATING: + schedule_path = paths["analysis_path"].parent / "auto_publish_schedule.json" + baseline_schedule = ( + baseline.get("schedule") if isinstance(baseline.get("schedule"), dict) else {} + ) + if not baseline_schedule or self._safe_file_evidence(schedule_path) != baseline_schedule: + return None + lease = job_service.current_job_lease() + if not lease: + raise job_service.JobLeaseLostError("发布草稿恢复缺少当前 Workflow Job 租约") + with get_connection() as connection: + rows = connection.execute( + "SELECT id, provider_response FROM publish_jobs WHERE task_id = ?", + (task_id,), + ).fetchall() + recovered_ids: list[str] = [] + for row in rows: + try: + provider_response = json.loads(str(row["provider_response"] or "{}")) + except json.JSONDecodeError: + continue + if ( + isinstance(provider_response, dict) + and str(provider_response.get("source") or "") == "auto_pipeline" + and str(provider_response.get("workflow_job_id") or "") == lease[0] + ): + recovered_ids.append(str(row["id"])) + if not recovered_ids: + return None + try: + job_evidence = self._publish_job_evidence(task_id, recovered_ids) + self._verify_publish_schedule_pairs(schedule_path, job_evidence) + except PipelineCheckpointError: + return None + from app.services.publish_service import get_publish_job + + provisional = { + "created": [get_publish_job(job_id) for job_id in recovered_ids], + "skipped": [], + "created_count": len(recovered_ids), + "skipped_count": 0, + } + outputs = self._checkpoint_outputs(task_id, step, provisional) + return self._restore_checkpoint_step(task_id, step, outputs) artifact_by_step = { TaskStatus.PREPARING_SOURCE: paths["task_dir"] / "source" / "source_reference.json", diff --git a/app/services/subtitle_auto_workflow_service.py b/app/services/subtitle_auto_workflow_service.py index 8821bbb..adad7d1 100644 --- a/app/services/subtitle_auto_workflow_service.py +++ b/app/services/subtitle_auto_workflow_service.py @@ -2,7 +2,9 @@ from __future__ import annotations +import hashlib import json +from pathlib import Path from typing import Any from app.db.database import get_connection @@ -10,7 +12,8 @@ from app.services import job_service from app.services.storage_service import get_artifact_paths, resolve_video_file_path from app.services.subtitle_data_service import ( - approve_revision, + SubtitleRevisionConflict, + approve_revisions_with_connection, ensure_clip_track, ensure_source_track, get_revision, @@ -29,6 +32,13 @@ def prepare_task_subtitle_review(task_id: str) -> dict[str, Any]: task = task_service.get_task(task_id, include_video_probe=False) if not task: raise ValueError("任务不存在") + allowed_statuses = { + TaskStatus.VIDEO_CUTTING.value, + TaskStatus.SUBTITLE_DRAFTING.value, + TaskStatus.PENDING_SUBTITLE_REVIEW.value, + } + if task.get("status") not in allowed_statuses: + raise ValueError("当前任务状态不能进入字幕审核，请刷新后重试") source_track = ensure_source_track(task_id) output_clips = [ item @@ -38,7 +48,64 @@ def prepare_task_subtitle_review(task_id: str) -> dict[str, Any]: if not output_clips: raise ValueError("没有可生成字幕草稿的成功切片") tracks = [ensure_clip_track(task_id, item["id"]) for item in output_clips] - task_service.update_task_status(task_id, TaskStatus.PENDING_SUBTITLE_REVIEW) + lease = job_service.current_job_lease() + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + active_job = connection.execute( + """ + SELECT id FROM workflow_jobs + WHERE task_id = ? AND status IN (?, ?) + AND (? = '' OR id != ?) + LIMIT 1 + """, + ( + task_id, + job_service.JOB_STATUS_QUEUED, + job_service.JOB_STATUS_RUNNING, + lease[0] if lease else "", + lease[0] if lease else "", + ), + ).fetchone() + if active_job: + connection.rollback() + raise ValueError("任务已有其他后台 Job，不能重复进入字幕审核") + lease_condition = "" + lease_params: tuple[str, ...] = () + if lease: + lease_condition = """ + AND EXISTS ( + SELECT 1 FROM workflow_jobs + WHERE id = ? AND task_id = ? AND status = 'running' + AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > strftime('%Y-%m-%dT%H:%M:%S+00:00', 'now') + AND cancel_requested = 0 + ) + """ + lease_params = (lease[0], task_id, lease[1], lease[2]) + cursor = connection.execute( + f""" + UPDATE tasks + SET status = ?, progress = ?, error_message = NULL, last_error = NULL, updated_at = ? + WHERE id = ? AND COALESCE(is_deleted, 0) = 0 + AND status IN (?, ?, ?) {lease_condition} + """, + ( + TaskStatus.PENDING_SUBTITLE_REVIEW.value, + task_service.STATUS_PROGRESS[TaskStatus.PENDING_SUBTITLE_REVIEW.value], + task_service._now_iso(), + task_id, + TaskStatus.VIDEO_CUTTING.value, + TaskStatus.SUBTITLE_DRAFTING.value, + TaskStatus.PENDING_SUBTITLE_REVIEW.value, + *lease_params, + ), + ) + if cursor.rowcount != 1: + connection.rollback() + if lease: + raise job_service.JobLeaseLostError("字幕草稿提交前 Workflow Job 租约已失效") + raise ValueError("任务状态已变化，不能进入字幕审核") + connection.commit() append_task_log(task_id, f"字幕草稿已生成：{len(tracks)} 条，流水线暂停等待人工审核") return { "status": "pending_subtitle_review", @@ -76,18 +143,16 @@ def enqueue_task_subtitle_render( raise ValueError("没有可烧录字幕的成功切片") items = [] + approvals: list[tuple[str, str]] = [] for output_clip in output_clips: track = ensure_clip_track(task_id, output_clip["id"]) revision_id = str(track.get("active_revision_id") or "") if not revision_id: raise ValueError(f"{output_clip.get('output_file_name') or output_clip['id']} 没有字幕 revision") - revision = get_revision(revision_id, include_cues=True) + revision = get_revision(revision_id) if int(revision.get("cue_count") or 0) <= 0: raise ValueError(f"{output_clip.get('output_file_name') or output_clip['id']} 没有可烧录的字幕内容") - if approve_active_revisions: - revision = approve_revision(track["id"], revision_id) - if revision.get("status") != "approved": - raise ValueError(f"{output_clip.get('output_file_name') or output_clip['id']} 的字幕尚未审核") + approvals.append((str(track["id"]), revision_id)) items.append( { "output_clip_id": output_clip["id"], @@ -96,19 +161,63 @@ def enqueue_task_subtitle_render( } ) - job, created = job_service.create_or_get_active_job( - task_id=task_id, - job_type=job_service.JOB_TYPE_SUBTITLE, - payload={ - "items": items, - "continue_pipeline": continue_pipeline, - "subtitle_delivery_mode": "subtitled", - }, - ) - if continue_pipeline and not bool((job.get("payload_json") or {}).get("continue_pipeline")): - raise ValueError("已有单条字幕任务正在运行，请等待完成或取消后再启动批量烧录") - if continue_pipeline: - _set_subtitle_delivery_mode(task_id, "subtitled") + job_payload = { + "items": items, + "continue_pipeline": continue_pipeline, + "subtitle_delivery_mode": "subtitled", + } + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + if continue_pipeline: + current_task = connection.execute( + "SELECT status FROM tasks WHERE id = ? AND COALESCE(is_deleted, 0) = 0", + (task_id,), + ).fetchone() + if not current_task or current_task["status"] != TaskStatus.PENDING_SUBTITLE_REVIEW.value: + connection.rollback() + raise ValueError("当前任务已离开字幕审核暂停状态，请刷新后重试") + active_resume = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE task_id = ? AND job_type = ? AND status IN (?, ?) + LIMIT 1 + """, + ( + task_id, + job_service.JOB_TYPE_AUTO_PIPELINE, + job_service.JOB_STATUS_QUEUED, + job_service.JOB_STATUS_RUNNING, + ), + ).fetchone() + if active_resume: + connection.rollback() + raise ValueError("后续自动流水线已排队或运行，不能重复执行字幕烧录") + if approve_active_revisions: + approve_revisions_with_connection(connection, approvals) + else: + _validate_approved_revisions_with_connection(connection, approvals) + job_id, created = job_service.create_or_get_active_job_with_connection( + connection, + task_id=task_id, + job_type=job_service.JOB_TYPE_SUBTITLE, + payload=job_payload, + ) + existing_payload_row = connection.execute( + "SELECT payload_json FROM workflow_jobs WHERE id = ?", + (job_id,), + ).fetchone() + try: + existing_payload = json.loads(existing_payload_row["payload_json"] or "{}") + except (TypeError, json.JSONDecodeError) as exc: + raise ValueError("已有字幕 Job 的 payload 已损坏，请先处理该任务") from exc + if existing_payload != job_payload: + raise ValueError("已有不同内容的字幕任务正在排队或运行，请等待完成或取消后重试") + if continue_pipeline: + _set_subtitle_delivery_mode_with_connection(connection, task_id, "subtitled") + connection.commit() + job = job_service.get_job(job_id) + if not job: + raise RuntimeError("字幕 Job 创建后无法读取") append_task_log( task_id, f"字幕批量烧录{'已加入队列' if created else '已在队列中'}：{len(items)} 条", @@ -149,23 +258,28 @@ def skip_task_subtitles_to_review(task_id: str) -> dict[str, Any]: active_subtitle_job = connection.execute( """ SELECT 1 FROM workflow_jobs - WHERE task_id = ? AND job_type = ? AND status IN (?, ?) + WHERE task_id = ? AND job_type IN (?, ?) AND status IN (?, ?) LIMIT 1 """, ( task_id, job_service.JOB_TYPE_SUBTITLE, + job_service.JOB_TYPE_AUTO_PIPELINE, job_service.JOB_STATUS_QUEUED, job_service.JOB_STATUS_RUNNING, ), ).fetchone() if active_subtitle_job: connection.rollback() - raise ValueError("字幕烧录仍在运行，请先取消并等待停止后再选择跳过字幕") + raise ValueError("字幕烧录或后续自动流水线仍在运行，请先取消并等待停止后再选择跳过字幕") try: config = json.loads(task["auto_config_json"] or "{}") - except json.JSONDecodeError: - config = {} + except (TypeError, json.JSONDecodeError) as exc: + connection.rollback() + raise ValueError("任务字幕配置已损坏，请先修复配置后重试") from exc + if not isinstance(config, dict): + connection.rollback() + raise ValueError("任务字幕配置格式无效，请先修复配置后重试") config["subtitle_delivery_mode"] = "original" config["subtitle_decided_at"] = now connection.execute( @@ -203,6 +317,7 @@ def execute_subtitle_render_job(job_id: str, task_id: str, payload: dict[str, An items = payload.get("items") or [] if not isinstance(items, list) or not items: raise ValueError("字幕 Job 没有待渲染条目") + reconcile_interrupted_subtitle_job(job_id) job = job_service.get_job(job_id) or {} checkpoint = job.get("checkpoint_json") if isinstance(job.get("checkpoint_json"), dict) else {} completed = dict(checkpoint.get("completed") or {}) @@ -214,6 +329,19 @@ def execute_subtitle_render_job(job_id: str, task_id: str, payload: dict[str, An raise ValueError("字幕 Job 条目缺少 output_clip_id 或 revision_id") if _checkpoint_result_is_valid(task_id, output_clip_id, revision_id, completed.get(output_clip_id)): continue + recovered = _find_recoverable_subtitle_result( + task_id, + output_clip_id, + revision_id, + workflow_job_id=job_id, + ) + if recovered: + completed[output_clip_id] = recovered + job_service.update_job_checkpoint( + job_id, + {"completed": completed, "completed_count": len(completed), "total_count": total}, + ) + continue if job_service.is_cancel_requested(job_id): raise SubtitleRenderCancelled("用户已取消字幕批量烧录") item_start = 5 + round(index / total * 88) @@ -241,16 +369,13 @@ def execute_subtitle_render_job(job_id: str, task_id: str, payload: dict[str, An {"completed": completed, "completed_count": len(completed), "total_count": total}, ) - resume_job = None - if payload.get("continue_pipeline"): - resume_job, _ = _enqueue_pipeline_resume(task_id) - append_task_log(task_id, "字幕成片全部验证通过，已排队恢复自动文案与发送中心流程") return { "task_id": task_id, "completed_count": len(completed), "total_count": total, "completed": completed, - "resume_job_id": resume_job["id"] if resume_job else "", + "resume_requested": bool(payload.get("continue_pipeline")), + "resume_job_id": "", } @@ -265,9 +390,9 @@ def cleanup_interrupted_subtitle_job( """父 Worker 强制终止子进程后，修正从属字幕记录并清理精确临时文件。""" from app.services.task_service import _now_iso - now = _now_iso() with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") + now = _now_iso() row = connection.execute( """ SELECT task_id FROM workflow_jobs @@ -287,15 +412,104 @@ def cleanup_interrupted_subtitle_job( """, (status, message, now, workflow_job_id), ) - directory = get_artifact_paths(str(row["task_id"]))["subtitled_dir"] - if directory.exists(): - for path in directory.glob(f"*.{workflow_job_id}.part.mp4"): - if path.is_file(): - path.unlink(missing_ok=True) connection.commit() + _, failures = _cleanup_subtitle_attempt_files(str(row["task_id"]), workflow_job_id) + if failures: + append_task_log( + str(row["task_id"]), + f"字幕中断产物有 {len(failures)} 个暂时无法删除，将在下次接管时重试：{'; '.join(failures)}", + ) return True +def reconcile_interrupted_subtitle_job(workflow_job_id: str) -> dict[str, Any]: + """新执行接管后收口旧 processing 记录，并再次清理本 Job 的临时文件。""" + from app.services.task_service import _now_iso + + lease = job_service.current_job_lease() + if not lease or lease[0] != workflow_job_id: + raise job_service.JobLeaseLostError(f"字幕 Job 缺少当前执行租约：{workflow_job_id}") + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + row = connection.execute( + """ + SELECT task_id FROM workflow_jobs + WHERE id = ? AND status = 'running' AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > ? AND cancel_requested = 0 + """, + (workflow_job_id, lease[1], lease[2], now), + ).fetchone() + if not row: + connection.rollback() + raise job_service.JobLeaseLostError(f"字幕 Job 接管前租约已失效：{workflow_job_id}") + cursor = connection.execute( + """ + UPDATE subtitle_jobs + SET status = 'failed', error_message = '上一次执行中断，已由当前 Worker 接管', updated_at = ? + WHERE workflow_job_id = ? AND status IN ('processing', 'queued') AND is_active = 0 + """, + (now, workflow_job_id), + ) + interrupted_count = cursor.rowcount + connection.commit() + deleted_count, failures = _cleanup_subtitle_attempt_files(str(row["task_id"]), workflow_job_id) + if failures: + append_task_log( + str(row["task_id"]), + f"字幕接管时有 {len(failures)} 个中断产物仍被占用，将保留为非活跃残留：{'; '.join(failures)}", + ) + return { + "interrupted_count": interrupted_count, + "deleted_artifact_count": deleted_count, + "cleanup_failures": failures, + } + + +def _cleanup_subtitle_attempt_files(task_id: str, workflow_job_id: str) -> tuple[int, list[str]]: + directory = get_artifact_paths(task_id)["subtitled_dir"] + if not directory.exists() or not directory.is_dir(): + return 0, [] + workflow_marker = hashlib.sha256(workflow_job_id.encode("utf-8")).hexdigest()[:12] + expected_temp_suffixes = { + f".{workflow_job_id}.part.mp4", + f".{workflow_marker}.part.mp4", + } + expected_final_marker = f"_subtitled_{workflow_marker}_" + with get_connection() as connection: + referenced_names = { + Path(str(row["output_file_path"])).name + for row in connection.execute( + """ + SELECT output_file_path FROM subtitle_jobs + WHERE task_id = ? AND COALESCE(output_file_path, '') != '' + """, + (task_id,), + ).fetchall() + } + deleted = 0 + failures: list[str] = [] + try: + candidates = list(directory.iterdir()) + except OSError as exc: + return 0, [f"{directory.name}: {exc}"] + for path in candidates: + try: + is_owned_temp = any(path.name.endswith(suffix) for suffix in expected_temp_suffixes) + is_owned_orphan_final = ( + expected_final_marker in path.name + and path.name.endswith(".mp4") + and path.name not in referenced_names + ) + if (not is_owned_temp and not is_owned_orphan_final) or not path.is_file(): + continue + path.unlink(missing_ok=True) + deleted += 1 + except OSError as exc: + failures.append(f"{path.name}: {exc}") + return deleted, failures + + def _checkpoint_result_is_valid( task_id: str, output_clip_id: str, @@ -312,7 +526,7 @@ def _checkpoint_result_is_valid( """ SELECT output_file_path FROM subtitle_jobs WHERE id = ? AND task_id = ? AND output_clip_id = ? AND revision_id = ? - AND status = 'completed' AND validation_status = 'verified' + AND status = 'completed' AND validation_status = 'verified' AND is_active = 1 """, (subtitle_job_id, task_id, output_clip_id, revision_id), ).fetchone() @@ -320,32 +534,83 @@ def _checkpoint_result_is_valid( return bool(path and path.exists() and path.is_file()) -def _set_subtitle_delivery_mode(task_id: str, mode: str) -> None: +def _find_recoverable_subtitle_result( + task_id: str, + output_clip_id: str, + revision_id: str, + *, + workflow_job_id: str, +) -> dict[str, str] | None: + """恢复“DB 已提交但 checkpoint 尚未写入”的同一执行结果。""" + with get_connection() as connection: + row = connection.execute( + """ + SELECT sj.id, sj.output_file_path + FROM subtitle_jobs sj + JOIN subtitle_tracks st + ON st.task_id = sj.task_id AND st.output_clip_id = sj.output_clip_id + AND st.track_type = 'clip' AND st.is_active = 1 + JOIN subtitle_revisions sr ON sr.id = sj.revision_id AND sr.track_id = st.id + WHERE sj.task_id = ? AND sj.output_clip_id = ? AND sj.revision_id = ? + AND sj.workflow_job_id = ? AND sj.status = 'completed' + AND sj.validation_status = 'verified' AND sj.is_active = 1 + AND st.active_revision_id = sj.revision_id AND sr.status = 'approved' + ORDER BY sj.updated_at DESC LIMIT 1 + """, + (task_id, output_clip_id, revision_id, workflow_job_id), + ).fetchone() + path = resolve_video_file_path(row["output_file_path"]) if row else None + if not path or not path.exists() or not path.is_file(): + return None + return { + "revision_id": revision_id, + "subtitle_job_id": str(row["id"]), + "output_file_path": str(row["output_file_path"]), + } + + +def _validate_approved_revisions_with_connection( + connection, + approvals: list[tuple[str, str]], +) -> None: + for track_id, revision_id in approvals: + row = connection.execute( + """ + SELECT st.active_revision_id, sr.track_id, sr.status, sr.cue_count + FROM subtitle_tracks st + JOIN subtitle_revisions sr ON sr.id = ? + WHERE st.id = ? AND st.is_active = 1 + """, + (revision_id, track_id), + ).fetchone() + if not row or row["track_id"] != track_id: + raise ValueError("字幕 revision 不属于当前字幕轨") + if str(row["active_revision_id"] or "") != revision_id: + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后重新烧录") + if row["status"] != "approved": + raise ValueError("存在尚未审核的字幕 revision") + if int(row["cue_count"] or 0) <= 0: + raise ValueError("存在没有可烧录内容的字幕 revision") + + +def _set_subtitle_delivery_mode_with_connection(connection, task_id: str, mode: str) -> None: if mode not in {"subtitled", "original"}: raise ValueError("字幕交付模式无效") from app.services.task_service import _now_iso now = _now_iso() - with get_connection() as connection: - row = connection.execute("SELECT auto_config_json FROM tasks WHERE id = ?", (task_id,)).fetchone() - if not row: - raise ValueError("任务不存在") - try: - config = json.loads(row["auto_config_json"] or "{}") - except json.JSONDecodeError: - config = {} - config["subtitle_delivery_mode"] = mode - config["subtitle_decided_at"] = now - connection.execute( - "UPDATE tasks SET auto_config_json = ?, updated_at = ? WHERE id = ?", - (json.dumps(config, ensure_ascii=False), now, task_id), - ) - connection.commit() - - -def _enqueue_pipeline_resume(task_id: str) -> tuple[dict[str, Any], bool]: - return job_service.create_or_get_active_job( - task_id=task_id, - job_type=job_service.JOB_TYPE_AUTO_PIPELINE, - payload={"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + row = connection.execute("SELECT auto_config_json FROM tasks WHERE id = ?", (task_id,)).fetchone() + if not row: + raise ValueError("任务不存在") + try: + config = json.loads(row["auto_config_json"] or "{}") + except (TypeError, json.JSONDecodeError) as exc: + raise ValueError("任务字幕配置已损坏，请先修复配置后重试") from exc + if not isinstance(config, dict): + raise ValueError("任务字幕配置格式无效，请先修复配置后重试") + config["subtitle_delivery_mode"] = mode + config["subtitle_decided_at"] = now + connection.execute( + "UPDATE tasks SET auto_config_json = ?, updated_at = ? WHERE id = ?", + (json.dumps(config, ensure_ascii=False), now, task_id), ) diff --git a/app/services/subtitle_data_service.py b/app/services/subtitle_data_service.py index 85893de..f45b1e6 100644 --- a/app/services/subtitle_data_service.py +++ b/app/services/subtitle_data_service.py @@ -102,8 +102,13 @@ def get_track(track_id: str) -> dict[str, Any]: def ensure_source_track(task_id: str, *, force: bool = False) -> dict[str, Any]: + cues, source_fingerprint, origin = _load_source_cues(task_id) + if not cues: + raise ValueError("当前任务没有可用的结构化转写或时间戳 Markdown") with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") if not connection.execute("SELECT id FROM tasks WHERE id = ?", (task_id,)).fetchone(): + connection.rollback() raise ValueError("任务不存在") existing = connection.execute( """ @@ -113,25 +118,25 @@ def ensure_source_track(task_id: str, *, force: bool = False) -> dict[str, Any]: """, (task_id,), ).fetchone() - - cues, source_fingerprint, origin = _load_source_cues(task_id) - if not cues: - raise ValueError("当前任务没有可用的结构化转写或时间戳 Markdown") - if existing and existing["source_fingerprint"] == source_fingerprint and existing["active_revision_id"]: - return get_track(existing["id"]) - if existing and existing["has_manual_edits"] and not force: - with get_connection() as connection: + if ( + existing + and existing["source_fingerprint"] == source_fingerprint + and existing["active_revision_id"] + ): + track_id = str(existing["id"]) + connection.commit() + return get_track(track_id) + if existing and existing["has_manual_edits"] and not force: connection.execute( "UPDATE subtitle_tracks SET sync_status = 'pending_source_refresh', updated_at = ? WHERE id = ?", (_now_iso(), existing["id"]), ) connection.commit() - return get_track(existing["id"]) + return get_track(str(existing["id"])) - now = _now_iso() - track_id = existing["id"] if existing else uuid4().hex - with get_connection() as connection: - connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + track_id = str(existing["id"]) if existing else uuid4().hex + base_revision_id = str(existing["active_revision_id"] or "") if existing else "" if not existing: connection.execute( """ @@ -149,20 +154,31 @@ def ensure_source_track(task_id: str, *, force: bool = False) -> dict[str, Any]: track_id, cues, origin=origin, - parent_revision_id=existing["active_revision_id"] if existing else None, + parent_revision_id=base_revision_id or None, status="draft", note="从结构化转写生成原片主时间轴" if origin == "asr" else "从旧版 Markdown 兼容生成", - activate=True, + activate=False, ) - connection.execute( + cursor = connection.execute( """ UPDATE subtitle_tracks SET source_fingerprint = ?, active_revision_id = ?, sync_status = 'up_to_date', has_manual_edits = 0, updated_at = ? - WHERE id = ? + WHERE id = ? AND is_active = 1 + AND ((active_revision_id = ?) OR (active_revision_id IS NULL AND ? = '')) """, - (source_fingerprint, revision["id"], now, track_id), + ( + source_fingerprint, + revision["id"], + now, + track_id, + base_revision_id, + base_revision_id, + ), ) + if cursor.rowcount != 1: + connection.rollback() + raise SubtitleRevisionConflict("原片字幕已产生新版本，请刷新后重试") connection.commit() with get_connection() as connection: @@ -179,6 +195,7 @@ def ensure_clip_track(task_id: str, output_clip_id: str) -> dict[str, Any]: source_track = ensure_source_track(task_id) output = ensure_output_clip_snapshot(task_id, output_clip_id) with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") existing = connection.execute( """ SELECT * FROM subtitle_tracks @@ -209,57 +226,92 @@ def ensure_clip_track(task_id: str, output_clip_id: str) -> dict[str, Any]: now, ), ) - connection.commit() else: track_id = existing["id"] + connection.commit() sync_clip_track(track_id, force=False) return get_track(track_id) def sync_clip_track(track_id: str, *, force: bool = False) -> dict[str, Any]: - track = get_track(track_id) - if track["track_type"] != "clip": + track_hint = get_track(track_id) + if track_hint["track_type"] != "clip": raise ValueError("只有切片字幕轨可以从原片同步") - source_track = get_track(track["source_track_id"]) - source_revision_id = source_track.get("active_revision_id") - if not source_revision_id: - raise ValueError("原片字幕还没有可同步 revision") - if track.get("has_manual_edits") and not force: - if track.get("source_revision_id") != source_revision_id: - with get_connection() as connection: + ensure_output_clip_snapshot(track_hint["task_id"], track_hint["output_clip_id"]) + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + track_row = connection.execute( + "SELECT * FROM subtitle_tracks WHERE id = ? AND is_active = 1", + (track_id,), + ).fetchone() + if not track_row: + connection.rollback() + raise ValueError("字幕轨不存在或已停用") + track = dict(track_row) + if track["track_type"] != "clip": + connection.rollback() + raise ValueError("只有切片字幕轨可以从原片同步") + source_track_row = connection.execute( + "SELECT * FROM subtitle_tracks WHERE id = ? AND track_type = 'source' AND is_active = 1", + (track["source_track_id"],), + ).fetchone() + if not source_track_row or not source_track_row["active_revision_id"]: + connection.rollback() + raise ValueError("原片字幕还没有可同步 revision") + source_track = dict(source_track_row) + source_revision_id = str(source_track["active_revision_id"]) + if track.get("has_manual_edits") and not force: + if track.get("source_revision_id") != source_revision_id: connection.execute( "UPDATE subtitle_tracks SET sync_status = 'pending_sync', updated_at = ? WHERE id = ?", (_now_iso(), track_id), ) - connection.commit() - return get_track(track_id) - - if track.get("source_revision_id") == source_revision_id and track.get("active_revision_id"): - return track + connection.commit() + return get_track(track_id) + if track.get("source_revision_id") == source_revision_id and track.get("active_revision_id"): + connection.commit() + return get_track(track_id) - output = ensure_output_clip_snapshot(track["task_id"], track["output_clip_id"]) - source_start_ms = int(output["source_start_ms"]) - source_end_ms = int(output["source_end_ms"]) - source_cues = _fetch_all_revision_cues(source_revision_id) - local_cues = inherit_cues_for_clip(source_cues, source_start_ms, source_end_ms) - with get_connection() as connection: - connection.execute("BEGIN IMMEDIATE") + output = connection.execute( + """ + SELECT source_start_ms, source_end_ms + FROM output_clip WHERE id = ? AND task_id = ? + """, + (track["output_clip_id"], track["task_id"]), + ).fetchone() + if not output or output["source_start_ms"] is None or output["source_end_ms"] is None: + connection.rollback() + raise ValueError("切片记录缺少原片时间范围") + source_cues = [ + dict(row) + for row in connection.execute( + "SELECT * FROM subtitle_cues WHERE revision_id = ? ORDER BY cue_index ASC", + (source_revision_id,), + ).fetchall() + ] + local_cues = inherit_cues_for_clip( + source_cues, + int(output["source_start_ms"]), + int(output["source_end_ms"]), + ) + base_revision_id = str(track.get("active_revision_id") or "") revision = _insert_revision_with_connection( connection, track_id, local_cues, origin="source_sync", - parent_revision_id=track.get("active_revision_id"), + parent_revision_id=base_revision_id or None, status="draft", note=f"继承原片 revision {source_revision_id}", - activate=True, + activate=False, ) - connection.execute( + cursor = connection.execute( """ UPDATE subtitle_tracks SET source_revision_id = ?, source_fingerprint = ?, active_revision_id = ?, sync_status = 'up_to_date', has_manual_edits = 0, updated_at = ? - WHERE id = ? + WHERE id = ? AND is_active = 1 + AND ((active_revision_id = ?) OR (active_revision_id IS NULL AND ? = '')) """, ( source_revision_id, @@ -267,8 +319,13 @@ def sync_clip_track(track_id: str, *, force: bool = False) -> dict[str, Any]: revision["id"], _now_iso(), track_id, + base_revision_id, + base_revision_id, ), ) + if cursor.rowcount != 1: + connection.rollback() + raise SubtitleRevisionConflict("切片字幕已产生新版本，请刷新后重试同步") connection.commit() return get_track(track_id) @@ -349,12 +406,20 @@ def create_manual_revision( cues: Iterable[Any], note: str = "", ) -> dict[str, Any]: - track = get_track(track_id) - if (track.get("active_revision_id") or None) != (base_revision_id or None): - raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后再保存，当前编辑没有覆盖新版本") normalized = [_cue_input_to_dict(cue) for cue in cues] with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") + track_row = connection.execute( + "SELECT * FROM subtitle_tracks WHERE id = ? AND is_active = 1", + (track_id,), + ).fetchone() + if not track_row: + connection.rollback() + raise ValueError("字幕轨不存在或已停用") + track = dict(track_row) + if (track.get("active_revision_id") or None) != (base_revision_id or None): + connection.rollback() + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后再保存，当前编辑没有覆盖新版本") revision = _insert_revision_with_connection( connection, track_id, @@ -363,18 +428,22 @@ def create_manual_revision( parent_revision_id=base_revision_id, status="draft", note=note or "字幕编辑器自动保存", - activate=True, + activate=False, ) - connection.execute( + cursor = connection.execute( """ UPDATE subtitle_tracks SET active_revision_id = ?, has_manual_edits = 1, sync_status = CASE WHEN track_type = 'clip' THEN 'manual' ELSE sync_status END, updated_at = ? - WHERE id = ? + WHERE id = ? AND is_active = 1 + AND ((active_revision_id = ?) OR (active_revision_id IS NULL AND ? IS NULL)) """, - (revision["id"], _now_iso(), track_id), + (revision["id"], _now_iso(), track_id, base_revision_id, base_revision_id), ) + if cursor.rowcount != 1: + connection.rollback() + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后再保存，当前编辑没有覆盖新版本") connection.commit() if track["track_type"] == "source": _sync_dependent_clip_tracks(track["task_id"]) @@ -477,27 +546,87 @@ def get_revision_cues( def approve_revision(track_id: str, revision_id: str) -> dict[str, Any]: - get_track(track_id) - revision = get_revision(revision_id, include_cues=True) - if revision["track_id"] != track_id: - raise ValueError("revision 不属于当前字幕轨") - if int((revision.get("quality") or {}).get("error_count") or 0) > 0: - raise ValueError("当前字幕仍有时间重叠错误，请修正后再审核") - now = _now_iso() with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") - connection.execute( - "UPDATE subtitle_revisions SET status = 'approved', approved_at = ? WHERE id = ?", - (now, revision_id), - ) - connection.execute( - "UPDATE subtitle_tracks SET active_revision_id = ?, updated_at = ? WHERE id = ?", - (revision_id, now, track_id), + approve_revisions_with_connection( + connection, + [(track_id, revision_id)], + require_non_empty=False, ) connection.commit() return get_revision(revision_id, include_cues=True) +def approve_revisions_with_connection( + connection, + approvals: Iterable[tuple[str, str]], + *, + require_non_empty: bool = True, +) -> list[dict[str, Any]]: + """在调用方事务内重新校验并批准一组当前 active revision。""" + now = _now_iso() + approved: list[dict[str, Any]] = [] + for track_id, revision_id in approvals: + track = connection.execute( + "SELECT id, active_revision_id FROM subtitle_tracks WHERE id = ? AND is_active = 1", + (track_id,), + ).fetchone() + if not track: + raise ValueError("字幕轨不存在或已停用") + if str(track["active_revision_id"] or "") != revision_id: + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后重新审核") + + revision = connection.execute( + "SELECT id, track_id, status, cue_count FROM subtitle_revisions WHERE id = ?", + (revision_id,), + ).fetchone() + if not revision: + raise ValueError("字幕 revision 不存在") + if revision["track_id"] != track_id: + raise ValueError("revision 不属于当前字幕轨") + if require_non_empty and int(revision["cue_count"] or 0) <= 0: + raise ValueError("当前字幕没有可审核内容") + + cues = [ + dict(row) + for row in connection.execute( + "SELECT * FROM subtitle_cues WHERE revision_id = ? ORDER BY cue_index ASC", + (revision_id,), + ).fetchall() + ] + if int(evaluate_subtitle_quality(cues).get("error_count") or 0) > 0: + raise ValueError("当前字幕仍有时间重叠错误，请修正后再审核") + + cursor = connection.execute( + """ + UPDATE subtitle_revisions + SET status = 'approved', approved_at = COALESCE(approved_at, ?) + WHERE id = ? AND track_id = ? + """, + (now, revision_id, track_id), + ) + if cursor.rowcount != 1: + raise SubtitleRevisionConflict("字幕 revision 已变化，请刷新后重新审核") + cursor = connection.execute( + """ + UPDATE subtitle_tracks SET active_revision_id = ?, updated_at = ? + WHERE id = ? AND active_revision_id = ? AND is_active = 1 + """, + (revision_id, now, track_id, revision_id), + ) + if cursor.rowcount != 1: + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后重新审核") + approved.append( + { + "id": revision_id, + "track_id": track_id, + "status": "approved", + "cue_count": int(revision["cue_count"] or 0), + } + ) + return approved + + def create_suggestion_revision( track_id: str, *, @@ -645,23 +774,40 @@ def import_subtitle_text( for event in document.events if event.type == "Dialogue" and event.end > event.start and event.plaintext.strip() ] - track = get_track(track_id) with get_connection() as connection: connection.execute("BEGIN IMMEDIATE") + track_row = connection.execute( + "SELECT * FROM subtitle_tracks WHERE id = ? AND is_active = 1", + (track_id,), + ).fetchone() + if not track_row: + connection.rollback() + raise ValueError("字幕轨不存在或已停用") + track = dict(track_row) + base_revision_id = str(track.get("active_revision_id") or "") revision = _insert_revision_with_connection( connection, track_id, cues, origin="import", - parent_revision_id=track.get("active_revision_id"), + parent_revision_id=base_revision_id or None, status="draft", note=note or f"导入 {format_name.upper()} 字幕", - activate=True, + activate=False, ) - connection.execute( - "UPDATE subtitle_tracks SET active_revision_id = ?, has_manual_edits = 1, sync_status = 'manual', updated_at = ? WHERE id = ?", - (revision["id"], _now_iso(), track_id), + cursor = connection.execute( + """ + UPDATE subtitle_tracks + SET active_revision_id = ?, has_manual_edits = 1, + sync_status = 'manual', updated_at = ? + WHERE id = ? AND is_active = 1 + AND ((active_revision_id = ?) OR (active_revision_id IS NULL AND ? = '')) + """, + (revision["id"], _now_iso(), track_id, base_revision_id, base_revision_id), ) + if cursor.rowcount != 1: + connection.rollback() + raise SubtitleRevisionConflict("字幕已产生新版本，请刷新后重新导入") connection.commit() if track["track_type"] == "source": _sync_dependent_clip_tracks(track["task_id"]) @@ -1268,6 +1414,7 @@ def _chinese_char_count(value: str) -> int: "SubtitleRevisionConflict", "apply_revision_operations", "approve_revision", + "approve_revisions_with_connection", "create_manual_revision", "ensure_clip_track", "ensure_output_clip_snapshot", diff --git a/app/services/subtitle_workflow_service.py b/app/services/subtitle_workflow_service.py index 0d7d498..db7d659 100644 --- a/app/services/subtitle_workflow_service.py +++ b/app/services/subtitle_workflow_service.py @@ -3,6 +3,7 @@ 从 task_service 中拆分出来的字幕样式、ASS 渲染和字幕烧录函数。 """ +import hashlib import json import shutil import subprocess @@ -253,23 +254,54 @@ def _activate_subtitle_job(task_id: str, output_clip_id: str, job_id: str) -> No connection.commit() -def _update_subtitle_job_status(job_id: str, status: str, error_message: str = "") -> None: +def _update_subtitle_job_status( + job_id: str, + status: str, + error_message: str = "", + *, + workflow_job_id: str | None = None, +) -> None: """更新字幕任务状态（不改变 is_active）""" from app.db.database import get_connection from app.services.task_service import _now_iso - now = _now_iso() with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + lease = job_service.current_job_lease() if workflow_job_id else None + if workflow_job_id: + if not lease or lease[0] != workflow_job_id: + connection.rollback() + raise job_service.JobLeaseLostError(f"字幕 Job 缺少当前执行租约：{workflow_job_id}") + active = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE id = ? AND status = 'running' AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > ? + """, + (workflow_job_id, lease[1], lease[2], now), + ).fetchone() + if not active: + connection.rollback() + raise job_service.JobLeaseLostError(f"字幕 Job 租约已失效：{workflow_job_id}") + condition = "id = ?" + params: tuple[str, ...] = (job_id,) + if workflow_job_id: + condition += " AND workflow_job_id = ? AND is_active = 0" + params = (job_id, workflow_job_id) if error_message: - connection.execute( - "UPDATE subtitle_jobs SET status = ?, error_message = ?, updated_at = ? WHERE id = ?", - (status, error_message, now, job_id), + cursor = connection.execute( + f"UPDATE subtitle_jobs SET status = ?, error_message = ?, updated_at = ? WHERE {condition}", + (status, error_message, now, *params), ) else: - connection.execute( - "UPDATE subtitle_jobs SET status = ?, updated_at = ? WHERE id = ?", - (status, now, job_id), + cursor = connection.execute( + f"UPDATE subtitle_jobs SET status = ?, updated_at = ? WHERE {condition}", + (status, now, *params), ) + if workflow_job_id and cursor.rowcount != 1: + connection.rollback() + raise job_service.JobLeaseLostError(f"字幕子任务已被其他执行收口：{job_id}") connection.commit() @@ -371,6 +403,10 @@ class SubtitleRenderCancelled(RuntimeError): pass +def _workflow_file_marker(workflow_job_id: str) -> str: + return hashlib.sha256(workflow_job_id.encode("utf-8")).hexdigest()[:12] + + def render_subtitles_for_output_clip( task_id: str, output_clip_id: str, @@ -409,9 +445,10 @@ def render_subtitles_for_output_clip( raise ValueError("只有已审核的字幕 revision 才能烧录") paths = get_artifact_paths(task_id) paths["subtitled_dir"].mkdir(parents=True, exist_ok=True) - render_token = uuid4().hex[:10] + workflow_marker = _workflow_file_marker(workflow_job_id) if workflow_job_id else "" + render_token = f"{workflow_marker}_{uuid4().hex[:10]}" if workflow_marker else uuid4().hex[:10] output_path = paths["subtitled_dir"] / f"{input_path.stem}_subtitled_{render_token}.mp4" - temp_owner = workflow_job_id or render_token + temp_owner = workflow_marker or render_token temporary_path = output_path.with_name(f".{output_path.stem}.{temp_owner}.part.mp4") # === 版本化：创建新的字幕 job，不覆盖旧的 === @@ -449,52 +486,45 @@ def render_subtitles_for_output_clip( source_duration=float(source_probe.get("duration") or 0), source_has_audio=bool(source_probe.get("has_audio")), ) - temporary_path.replace(output_path) + _finalize_subtitle_job( + task_id=task_id, + output_clip_id=output_clip_id, + revision_id=selected_revision_id, + subtitle_job_id=job["id"], + workflow_job_id=workflow_job_id, + subtitle_path=subtitle_path, + temporary_path=temporary_path, + output_path=output_path, + validation=validation, + encoder=encoder, + audio_mode=audio_mode, + ) except job_service.JobLeaseLostError: temporary_path.unlink(missing_ok=True) raise except SubtitleRenderCancelled as exc: temporary_path.unlink(missing_ok=True) - _update_subtitle_job_status(job["id"], "cancelled", error_message=str(exc)) + _update_subtitle_job_status( + job["id"], + "cancelled", + error_message=str(exc), + workflow_job_id=workflow_job_id, + ) append_task_log(task_id, f"字幕烧录已取消：{input_path.name}") raise except Exception as exc: temporary_path.unlink(missing_ok=True) error = str(exc) # 失败时：标记当前 job 为 failed，不激活，旧字幕保持 active - _update_subtitle_job_status(job["id"], "failed", error_message=error) + _update_subtitle_job_status( + job["id"], + "failed", + error_message=error, + workflow_job_id=workflow_job_id, + ) append_task_log(task_id, f"自动加字幕失败：{input_path.name}，原因：{error}") raise - # 成功：更新 job 信息并切换为 active - # 验证成功后才写入最终路径并激活，失败时旧 active 字幕成片保持不变。 - from app.db.database import get_connection - from app.services.task_service import _now_iso - - now = _now_iso() - with get_connection() as connection: - connection.execute( - """ - UPDATE subtitle_jobs - SET status = 'completed', subtitle_file_path = ?, output_file_path = ?, - error_message = '', validation_status = 'verified', validation_json = ?, - encoder = ?, verified_at = ?, updated_at = ? - WHERE id = ? - """, - ( - str(subtitle_path), - str(output_path), - json.dumps({**validation, "audio_mode": audio_mode}, ensure_ascii=False), - encoder, - now, - now, - job["id"], - ), - ) - connection.commit() - - # 激活当前字幕 job，旧字幕 job 标记为非活跃 - _activate_subtitle_job(task_id, output_clip_id, job["id"]) append_task_log(task_id, f"自动加字幕完成：{output_path.name}") job = _subtitle_job_for_output(task_id, output_clip_id, active_only=False) or job @@ -507,6 +537,132 @@ def render_subtitles_for_output_clip( } +def _finalize_subtitle_job( + *, + task_id: str, + output_clip_id: str, + revision_id: str, + subtitle_job_id: str, + workflow_job_id: str | None, + subtitle_path: Path, + temporary_path: Path, + output_path: Path, + validation: dict[str, Any], + encoder: str, + audio_mode: str, +) -> None: + """在 lease/当前 revision 保护下原子切换最终文件与 active 字幕记录。""" + from app.db.database import get_connection + from app.services.task_service import _now_iso + + lease = job_service.current_job_lease() if workflow_job_id else None + if workflow_job_id and (not lease or lease[0] != workflow_job_id): + raise job_service.JobLeaseLostError(f"字幕 Job 缺少当前执行租约：{workflow_job_id}") + final_file_created = False + try: + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + now = _now_iso() + if workflow_job_id: + active_lease = connection.execute( + """ + SELECT 1 FROM workflow_jobs + WHERE id = ? AND status = 'running' AND lease_owner = ? AND lease_token = ? + AND lease_expires_at > ? AND cancel_requested = 0 + """, + (workflow_job_id, lease[1], lease[2], now), + ).fetchone() + if not active_lease: + raise job_service.JobLeaseLostError(f"字幕 Job 最终提交前租约已失效：{workflow_job_id}") + current_revision = connection.execute( + """ + SELECT sr.status + FROM subtitle_tracks st + JOIN subtitle_revisions sr ON sr.id = st.active_revision_id + WHERE st.task_id = ? AND st.output_clip_id = ? AND st.track_type = 'clip' + AND st.is_active = 1 AND st.active_revision_id = ? AND sr.track_id = st.id + """, + (task_id, output_clip_id, revision_id), + ).fetchone() + if not current_revision or current_revision["status"] != "approved": + raise ValueError("字幕 revision 已变化，旧渲染结果不会被激活") + subtitle_job = connection.execute( + """ + SELECT 1 FROM subtitle_jobs + WHERE id = ? AND task_id = ? AND output_clip_id = ? AND revision_id = ? + AND status = 'processing' AND is_active = 0 + AND ((? IS NULL AND workflow_job_id IS NULL) OR workflow_job_id = ?) + """, + ( + subtitle_job_id, + task_id, + output_clip_id, + revision_id, + workflow_job_id, + workflow_job_id, + ), + ).fetchone() + if not subtitle_job: + raise RuntimeError("字幕子任务已被其他执行收口，拒绝激活旧结果") + if output_path.exists(): + raise RuntimeError("字幕最终输出路径已存在，拒绝覆盖") + temporary_path.replace(output_path) + final_file_created = True + connection.execute( + """ + UPDATE subtitle_jobs SET is_active = 0, updated_at = ? + WHERE task_id = ? AND output_clip_id = ? AND id != ? + """, + (now, task_id, output_clip_id, subtitle_job_id), + ) + cursor = connection.execute( + """ + UPDATE subtitle_jobs + SET status = 'completed', subtitle_file_path = ?, output_file_path = ?, + error_message = '', validation_status = 'verified', validation_json = ?, + encoder = ?, verified_at = ?, updated_at = ?, is_active = 1 + WHERE id = ? AND task_id = ? AND output_clip_id = ? AND revision_id = ? + AND status = 'processing' AND is_active = 0 + """, + ( + str(subtitle_path), + str(output_path), + json.dumps({**validation, "audio_mode": audio_mode}, ensure_ascii=False), + encoder, + now, + now, + subtitle_job_id, + task_id, + output_clip_id, + revision_id, + ), + ) + if cursor.rowcount != 1: + raise RuntimeError("字幕最终状态提交冲突，拒绝激活旧结果") + connection.commit() + except Exception as exc: + if final_file_created: + try: + with get_connection() as verification_connection: + persisted = verification_connection.execute( + """ + SELECT 1 FROM subtitle_jobs + WHERE id = ? AND status = 'completed' AND validation_status = 'verified' + AND is_active = 1 AND output_file_path = ? + """, + (subtitle_job_id, str(output_path)), + ).fetchone() + except Exception as verification_exc: + exc.add_note(f"无法确认字幕最终提交是否持久化，已保留文件供恢复：{verification_exc}") + else: + if not persisted: + try: + output_path.unlink(missing_ok=True) + except OSError as cleanup_exc: + exc.add_note(f"回滚字幕最终文件失败：{cleanup_exc}") + raise + + def _probe_media(path: Path) -> dict[str, Any]: ffprobe = shutil.which("ffprobe") if not ffprobe: diff --git a/docs/agent_tasks/2026-08-25-p1-3d-subtitle-atomicity.md b/docs/agent_tasks/2026-08-25-p1-3d-subtitle-atomicity.md new file mode 100644 index 0000000..df807ab --- /dev/null +++ b/docs/agent_tasks/2026-08-25-p1-3d-subtitle-atomicity.md @@ -0,0 +1,70 @@ +# P1.3d 字幕批准原子性与跨进程恢复任务 + +## 背景 + +工程审计确认字幕批量烧录仍有三个稳定性缺口：批量批准逐条提交、渲染完成与 active 结果切换分两次提交、进程被接管后旧执行仍可能落下最终文件或激活旧结果。现有实现已经使用 workflow job lease、attempt 专属 `.part.mp4` 和 checkpoint，本轮在这些机制上收口，不引入新队列或新架构。 + +## 目标 + +1. 批量批准、固定待渲染 revision、创建或复用字幕 workflow job，以及全自动任务的字幕交付模式写入形成单个数据库事务。 +2. 任一 revision 校验或 job 创建失败时，整批批准不得出现部分成功。 +3. 只有仍持有当前 workflow job lease 的执行可以把验证通过的临时文件切换为最终文件并激活 subtitle job。 +4. 新执行接管或重试时，清理同一 workflow job 遗留的未激活 processing 记录和 attempt 临时文件；不得删除已激活结果、已验证最终文件或其他 job 的临时文件。 +5. checkpoint 只接受当前已激活、已验证且文件仍存在的字幕结果，避免恢复时复用已被后续版本替代的记录。 +6. 字幕 Job 完成与后续自动流水线 Job 的创建/复用必须同事务提交，取消或 lease 失效时两者都不得落库。 + +## 允许修改范围 + +- `app/services/subtitle_data_service.py` +- `app/services/subtitle_auto_workflow_service.py` +- `app/services/subtitle_workflow_service.py` +- 为复用原子 job 创建所必需的 `app/services/job_service.py` +- 为父进程异常退出收口所必需的 `app/services/job_worker.py` +- 为剩余 Workflow lease/follow-up/发布草稿跨进程边界所必需的 `app/services/auto_publish_service.py` +- 为发布草稿已提交但 checkpoint 尚未写入恢复所必需的 `app/services/pipeline_engine.py` +- 字幕、job fencing、pipeline checkpoint 相关测试 +- `.codemap/modules.json`、`PROJECT_AUDIT.md`、`DEVELOPMENT_LOG.md`、`NEXT_STEPS.md` + +## 禁止修改范围 + +- 不改变用户可见的字幕审核流程和页面结构。 +- 不新增数据库 Schema，不迁移或写入活动 `data/workflow.sqlite3`。 +- 不执行真实 FFmpeg、AI Provider、Chrome 或平台投稿。 +- 不处理 P1.4 的通用超时/重试策略，不处理 P1.5 的管理员门禁和 XSS。 +- 不做字幕模块的大规模拆分或格式化。 + +## 已确定实现要求 + +- 所有批量输入先完成无副作用收集；正式批准必须在 `BEGIN IMMEDIATE` 中重新校验 track、revision、cue 数量与时间重叠。 +- workflow job 的“查找活动任务或插入新任务”必须复用同一事务连接，不能在批准提交后才创建。 +- 渲染成功落库必须检查 `workflow_job_id + status=running + lease_owner + lease_token + 未过期`；无 workflow job 的同步单条渲染保持兼容。 +- 文件最终切换与数据库 active 切换在同一短事务保护区完成；后续数据库失败时删除本次唯一命名的孤儿最终文件。 +- 清理函数只接受受管理字幕目录中的、名称包含精确 workflow job 标记的 `.part.mp4` 或无数据库引用的本次最终文件，单文件失败需可诊断且不能伪装成清理成功。 +- 所有 lease 校验时间必须在取得数据库写锁后计算；续跑 Job 与当前字幕 Job 终态必须原子提交。 +- 源轨生成、切片轨同步和字幕导入必须在写锁内重读 active revision，不能用事务外旧基线覆盖人工新版本。 +- ASS 文件按 revision 确定性生成并可复用，不作为 attempt 临时文件删除；只有带本次 workflow 标记的临时/孤儿视频属于中断清理范围。 + +## 验收标准 + +- 批次第 N 项校验失败，前 N-1 项仍保持原状态，且不创建 workflow job。 +- job 插入失败时没有批准或 delivery mode 的半提交。 +- stale lease 不能完成、激活或覆盖新执行；旧 active 结果保持不变。 +- 新 lease 可标记遗留 processing 记录并清理自己的 `.part.mp4`，其他 job 文件不受影响。 +- 有效 checkpoint 可恢复；未激活、revision 不匹配、文件缺失或已被替代的 checkpoint 会重做。 +- Pytest 必须明确使用唯一临时 `test_workflow.sqlite3`；活动数据库不得作为测试目标。 + +## 测试命令 + +```powershell +.venv\Scripts\python.exe -m pytest -q tests/test_subtitle_editor.py tests/test_subtitle_auto_workflow.py tests/test_job_fencing.py tests/test_job_queue.py tests/test_auto_pipeline.py tests/test_pipeline_checkpoint.py tests/test_pipeline_state_stability.py tests/test_task_state_machine.py tests/test_publish_task_linkage.py tests/test_versioning_rollback.py +.venv\Scripts\python.exe -m ruff check app tests +.venv\Scripts\python.exe -m compileall -q app +``` + +## 返回格式 + +- 修改文件与关键行为变化 +- 测试命令、通过数、失败证据 +- 活动数据库未变化证据 +- Codemap 受影响模块复评 +- commit、分支、push 与 PR 链接 diff --git a/tests/test_auto_pipeline.py b/tests/test_auto_pipeline.py index 57010a6..a11c4dc 100644 --- a/tests/test_auto_pipeline.py +++ b/tests/test_auto_pipeline.py @@ -340,7 +340,13 @@ def test_daily_window_schedule_supports_seven_to_midnight_without_looping(): def test_create_auto_publish_job_records_scheduled_at(): task = _create_auto_task("test-auto-publish-job") - clip_path = _fake_video("publish_clip.mp4") + paths = get_artifact_paths(task["id"]) + paths["clips_dir"].mkdir(parents=True, exist_ok=True) + paths["covers_dir"].mkdir(parents=True, exist_ok=True) + clip_path = paths["clips_dir"] / "publish_clip.mp4" + cover_path = paths["covers_dir"] / "publish_clip_cover.jpg" + clip_path.write_bytes(b"fake mp4") + cover_path.write_bytes(b"fake jpg") with get_connection() as connection: now = "2026-06-23T08:00:00+00:00" connection.execute( @@ -358,7 +364,7 @@ def test_create_auto_publish_job_records_scheduled_at(): { "output_clip": {"id": "out-1", "output_file_path": str(clip_path)}, "cover": { - "cover_file_path": str(_fake_cover("publish_clip_cover.jpg")), + "cover_file_path": str(cover_path), "cover_time_seconds": 12.5, "cover_source": "ai_frame", }, @@ -374,14 +380,45 @@ def test_create_auto_publish_job_records_scheduled_at(): "scheduled_at": "2026-06-23T08:10:00+00:00", } ] - result = create_auto_publish_jobs( - task, - scheduled_items, - subtitle_delivery_mode="original", - workflow_job_id="test-auto-workflow-job", - ) + workflow_job = job_service.create_job(task["id"], job_service.JOB_TYPE_AUTO_PIPELINE) + claimed = job_service.claim_job(workflow_job["id"], "auto-publish-test-worker") + with job_service.job_lease_context( + workflow_job["id"], + "auto-publish-test-worker", + claimed["lease_token"], + ): + result = create_auto_publish_jobs( + task, + scheduled_items, + subtitle_delivery_mode="original", + workflow_job_id=workflow_job["id"], + ) + cover_path.write_bytes(b"changed cover") + with pytest.raises(ValueError, match="证据已失效"): + create_auto_publish_jobs( + task, + scheduled_items, + subtitle_delivery_mode="original", + workflow_job_id=workflow_job["id"], + ) + cover_path.write_bytes(b"fake jpg") + changed_copy = [ + { + **scheduled_items[0], + "metadata": {**scheduled_items[0]["metadata"], "title": "新的排期标题"}, + } + ] + with pytest.raises(ValueError, match="文案证据已失效"): + create_auto_publish_jobs( + task, + changed_copy, + subtitle_delivery_mode="original", + workflow_job_id=workflow_job["id"], + ) assert result["created_count"] == 1 - assert result["created"][0]["provider_payload"]["workflow_job_id"] == "test-auto-workflow-job" + assert result["created"][0]["provider_payload"]["workflow_job_id"] == workflow_job["id"] + assert result["created"][0]["provider_payload"]["video_file_fingerprint"] + assert result["created"][0]["provider_payload"]["cover_file_fingerprint"] with get_connection() as connection: row = connection.execute( "SELECT scheduled_at, status, video_source, cover_mode, cover_time_seconds, cover_file_path FROM publish_jobs WHERE task_id = ?", @@ -396,6 +433,47 @@ def test_create_auto_publish_job_records_scheduled_at(): assert row["cover_file_path"].endswith("publish_clip_cover.jpg") +def test_workflow_publish_rejects_artifacts_outside_task_directory(): + task = _create_auto_task("test-auto-publish-unmanaged") + clip_path = _fake_video("unmanaged_publish_clip.mp4") + cover_path = _fake_cover("unmanaged_publish_cover.jpg") + workflow_job = job_service.create_job(task["id"], job_service.JOB_TYPE_AUTO_PIPELINE) + claimed = job_service.claim_job(workflow_job["id"], "unmanaged-publish-worker") + + with job_service.job_lease_context( + workflow_job["id"], + "unmanaged-publish-worker", + claimed["lease_token"], + ): + with pytest.raises(ValueError, match="原片切片文件不存在"): + create_auto_publish_jobs( + task, + [ + { + "output_clip": {"id": "unmanaged-out", "output_file_path": str(clip_path)}, + "cover": {"cover_file_path": str(cover_path), "cover_time_seconds": 1}, + "metadata": { + "platform": "douyin", + "title": "越界测试", + "caption": "越界测试", + "hashtags": ["测试"], + "risk_flags": [], + }, + "scheduled_at": "", + } + ], + subtitle_delivery_mode="original", + workflow_job_id=workflow_job["id"], + ) + + with get_connection() as connection: + count = connection.execute( + "SELECT COUNT(*) FROM publish_jobs WHERE task_id = ?", + (task["id"],), + ).fetchone()[0] + assert count == 0 + + def test_create_auto_publish_job_without_schedule_waits_for_send_center(): task = _create_auto_task("test-auto-publish-waiting") clip_path = _fake_video("publish_waiting_clip.mp4") diff --git a/tests/test_job_fencing.py b/tests/test_job_fencing.py index a09b7dc..be17c86 100644 --- a/tests/test_job_fencing.py +++ b/tests/test_job_fencing.py @@ -130,6 +130,29 @@ def forbidden_handler(_task_id: str): _cleanup(task_id) +def test_worker_marks_job_failed_when_subprocess_cannot_start(monkeypatch) -> None: + task_id, created = _create_task_and_job() + runner = job_worker.WorkflowJobRunner() + try: + claimed = job_service.claim_job(created["id"], runner.owner) + assert claimed and claimed["lease_token"] + monkeypatch.setattr( + job_worker, + "popen_process_group", + lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("spawn denied")), + ) + + runner._run_job_subprocess(created["id"]) + + failed = job_service.get_job(created["id"]) + assert failed["status"] == job_service.JOB_STATUS_FAILED + assert "无法启动 Job 子进程：spawn denied" in failed["error_message"] + assert failed["lease_owner"] is None + assert failed["lease_token"] is None + finally: + _cleanup(task_id) + + def test_claim_next_does_not_fail_live_max_attempt_worker() -> None: active_task_id, active_created = _create_task_and_job(max_attempts=1) queued_task_id, queued_created = _create_task_and_job() diff --git a/tests/test_pipeline_checkpoint.py b/tests/test_pipeline_checkpoint.py index 94320fe..d4f8086 100644 --- a/tests/test_pipeline_checkpoint.py +++ b/tests/test_pipeline_checkpoint.py @@ -289,6 +289,34 @@ def capture_schedule(task_id: str, context: dict) -> dict: TaskStatus.PUBLISH_JOB_CREATING.value, ] publish_outputs = stored["steps"][TaskStatus.PUBLISH_JOB_CREATING.value]["outputs"] + with job_service.job_lease_context(job["id"], "metadata-owner", job["lease_token"]): + recovered_publish = engine._reconcile_interrupted_step( + task["id"], + TaskStatus.PUBLISH_JOB_CREATING, + {"baseline": {"schedule": publish_outputs["schedule_input"]}}, + ) + assert recovered_publish is not None + assert recovered_publish["created_count"] == 1 + assert recovered_publish["created"][0]["id"] == publish_outputs["created_ids"][0] + expected_job = publish_outputs["job_evidence"][0] + with get_connection() as connection: + connection.execute( + "UPDATE output_clip SET is_active = 0 WHERE id = ?", + (expected_job["output_clip_id"],), + ) + connection.commit() + with pytest.raises(PipelineCheckpointError, match="切片已失活"): + engine._restore_checkpoint_step( + task["id"], + TaskStatus.PUBLISH_JOB_CREATING, + publish_outputs, + ) + with get_connection() as connection: + connection.execute( + "UPDATE output_clip SET is_active = 1 WHERE id = ?", + (expected_job["output_clip_id"],), + ) + connection.commit() with get_connection() as connection: connection.execute( "UPDATE publish_jobs SET platform = 'bilibili' WHERE id = ?", @@ -301,7 +329,6 @@ def capture_schedule(task_id: str, context: dict) -> dict: TaskStatus.PUBLISH_JOB_CREATING, publish_outputs, ) - expected_job = publish_outputs["job_evidence"][0] with get_connection() as connection: connection.execute( "UPDATE publish_jobs SET platform = ?, title = 'tampered title' WHERE id = ?", diff --git a/tests/test_subtitle_auto_workflow.py b/tests/test_subtitle_auto_workflow.py index a27237a..0e1e8b8 100644 --- a/tests/test_subtitle_auto_workflow.py +++ b/tests/test_subtitle_auto_workflow.py @@ -2,6 +2,7 @@ import hashlib import json +import sqlite3 import shutil import subprocess from pathlib import Path @@ -26,12 +27,14 @@ enqueue_task_subtitle_render, execute_subtitle_render_job, prepare_task_subtitle_review, + reconcile_interrupted_subtitle_job, skip_task_subtitles_and_resume, skip_task_subtitles_to_review, ) from app.services.subtitle_data_service import ( accept_suggestion_revision, approve_revision, + create_manual_revision, ensure_clip_track, get_revision, get_track, @@ -39,6 +42,7 @@ from app.services.subtitle_workflow_service import ( _create_subtitle_job, _build_ffmpeg_render_command, + _finalize_subtitle_job, _probe_media, _render_with_fallback, _validate_rendered_media, @@ -156,6 +160,27 @@ def _create_task(tmp_path: Path, *, status: str = "VIDEO_CUTTING") -> tuple[str, return task_id, output_id, output_path +def _add_output_clip(task_id: str, tmp_path: Path, *, file_name: str = "z-clip.mp4") -> tuple[str, Path]: + output_id = f"out-{uuid4().hex[:8]}" + output_path = tmp_path / f"{output_id}.mp4" + output_path.write_bytes(b"clip") + now = "2026-08-24T00:00:01+00:00" + with get_connection() as connection: + 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 (?, ?, ?, ?, 'completed', 1, 0, 5000, 5000, + 'source-v1', 'cut_commit', ?, ?) + """, + (output_id, task_id, str(output_path), file_name, now, now), + ) + connection.commit() + return output_id, output_path + + def test_schema_contains_async_render_validation_fields(): with get_connection() as connection: columns = {row[1] for row in connection.execute("PRAGMA table_info(subtitle_jobs)")} @@ -238,6 +263,15 @@ def test_prepare_review_creates_clip_draft_and_pauses(tmp_path: Path): assert get_task(task_id, include_video_probe=False)["status"] == TaskStatus.PENDING_SUBTITLE_REVIEW.value +def test_prepare_review_cannot_reopen_terminal_task(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path, status=TaskStatus.READY_TO_PUBLISH.value) + + with pytest.raises(ValueError, match="不能进入字幕审核"): + prepare_task_subtitle_review(task_id) + + assert get_task(task_id, include_video_probe=False)["status"] == TaskStatus.READY_TO_PUBLISH.value + + def test_review_page_and_api_enqueue_batch_job(tmp_path: Path): task_id, _, _ = _create_task(tmp_path) prepare_task_subtitle_review(task_id) @@ -336,6 +370,108 @@ def test_batch_approval_pins_revisions_and_records_delivery_mode(tmp_path: Path) assert config["subtitle_delivery_mode"] == "subtitled" +def test_batch_approval_rolls_back_when_later_revision_is_invalid(tmp_path: Path): + task_id, first_output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + second_output_id, _ = _add_output_clip(task_id, tmp_path) + first_track = ensure_clip_track(task_id, first_output_id) + second_track = ensure_clip_track(task_id, second_output_id) + with get_connection() as connection: + second_cues = connection.execute( + "SELECT id FROM subtitle_cues WHERE revision_id = ? ORDER BY cue_index ASC", + (second_track["active_revision_id"],), + ).fetchall() + connection.execute( + "UPDATE subtitle_cues SET start_ms = 1000 WHERE id = ?", + (second_cues[1]["id"],), + ) + connection.commit() + + with pytest.raises(ValueError, match="时间重叠"): + enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + + with get_connection() as connection: + statuses = { + row["id"]: row["status"] + for row in connection.execute( + "SELECT id, status FROM subtitle_revisions WHERE id IN (?, ?)", + (first_track["active_revision_id"], second_track["active_revision_id"]), + ).fetchall() + } + job_count = connection.execute( + "SELECT COUNT(*) FROM workflow_jobs WHERE task_id = ? AND job_type = ?", + (task_id, job_service.JOB_TYPE_SUBTITLE), + ).fetchone()[0] + assert statuses[first_track["active_revision_id"]] == "draft" + assert statuses[second_track["active_revision_id"]] == "draft" + assert job_count == 0 + + +def test_job_insert_failure_rolls_back_approval_and_delivery_mode(tmp_path: Path): + task_id, output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + track = ensure_clip_track(task_id, output_id) + with get_connection() as connection: + connection.execute( + """ + CREATE TRIGGER fail_subtitle_workflow_job_insert + BEFORE INSERT ON workflow_jobs + WHEN NEW.job_type = 'subtitle' + BEGIN + SELECT RAISE(ABORT, 'forced subtitle job insert failure'); + END + """ + ) + connection.commit() + try: + with pytest.raises(sqlite3.IntegrityError, match="forced subtitle job insert failure"): + enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + finally: + with get_connection() as connection: + connection.execute("DROP TRIGGER IF EXISTS fail_subtitle_workflow_job_insert") + connection.commit() + + with get_connection() as connection: + revision_status = connection.execute( + "SELECT status FROM subtitle_revisions WHERE id = ?", + (track["active_revision_id"],), + ).fetchone()[0] + config = json.loads( + connection.execute("SELECT auto_config_json FROM tasks WHERE id = ?", (task_id,)).fetchone()[0] + ) + job_count = connection.execute( + "SELECT COUNT(*) FROM workflow_jobs WHERE task_id = ? AND job_type = ?", + (task_id, job_service.JOB_TYPE_SUBTITLE), + ).fetchone()[0] + assert revision_status == "draft" + assert "subtitle_delivery_mode" not in config + assert job_count == 0 + + +def test_corrupt_delivery_config_fails_closed_without_partial_approval(tmp_path: Path): + task_id, output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + track = ensure_clip_track(task_id, output_id) + with get_connection() as connection: + connection.execute("UPDATE tasks SET auto_config_json = '{broken' WHERE id = ?", (task_id,)) + connection.commit() + + with pytest.raises(ValueError, match="配置已损坏"): + enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + + with get_connection() as connection: + assert connection.execute( + "SELECT status FROM subtitle_revisions WHERE id = ?", (track["active_revision_id"],) + ).fetchone()[0] == "draft" + assert connection.execute( + "SELECT auto_config_json FROM tasks WHERE id = ?", (task_id,) + ).fetchone()[0] == "{broken" + assert connection.execute( + "SELECT COUNT(*) FROM workflow_jobs WHERE task_id = ? AND job_type = ?", + (task_id, job_service.JOB_TYPE_SUBTITLE), + ).fetchone()[0] == 0 + + def test_skip_is_explicit_and_enters_clip_review_without_resume_job(tmp_path: Path): task_id, _, _ = _create_task(tmp_path) prepare_task_subtitle_review(task_id) @@ -360,6 +496,22 @@ def test_legacy_skip_endpoint_uses_new_review_semantics(tmp_path: Path): assert result["review_url"].endswith("/clips/review") +def test_skip_with_corrupt_config_keeps_review_gate_and_original_payload(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + with get_connection() as connection: + connection.execute("UPDATE tasks SET auto_config_json = '[broken' WHERE id = ?", (task_id,)) + connection.commit() + with pytest.raises(ValueError, match="配置已损坏"): + skip_task_subtitles_to_review(task_id) + with get_connection() as connection: + row = connection.execute( + "SELECT status, auto_config_json FROM tasks WHERE id = ?", (task_id,) + ).fetchone() + assert row["status"] == TaskStatus.PENDING_SUBTITLE_REVIEW.value + assert row["auto_config_json"] == "[broken" + + def test_skip_rejects_active_subtitle_job(tmp_path: Path): task_id, _, _ = _create_task(tmp_path) prepare_task_subtitle_review(task_id) @@ -368,6 +520,63 @@ def test_skip_rejects_active_subtitle_job(tmp_path: Path): skip_task_subtitles_and_resume(task_id) +def test_skip_rejects_queued_pipeline_resume_after_subtitle_completion(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + job_service.create_job( + task_id, + job_service.JOB_TYPE_AUTO_PIPELINE, + {"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + ) + + with pytest.raises(ValueError, match="后续自动流水线仍在运行"): + skip_task_subtitles_to_review(task_id) + + with get_connection() as connection: + row = connection.execute( + "SELECT status, auto_config_json FROM tasks WHERE id = ?", + (task_id,), + ).fetchone() + assert row["status"] == TaskStatus.PENDING_SUBTITLE_REVIEW.value + assert json.loads(row["auto_config_json"] or "{}").get("subtitle_delivery_mode") != "original" + + +def test_render_enqueue_rejects_existing_pipeline_resume(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + job_service.create_job( + task_id, + job_service.JOB_TYPE_AUTO_PIPELINE, + {"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + ) + + with pytest.raises(ValueError, match="不能重复执行字幕烧录"): + enqueue_task_subtitle_render( + task_id, + approve_active_revisions=True, + continue_pipeline=True, + ) + + with get_connection() as connection: + subtitle_jobs = connection.execute( + "SELECT COUNT(*) FROM workflow_jobs WHERE task_id = ? AND job_type = ?", + (task_id, job_service.JOB_TYPE_SUBTITLE), + ).fetchone()[0] + revision_statuses = { + row[0] + for row in connection.execute( + """ + SELECT sr.status FROM subtitle_revisions sr + JOIN subtitle_tracks st ON st.id = sr.track_id + WHERE st.task_id = ? AND st.active_revision_id = sr.id + """, + (task_id,), + ).fetchall() + } + assert subtitle_jobs == 0 + assert revision_statuses == {"draft"} + + def test_pending_subtitle_review_cannot_sync_publish_center(tmp_path: Path): task_id, _, _ = _create_task(tmp_path) prepare_task_subtitle_review(task_id) @@ -375,7 +584,10 @@ def test_pending_subtitle_review_cannot_sync_publish_center(tmp_path: Path): sync_task_publish_jobs(task_id) -def test_batch_execution_checkpoints_and_queues_resume(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): +def test_batch_execution_checkpoints_and_completes_with_resume_atomically( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): task_id, output_id, _ = _create_task(tmp_path) prepare_task_subtitle_review(task_id) queued = enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) @@ -394,10 +606,126 @@ def test_batch_execution_checkpoints_and_queues_resume(tmp_path: Path, monkeypat claimed = job_service.claim_job(queued["job_id"], "subtitle-test-worker") with job_service.job_lease_context(queued["job_id"], "subtitle-test-worker", claimed["lease_token"]): result = execute_subtitle_render_job(queued["job_id"], task_id, queued["job"]["payload_json"]) + completed_job, resume_job, created = job_service.mark_job_completed_with_followup( + queued["job_id"], + result, + followup_task_id=task_id, + followup_job_type=job_service.JOB_TYPE_AUTO_PIPELINE, + followup_payload={"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + result_followup_key="resume_job_id", + ) checkpoint = job_service.get_job(queued["job_id"])["checkpoint_json"] assert result["completed_count"] == 1 assert checkpoint["completed"][output_id]["revision_id"] - assert result["resume_job_id"] + assert result["resume_requested"] is True + assert result["resume_job_id"] == "" + assert created is True + assert completed_job["status"] == job_service.JOB_STATUS_COMPLETED + assert completed_job["result_json"]["resume_job_id"] == resume_job["id"] + assert resume_job["job_type"] == job_service.JOB_TYPE_AUTO_PIPELINE + assert resume_job["payload_json"]["start_step"] == TaskStatus.METADATA_GENERATING.value + + +def test_cancelled_subtitle_job_cannot_create_resume_job(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + queued = enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + claimed = job_service.claim_job(queued["job_id"], "subtitle-cancelled-worker") + + job_service.request_job_cancel(queued["job_id"]) + with job_service.job_lease_context( + queued["job_id"], + "subtitle-cancelled-worker", + claimed["lease_token"], + ): + with pytest.raises(job_service.JobLeaseLostError, match="租约已失效或已取消"): + job_service.mark_job_completed_with_followup( + queued["job_id"], + {"resume_requested": True}, + followup_task_id=task_id, + followup_job_type=job_service.JOB_TYPE_AUTO_PIPELINE, + followup_payload={"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + result_followup_key="resume_job_id", + ) + + with get_connection() as connection: + resume_count = connection.execute( + "SELECT COUNT(*) FROM workflow_jobs WHERE task_id = ? AND job_type = ?", + (task_id, job_service.JOB_TYPE_AUTO_PIPELINE), + ).fetchone()[0] + assert resume_count == 0 + assert job_service.get_job(queued["job_id"])["cancel_requested"] == 1 + + +def test_incompatible_existing_resume_job_rolls_back_subtitle_completion(tmp_path: Path): + task_id, _, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + queued = enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + existing_resume = job_service.create_job( + task_id, + job_service.JOB_TYPE_AUTO_PIPELINE, + {"retry": True, "start_step": TaskStatus.AI_ANALYZING.value}, + ) + claimed = job_service.claim_job(queued["job_id"], "subtitle-payload-worker") + + with job_service.job_lease_context( + queued["job_id"], + "subtitle-payload-worker", + claimed["lease_token"], + ): + with pytest.raises(ValueError, match="执行参数不同"): + job_service.mark_job_completed_with_followup( + queued["job_id"], + {"resume_requested": True}, + followup_task_id=task_id, + followup_job_type=job_service.JOB_TYPE_AUTO_PIPELINE, + followup_payload={"retry": False, "start_step": TaskStatus.METADATA_GENERATING.value}, + result_followup_key="resume_job_id", + ) + + current = job_service.get_job(queued["job_id"]) + assert current["status"] == job_service.JOB_STATUS_RUNNING + assert current["result_json"] == {} + assert job_service.get_job(existing_resume["id"])["payload_json"] == { + "retry": True, + "start_step": TaskStatus.AI_ANALYZING.value, + } + + +def test_restart_recovers_completed_active_result_before_checkpoint(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + task_id, output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + queued = enqueue_task_subtitle_render(task_id, approve_active_revisions=True, continue_pipeline=True) + workflow_job_id = queued["job_id"] + revision_id = queued["job"]["payload_json"]["items"][0]["revision_id"] + rendered_path = tmp_path / "already-committed.mp4" + rendered_path.write_bytes(b"verified") + child = _create_subtitle_job( + task_id, + output_id, + "completed", + output_file_path=str(rendered_path), + revision_id=revision_id, + workflow_job_id=workflow_job_id, + is_active=1, + ) + with get_connection() as connection: + connection.execute( + "UPDATE subtitle_jobs SET validation_status = 'verified', verified_at = updated_at WHERE id = ?", + (child["id"],), + ) + connection.commit() + renderer = Mock(side_effect=AssertionError("已提交结果不得重复渲染")) + monkeypatch.setattr( + "app.services.subtitle_auto_workflow_service.render_subtitles_for_output_clip", + renderer, + ) + claimed = job_service.claim_job(workflow_job_id, "subtitle-recovery-worker") + with job_service.job_lease_context(workflow_job_id, claimed["lease_owner"], claimed["lease_token"]): + result = execute_subtitle_render_job(workflow_job_id, task_id, queued["job"]["payload_json"]) + renderer.assert_not_called() + assert result["completed"][output_id]["subtitle_job_id"] == child["id"] + assert job_service.get_job(workflow_job_id)["checkpoint_json"]["completed"][output_id]["subtitle_job_id"] == child["id"] def test_cancel_cleanup_is_precise_and_retry_keeps_checkpoint( @@ -469,6 +797,176 @@ def test_cancel_cleanup_is_precise_and_retry_keeps_checkpoint( assert retried["checkpoint_json"] == checkpoint +def test_takeover_rejects_stale_finalize_and_reconciles_owned_temp( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +): + task_id, output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + track = ensure_clip_track(task_id, output_id) + approve_revision(track["id"], track["active_revision_id"]) + old_final = tmp_path / "old-active.mp4" + old_final.write_bytes(b"old") + old_active = _create_subtitle_job( + task_id, + output_id, + "completed", + output_file_path=str(old_final), + revision_id=track["active_revision_id"], + is_active=1, + ) + queued = enqueue_task_subtitle_render(task_id, continue_pipeline=True) + workflow_job_id = queued["job_id"] + stale_claim = job_service.claim_job(workflow_job_id, "subtitle-stale-worker") + child = _create_subtitle_job( + task_id, + output_id, + "processing", + revision_id=track["active_revision_id"], + workflow_job_id=workflow_job_id, + is_active=0, + ) + managed_dir = tmp_path / "managed-subtitled" + managed_dir.mkdir() + temporary_path = managed_dir / f".new-output.{workflow_job_id}.part.mp4" + temporary_path.write_bytes(b"new") + workflow_marker = hashlib.sha256(workflow_job_id.encode("utf-8")).hexdigest()[:12] + orphan_final = managed_dir / f"clip_subtitled_{workflow_marker}_orphan.mp4" + orphan_final.write_bytes(b"orphan") + referenced_final = managed_dir / f"clip_subtitled_{workflow_marker}_referenced.mp4" + referenced_final.write_bytes(b"referenced") + with get_connection() as connection: + connection.execute( + "UPDATE subtitle_jobs SET output_file_path = ? WHERE id = ?", + (str(referenced_final), old_active["id"]), + ) + connection.commit() + final_path = managed_dir / "new-output.mp4" + subtitle_path = managed_dir / "revision.ass" + subtitle_path.write_text("subtitle", encoding="utf-8") + monkeypatch.setattr( + "app.services.subtitle_auto_workflow_service.get_artifact_paths", + lambda _task_id: {"subtitled_dir": managed_dir}, + ) + with get_connection() as connection: + connection.execute( + "UPDATE workflow_jobs SET lease_expires_at = '2000-01-01T00:00:00+00:00' WHERE id = ?", + (workflow_job_id,), + ) + connection.commit() + current_claim = job_service.claim_job(workflow_job_id, "subtitle-current-worker") + + with job_service.job_lease_context( + workflow_job_id, + stale_claim["lease_owner"], + stale_claim["lease_token"], + ): + with pytest.raises(job_service.JobLeaseLostError, match="租约已失效"): + _finalize_subtitle_job( + task_id=task_id, + output_clip_id=output_id, + revision_id=track["active_revision_id"], + subtitle_job_id=child["id"], + workflow_job_id=workflow_job_id, + subtitle_path=subtitle_path, + temporary_path=temporary_path, + output_path=final_path, + validation={"duration": 3.0}, + encoder="libx264", + audio_mode="aac", + ) + assert final_path.exists() is False + assert temporary_path.exists() is True + with get_connection() as connection: + assert connection.execute( + "SELECT is_active FROM subtitle_jobs WHERE id = ?", (old_active["id"],) + ).fetchone()[0] == 1 + assert connection.execute( + "SELECT status FROM subtitle_jobs WHERE id = ?", (child["id"],) + ).fetchone()[0] == "processing" + connection.execute( + "UPDATE subtitle_jobs SET status = 'queued' WHERE id = ?", + (child["id"],), + ) + connection.commit() + + with job_service.job_lease_context( + workflow_job_id, + current_claim["lease_owner"], + current_claim["lease_token"], + ): + recovery = reconcile_interrupted_subtitle_job(workflow_job_id) + assert recovery["interrupted_count"] == 1 + assert recovery["deleted_artifact_count"] == 2 + assert temporary_path.exists() is False + assert orphan_final.exists() is False + assert referenced_final.read_bytes() == b"referenced" + assert old_final.read_bytes() == b"old" + with get_connection() as connection: + assert connection.execute( + "SELECT status FROM subtitle_jobs WHERE id = ?", (child["id"],) + ).fetchone()[0] == "failed" + + +def test_old_revision_cannot_activate_after_editor_moves_forward(tmp_path: Path): + task_id, output_id, _ = _create_task(tmp_path) + prepare_task_subtitle_review(task_id) + track = ensure_clip_track(task_id, output_id) + approved = approve_revision(track["id"], track["active_revision_id"]) + old_final = tmp_path / "current-active.mp4" + old_final.write_bytes(b"current") + current_active = _create_subtitle_job( + task_id, + output_id, + "completed", + output_file_path=str(old_final), + revision_id=approved["id"], + is_active=1, + ) + child = _create_subtitle_job( + task_id, + output_id, + "processing", + revision_id=approved["id"], + is_active=0, + ) + newer = create_manual_revision( + track["id"], + base_revision_id=approved["id"], + cues=approved["cues"], + note="渲染期间的新编辑", + ) + temporary_path = tmp_path / ".old-revision.part.mp4" + temporary_path.write_bytes(b"old-revision-render") + final_path = tmp_path / "old-revision-final.mp4" + subtitle_path = tmp_path / "old-revision.ass" + subtitle_path.write_text("subtitle", encoding="utf-8") + + with pytest.raises(ValueError, match="revision 已变化"): + _finalize_subtitle_job( + task_id=task_id, + output_clip_id=output_id, + revision_id=approved["id"], + subtitle_job_id=child["id"], + workflow_job_id=None, + subtitle_path=subtitle_path, + temporary_path=temporary_path, + output_path=final_path, + validation={"duration": 3.0}, + encoder="libx264", + audio_mode="aac", + ) + assert final_path.exists() is False + assert get_track(track["id"])["active_revision_id"] == newer["id"] + with get_connection() as connection: + assert connection.execute( + "SELECT is_active FROM subtitle_jobs WHERE id = ?", (current_active["id"],) + ).fetchone()[0] == 1 + assert connection.execute( + "SELECT is_active FROM subtitle_jobs WHERE id = ?", (child["id"],) + ).fetchone()[0] == 0 + + def test_ffmpeg_command_maps_optional_audio_and_forces_compatible_video(tmp_path: Path): command = _build_ffmpeg_render_command( tmp_path / "input.mp4",