From f21d97262583178c8ebece87785d1a048cc1af9a Mon Sep 17 00:00:00 2001 From: Codex Date: Sun, 23 Aug 2026 22:29:34 +0800 Subject: [PATCH 01/28] =?UTF-8?q?=E6=96=B0=E5=A2=9E=EF=BC=9A=E9=95=BF?= =?UTF-8?q?=E7=9B=B4=E6=92=AD=E4=BB=BB=E5=8A=A1=E5=9F=BA=E7=A1=80=E8=AE=BE?= =?UTF-8?q?=E6=96=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- DEVELOPMENT_LOG.md | 12 + NEXT_STEPS.md | 9 + app/db/database.py | 126 +++++++- app/main.py | 15 + app/models/task.py | 8 +- app/routers/tasks.py | 68 ++-- app/services/database_backup_service.py | 37 +++ app/services/job_service.py | 179 ++++++++++- app/services/job_worker.py | 149 ++++++++- app/services/job_worker_process.py | 24 ++ app/services/managed_process_service.py | 36 +++ app/services/media_preflight_service.py | 179 +++++++++++ app/services/pipeline_engine.py | 35 ++- app/services/storage_service.py | 1 + app/services/task_lifecycle_service.py | 47 ++- app/services/task_service.py | 12 +- app/services/transcript_service.py | 292 ++++++++++++++++-- app/services/transcript_workflow_service.py | 52 +++- .../transcription_checkpoint_service.py | 180 +++++++++++ app/static/css/styles.css | 45 +++ app/static/js/app.js | 150 +++++++-- app/templates/new_task.html | 52 +++- app/templates/task_detail.html | 6 +- docs/ARCHITECTURE.md | 13 +- docs/DATABASE_SCHEMA.md | 10 + docs/TASK_FLOW.md | 9 + docs/UI_REFERENCE.md | 8 + .../2026-08-23-long-live-foundation.md | 54 ++++ tests/test_auto_pipeline.py | 14 +- tests/test_long_live_foundation.py | 158 ++++++++++ tests/test_media_storage_lifecycle.py | 14 +- tests/test_selection_profile_default.py | 142 +++------ tests/test_split_services.py | 16 +- tests/test_task_defaults.py | 7 +- tests/test_versioning_rollback.py | 1 + 35 files changed, 1942 insertions(+), 218 deletions(-) create mode 100644 app/services/job_worker_process.py create mode 100644 app/services/managed_process_service.py create mode 100644 app/services/media_preflight_service.py create mode 100644 app/services/transcription_checkpoint_service.py create mode 100644 docs/agent_tasks/2026-08-23-long-live-foundation.md create mode 100644 tests/test_long_live_foundation.py diff --git a/DEVELOPMENT_LOG.md b/DEVELOPMENT_LOG.md index 4028a04..9ecb3f0 100644 --- a/DEVELOPMENT_LOG.md +++ b/DEVELOPMENT_LOG.md @@ -1,5 +1,17 @@ # Development Log +## 2026-08-23 长直播基础设施（PR 1） + +- 新建任务移除综艺隐藏默认值，页面、multipart 上传与 JSON API 都要求显式选择三种模式；历史任务保留原模式。 +- 新建页接入受根目录保护的本地/NAS已有文件浏览，4 GB 以上素材引导使用引用模式，不复制外部唯一原片。 +- 增加媒体/磁盘预检、6 小时验收范围提示，以及长直播每小时密度和总量参数。 +- 工作流 Job 增加 lease、heartbeat、取消、重试和 checkpoint；应用启动单重型 worker，每个任务由独立 Python 子进程执行，全自动流水线和转写进入持久化队列。 +- 转写新增 run/chunk、源指纹、逐块原子提交和校验恢复；本地/火山结果统一支持置信度和词级毫秒时间戳。 +- 音频提取改为临时输出原子替换，并支持真实进度、无进展超时和终止进程树；显式 CPU 配置保持优先。 +- 本阶段不实现 `long_live_talk` 分层 AI 选片和新字幕编辑器；分别留给 PR 2 与 PR 3。 +- 验证通过：Ruff、Python 编译、`app.js` 语法、PR 1 定向测试 `85 passed`、完整测试 `455 passed`。 +- 六小时低码率实测通过：H.264/AAC 原片为 `21600s`，媒体/空间预检正常；实际提取 `16kHz` 单声道 PCM WAV 仍为 `21600s`（约 691 MB），验收临时媒体已清理。 + ## 2026-08-23 v2.1.0 主线整合与片段审核稳定性增强 - 审计全部本地/远端分支、开放 PR 和 5 个 worktree：图片文档、续接排期、发送中心文案规则等近期分支已进入 `origin/master`；当前 PR #38 是唯一需要整体合入的有效新主线。存在冲突、失败 CI 或旧 opencli 架构的历史分支不做盲目整分支合并。 diff --git a/NEXT_STEPS.md b/NEXT_STEPS.md index 1183418..a4e75a2 100644 --- a/NEXT_STEPS.md +++ b/NEXT_STEPS.md @@ -1,5 +1,14 @@ # Next Steps +## 2026-08-23 长直播四阶段进度 + +- [x] PR 1：模式必选、已有文件入口、媒体/磁盘预检、持久化重型 Job、转写断点与词级时间戳。 +- [ ] PR 2：`long_live_talk` 5 分钟重叠窗口、每小时覆盖、全局配额、去重、90% 覆盖门禁和窗口级恢复。 +- [ ] PR 3：统一字幕 track/revision/cue、pysubs2 导入导出、wavesurfer 波形与专业编辑器。 +- [ ] PR 4：字幕审核暂停、AI 建议 revision、异步批量烧录、NVENC 回退和发送中心门禁。 + +当前验证重点：完成 PR 1 专项与全量回归；不要在 PR 1 宣称长直播选片或新字幕系统已经可用。 + ## 2026-08-23 v2.1.0 主线同步后检查 1. 最终提交进入 `master` 后先查看 GitHub CI；只有 Python、Windows host smoke 和 Docker image smoke 全部通过，才把该提交作为 2.1.0 候选版本。 diff --git a/app/db/database.py b/app/db/database.py index aa9f677..8bbab47 100644 --- a/app/db/database.py +++ b/app/db/database.py @@ -5,7 +5,7 @@ from contextlib import contextmanager from app.core.config import settings -from app.services.database_backup_service import create_publish_migration_backup +from app.services.database_backup_service import create_publish_migration_backup, create_schema_migration_backup DEFAULT_AI_PROMPT_PRESET_ID = "preset_001" @@ -32,6 +32,13 @@ def init_db() -> None: settings.data_dir.mkdir(parents=True, exist_ok=True) settings.tasks_dir.mkdir(parents=True, exist_ok=True) + if _requires_long_live_schema_migration(settings.database_path): + create_schema_migration_backup( + settings.database_path, + settings.data_dir / "backups", + "long-live-foundation", + ) + with get_connection() as connection: connection.executescript( """ @@ -47,6 +54,8 @@ def init_db() -> None: candidate_clip_count INTEGER NOT NULL DEFAULT 12, selection_profile TEXT NOT NULL DEFAULT 'general', final_clip_target INTEGER NOT NULL DEFAULT 5, + highlight_density_per_hour INTEGER NOT NULL DEFAULT 4, + highlight_total_limit INTEGER NOT NULL DEFAULT 30, ai_preference TEXT, ai_prompt_preset_id TEXT NOT NULL DEFAULT 'preset_001', auto_mode INTEGER NOT NULL DEFAULT 0, @@ -308,6 +317,55 @@ def init_db() -> None: updated_at TEXT NOT NULL, started_at TEXT, finished_at TEXT, + attempt_count INTEGER NOT NULL DEFAULT 0, + max_attempts INTEGER NOT NULL DEFAULT 3, + next_attempt_at TEXT, + lease_owner TEXT, + lease_expires_at TEXT, + heartbeat_at TEXT, + cancel_requested INTEGER NOT NULL DEFAULT 0, + checkpoint_json TEXT, + checkpoint_updated_at TEXT, + FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + + CREATE TABLE IF NOT EXISTS transcription_runs ( + id TEXT PRIMARY KEY, + task_id TEXT NOT NULL, + source_fingerprint TEXT NOT NULL, + provider TEXT NOT NULL, + model TEXT NOT NULL DEFAULT '', + device TEXT NOT NULL DEFAULT '', + compute_type TEXT NOT NULL DEFAULT '', + chunk_seconds INTEGER NOT NULL, + overlap_seconds INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'processing', + total_chunks INTEGER NOT NULL DEFAULT 0, + completed_chunks INTEGER NOT NULL DEFAULT 0, + is_active INTEGER NOT NULL DEFAULT 0, + error_message TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + completed_at TEXT, + FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + + CREATE TABLE IF NOT EXISTS transcription_chunks ( + id TEXT PRIMARY KEY, + run_id TEXT NOT NULL, + task_id TEXT NOT NULL, + chunk_index INTEGER NOT NULL, + start_ms INTEGER NOT NULL, + end_ms INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', + attempt_count INTEGER NOT NULL DEFAULT 0, + result_json TEXT, + result_checksum TEXT, + error_message TEXT, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, + UNIQUE(run_id, chunk_index), + FOREIGN KEY(run_id) REFERENCES transcription_runs(id) ON DELETE CASCADE, FOREIGN KEY(task_id) REFERENCES tasks(id) ); @@ -354,6 +412,7 @@ def init_db() -> None: _migrate_publish_job_events_table(connection) _restore_legacy_user_cancelled_publish_jobs(connection) _migrate_workflow_jobs_table(connection) + _migrate_transcription_tables(connection) _migrate_cut_runs_table(connection) _seed_ai_prompt_presets(connection) _seed_subtitle_style_preset(connection) @@ -362,6 +421,26 @@ def init_db() -> None: connection.commit() +def _requires_long_live_schema_migration(database_path) -> bool: + """只对已经存在且确实缺少新结构的数据库创建一次迁移前备份。""" + if not database_path.exists() or database_path.stat().st_size == 0: + return False + try: + connection = sqlite3.connect(f"{database_path.resolve().as_uri()}?mode=ro", uri=True, timeout=10) + task_columns = {row[1] for row in connection.execute("PRAGMA table_info(tasks)").fetchall()} + job_columns = {row[1] for row in connection.execute("PRAGMA table_info(workflow_jobs)").fetchall()} + table_names = {row[0] for row in connection.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()} + finally: + if "connection" in locals(): + connection.close() + return ( + "highlight_density_per_hour" not in task_columns + or "lease_owner" not in job_columns + or "transcription_runs" not in table_names + or "transcription_chunks" not in table_names + ) + + def _get_table_columns(connection: sqlite3.Connection, table_name: str) -> set[str]: rows = connection.execute(f"PRAGMA table_info({table_name})").fetchall() return {row["name"] for row in rows} @@ -393,6 +472,10 @@ def _create_indexes(connection: sqlite3.Connection) -> None: "CREATE INDEX IF NOT EXISTS idx_publish_jobs_execution ON publish_jobs(execution_id)", "CREATE INDEX IF NOT EXISTS idx_publish_jobs_history_visibility ON publish_jobs(history_hidden, platform, status, created_at)", "CREATE INDEX IF NOT EXISTS idx_publish_job_events_job_time ON publish_job_events(job_id, occurred_at)", + "CREATE INDEX IF NOT EXISTS idx_workflow_jobs_claim ON workflow_jobs(status, next_attempt_at, created_at)", + "CREATE INDEX IF NOT EXISTS idx_workflow_jobs_task_type_status ON workflow_jobs(task_id, job_type, status)", + "CREATE INDEX IF NOT EXISTS idx_transcription_runs_task_active ON transcription_runs(task_id, is_active, updated_at)", + "CREATE INDEX IF NOT EXISTS idx_transcription_chunks_run_status ON transcription_chunks(run_id, status, chunk_index)", """CREATE UNIQUE INDEX IF NOT EXISTS uq_publish_jobs_active_clip_platform_mode ON publish_jobs(output_clip_id, platform, publish_mode) WHERE status IN ('DRAFT', 'WAITING', 'SCHEDULED', 'PUBLISHING', 'NEED_REVIEW') @@ -420,6 +503,8 @@ def _migrate_tasks_table(connection: sqlite3.Connection) -> None: "candidate_clip_count": "ALTER TABLE tasks ADD COLUMN candidate_clip_count INTEGER NOT NULL DEFAULT 12", "selection_profile": "ALTER TABLE tasks ADD COLUMN selection_profile TEXT NOT NULL DEFAULT 'general'", "final_clip_target": "ALTER TABLE tasks ADD COLUMN final_clip_target INTEGER NOT NULL DEFAULT 5", + "highlight_density_per_hour": "ALTER TABLE tasks ADD COLUMN highlight_density_per_hour INTEGER NOT NULL DEFAULT 4", + "highlight_total_limit": "ALTER TABLE tasks ADD COLUMN highlight_total_limit INTEGER NOT NULL DEFAULT 30", "ai_preference": "ALTER TABLE tasks ADD COLUMN ai_preference TEXT", "ai_prompt_preset_id": "ALTER TABLE tasks ADD COLUMN ai_prompt_preset_id TEXT NOT NULL DEFAULT 'preset_001'", "auto_mode": "ALTER TABLE tasks ADD COLUMN auto_mode INTEGER NOT NULL DEFAULT 0", @@ -490,8 +575,10 @@ def _migrate_tasks_table(connection: sqlite3.Connection) -> None: UPDATE tasks SET task_dir_name = id WHERE task_dir_name IS NULL OR task_dir_name = ''; UPDATE tasks SET is_deleted = 0 WHERE is_deleted IS NULL; UPDATE tasks SET ai_prompt_preset_id = 'preset_001' WHERE ai_prompt_preset_id IS NULL OR ai_prompt_preset_id = ''; - UPDATE tasks SET selection_profile = 'general' WHERE selection_profile NOT IN ('general', 'variety_comedy') OR selection_profile IS NULL OR selection_profile = ''; + UPDATE tasks SET selection_profile = 'general' WHERE selection_profile NOT IN ('general', 'variety_comedy', 'long_live_talk') OR selection_profile IS NULL OR selection_profile = ''; UPDATE tasks SET final_clip_target = 5 WHERE final_clip_target IS NULL OR final_clip_target < 1 OR final_clip_target > 12; + UPDATE tasks SET highlight_density_per_hour = 4 WHERE highlight_density_per_hour IS NULL OR highlight_density_per_hour < 1 OR highlight_density_per_hour > 10; + UPDATE tasks SET highlight_total_limit = 30 WHERE highlight_total_limit IS NULL OR highlight_total_limit < 1 OR highlight_total_limit > 50; UPDATE tasks SET source_type = 'upload' WHERE source_type NOT IN ('upload', 'nas') OR source_type IS NULL OR source_type = ''; UPDATE tasks SET platform = 'douyin' WHERE platform IN ('抖音', 'douyin'); @@ -1135,12 +1222,47 @@ def _migrate_workflow_jobs_table(connection: sqlite3.Connection) -> None: "updated_at": "ALTER TABLE workflow_jobs ADD COLUMN updated_at TEXT NOT NULL DEFAULT ''", "started_at": "ALTER TABLE workflow_jobs ADD COLUMN started_at TEXT", "finished_at": "ALTER TABLE workflow_jobs ADD COLUMN finished_at TEXT", + "attempt_count": "ALTER TABLE workflow_jobs ADD COLUMN attempt_count INTEGER NOT NULL DEFAULT 0", + "max_attempts": "ALTER TABLE workflow_jobs ADD COLUMN max_attempts INTEGER NOT NULL DEFAULT 3", + "next_attempt_at": "ALTER TABLE workflow_jobs ADD COLUMN next_attempt_at TEXT", + "lease_owner": "ALTER TABLE workflow_jobs ADD COLUMN lease_owner TEXT", + "lease_expires_at": "ALTER TABLE workflow_jobs ADD COLUMN lease_expires_at TEXT", + "heartbeat_at": "ALTER TABLE workflow_jobs ADD COLUMN heartbeat_at TEXT", + "cancel_requested": "ALTER TABLE workflow_jobs ADD COLUMN cancel_requested INTEGER NOT NULL DEFAULT 0", + "checkpoint_json": "ALTER TABLE workflow_jobs ADD COLUMN checkpoint_json TEXT", + "checkpoint_updated_at": "ALTER TABLE workflow_jobs ADD COLUMN checkpoint_updated_at TEXT", } for column, statement in migrations.items(): if column not in columns: connection.execute(statement) +def _migrate_transcription_tables(connection: sqlite3.Connection) -> None: + """创建转写分块 checkpoint 表；重复启动不会覆盖现有结果。""" + connection.executescript( + """ + CREATE TABLE IF NOT EXISTS transcription_runs ( + id TEXT PRIMARY KEY, task_id TEXT NOT NULL, source_fingerprint TEXT NOT NULL, + provider TEXT NOT NULL, model TEXT NOT NULL DEFAULT '', device TEXT NOT NULL DEFAULT '', + compute_type TEXT NOT NULL DEFAULT '', chunk_seconds INTEGER NOT NULL, + overlap_seconds INTEGER NOT NULL, status TEXT NOT NULL DEFAULT 'processing', + total_chunks INTEGER NOT NULL DEFAULT 0, completed_chunks INTEGER NOT NULL DEFAULT 0, + is_active INTEGER NOT NULL DEFAULT 0, error_message TEXT, created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, completed_at TEXT, FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + CREATE TABLE IF NOT EXISTS transcription_chunks ( + id TEXT PRIMARY KEY, run_id TEXT NOT NULL, task_id TEXT NOT NULL, + chunk_index INTEGER NOT NULL, start_ms INTEGER NOT NULL, end_ms INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'queued', attempt_count INTEGER NOT NULL DEFAULT 0, + result_json TEXT, result_checksum TEXT, error_message TEXT, created_at TEXT NOT NULL, + updated_at TEXT NOT NULL, UNIQUE(run_id, chunk_index), + FOREIGN KEY(run_id) REFERENCES transcription_runs(id) ON DELETE CASCADE, + FOREIGN KEY(task_id) REFERENCES tasks(id) + ); + """ + ) + + def _migrate_cut_runs_table(connection: sqlite3.Connection) -> None: """cut_runs 表的列级迁移，兼容未来新增字段""" columns = _get_table_columns(connection, "cut_runs") diff --git a/app/main.py b/app/main.py index c06b7df..b67d83e 100644 --- a/app/main.py +++ b/app/main.py @@ -4,6 +4,7 @@ from fastapi import FastAPI, Request, Response from fastapi.responses import FileResponse, JSONResponse +from fastapi.exceptions import RequestValidationError from fastapi.staticfiles import StaticFiles from app.core.config import settings @@ -11,6 +12,7 @@ from app.routers import ai_prompts, files, media, pages, publish, settings as settings_router, tasks from app.services.publish_scheduler import start_scheduler_background from app.services.storage_service import configure_runtime_media_storage +from app.services.job_worker import WorkflowJobRunner # /media 和 /static 的 Origin 白名单 @@ -50,11 +52,15 @@ async def lifespan(app: FastAPI): previous_temp_env = {name: os.environ.get(name) for name in ("TEMP", "TMP")} app.state.media_storage = configure_runtime_media_storage() init_db() + workflow_job_runner = WorkflowJobRunner() + workflow_job_runner.start() + app.state.workflow_job_runner = workflow_job_runner scheduler = await start_scheduler_background() app.state.publish_scheduler = scheduler try: yield finally: + workflow_job_runner.stop() if scheduler: scheduler.stop() tempfile.tempdir = previous_temp @@ -73,6 +79,15 @@ async def lifespan(app: FastAPI): ) +@app.exception_handler(RequestValidationError) +async def request_validation_error_handler(_request: Request, exc: RequestValidationError) -> JSONResponse: + for error in exc.errors(): + location = tuple(error.get("loc") or ()) + if "selection_profile" in location: + return JSONResponse(status_code=422, content={"detail": "请选择有效的选片模式"}) + return JSONResponse(status_code=422, content={"detail": exc.errors()}) + + @app.middleware("http") async def security_middleware(request: Request, call_next): request_path = request.url.path diff --git a/app/models/task.py b/app/models/task.py index acc1050..abe3967 100644 --- a/app/models/task.py +++ b/app/models/task.py @@ -45,8 +45,10 @@ class TaskCreate(BaseModel): nas_file_path: Optional[str] = None max_clip_duration: int = Field(default=10, ge=1, le=60) candidate_clip_count: int = Field(default=12, ge=1, le=50) - selection_profile: Literal["general", "variety_comedy"] = "variety_comedy" + selection_profile: Literal["general", "variety_comedy", "long_live_talk"] final_clip_target: int = Field(default=5, ge=1, le=12) + highlight_density_per_hour: int = Field(default=4, ge=1, le=10) + highlight_total_limit: int = Field(default=30, ge=1, le=50) ai_preference: Optional[str] = None auto_mode: bool = False auto_clip_count: str = Field(default="auto", max_length=10) @@ -107,8 +109,10 @@ class TaskCandidateClipCountUpdate(BaseModel): class TaskSelectionSettingsUpdate(BaseModel): - selection_profile: Literal["general", "variety_comedy"] = "variety_comedy" + selection_profile: Literal["general", "variety_comedy", "long_live_talk"] final_clip_target: int = Field(default=5, ge=1, le=12) + highlight_density_per_hour: int = Field(default=4, ge=1, le=10) + highlight_total_limit: int = Field(default=30, ge=1, le=50) class ClipFeedbackCreate(BaseModel): diff --git a/app/routers/tasks.py b/app/routers/tasks.py index eabffb5..531488b 100644 --- a/app/routers/tasks.py +++ b/app/routers/tasks.py @@ -27,7 +27,6 @@ ) from app.services.task_lifecycle_service import TaskDeletionConflictError from app.services import job_service -from app.services import job_worker router = APIRouter(prefix="/api/tasks", tags=["tasks"]) @@ -56,8 +55,10 @@ async def create_upload_task( platform: str = Form("general"), max_clip_duration: int = Form(10), candidate_clip_count: int = Form(12), - selection_profile: str = Form("variety_comedy"), + selection_profile: str | None = Form(None), final_clip_target: int = Form(5), + highlight_density_per_hour: int = Form(4), + highlight_total_limit: int = Form(30), ai_preference: str | None = Form(None), auto_mode: bool = Form(False), auto_clip_count: str = Form("auto"), @@ -71,6 +72,8 @@ async def create_upload_task( auto_metadata_use_ai: bool = Form(False), video_file: UploadFile = File(...), ) -> dict: + if not selection_profile: + raise HTTPException(status_code=422, detail="请选择选片模式") task_id = uuid4().hex[:12] task_dir_name = allocate_task_dir_name(task_name, exclude_task_id=task_id) task_record_created = False @@ -91,6 +94,8 @@ async def create_upload_task( candidate_clip_count=candidate_clip_count, selection_profile=selection_profile, final_clip_target=final_clip_target, + highlight_density_per_hour=highlight_density_per_hour, + highlight_total_limit=highlight_total_limit, ai_preference=ai_preference, auto_mode=auto_mode, auto_clip_count=auto_clip_count, @@ -205,6 +210,8 @@ async def patch_task_selection_settings(task_id: str, payload: TaskSelectionSett task_id, payload.selection_profile, payload.final_clip_target, + payload.highlight_density_per_hour, + payload.highlight_total_limit, ) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc @@ -242,12 +249,23 @@ async def process_transcript_workflow( provider: str | None = Query(default=None, pattern="^(remote|local)$"), ) -> dict: try: - return task_service.process_task_transcript_workflow( - task_id, - background_tasks=background_tasks, - force=force, - provider=provider, + task = task_service.get_task(task_id, include_video_probe=False) + if not task: + raise ValueError("任务不存在") + if task.get("transcript_exists") and not force: + return {"status": "completed", "message": "转写已经生成，无需重复处理。", "task": task} + job, created = job_service.create_or_get_active_job( + task_id=task_id, + job_type=job_service.JOB_TYPE_TRANSCRIPT, + payload={"force": force, "provider": provider}, ) + return { + "status": job["status"], + "message": "转写任务已加入持久化队列" if created else "已有转写任务正在排队或运行", + "job_id": job["id"], + "job": job, + "task": task, + } except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc except RuntimeError as exc: @@ -257,6 +275,13 @@ async def process_transcript_workflow( @router.post("/{task_id}/process/transcript-cancel") async def cancel_transcript(task_id: str) -> dict: try: + active_jobs = [ + job for job in job_service.list_jobs(task_id=task_id) + if job.get("job_type") == job_service.JOB_TYPE_TRANSCRIPT + and job.get("status") in {job_service.JOB_STATUS_QUEUED, job_service.JOB_STATUS_RUNNING} + ] + if active_jobs: + job_service.request_job_cancel(active_jobs[0]["id"]) return task_service.cancel_task_transcript(task_id) except ValueError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc @@ -436,7 +461,6 @@ async def save_subtitle_style(payload: SubtitleStyleUpdate) -> dict: @router.post("/{task_id}/process/cuts-async") async def process_video_cuts_async( task_id: str, - background_tasks: BackgroundTasks, ) -> dict: """自动切片异步版：创建 job 后立即返回，后台执行切割""" # 先验证任务存在 @@ -453,9 +477,6 @@ async def process_video_cuts_async( except Exception as exc: raise HTTPException(status_code=500, detail=f"创建切片任务失败：{exc}") from exc - if created: - background_tasks.add_task(_run_job_in_background, job["id"]) - return { "status": job["status"], "message": ( @@ -478,6 +499,22 @@ async def get_job_status(job_id: str) -> dict: return job +@router.post("/jobs/{job_id}/cancel") +async def cancel_job(job_id: str) -> dict: + job = job_service.request_job_cancel(job_id) + if not job: + raise HTTPException(status_code=404, detail="job 不存在") + return {"status": job["status"], "message": job.get("message"), "job": job} + + +@router.post("/jobs/{job_id}/retry") +async def retry_job(job_id: str) -> dict: + job = job_service.retry_job(job_id) + if not job: + raise HTTPException(status_code=409, detail="只有失败或已取消的 job 可以重试") + return {"status": job["status"], "message": job.get("message"), "job": job} + + @router.get("/{task_id}/jobs") async def list_task_jobs(task_id: str, status: str | None = Query(default=None)) -> dict: """查看某个任务下的所有 job 记录""" @@ -490,12 +527,3 @@ async def list_task_jobs(task_id: str, status: str | None = Query(default=None)) "jobs": jobs, "count": len(jobs), } - - -def _run_job_in_background(job_id: str) -> None: - """后台线程执行 job""" - try: - job_worker.execute_job(job_id) - except Exception: - # 错误已由 worker 记录到 job 的 error_message 字段 - pass diff --git a/app/services/database_backup_service.py b/app/services/database_backup_service.py index f868c51..b0cead4 100644 --- a/app/services/database_backup_service.py +++ b/app/services/database_backup_service.py @@ -265,6 +265,43 @@ def create_publish_migration_backup( source_connection.close() +def create_schema_migration_backup(database_path: Path, backup_dir: Path, label: str) -> Path: + """使用 SQLite Online Backup API 创建一次带完整性检查的结构迁移备份。""" + database_path = database_path.resolve() + backup_dir = backup_dir.resolve() + backup_dir.mkdir(parents=True, exist_ok=True) + safe_label = "".join(character for character in label.lower() if character.isalnum() or character == "-") + timestamp = datetime.now(BACKUP_TIMEZONE).strftime("%Y%m%d-%H%M%S-%f") + final_path = backup_dir / f"workflow-before-{safe_label}-{timestamp}-{uuid4().hex[:8]}.sqlite3" + temporary_path = final_path.with_suffix(final_path.suffix + ".tmp") + source_connection: sqlite3.Connection | None = None + backup_connection: sqlite3.Connection | None = None + try: + source_connection = sqlite3.connect(f"{database_path.as_uri()}?mode=ro", uri=True, timeout=10) + backup_connection = sqlite3.connect(str(temporary_path), timeout=10) + source_connection.backup(backup_connection) + backup_connection.close() + backup_connection = None + source_connection.close() + source_connection = None + integrity = sqlite_quick_check(temporary_path) + if integrity != "ok": + raise BackupSafetyError(f"新备份完整性检查失败：{integrity}") + os.replace(temporary_path, final_path) + return final_path + except Exception as exc: + if temporary_path.exists(): + temporary_path.unlink() + if isinstance(exc, BackupSafetyError): + raise + raise BackupSafetyError(f"创建结构迁移前备份失败：{exc}") from exc + finally: + if backup_connection is not None: + backup_connection.close() + if source_connection is not None: + source_connection.close() + + def create_media_cleanup_backup( database_path: Path, backup_dir: Path, diff --git a/app/services/job_service.py b/app/services/job_service.py index 763c7ca..38f2652 100644 --- a/app/services/job_service.py +++ b/app/services/job_service.py @@ -5,7 +5,7 @@ """ import json -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from typing import Optional from uuid import uuid4 @@ -17,6 +17,7 @@ JOB_TYPE_TRANSCRIPT = "transcript" JOB_TYPE_SUBTITLE = "subtitle" JOB_TYPE_PUBLISH = "publish" +JOB_TYPE_AUTO_PIPELINE = "auto_pipeline" # ── job 状态枚举 ───────────────────────────────────────────────── JOB_STATUS_QUEUED = "queued" @@ -39,6 +40,7 @@ JOB_TYPE_TRANSCRIPT: "转写", JOB_TYPE_SUBTITLE: "字幕", JOB_TYPE_PUBLISH: "发布", + JOB_TYPE_AUTO_PIPELINE: "全自动流水线", } @@ -49,7 +51,7 @@ def _now_iso() -> str: def _row_to_dict(row) -> dict: """将 sqlite3.Row 转为普通字典，并解析 JSON 字段""" job = dict(row) - for field in ("payload_json", "result_json"): + for field in ("payload_json", "result_json", "checkpoint_json"): raw = job.get(field) if isinstance(raw, str) and raw: try: @@ -241,6 +243,160 @@ def mark_job_running(job_id: str) -> dict | None: return get_job(job_id) +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") + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + "SELECT status, lease_expires_at, cancel_requested, attempt_count, max_attempts FROM workflow_jobs WHERE id = ?", + (job_id,), + ).fetchone() + if not row: + connection.commit() + return None + expired = row["status"] == JOB_STATUS_RUNNING and ( + not row["lease_expires_at"] or str(row["lease_expires_at"]) <= now_iso + ) + claimable = row["status"] == JOB_STATUS_QUEUED or expired + if not claimable or int(row["cancel_requested"] or 0) or int(row["attempt_count"] or 0) >= int(row["max_attempts"] or 3): + connection.commit() + return None + connection.execute( + """ + UPDATE workflow_jobs + SET status = ?, progress = CASE WHEN progress < 10 THEN 10 ELSE progress END, + message = '任务已开始执行', started_at = COALESCE(started_at, ?), + updated_at = ?, heartbeat_at = ?, lease_owner = ?, lease_expires_at = ?, + attempt_count = attempt_count + 1 + WHERE id = ? + """, + (JOB_STATUS_RUNNING, now_iso, now_iso, now_iso, lease_owner, lease_expires_at, job_id), + ) + connection.commit() + return get_job(job_id) + + +def claim_next_job(lease_owner: str, lease_seconds: int = 120) -> dict | None: + """按创建时间领取一个重型任务，保证本地默认串行。""" + now = _now_iso() + with get_connection() as connection: + connection.execute( + """ + UPDATE workflow_jobs SET status = ?, error_message = '已达到最大尝试次数', + message = '任务失败：已达到最大尝试次数', finished_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL + WHERE status IN (?, ?) AND attempt_count >= max_attempts + """, + (JOB_STATUS_FAILED, now, now, JOB_STATUS_QUEUED, JOB_STATUS_RUNNING), + ) + connection.commit() + row = connection.execute( + """ + SELECT id FROM workflow_jobs + WHERE cancel_requested = 0 + AND attempt_count < max_attempts + AND ( + (status = ? AND (next_attempt_at IS NULL OR next_attempt_at <= ?)) + OR (status = ? AND (lease_expires_at IS NULL OR lease_expires_at <= ?)) + ) + ORDER BY created_at ASC + LIMIT 1 + """, + (JOB_STATUS_QUEUED, now, JOB_STATUS_RUNNING, now), + ).fetchone() + return claim_job(row["id"], lease_owner, lease_seconds) if row else None + + +def heartbeat_job(job_id: str, lease_owner: str, lease_seconds: int = 120) -> bool: + now = datetime.now(timezone.utc) + now_iso = now.isoformat(timespec="seconds") + expires = (now + timedelta(seconds=max(30, lease_seconds))).isoformat(timespec="seconds") + with get_connection() as connection: + cursor = connection.execute( + """ + UPDATE workflow_jobs SET heartbeat_at = ?, lease_expires_at = ?, updated_at = ? + WHERE id = ? AND status = ? AND lease_owner = ? + """, + (now_iso, expires, now_iso, job_id, JOB_STATUS_RUNNING, lease_owner), + ) + connection.commit() + return cursor.rowcount == 1 + + +def update_job_checkpoint(job_id: str, checkpoint: dict) -> dict | None: + now = _now_iso() + with get_connection() as connection: + connection.execute( + "UPDATE workflow_jobs SET checkpoint_json = ?, checkpoint_updated_at = ?, updated_at = ? WHERE id = ?", + (json.dumps(checkpoint, ensure_ascii=False), now, now, job_id), + ) + connection.commit() + return get_job(job_id) + + +def request_job_cancel(job_id: str) -> dict | None: + now = _now_iso() + with get_connection() as connection: + row = connection.execute("SELECT status FROM workflow_jobs WHERE id = ?", (job_id,)).fetchone() + if not row: + return None + if row["status"] == JOB_STATUS_QUEUED: + connection.execute( + """UPDATE workflow_jobs SET status = ?, cancel_requested = 1, message = '任务已取消', + finished_at = ?, updated_at = ?, lease_owner = NULL, lease_expires_at = NULL WHERE id = ?""", + (JOB_STATUS_CANCELLED, now, now, job_id), + ) + elif row["status"] == JOB_STATUS_RUNNING: + connection.execute( + "UPDATE workflow_jobs SET cancel_requested = 1, message = '正在停止任务', updated_at = ? WHERE id = ?", + (now, job_id), + ) + connection.commit() + return get_job(job_id) + + +def is_cancel_requested(job_id: str) -> bool: + job = get_job(job_id) + return bool(job and int(job.get("cancel_requested") or 0)) + + +def retry_job(job_id: str) -> dict | None: + now = _now_iso() + with get_connection() as connection: + cursor = connection.execute( + """ + UPDATE workflow_jobs + SET status = ?, progress = 0, message = '任务已重新加入队列', error_message = NULL, + finished_at = NULL, next_attempt_at = ?, cancel_requested = 0, + lease_owner = NULL, lease_expires_at = NULL, heartbeat_at = NULL, + attempt_count = 0, updated_at = ? + WHERE id = ? AND status IN (?, ?) + """, + (JOB_STATUS_QUEUED, now, now, job_id, JOB_STATUS_FAILED, JOB_STATUS_CANCELLED), + ) + connection.commit() + return get_job(job_id) if cursor.rowcount else None + + +def release_job_lease(job_id: str, lease_owner: str) -> bool: + """Web 正常停止时释放子进程 Job，供新进程立即恢复。""" + now = _now_iso() + with get_connection() as connection: + cursor = connection.execute( + """ + UPDATE workflow_jobs SET status = ?, message = '应用停止，等待新 worker 恢复', + lease_owner = NULL, lease_expires_at = NULL, heartbeat_at = NULL, updated_at = ? + WHERE id = ? AND status = ? AND lease_owner = ? + """, + (JOB_STATUS_QUEUED, now, job_id, JOB_STATUS_RUNNING, lease_owner), + ) + connection.commit() + return cursor.rowcount == 1 + + def update_job_progress(job_id: str, progress: int, message: Optional[str] = None) -> dict | None: """更新 job 进度（0-100）和消息""" now = _now_iso() @@ -271,7 +427,7 @@ def mark_job_completed(job_id: str, result: Optional[dict] = None) -> dict | Non UPDATE workflow_jobs SET status = ?, progress = 100, message = '任务已完成', result_json = ?, - finished_at = ?, updated_at = ? + finished_at = ?, updated_at = ?, lease_owner = NULL, lease_expires_at = NULL WHERE id = ? """, (JOB_STATUS_COMPLETED, result_json, now, now, job_id), @@ -291,7 +447,7 @@ def mark_job_failed(job_id: str, error_message: str) -> dict | None: """ UPDATE workflow_jobs SET status = ?, error_message = ?, message = ?, - finished_at = ?, updated_at = ? + finished_at = ?, updated_at = ?, lease_owner = NULL, lease_expires_at = NULL WHERE id = ? """, (JOB_STATUS_FAILED, error_message, f"任务失败：{error_message}", now, now, job_id), @@ -301,3 +457,18 @@ def mark_job_failed(job_id: str, error_message: str) -> dict | None: if cursor.rowcount == 0: return None return get_job(job_id) + + +def mark_job_cancelled(job_id: str, message: str = "任务已取消") -> dict | None: + now = _now_iso() + with get_connection() as connection: + cursor = connection.execute( + """ + UPDATE workflow_jobs SET status = ?, message = ?, finished_at = ?, updated_at = ?, + lease_owner = NULL, lease_expires_at = NULL + WHERE id = ? + """, + (JOB_STATUS_CANCELLED, message, now, now, job_id), + ) + connection.commit() + return get_job(job_id) if cursor.rowcount else None diff --git a/app/services/job_worker.py b/app/services/job_worker.py index 2bd8e83..91a928e 100644 --- a/app/services/job_worker.py +++ b/app/services/job_worker.py @@ -1,19 +1,31 @@ -"""轻量本地工作流任务队列 —— Job Worker +"""单进程持久化工作流 Worker。 -负责在后台执行 job 记录对应的长任务。 -第一轮仅接入自动切片（video_cut），后续再逐步迁移其他流程。 +数据库 lease 让应用重启后可接管过期任务；默认只运行一个本地重型任务。 """ +from __future__ import annotations + +import os +import socket +import subprocess +import sys +import threading +import time +from uuid import uuid4 + from app.services import job_service from app.services import task_service +from app.core.config import settings +from app.services.managed_process_service import popen_process_group, terminate_process_tree -def execute_job(job_id: str) -> dict: +def execute_job(job_id: str, *, lease_owner: str | None = None, already_claimed: bool = False) -> dict: """根据 job 记录执行对应的后台任务 这是一个同步函数，调用方应自行决定放在线程池还是 BackgroundTasks 中运行。 返回最终的 job 记录。 """ + owner = lease_owner or f"{socket.gethostname()}:{os.getpid()}:{uuid4().hex[:8]}" job = job_service.get_job(job_id) if not job: raise ValueError(f"job 不存在：{job_id}") @@ -23,14 +35,18 @@ def execute_job(job_id: str) -> dict: job_type = job.get("job_type") task_id = job.get("task_id") - # 标记开始执行 - job = job_service.mark_job_running(job_id) - if not job: - raise RuntimeError(f"无法更新 job 状态：{job_id}") + if not already_claimed: + job = job_service.claim_job(job_id, owner) + if not job: + return job_service.get_job(job_id) try: if job_type == job_service.JOB_TYPE_VIDEO_CUT: _execute_video_cut(job_id, task_id) + elif job_type == job_service.JOB_TYPE_TRANSCRIPT: + _execute_transcript(job_id, task_id, job.get("payload_json") or {}) + elif job_type == job_service.JOB_TYPE_AUTO_PIPELINE: + _execute_auto_pipeline(job_id, task_id, job.get("payload_json") or {}) else: raise ValueError(f"暂不支持的 job 类型：{job_type}") except Exception as exc: @@ -64,3 +80,120 @@ def _execute_video_cut(job_id: str, task_id: str) -> None: } job_service.mark_job_completed(job_id, result_summary) + + +def _execute_transcript(job_id: str, task_id: str, payload: dict) -> None: + from app.services import transcript_workflow_service + from app.services.storage_service import get_artifact_paths + from app.services.transcript_service import read_transcript_progress + + job_service.update_job_progress(job_id, 5, "正在准备音频和转写 checkpoint") + transcript_workflow_service.process_task_transcript_workflow( + task_id, + background_tasks=None, + force=bool(payload.get("force")), + provider=payload.get("provider"), + job_id=job_id, + ) + if job_service.is_cancel_requested(job_id): + job_service.mark_job_cancelled(job_id, "转写已取消，已完成分块 checkpoint 会保留") + return + progress = read_transcript_progress(get_artifact_paths(task_id)["transcript_path"]) + if progress.get("status") == "failed": + raise RuntimeError(str(progress.get("message") or "转写失败")) + job_service.mark_job_completed( + job_id, + {"task_id": task_id, "transcript_status": progress.get("status"), "checkpoint_retained": True}, + ) + + +def _execute_auto_pipeline(job_id: str, task_id: str, payload: dict) -> None: + from app.services.pipeline_engine import run_auto_pipeline + + result = run_auto_pipeline( + task_id, + retry=bool(payload.get("retry")), + start_step=payload.get("start_step"), + job_id=job_id, + ) + if job_service.is_cancel_requested(job_id): + job_service.mark_job_cancelled(job_id, "全自动流水线已取消") + return + if result.get("status") == "failed": + raise RuntimeError(str(result.get("last_error") or "全自动流水线失败")) + job_service.mark_job_completed(job_id, result) + + +class WorkflowJobRunner: + """应用生命周期内的单 worker 线程。""" + + def __init__(self, poll_seconds: float = 1.0) -> None: + self.poll_seconds = poll_seconds + self.owner = f"{socket.gethostname()}:{os.getpid()}:{uuid4().hex[:8]}" + self._stop_event = threading.Event() + self._thread: threading.Thread | None = None + + def start(self) -> None: + if self._thread and self._thread.is_alive(): + return + self._thread = threading.Thread(target=self._run, name="workflow-job-worker", daemon=True) + self._thread.start() + + def stop(self) -> None: + self._stop_event.set() + if self._thread: + self._thread.join(timeout=5) + + def _run(self) -> None: + while not self._stop_event.is_set(): + job = job_service.claim_next_job(self.owner) + if not job: + self._stop_event.wait(self.poll_seconds) + continue + self._run_job_subprocess(job["id"]) + + def _run_job_subprocess(self, job_id: str) -> None: + process = popen_process_group( + [sys.executable, "-m", "app.services.job_worker_process", job_id, self.owner], + cwd=str(settings.project_root), + stdout=subprocess.DEVNULL, + stderr=subprocess.DEVNULL, + ) + last_heartbeat = 0.0 + last_progress_at = time.monotonic() + previous_progress: tuple[int, str] | None = None + no_progress_timeout = max( + 900, + settings.ffmpeg_audio_extract_timeout, + settings.ffmpeg_cut_timeout, + settings.volcengine_asr_timeout_seconds, + ) + while process.poll() is None: + if self._stop_event.wait(1): + terminate_process_tree(process) + job_service.release_job_lease(job_id, self.owner) + return + job = job_service.get_job(job_id) + if not job: + terminate_process_tree(process) + return + if job_service.is_cancel_requested(job_id): + terminate_process_tree(process) + job_service.mark_job_cancelled(job_id, "任务已取消，子进程树已终止") + return + progress_state = (int(job.get("progress") or 0), str(job.get("message") or "")) + if previous_progress != progress_state: + previous_progress = progress_state + last_progress_at = time.monotonic() + if time.monotonic() - last_progress_at > no_progress_timeout: + terminate_process_tree(process) + job_service.mark_job_failed(job_id, f"任务连续 {no_progress_timeout} 秒没有进展，已终止子进程树") + return + if time.monotonic() - last_heartbeat >= 20: + job_service.heartbeat_job(job_id, self.owner) + last_heartbeat = time.monotonic() + final_job = job_service.get_job(job_id) + if process.returncode != 0 and final_job and final_job.get("status") == job_service.JOB_STATUS_RUNNING: + job_service.mark_job_failed(job_id, f"Job 子进程异常退出，退出码：{process.returncode}") + elif process.returncode == 0 and final_job and final_job.get("status") == job_service.JOB_STATUS_RUNNING: + job_service.mark_job_failed(job_id, "Job 子进程已退出但没有写入终态") diff --git a/app/services/job_worker_process.py b/app/services/job_worker_process.py new file mode 100644 index 0000000..509c3f3 --- /dev/null +++ b/app/services/job_worker_process.py @@ -0,0 +1,24 @@ +"""持久化 Job 的独立子进程入口。""" + +from __future__ import annotations + +import sys + +from app.services.job_worker import execute_job + + +def main() -> int: + if len(sys.argv) != 3: + print("用法：python -m app.services.job_worker_process ", file=sys.stderr) + return 2 + job_id, lease_owner = sys.argv[1], sys.argv[2] + try: + execute_job(job_id, lease_owner=lease_owner, already_claimed=True) + except Exception as exc: + print(str(exc), file=sys.stderr) + return 1 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/app/services/managed_process_service.py b/app/services/managed_process_service.py new file mode 100644 index 0000000..199f589 --- /dev/null +++ b/app/services/managed_process_service.py @@ -0,0 +1,36 @@ +"""可取消、可检测无进展超时的本地子进程工具。""" + +from __future__ import annotations + +import os +import signal +import subprocess + + +def popen_process_group(command: list[str], **kwargs) -> subprocess.Popen: + if os.name == "nt": + kwargs["creationflags"] = kwargs.get("creationflags", 0) | subprocess.CREATE_NEW_PROCESS_GROUP + else: + kwargs["start_new_session"] = True + return subprocess.Popen(command, **kwargs) + + +def terminate_process_tree(process: subprocess.Popen) -> None: + if process.poll() is not None: + return + if os.name == "nt": + subprocess.run( + ["taskkill", "/PID", str(process.pid), "/T", "/F"], + capture_output=True, + check=False, + timeout=15, + ) + else: + try: + os.killpg(os.getpgid(process.pid), signal.SIGTERM) + process.wait(timeout=5) + except (ProcessLookupError, subprocess.TimeoutExpired): + try: + os.killpg(os.getpgid(process.pid), signal.SIGKILL) + except ProcessLookupError: + pass diff --git a/app/services/media_preflight_service.py b/app/services/media_preflight_service.py new file mode 100644 index 0000000..19eab61 --- /dev/null +++ b/app/services/media_preflight_service.py @@ -0,0 +1,179 @@ +"""长素材媒体与磁盘预检。 + +创建任务前只做只读探测，不改写原片。超过 6 小时是提示，不是拒绝条件。 +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +from dataclasses import asdict, dataclass +from pathlib import Path + +from app.core.config import settings + + +MAX_TESTED_DURATION_SECONDS = 6 * 60 * 60 +PCM_BYTES_PER_SECOND = 16_000 * 2 +MIN_SAFETY_MARGIN_BYTES = 1024 * 1024 * 1024 + + +@dataclass(frozen=True) +class MediaPreflight: + path: str + duration_seconds: float + file_size_bytes: int + video_codec: str + audio_codec: str + width: int + height: int + frame_rate: float + audio_channels: int + audio_sample_rate: int + required_free_bytes: int + available_free_bytes: int + warnings: list[str] + + def to_dict(self) -> dict: + return asdict(self) + + +def _parse_rate(value: str | None) -> float: + text = str(value or "0/1") + try: + numerator, denominator = text.split("/", 1) + return float(numerator) / max(float(denominator), 1.0) + except (TypeError, ValueError, ZeroDivisionError): + return 0.0 + + +def _run_decode_sample(path: Path, start_seconds: float) -> None: + ffmpeg = shutil.which("ffmpeg") + if not ffmpeg: + raise ValueError("未找到 FFmpeg，无法验证视频可解码性") + command = [ffmpeg, "-v", "error", "-xerror"] + if start_seconds > 0: + command.extend(["-ss", f"{start_seconds:.3f}"]) + command.extend( + ["-i", str(path), "-t", "3", "-map", "0:v:0", "-map", "0:a:0", "-f", "null", "-"] + ) + completed = subprocess.run( + command, + capture_output=True, + text=True, + timeout=settings.ffprobe_timeout, + check=False, + ) + if completed.returncode != 0: + message = (completed.stderr or completed.stdout or "未知解码错误").strip() + raise ValueError(f"视频无法正常解码：{message[-500:]}") + + +def probe_media(path_value: str | Path) -> dict: + path = Path(path_value).resolve() + ffprobe = shutil.which("ffprobe") + if not ffprobe: + raise ValueError("未找到 FFprobe，无法创建视频任务") + try: + completed = subprocess.run( + [ + ffprobe, + "-v", + "error", + "-show_streams", + "-show_format", + "-of", + "json", + str(path), + ], + capture_output=True, + text=True, + timeout=settings.ffprobe_timeout, + check=False, + ) + except subprocess.TimeoutExpired as exc: + raise ValueError(f"媒体探测超过 {settings.ffprobe_timeout} 秒，文件可能损坏") from exc + if completed.returncode != 0: + message = (completed.stderr or "FFprobe 无法读取文件").strip() + raise ValueError(f"媒体探测失败：{message[-500:]}") + try: + payload = json.loads(completed.stdout or "{}") + except json.JSONDecodeError as exc: + raise ValueError("FFprobe 返回了无法解析的媒体信息") from exc + + streams = payload.get("streams") or [] + video_stream = next((item for item in streams if item.get("codec_type") == "video"), None) + audio_stream = next((item for item in streams if item.get("codec_type") == "audio"), None) + if not video_stream: + raise ValueError("源文件没有视频轨，不能创建视频处理任务") + if not audio_stream: + raise ValueError("源文件没有音轨，无法进行语言转写和高光选片") + try: + duration = float((payload.get("format") or {}).get("duration") or video_stream.get("duration") or 0) + except (TypeError, ValueError): + duration = 0.0 + if duration <= 0: + raise ValueError("源文件时长无效，无法创建任务") + + _run_decode_sample(path, 0) + if duration > 10: + _run_decode_sample(path, max(0.0, duration - 5.0)) + return { + "duration_seconds": duration, + "file_size_bytes": path.stat().st_size, + "video_codec": str(video_stream.get("codec_name") or "unknown"), + "audio_codec": str(audio_stream.get("codec_name") or "unknown"), + "width": int(video_stream.get("width") or 0), + "height": int(video_stream.get("height") or 0), + "frame_rate": _parse_rate(video_stream.get("avg_frame_rate") or video_stream.get("r_frame_rate")), + "audio_channels": int(audio_stream.get("channels") or 0), + "audio_sample_rate": int(audio_stream.get("sample_rate") or 0), + } + + +def estimate_required_bytes( + *, + duration_seconds: float, + source_size_bytes: int, + total_output_limit: int, +) -> int: + pcm_bytes = int(duration_seconds * PCM_BYTES_PER_SECOND) + # 默认按每条最多 5 分钟估算切片；同时预留一份字幕成片。 + output_fraction = min(1.0, total_output_limit * 300 / max(duration_seconds, 1.0)) + clip_and_subtitle_bytes = int(source_size_bytes * output_fraction * 2.2) + working_margin = max(MIN_SAFETY_MARGIN_BYTES, int((pcm_bytes + clip_and_subtitle_bytes) * 0.25)) + return pcm_bytes + clip_and_subtitle_bytes + working_margin + + +def preflight_media( + path_value: str | Path, + *, + total_output_limit: int, +) -> MediaPreflight: + path = Path(path_value).resolve() + probe = probe_media(path) + required = estimate_required_bytes( + duration_seconds=probe["duration_seconds"], + source_size_bytes=probe["file_size_bytes"], + total_output_limit=total_output_limit, + ) + storage_anchor = settings.tasks_dir + storage_anchor.mkdir(parents=True, exist_ok=True) + available = shutil.disk_usage(storage_anchor).free + if available < required: + required_gib = required / (1024 ** 3) + available_gib = available / (1024 ** 3) + raise ValueError( + f"任务存储空间不足：预计至少需要 {required_gib:.1f} GiB，当前可用 {available_gib:.1f} GiB" + ) + warnings: list[str] = [] + if probe["duration_seconds"] > MAX_TESTED_DURATION_SECONDS: + warnings.append("素材超过当前 6 小时验收范围，可以创建，但请重点关注耗时和磁盘空间") + return MediaPreflight( + path=str(path), + required_free_bytes=required, + available_free_bytes=available, + warnings=warnings, + **probe, + ) diff --git a/app/services/pipeline_engine.py b/app/services/pipeline_engine.py index 79ae240..26a181b 100644 --- a/app/services/pipeline_engine.py +++ b/app/services/pipeline_engine.py @@ -78,6 +78,7 @@ def run( task_id: str, retry: bool = False, start_step: TaskStatus | str | None = None, + job_id: str | None = None, ) -> dict: task = self._get_task(task_id) if not task.get("auto_mode"): @@ -112,6 +113,19 @@ def run( for step in steps: try: + if job_id: + from app.services import job_service + if job_service.is_cancel_requested(job_id): + raise RuntimeError("用户已取消全自动流水线") + step_index = STEP_STATUSES.index(step) + job_service.update_job_progress( + job_id, + 5 + round(step_index / max(1, len(STEP_STATUSES)) * 90), + f"正在执行：{step.value}", + ) + job = job_service.get_job(job_id) + if job and job.get("lease_owner"): + job_service.heartbeat_job(job_id, str(job["lease_owner"])) task_service.update_task_status(task_id, step) context[step.value] = handlers[step](task_id, context) except Exception as exc: @@ -586,8 +600,9 @@ def run_auto_pipeline( task_id: str, retry: bool = False, start_step: TaskStatus | str | None = None, + job_id: str | None = None, ) -> dict: - return PipelineEngine().run(task_id, retry=retry, start_step=start_step) + return PipelineEngine().run(task_id, retry=retry, start_step=start_step, job_id=job_id) def start_auto_pipeline( @@ -597,6 +612,20 @@ def start_auto_pipeline( start_step: TaskStatus | str | None = None, ) -> dict: if background_tasks is not None: - background_tasks.add_task(run_auto_pipeline, task_id, retry, start_step) - return {"status": "started", "message": "全自动流水线已在后台启动。", "task_id": task_id} + from app.services import job_service + + job, created = job_service.create_or_get_active_job( + task_id=task_id, + job_type=job_service.JOB_TYPE_AUTO_PIPELINE, + payload={ + "retry": retry, + "start_step": start_step.value if isinstance(start_step, TaskStatus) else start_step, + }, + ) + return { + "status": job["status"], + "message": "全自动流水线已加入持久化队列。" if created else "全自动流水线已经在排队或运行。", + "task_id": task_id, + "job_id": job["id"], + } return run_auto_pipeline(task_id, retry=retry, start_step=start_step) diff --git a/app/services/storage_service.py b/app/services/storage_service.py index 56065aa..3b8b82b 100644 --- a/app/services/storage_service.py +++ b/app/services/storage_service.py @@ -605,6 +605,7 @@ def browse_video_directory(path_value: str | None) -> dict: return { "path": str(base_path), + "parent_path": str(base_path.parent) if _is_path_within_roots(base_path.parent, roots=allowed_roots) else "", "exists": True, "directories": directories, "files": files, diff --git a/app/services/task_lifecycle_service.py b/app/services/task_lifecycle_service.py index ca1f9de..1026884 100644 --- a/app/services/task_lifecycle_service.py +++ b/app/services/task_lifecycle_service.py @@ -16,6 +16,7 @@ validate_source_video_path, ) from app.services.task_log_service import append_task_log +from app.services.media_preflight_service import preflight_media class TaskDeletionConflictError(RuntimeError): @@ -48,14 +49,21 @@ def create_task_record(payload: TaskCreate, task_id: str | None = None, task_dir exclude_task_id=resolved_task_id, ) now = _now_iso() - create_task_directory(resolved_task_id, resolved_task_dir_name) - source_path = payload.nas_file_path if payload.source_type == "nas" else payload.original_video_path has_source_file = bool(source_path) + media_preflight = None if source_path: valid, error_message = validate_source_video_path(source_path) if not valid: raise ValueError(error_message) + output_limit = ( + payload.highlight_total_limit + if payload.selection_profile == "long_live_talk" + else payload.candidate_clip_count + ) + media_preflight = preflight_media(source_path, total_output_limit=output_limit) + + create_task_directory(resolved_task_id, resolved_task_dir_name) initial_status = TaskStatus.CREATED.value if payload.auto_mode else ( TaskStatus.pending_processing.value if has_source_file else TaskStatus.pending_video.value @@ -89,6 +97,8 @@ def create_task_record(payload: TaskCreate, task_id: str | None = None, task_dir "candidate_clip_count": payload.candidate_clip_count, "selection_profile": payload.selection_profile, "final_clip_target": payload.final_clip_target, + "highlight_density_per_hour": payload.highlight_density_per_hour, + "highlight_total_limit": payload.highlight_total_limit, "ai_preference": payload.ai_preference, "ai_prompt_preset_id": "preset_001", "auto_mode": 1 if payload.auto_mode else 0, @@ -123,7 +133,7 @@ def create_task_record(payload: TaskCreate, task_id: str | None = None, task_dir append_task_log(resolved_task_id, "任务已创建") if payload.auto_mode: append_task_log(resolved_task_id, "已开启全自动模式，等待流水线启动") - return { + result = { "id": resolved_task_id, "task_name": payload.task_name, "task_dir_name": resolved_task_dir_name, @@ -133,6 +143,9 @@ def create_task_record(payload: TaskCreate, task_id: str | None = None, task_dir "detail_url": f"/tasks/{resolved_task_id}", "message": "任务已创建并写入数据库。", } + if media_preflight: + result["media_preflight"] = media_preflight.to_dict() + return result def update_task_status( @@ -223,30 +236,48 @@ def update_task_selection_settings( task_id: str, selection_profile: str, final_clip_target: int, + highlight_density_per_hour: int = 4, + highlight_total_limit: int = 30, ) -> dict: from app.services.task_service import _now_iso, get_task # noqa: F811 task = get_task(task_id, include_video_probe=False) if not task: raise ValueError("任务不存在") - if selection_profile not in {"general", "variety_comedy"}: - raise ValueError("选片模式只能是康熙笑点选片模式或通用模式（历史任务）") + if selection_profile not in {"general", "variety_comedy", "long_live_talk"}: + raise ValueError("选片模式只能是通用内容价值、康熙笑点选片模式或长直播高光") if final_clip_target < 1 or final_clip_target > 12: raise ValueError("最终启用目标必须在 1 到 12 条之间") + if not 1 <= highlight_density_per_hour <= 10: + raise ValueError("每小时高光密度必须在 1 到 10 条之间") + if not 1 <= highlight_total_limit <= 50: + raise ValueError("高光总上限必须在 1 到 50 条之间") now = _now_iso() with get_connection() as connection: connection.execute( """ UPDATE tasks - SET selection_profile = ?, final_clip_target = ?, updated_at = ? + SET selection_profile = ?, final_clip_target = ?, + highlight_density_per_hour = ?, highlight_total_limit = ?, updated_at = ? WHERE id = ? """, - (selection_profile, final_clip_target, now, task_id), + ( + selection_profile, + final_clip_target, + highlight_density_per_hour, + highlight_total_limit, + now, + task_id, + ), ) connection.commit() - profile_label = "康熙笑点选片模式" if selection_profile == "variety_comedy" else "通用模式（历史任务）" + profile_label = { + "general": "通用内容价值", + "variety_comedy": "康熙笑点选片模式", + "long_live_talk": "长直播高光（语言类）", + }[selection_profile] append_task_log(task_id, f"已更新选片模式：{profile_label}，最终启用目标：{final_clip_target} 条") return { "status": "ok", diff --git a/app/services/task_service.py b/app/services/task_service.py index 0678e39..6111301 100644 --- a/app/services/task_service.py +++ b/app/services/task_service.py @@ -494,10 +494,14 @@ def _row_to_task(row: Row, include_video_probe: bool = False) -> dict: "progress": progress, "candidate_count": task.get("candidate_clip_count") or 0, "selection_profile": task.get("selection_profile") or "general", - "selection_profile_label": ( - "康熙笑点选片模式" if task.get("selection_profile") == "variety_comedy" else "通用模式（历史任务）" - ), + "selection_profile_label": { + "general": "通用内容价值", + "variety_comedy": "康熙笑点选片模式", + "long_live_talk": "长直播高光（语言类）", + }.get(task.get("selection_profile") or "general", "通用内容价值"), "final_clip_target": int(task.get("final_clip_target") or 5), + "highlight_density_per_hour": int(task.get("highlight_density_per_hour") or 4), + "highlight_total_limit": int(task.get("highlight_total_limit") or 30), "duration": video_meta["duration"], "video_size": video_meta["video_size"], "owner": "本地用户", @@ -536,6 +540,7 @@ def list_tasks(include_deleted: bool = False) -> list[dict]: SELECT id, task_name, task_dir_name, source_type, platform, original_video_path, nas_file_path, max_clip_duration, candidate_clip_count, selection_profile, final_clip_target, + highlight_density_per_hour, highlight_total_limit, ai_preference, ai_prompt_preset_id, auto_mode, auto_config_json, status, progress, error_message, last_error, is_deleted, deleted_at, created_at, updated_at @@ -572,6 +577,7 @@ def get_task(task_id: str, include_video_probe: bool = True) -> dict | None: SELECT id, task_name, task_dir_name, source_type, platform, original_video_path, nas_file_path, max_clip_duration, candidate_clip_count, selection_profile, final_clip_target, + highlight_density_per_hour, highlight_total_limit, ai_preference, ai_prompt_preset_id, auto_mode, auto_config_json, status, progress, error_message, last_error, is_deleted, deleted_at, created_at, updated_at diff --git a/app/services/transcript_service.py b/app/services/transcript_service.py index 8974504..e6f7251 100644 --- a/app/services/transcript_service.py +++ b/app/services/transcript_service.py @@ -6,6 +6,9 @@ import re import shutil import subprocess +import queue +import threading +import time from tempfile import TemporaryDirectory from typing import Callable from urllib.error import HTTPError, URLError @@ -13,6 +16,8 @@ from uuid import uuid4 from app.core.config import settings +from app.services.transcription_checkpoint_service import TranscriptionCheckpoint +from app.services.managed_process_service import popen_process_group, terminate_process_tree TIME_TABLE_PATTERN = re.compile( @@ -26,11 +31,21 @@ PLACEHOLDER_TEXT_MARKERS = ("这里会保存", "后续可接入", "占位转写内容") +@dataclass(frozen=True) +class TranscriptWord: + start_ms: int + end_ms: int + text: str + confidence: float | None = None + + @dataclass(frozen=True) class TranscriptSegment: start_seconds: float end_seconds: float text: str + confidence: float | None = None + words: tuple[TranscriptWord, ...] = () @dataclass(frozen=True) @@ -65,11 +80,22 @@ def duration_seconds(self) -> float: _RESERVED_REMOTE_PROVIDERS = ("aliyun", "tencent", "xunfei") -def run_ffmpeg_audio_extract(video_path: Path, output_path: Path) -> dict[str, str]: +def run_ffmpeg_audio_extract( + video_path: Path, + output_path: Path, + *, + cancel_check: Callable[[], bool] | None = None, + progress_callback: Callable[[int], None] | None = None, +) -> dict[str, str]: output_path.parent.mkdir(parents=True, exist_ok=True) + temporary_path = output_path.with_name(f"{output_path.stem}.part{output_path.suffix}") + temporary_path.unlink(missing_ok=True) command = [ "ffmpeg", "-y", + "-nostats", + "-loglevel", + "error", "-i", str(video_path), "-vn", @@ -79,12 +105,67 @@ def run_ffmpeg_audio_extract(video_path: Path, output_path: Path) -> dict[str, s "16000", "-ac", "1", - str(output_path), + "-progress", + "pipe:1", + str(temporary_path), ] - result = subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="replace", - timeout=settings.ffmpeg_audio_extract_timeout) - if result.returncode != 0: - raise RuntimeError(result.stderr.strip() or "FFmpeg 音频提取失败") + duration = get_audio_duration_seconds(video_path) + process = popen_process_group( + command, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + ) + output_lines: queue.Queue[str] = queue.Queue() + stderr_lines: list[str] = [] + + def read_stream(stream, target) -> None: + if not stream: + return + for line in iter(stream.readline, ""): + target(line.rstrip()) + + stdout_thread = threading.Thread(target=read_stream, args=(process.stdout, output_lines.put), daemon=True) + stderr_thread = threading.Thread(target=read_stream, args=(process.stderr, stderr_lines.append), daemon=True) + stdout_thread.start() + stderr_thread.start() + last_progress_at = time.monotonic() + try: + while process.poll() is None: + if cancel_check and cancel_check(): + terminate_process_tree(process) + raise RuntimeError("用户已取消音频提取") + try: + line = output_lines.get(timeout=0.5) + except queue.Empty: + line = "" + if line: + last_progress_at = time.monotonic() + if line.startswith(("out_time_ms=", "out_time_us=")): + try: + microseconds = int(line.split("=", 1)[1]) + percent = round((microseconds / 1_000_000) / max(duration, 0.001) * 100) + if progress_callback: + progress_callback(max(0, min(99, percent))) + except ValueError: + pass + if time.monotonic() - last_progress_at > settings.ffmpeg_audio_extract_timeout: + terminate_process_tree(process) + raise RuntimeError(f"FFmpeg 连续 {settings.ffmpeg_audio_extract_timeout} 秒没有进展，已终止进程树") + stdout_thread.join(timeout=2) + stderr_thread.join(timeout=2) + if process.returncode != 0: + raise RuntimeError("\n".join(stderr_lines[-20:]).strip() or "FFmpeg 音频提取失败") + temporary_path.replace(output_path) + if progress_callback: + progress_callback(100) + except Exception: + if process.poll() is None: + terminate_process_tree(process) + temporary_path.unlink(missing_ok=True) + raise return { "status": "ok", "message": "音频提取完成", @@ -106,6 +187,20 @@ def write_transcript_markdown( transcript_path.parent.mkdir(parents=True, exist_ok=True) progress_path = get_transcript_progress_path(transcript_path) _set_configured_transcription_runtime(provider) + checkpoint = None + task_id = str(task.get("id") or "").strip() + if task_id: + source_path = task.get("nas_file_path") or task.get("original_video_path") or audio_path + checkpoint = TranscriptionCheckpoint( + task_id=task_id, + source_path=source_path, + provider=_ACTIVE_TRANSCRIPTION_PROVIDER, + model=_ACTIVE_TRANSCRIPTION_MODEL, + device=_ACTIVE_TRANSCRIPTION_DEVICE, + compute_type=_ACTIVE_TRANSCRIPTION_COMPUTE_TYPE, + chunk_seconds=settings.transcription_chunk_seconds, + overlap_seconds=settings.transcription_chunk_overlap_seconds, + ) _emit_transcript_progress( progress_path, progress_callback, @@ -121,6 +216,7 @@ def write_transcript_markdown( progress_path, progress_callback, provider=provider, + checkpoint=checkpoint, ) content = build_transcript_markdown(task, audio_path, segments) temp_path = transcript_path.with_name(f"{transcript_path.name}.tmp") @@ -153,11 +249,14 @@ def transcribe_audio_with_configured_provider( progress_callback: Callable[[dict], None] | None = None, provider: str | None = None, allow_fallback: bool = False, + checkpoint: TranscriptionCheckpoint | None = None, ) -> list[TranscriptSegment]: provider = _normalize_provider_name(provider or settings.transcription_provider) fallback_provider = _normalize_provider_name(settings.transcription_fallback_provider) try: - return transcribe_audio_with_provider(audio_path, working_dir, progress_path, provider, progress_callback) + return transcribe_audio_with_provider( + audio_path, working_dir, progress_path, provider, progress_callback, checkpoint=checkpoint + ) except Exception as exc: if allow_fallback and fallback_provider and fallback_provider != provider: _emit_transcript_progress( @@ -191,6 +290,7 @@ def transcribe_audio_with_provider( progress_path: Path, provider: str, progress_callback: Callable[[dict], None] | None = None, + checkpoint: TranscriptionCheckpoint | None = None, ) -> list[TranscriptSegment]: provider = _normalize_provider_name(provider) if provider == "local": @@ -201,9 +301,13 @@ def transcribe_audio_with_provider( device=(_WHISPER_MODEL_KEY or _primary_model_key())[1], compute_type=(_WHISPER_MODEL_KEY or _primary_model_key())[2], ) - return transcribe_audio_in_chunks(audio_path, working_dir, progress_path, progress_callback) + return transcribe_audio_in_chunks( + audio_path, working_dir, progress_path, progress_callback, checkpoint=checkpoint + ) if provider == "volcengine": - return transcribe_audio_with_volcengine(audio_path, working_dir, progress_path, progress_callback) + return transcribe_audio_with_volcengine( + audio_path, working_dir, progress_path, progress_callback, checkpoint=checkpoint + ) if provider in _RESERVED_REMOTE_PROVIDERS: raise RuntimeError(f"{_provider_label(provider)} 转写接口已预留，但当前版本还没有完整接入") raise RuntimeError(f"未知转写服务商：{provider or '空'}") @@ -214,6 +318,7 @@ def transcribe_audio_in_chunks( working_dir: Path, progress_path: Path, progress_callback: Callable[[dict], None] | None = None, + checkpoint: TranscriptionCheckpoint | None = None, ) -> list[TranscriptSegment]: duration_seconds = get_audio_duration_seconds(audio_path) chunks = build_transcript_chunks( @@ -223,6 +328,8 @@ def transcribe_audio_in_chunks( ) if not chunks: raise RuntimeError("本地语音转写失败：音频时长无效，无法分段") + if checkpoint: + checkpoint.ensure_run(chunks) _emit_transcript_progress( progress_path, progress_callback, @@ -237,6 +344,17 @@ def transcribe_audio_in_chunks( with TemporaryDirectory(prefix=_TRANSCRIPT_CHUNK_DIR_PREFIX, dir=working_dir) as temp_dir: temp_dir_path = Path(temp_dir) for chunk in chunks: + completed_segments = checkpoint.load_completed(chunk.index, _segment_from_checkpoint) if checkpoint else None + if completed_segments is not None: + all_segments.extend( + _offset_chunk_segments(completed_segments, chunk, settings.transcription_chunk_overlap_seconds) + ) + _emit_transcript_progress( + progress_path, progress_callback, status="running", current_chunk=chunk.index, + total_chunks=len(chunks), percent=_chunk_percent(chunk.index, len(chunks)), + message=f"已复用第 {chunk.index}/{len(chunks)} 段 checkpoint", + ) + continue _emit_transcript_progress( progress_path, progress_callback, @@ -257,7 +375,14 @@ def transcribe_audio_in_chunks( percent=_chunk_start_percent(chunk.index, len(chunks)), message=f"正在加载模型并转写第 {chunk.index}/{len(chunks)} 段", ) - chunk_segments = transcribe_audio(chunk_path, allow_empty=True) + try: + chunk_segments = transcribe_audio(chunk_path, allow_empty=True) + if checkpoint: + checkpoint.save_completed(chunk.index, chunk_segments) + except Exception as exc: + if checkpoint: + checkpoint.save_failed(chunk.index, str(exc)) + raise all_segments.extend(_offset_chunk_segments(chunk_segments, chunk, settings.transcription_chunk_overlap_seconds)) _emit_transcript_progress( progress_path, @@ -271,6 +396,8 @@ def transcribe_audio_in_chunks( if not all_segments: raise RuntimeError("本地语音转写完成，但没有识别到可用语音内容") + if checkpoint: + checkpoint.complete() return sorted(all_segments, key=lambda segment: (segment.start_seconds, segment.end_seconds)) @@ -279,6 +406,7 @@ def transcribe_audio_with_volcengine( working_dir: Path, progress_path: Path, progress_callback: Callable[[dict], None] | None = None, + checkpoint: TranscriptionCheckpoint | None = None, ) -> list[TranscriptSegment]: _ensure_volcengine_configured() _set_active_transcription_runtime( @@ -297,6 +425,8 @@ def transcribe_audio_with_volcengine( ) if not chunks: raise RuntimeError("火山引擎远程转写失败：音频时长无效，无法分段") + if checkpoint: + checkpoint.ensure_run(chunks) _emit_transcript_progress( progress_path, progress_callback, @@ -311,6 +441,17 @@ def transcribe_audio_with_volcengine( with TemporaryDirectory(prefix=_TRANSCRIPT_CHUNK_DIR_PREFIX, dir=working_dir) as temp_dir: temp_dir_path = Path(temp_dir) for chunk in chunks: + completed_segments = checkpoint.load_completed(chunk.index, _segment_from_checkpoint) if checkpoint else None + if completed_segments is not None: + all_segments.extend( + _offset_chunk_segments(completed_segments, chunk, settings.transcription_chunk_overlap_seconds) + ) + _emit_transcript_progress( + progress_path, progress_callback, status="running", current_chunk=chunk.index, + total_chunks=len(chunks), percent=_chunk_percent(chunk.index, len(chunks)), + message=f"已复用第 {chunk.index}/{len(chunks)} 段 checkpoint", + ) + continue _emit_transcript_progress( progress_path, progress_callback, @@ -331,7 +472,14 @@ def transcribe_audio_with_volcengine( percent=_chunk_start_percent(chunk.index, len(chunks)), message=f"正在请求火山引擎转写第 {chunk.index}/{len(chunks)} 段", ) - chunk_segments = transcribe_audio_with_volcengine_flash(chunk_path, allow_empty=True) + try: + chunk_segments = transcribe_audio_with_volcengine_flash(chunk_path, allow_empty=True) + if checkpoint: + checkpoint.save_completed(chunk.index, chunk_segments) + except Exception as exc: + if checkpoint: + checkpoint.save_failed(chunk.index, str(exc)) + raise all_segments.extend(_offset_chunk_segments(chunk_segments, chunk, settings.transcription_chunk_overlap_seconds)) _emit_transcript_progress( progress_path, @@ -345,6 +493,8 @@ def transcribe_audio_with_volcengine( if not all_segments: raise RuntimeError("火山引擎远程转写完成，但没有识别到可用语音内容") + if checkpoint: + checkpoint.complete() return sorted(all_segments, key=lambda segment: (segment.start_seconds, segment.end_seconds)) @@ -561,7 +711,34 @@ def _segment_from_volcengine_utterance(item: dict) -> TranscriptSegment | None: end_seconds = _volcengine_utterance_time_to_seconds(item, "end_time", "end") if end_seconds <= start_seconds: end_seconds = start_seconds + 1 - return TranscriptSegment(start_seconds=start_seconds, end_seconds=end_seconds, text=text) + raw_words = item.get("words") or item.get("word_info") or [] + words: list[TranscriptWord] = [] + if isinstance(raw_words, list): + for word in raw_words: + if not isinstance(word, dict): + continue + word_text = normalize_transcript_text(str(word.get("text") or word.get("word") or "")) + if not word_text: + continue + word_start = _volcengine_utterance_time_to_seconds(word, "start_time", "start") + word_end = _volcengine_utterance_time_to_seconds(word, "end_time", "end") + confidence = word.get("confidence") or word.get("probability") + words.append( + TranscriptWord( + start_ms=round(word_start * 1000), + end_ms=round(max(word_start, word_end) * 1000), + text=word_text, + confidence=float(confidence) if confidence is not None else None, + ) + ) + segment_confidence = item.get("confidence") + return TranscriptSegment( + start_seconds=start_seconds, + end_seconds=end_seconds, + text=text, + confidence=float(segment_confidence) if segment_confidence is not None else None, + words=tuple(words), + ) def _volcengine_utterance_time_to_seconds(item: dict, millisecond_key: str, fallback_key: str) -> float: @@ -608,11 +785,41 @@ def _offset_chunk_segments( start_seconds=adjusted_start, end_seconds=adjusted_end, text=segment.text, + confidence=segment.confidence, + words=tuple( + TranscriptWord( + start_ms=word.start_ms + round(chunk.start_seconds * 1000), + end_ms=word.end_ms + round(chunk.start_seconds * 1000), + text=word.text, + confidence=word.confidence, + ) + for word in segment.words + ), ) ) return adjusted_segments +def _segment_from_checkpoint(item: dict) -> TranscriptSegment: + words = tuple( + TranscriptWord( + start_ms=int(word.get("start_ms") or 0), + end_ms=int(word.get("end_ms") or 0), + text=str(word.get("text") or ""), + confidence=float(word["confidence"]) if word.get("confidence") is not None else None, + ) + for word in (item.get("words") or []) + if isinstance(word, dict) + ) + return TranscriptSegment( + start_seconds=float(item.get("start_seconds") or 0), + end_seconds=float(item.get("end_seconds") or 0), + text=str(item.get("text") or ""), + confidence=float(item["confidence"]) if item.get("confidence") is not None else None, + words=words, + ) + + def get_transcript_progress_path(transcript_path: Path) -> Path: return transcript_path.with_name(_TRANSCRIPT_PROGRESS_FILE_NAME) @@ -743,16 +950,35 @@ def _transcribe_audio_with_model(model, audio_path: Path) -> list[TranscriptSegm language=settings.transcription_language or None, vad_filter=True, beam_size=5, + word_timestamps=True, ) - return [ - TranscriptSegment( - start_seconds=float(segment.start), - end_seconds=float(segment.end), - text=normalize_transcript_text(segment.text), + results: list[TranscriptSegment] = [] + for segment in raw_segments: + text = normalize_transcript_text(segment.text) + if not text: + continue + words = tuple( + TranscriptWord( + start_ms=round(float(word.start) * 1000), + end_ms=round(float(word.end) * 1000), + text=normalize_transcript_text(word.word), + confidence=float(word.probability) if getattr(word, "probability", None) is not None else None, + ) + for word in (getattr(segment, "words", None) or []) + if normalize_transcript_text(getattr(word, "word", "")) ) - for segment in raw_segments - if normalize_transcript_text(segment.text) - ] + avg_logprob = getattr(segment, "avg_logprob", None) + confidence = max(0.0, min(1.0, 2.718281828 ** float(avg_logprob))) if avg_logprob is not None else None + results.append( + TranscriptSegment( + start_seconds=float(segment.start), + end_seconds=float(segment.end), + text=text, + confidence=confidence, + words=words, + ) + ) + return results def _get_whisper_model(model_key: tuple[str, str, str] | None = None): @@ -796,13 +1022,35 @@ def _get_whisper_model(model_key: tuple[str, str, str] | None = None): def _primary_model_key() -> tuple[str, str, str]: + configured_device = settings.transcription_device.strip().lower() + device = _detect_transcription_device() if configured_device == "auto" else configured_device + configured_compute = settings.transcription_compute_type.strip().lower() + compute_type = ("float16" if device == "cuda" else "int8") if configured_compute == "auto" else configured_compute return ( settings.transcription_model, - settings.transcription_device, - settings.transcription_compute_type, + device, + compute_type, ) +def _detect_transcription_device() -> str: + """仅在显式配置 TRANSCRIPTION_DEVICE=auto 时检测 CUDA；cpu 配置永不覆盖。""" + executable = shutil.which("nvidia-smi") + if not executable: + return "cpu" + try: + result = subprocess.run( + [executable, "--query-gpu=name", "--format=csv,noheader"], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + except (OSError, subprocess.TimeoutExpired): + return "cpu" + return "cuda" if result.returncode == 0 and result.stdout.strip() else "cpu" + + def _cpu_fallback_model_key() -> tuple[str, str, str]: return ( settings.transcription_cpu_fallback_model, diff --git a/app/services/transcript_workflow_service.py b/app/services/transcript_workflow_service.py index 1b011dc..b455adf 100644 --- a/app/services/transcript_workflow_service.py +++ b/app/services/transcript_workflow_service.py @@ -152,7 +152,7 @@ def get_task_transcript_status(task_id: str) -> dict: # ---------- 音频提取 ---------- -def process_task_audio(task_id: str) -> dict: +def process_task_audio(task_id: str, job_id: str | None = None) -> dict: from app.services.task_service import get_task, update_task_status # noqa: F811 task = get_task(task_id) @@ -169,7 +169,27 @@ def process_task_audio(task_id: str) -> dict: update_task_status(task_id, TaskStatus.audio_extracting) append_task_log(task_id, "开始使用 FFmpeg 提取音频") try: - result = run_ffmpeg_audio_extract(source_path, paths["audio_path"]) + cancel_check = None + audio_progress_callback = None + if job_id: + from app.services import job_service + + def cancel_check() -> bool: + return job_service.is_cancel_requested(job_id) + + def audio_progress_callback(percent: int) -> None: + job_service.update_job_progress( + job_id, max(1, min(20, round(percent * 0.2))), f"正在提取音频：{percent}%" + ) + job = job_service.get_job(job_id) + if job and job.get("lease_owner"): + job_service.heartbeat_job(job_id, str(job["lease_owner"])) + result = run_ffmpeg_audio_extract( + source_path, + paths["audio_path"], + cancel_check=cancel_check, + progress_callback=audio_progress_callback, + ) except Exception as exc: error = str(exc) update_task_status(task_id, TaskStatus.failed, error) @@ -187,6 +207,7 @@ def process_task_transcript( task_id: str, background_tasks: Any | None = None, provider: str | None = None, + job_id: str | None = None, ) -> dict: from app.services.task_service import get_task, update_task_status # noqa: F811 @@ -229,9 +250,9 @@ def process_task_transcript( _CANCEL_TRANSCRIPT_TASKS.discard(task_id) _RUNNING_TRANSCRIPT_TASKS.add(task_id) if background_tasks is not None: - background_tasks.add_task(_run_task_transcript_background, task_id, provider_name) + background_tasks.add_task(_run_task_transcript_background, task_id, provider_name, job_id) else: - _run_task_transcript_background(task_id, provider_name) + _run_task_transcript_background(task_id, provider_name, job_id) return { "status": "started", "message": f"已开始{provider_label}分段转写，请稍后刷新查看进度。", @@ -285,6 +306,7 @@ def process_task_transcript_workflow( background_tasks: Any | None = None, force: bool = False, provider: str | None = None, + job_id: str | None = None, ) -> dict: from app.services.task_service import get_task # noqa: F811 @@ -302,15 +324,21 @@ def process_task_transcript_workflow( if not paths["audio_path"].exists(): append_task_log(task_id, "一键处理：未发现音频文件，先自动提取音频") - process_task_audio(task_id) + process_task_audio(task_id, job_id=job_id) if force: append_task_log(task_id, "用户明确要求重新生成转写 Markdown") - return process_task_transcript(task_id, background_tasks=background_tasks, provider=provider) + return process_task_transcript( + task_id, background_tasks=background_tasks, provider=provider, job_id=job_id + ) -def _run_task_transcript_background(task_id: str, provider: str | None = None) -> None: +def _run_task_transcript_background( + task_id: str, + provider: str | None = None, + job_id: str | None = None, +) -> None: from app.services.task_service import get_task, update_task_status # noqa: F811 task = get_task(task_id) @@ -322,6 +350,16 @@ def _run_task_transcript_background(task_id: str, provider: str | None = None) - provider_label = _transcription_choice_label(provider_name) def progress_callback(progress: dict) -> None: + if job_id: + from app.services import job_service + if job_service.is_cancel_requested(job_id): + raise TranscriptCancelledError("用户已停止当前转写任务") + percent = int(progress.get("percent") or 0) + job_percent = 20 + round(max(0, min(100, percent)) * 0.79) + job_service.update_job_progress(job_id, job_percent, str(progress.get("message") or "转写中")) + job = job_service.get_job(job_id) + if job and job.get("lease_owner"): + job_service.heartbeat_job(job_id, str(job["lease_owner"])) if task_id in _CANCEL_TRANSCRIPT_TASKS: raise TranscriptCancelledError("用户已停止当前转写任务") message = progress.get("message") or "转写进度已更新" diff --git a/app/services/transcription_checkpoint_service.py b/app/services/transcription_checkpoint_service.py new file mode 100644 index 0000000..81af868 --- /dev/null +++ b/app/services/transcription_checkpoint_service.py @@ -0,0 +1,180 @@ +"""SQLite 转写分块 checkpoint。 + +每个块成功后独立提交；进程失败或重启时只读取同一源指纹和同一运行配置下的成功块。 +""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import asdict +from datetime import datetime, timezone +from pathlib import Path +from uuid import uuid4 + +from app.db.database import get_connection + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def fingerprint_file(path_value: str | Path) -> str: + path = Path(path_value).resolve() + digest = hashlib.sha256() + size = path.stat().st_size + digest.update(str(size).encode("ascii")) + with path.open("rb") as source: + digest.update(source.read(1024 * 1024)) + if size > 1024 * 1024: + source.seek(max(0, size - 1024 * 1024)) + digest.update(source.read(1024 * 1024)) + return digest.hexdigest() + + +class TranscriptionCheckpoint: + def __init__( + self, + *, + task_id: str, + source_path: str | Path, + provider: str, + model: str, + device: str, + compute_type: str, + chunk_seconds: int, + overlap_seconds: int, + ) -> None: + self.task_id = task_id + self.source_fingerprint = fingerprint_file(source_path) + self.provider = provider + self.model = model + self.device = device + self.compute_type = compute_type + self.chunk_seconds = chunk_seconds + self.overlap_seconds = overlap_seconds + self.run_id = "" + + def ensure_run(self, chunks) -> str: + if self.run_id: + return self.run_id + now = _now_iso() + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + row = connection.execute( + """ + SELECT id FROM transcription_runs + WHERE task_id = ? AND source_fingerprint = ? AND provider = ? AND model = ? + AND device = ? AND compute_type = ? AND chunk_seconds = ? AND overlap_seconds = ? + AND status IN ('processing', 'failed', 'completed') + ORDER BY updated_at DESC LIMIT 1 + """, + ( + self.task_id, self.source_fingerprint, self.provider, self.model, + self.device, self.compute_type, self.chunk_seconds, self.overlap_seconds, + ), + ).fetchone() + connection.execute( + "UPDATE transcription_runs SET is_active = 0 WHERE task_id = ?", + (self.task_id,), + ) + if row: + self.run_id = row["id"] + connection.execute( + "UPDATE transcription_runs SET status = 'processing', is_active = 1, error_message = NULL, total_chunks = ?, updated_at = ? WHERE id = ?", + (len(chunks), now, self.run_id), + ) + else: + self.run_id = uuid4().hex + connection.execute( + """ + INSERT INTO transcription_runs ( + id, task_id, source_fingerprint, provider, model, device, compute_type, + chunk_seconds, overlap_seconds, status, total_chunks, completed_chunks, + is_active, error_message, created_at, updated_at, completed_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'processing', ?, 0, 1, NULL, ?, ?, NULL) + """, + ( + self.run_id, self.task_id, self.source_fingerprint, self.provider, + self.model, self.device, self.compute_type, self.chunk_seconds, + self.overlap_seconds, len(chunks), now, now, + ), + ) + for chunk in chunks: + connection.execute( + """ + INSERT OR IGNORE INTO transcription_chunks ( + id, run_id, task_id, chunk_index, start_ms, end_ms, status, + attempt_count, result_json, result_checksum, error_message, created_at, updated_at + ) VALUES (?, ?, ?, ?, ?, ?, 'queued', 0, NULL, NULL, NULL, ?, ?) + """, + ( + uuid4().hex, self.run_id, self.task_id, chunk.index, + round(chunk.start_seconds * 1000), round(chunk.end_seconds * 1000), now, now, + ), + ) + connection.commit() + return self.run_id + + def load_completed(self, chunk_index: int, segment_factory) -> list | None: + with get_connection() as connection: + row = connection.execute( + "SELECT result_json, result_checksum FROM transcription_chunks WHERE run_id = ? AND chunk_index = ? AND status = 'completed'", + (self.run_id, chunk_index), + ).fetchone() + if not row or not row["result_json"]: + return None + raw = str(row["result_json"]) + if hashlib.sha256(raw.encode("utf-8")).hexdigest() != str(row["result_checksum"] or ""): + return None + payload = json.loads(raw) + return [segment_factory(item) for item in payload] + + def save_completed(self, chunk_index: int, segments) -> None: + now = _now_iso() + raw = json.dumps([asdict(segment) for segment in segments], ensure_ascii=False, separators=(",", ":")) + checksum = hashlib.sha256(raw.encode("utf-8")).hexdigest() + with get_connection() as connection: + connection.execute("BEGIN IMMEDIATE") + connection.execute( + """ + UPDATE transcription_chunks + SET status = 'completed', attempt_count = attempt_count + 1, result_json = ?, + result_checksum = ?, error_message = NULL, updated_at = ? + WHERE run_id = ? AND chunk_index = ? + """, + (raw, checksum, now, self.run_id, chunk_index), + ) + completed = connection.execute( + "SELECT COUNT(*) FROM transcription_chunks WHERE run_id = ? AND status = 'completed'", + (self.run_id,), + ).fetchone()[0] + connection.execute( + "UPDATE transcription_runs SET completed_chunks = ?, updated_at = ? WHERE id = ?", + (completed, now, self.run_id), + ) + connection.commit() + + def save_failed(self, chunk_index: int, error: str) -> None: + now = _now_iso() + with get_connection() as connection: + connection.execute( + """UPDATE transcription_chunks SET status = 'failed', attempt_count = attempt_count + 1, + error_message = ?, updated_at = ? WHERE run_id = ? AND chunk_index = ?""", + (error[:2000], now, self.run_id, chunk_index), + ) + connection.execute( + "UPDATE transcription_runs SET status = 'failed', error_message = ?, updated_at = ? WHERE id = ?", + (error[:2000], now, self.run_id), + ) + connection.commit() + + def complete(self) -> None: + now = _now_iso() + with get_connection() as connection: + connection.execute( + """UPDATE transcription_runs SET status = 'completed', is_active = 1, + error_message = NULL, completed_at = ?, updated_at = ? WHERE id = ?""", + (now, now, self.run_id), + ) + connection.commit() diff --git a/app/static/css/styles.css b/app/static/css/styles.css index 7caf64f..b9281bf 100644 --- a/app/static/css/styles.css +++ b/app/static/css/styles.css @@ -854,6 +854,51 @@ dd { line-height: 1.6; } +.segmented-control { + display: flex; + flex-wrap: wrap; + gap: 10px; + margin-bottom: 14px; +} + +.segmented-control label { + display: inline-flex; + flex-direction: row; + align-items: center; + gap: 8px; + padding: 10px 14px; + border: 1px solid var(--border); + border-radius: 999px; + background: #fff; + cursor: pointer; +} + +.file-browser-list { + display: grid; + max-height: 300px; + overflow: auto; + border: 1px solid var(--border); + border-radius: 10px; + background: #fff; +} + +.file-browser-item { + width: 100%; + padding: 11px 13px; + border: 0; + border-bottom: 1px solid var(--border); + background: transparent; + color: var(--text); + text-align: left; + cursor: pointer; +} + +.file-browser-item:hover, +.file-browser-item.is-selected { + background: rgba(37, 111, 255, 0.08); + color: var(--blue); +} + .file-upload-button { display: inline-flex; grid-auto-flow: column; diff --git a/app/static/js/app.js b/app/static/js/app.js index 1e4b73d..aebcb4e 100644 --- a/app/static/js/app.js +++ b/app/static/js/app.js @@ -34,6 +34,13 @@ window.apiFetch = apiFetch; const newTaskForm = document.querySelector("#new-task-form"); const newTaskAutoMode = newTaskForm?.querySelector("input[name='auto_mode']"); const newTaskSubmitButton = document.querySelector("#new-task-submit-button"); +const newTaskSourceInputs = newTaskForm?.querySelectorAll("input[name='source_type']") || []; +const uploadSourcePanel = document.querySelector("#upload-source-panel"); +const existingSourcePanel = document.querySelector("#existing-source-panel"); +const nasFilePathInput = document.querySelector("#nas-file-path"); +const selectionProfileInput = document.querySelector("#selection-profile"); +const longLiveSettings = document.querySelector("#long-live-settings"); +let currentBrowseParentPath = ""; function updateNewTaskSubmitLabel() { if (!newTaskSubmitButton) return; @@ -45,6 +52,85 @@ if (newTaskAutoMode) { updateNewTaskSubmitLabel(); } +function selectedNewTaskSource() { + return newTaskForm?.querySelector("input[name='source_type']:checked")?.value || "upload"; +} + +function updateNewTaskSourcePanels() { + const useExisting = selectedNewTaskSource() === "nas"; + if (uploadSourcePanel) uploadSourcePanel.hidden = useExisting; + if (existingSourcePanel) existingSourcePanel.hidden = !useExisting; + const videoFileInput = document.querySelector("#video-file-input"); + if (videoFileInput) videoFileInput.required = !useExisting; + if (nasFilePathInput) nasFilePathInput.required = useExisting; + if (useExisting && !document.querySelector("#browse-current-path")?.dataset.loaded) { + browseVideoDirectory(""); + } +} + +newTaskSourceInputs.forEach((input) => input.addEventListener("change", updateNewTaskSourcePanels)); +updateNewTaskSourcePanels(); + +if (selectionProfileInput && longLiveSettings) { + const updateLongLiveSettings = () => { + longLiveSettings.hidden = selectionProfileInput.value !== "long_live_talk"; + }; + selectionProfileInput.addEventListener("change", updateLongLiveSettings); + updateLongLiveSettings(); +} + +async function browseVideoDirectory(path = "") { + const browser = document.querySelector("#video-file-browser"); + const currentPath = document.querySelector("#browse-current-path"); + if (!browser || !currentPath) return; + browser.textContent = "正在读取目录..."; + try { + const suffix = path ? `?path=${encodeURIComponent(path)}` : ""; + const data = await apiFetch(`/api/files/browse${suffix}`); + currentPath.textContent = `当前目录：${data.path || "不可用"}`; + currentPath.dataset.loaded = "true"; + currentBrowseParentPath = data.parent_path || ""; + browser.replaceChildren(); + if (data.error) { + const error = document.createElement("p"); + error.className = "form-hint error-text"; + error.textContent = data.error; + browser.append(error); + } + for (const directory of data.directories || []) { + const button = document.createElement("button"); + button.type = "button"; + button.className = "file-browser-item directory-item"; + button.textContent = `📁 ${directory.name}`; + button.addEventListener("click", () => browseVideoDirectory(directory.path)); + browser.append(button); + } + for (const file of data.files || []) { + const button = document.createElement("button"); + button.type = "button"; + button.className = "file-browser-item video-item"; + button.textContent = `🎬 ${file.name} · ${(Number(file.size || 0) / (1024 ** 3)).toFixed(2)} GB`; + button.addEventListener("click", () => { + nasFilePathInput.value = file.path; + browser.querySelectorAll(".is-selected").forEach((item) => item.classList.remove("is-selected")); + button.classList.add("is-selected"); + }); + browser.append(button); + } + if (!(data.directories || []).length && !(data.files || []).length && !data.error) { + browser.textContent = "此目录下没有可用的视频文件。"; + } + } catch (error) { + browser.textContent = `目录读取失败：${error.message}`; + } +} + +document.querySelector("#browse-refresh-button")?.addEventListener("click", () => { + const current = document.querySelector("#browse-current-path")?.textContent?.replace("当前目录：", "") || ""; + browseVideoDirectory(current); +}); +document.querySelector("#browse-parent-button")?.addEventListener("click", () => browseVideoDirectory(currentBrowseParentPath)); + if (newTaskForm) { newTaskForm.addEventListener("submit", async (event) => { event.preventDefault(); @@ -58,28 +144,38 @@ if (newTaskForm) { result.textContent = "正在创建任务..."; try { - if (!videoFileInput.files.length) { - throw new Error("请选择要上传的视频文件"); - } - const uploadData = new FormData(); - uploadData.append("task_name", payload.task_name || ""); - uploadData.append("platform", payload.platform || "general"); - uploadData.append("max_clip_duration", payload.max_clip_duration || "10"); - uploadData.append("candidate_clip_count", payload.candidate_clip_count || "12"); - uploadData.append("selection_profile", payload.selection_profile || "variety_comedy"); - uploadData.append("final_clip_target", payload.final_clip_target || "5"); - uploadData.append("ai_preference", ""); - uploadData.append("auto_mode", payload.auto_mode === "true" ? "true" : "false"); - uploadData.append("auto_metadata_use_ai", "false"); - uploadData.append("video_file", videoFileInput.files[0]); - const response = await fetch("/api/tasks/upload", { - method: "POST", - body: uploadData, - }); - - const data = await response.json(); - if (!response.ok) { - throw new Error(data.detail || "任务创建失败"); + if (!payload.selection_profile) throw new Error("请选择选片模式"); + let data; + if (selectedNewTaskSource() === "upload") { + if (!videoFileInput.files.length) throw new Error("请选择要上传的视频文件"); + const uploadData = new FormData(); + for (const key of [ + "task_name", "platform", "max_clip_duration", "candidate_clip_count", + "selection_profile", "final_clip_target", "highlight_density_per_hour", "highlight_total_limit", + ]) uploadData.append(key, payload[key] || ""); + uploadData.append("ai_preference", ""); + uploadData.append("auto_mode", payload.auto_mode === "true" ? "true" : "false"); + uploadData.append("auto_metadata_use_ai", "false"); + uploadData.append("video_file", videoFileInput.files[0]); + data = await apiFetch("/api/tasks/upload", { method: "POST", body: uploadData }); + } else { + if (!payload.nas_file_path) throw new Error("请选择本地或 NAS 中的已有视频文件"); + data = await apiFetch("/api/tasks", { + method: "POST", + body: JSON.stringify({ + task_name: payload.task_name, + source_type: "nas", + platform: payload.platform || "general", + nas_file_path: payload.nas_file_path, + max_clip_duration: Number(payload.max_clip_duration || 10), + candidate_clip_count: Number(payload.candidate_clip_count || 12), + selection_profile: payload.selection_profile, + final_clip_target: Number(payload.final_clip_target || 5), + highlight_density_per_hour: Number(payload.highlight_density_per_hour || 4), + highlight_total_limit: Number(payload.highlight_total_limit || 30), + auto_mode: payload.auto_mode === "true", + }), + }); } result.textContent = `${data.message}${payload.auto_mode === "true" ? " 全自动流水线已启动。" : ""} 正在进入详情页...`; window.location.href = data.detail_url; @@ -333,9 +429,11 @@ const aiAnalysisSummary = document.querySelector("#ai-analysis-summary"); const aiCandidateCountPill = document.querySelector("#ai-candidate-count-pill"); const aiCandidateCountInput = document.querySelector("#ai-candidate-count-input"); const aiSelectionProfile = aiAnalysisForm - ? aiAnalysisForm.dataset.selectionProfile || "variety_comedy" - : "variety_comedy"; + ? aiAnalysisForm.dataset.selectionProfile || "general" + : "general"; const aiFinalClipTarget = document.querySelector("#ai-final-clip-target"); +const aiHighlightDensity = document.querySelector("#ai-highlight-density"); +const aiHighlightTotalLimit = document.querySelector("#ai-highlight-total-limit"); const showAiHistoryButton = document.querySelector("#show-ai-history-button"); const refreshAiHistoryButton = document.querySelector("#refresh-ai-history-button"); const aiAnalysisHistory = document.querySelector("#ai-analysis-history"); @@ -1583,8 +1681,10 @@ async function saveTaskSelectionSettings() { method: "PATCH", headers: { "Content-Type": "application/json" }, body: JSON.stringify({ - selection_profile: aiSelectionProfile || "variety_comedy", + selection_profile: aiSelectionProfile, final_clip_target: finalTarget, + highlight_density_per_hour: Number(aiHighlightDensity?.value || 4), + highlight_total_limit: Number(aiHighlightTotalLimit?.value || 30), }), }); const data = await response.json(); diff --git a/app/templates/new_task.html b/app/templates/new_task.html index 685d1f2..d2f36b0 100644 --- a/app/templates/new_task.html +++ b/app/templates/new_task.html @@ -30,11 +30,14 @@

基本信息

视频来源 - -
+
+ + +
+
上传本机视频 - 选择视频后，会直接写入 E 盘任务的 source 目录；上传临时文件也在 E 盘。 + 默认最大 4 GB。更大的直播录像请使用“已有文件”，避免浏览器上传中断。

尚未选择视频文件。

+