feat: VRSub 单体应用（WOV 单机版）初始提交

为视频生成 VR 双眼字幕的单体实现：FastAPI 后端、调度器与全部节点（提音/转写/翻译/ASS/抽帧/OCR/LLM 过滤）在单进程内运行。 - 节点协议（wov_sdk 数据模型）与分布式版保持一致，预留回退桥梁 - 工作流即数据：DAG 存于 workflows/*.json，模型/链路改动只改数据 - 调度器：拓扑顺序执行、断点续跑（产物重建）、任务暂停/继续 - 抽帧按帧间隔（select 按帧号精确取帧），VLM OCR 与 LLM 过滤使用自适应线程池弹性并发，并打印数据处理速度进度日志 - 100% 行覆盖率（pytest --cov-fail-under=100）
2026-08-16 23:58:25 +08:00
commit 4746e0363f
75 changed files with 10969 additions and 0 deletions
@@ -0,0 +1,308 @@
+"""工作流调度器。
+
+轮询 SQLite 中的排队任务，按工作流 DAG 的拓扑顺序依次调用进程内节点
+处理器，并把节点产物登记为任务产物。单体版使用单线程顺序执行，节点在
+同一进程内直接调用，不再经过子进程与 HTTP 协议。
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from datetime import datetime, timezone
+from pathlib import Path
+
+from wov_sdk.models import InvokeRequest, WorkflowDefinition
+
+from wov_app import registry
+from wov_app.config import SCHEDULER_INTERVAL_SECONDS
+from wov_app.db import Database
+from wov_app.logging import get_logger
+
+# 调度器运行日志：节点进度、暂停/续跑等状态变化。
+logger = get_logger("scheduler")
+
+def _now_iso() -> str:
+    """返回当前 UTC 时间的 ISO 格式字符串。"""
+    return datetime.now(timezone.utc).isoformat()
+
+
+def topological_sort(definition: WorkflowDefinition) -> list[str]:
+    """对工作流 DAG 做拓扑排序，返回可执行的节点 ID 顺序。"""
+    nodes = {node.id: node for node in definition.nodes}
+    # 统计每个节点的入度，并记录依赖关系。
+    indegree = {node_id: 0 for node_id in nodes}
+    dependents: dict[str, list[str]] = {node_id: [] for node_id in nodes}
+
+    for edge in definition.edges:
+        # 边引用了不存在的节点时直接报错。
+        if edge.from_node not in nodes or edge.to_node not in nodes:
+            raise ValueError(f"unknown edge: {edge.from_node} -> {edge.to_node}")
+        indegree[edge.to_node] += 1
+        dependents[edge.from_node].append(edge.to_node)
+
+    # Kahn 算法：从入度为 0 的节点开始逐层取出。
+    queue = [node_id for node_id, degree in indegree.items() if degree == 0]
+    ordered: list[str] = []
+    while queue:
+        current = queue.pop(0)
+        ordered.append(current)
+        for dependent in dependents[current]:
+            indegree[dependent] -= 1
+            if indegree[dependent] == 0:
+                queue.append(dependent)
+
+    # 排序结果数量不足说明存在环，无法确定执行顺序。
+    if len(ordered) != len(nodes):
+        raise ValueError("workflow contains a cycle")
+    return ordered
+
+
+class WorkflowScheduler:
+    """后台任务调度器：单线程轮询并执行排队中的工作流运行。"""
+
+    def __init__(
+        self,
+        db: Database,
+        storage_dir: Path,
+        interval_seconds: float | None = None,
+    ) -> None:
+        """保存依赖并初始化轮询线程控制字段。"""
+        self.db = db
+        self.storage_dir = storage_dir
+        self.interval_seconds = interval_seconds or SCHEDULER_INTERVAL_SECONDS
+        self._thread: threading.Thread | None = None
+        self._stopping = False
+
+    def start(self) -> None:
+        """启动调度线程；重复调用无副作用。"""
+        if self._thread is not None:
+            return
+        self._stopping = False
+        self._thread = threading.Thread(
+            target=self._loop,
+            name="wov-workflow-scheduler",
+            daemon=True,
+        )
+        self._thread.start()
+
+    def stop(self) -> None:
+        """请求停止并等待轮询线程退出。"""
+        self._stopping = True
+        if self._thread is not None:
+            self._thread.join(timeout=5)
+            self._thread = None
+
+    def _loop(self) -> None:
+        """轮询循环：有排队任务就立即执行，否则休眠一个间隔。"""
+        while not self._stopping:
+            run = self.db.next_queued_run()
+            if run is not None:
+                self.execute_run(run["id"])
+            else:
+                time.sleep(self.interval_seconds)
+
+    def _resolve_ref(
+        self,
+        ref: str,
+        run_input_uri: str | None,
+        outputs_by_node: dict[str, dict[str, str]],
+    ) -> str | None:
+        """解析输入引用：input.xxx 取任务入口，node.key 取前序节点产物。"""
+        # 入口引用以 input. 为前缀。
+        if ref.startswith("input."):
+            return run_input_uri
+        # 其余引用必须形如 "节点ID.输出名"。
+        node_id, separator, key = ref.partition(".")
+        if not separator:
+            return None
+        return outputs_by_node.get(node_id, {}).get(key)
+
+    def execute_run(self, run_id: str) -> None:
+        """执行单个任务：加载 DAG、按拓扑顺序调用节点并登记产物。"""
+        run = self.db.get_run(run_id)
+        # 任务不存在或不在可执行状态（排队/暂停）时直接返回，避免重复执行。
+        if run is None or run["status"] not in ("QUEUED", "PAUSED"):
+            return
+
+        # 工作流或版本记录丢失时把任务标记为失败。
+        workflow = self.db.get_workflow(run["workflow_id"])
+        if workflow is None:
+            self.db.update_run(run_id, status="FAILED", error="workflow not found", updated_at=_now_iso())
+            return
+
+        version = self.db.get_workflow_version(run["workflow_id"], run["workflow_version"])
+        if version is None:
+            self.db.update_run(run_id, status="FAILED", error="workflow version not found", updated_at=_now_iso())
+            return
+
+        # 解析并校验 DAG，随后计算拓扑执行顺序。
+        definition = WorkflowDefinition.from_dict(version["definition"])
+        definition.validate()
+        ordered = topological_sort(definition)
+        # 从已登记产物重建已完成节点的输出，支持暂停后断点续跑。
+        outputs_by_node = self.db.restore_run_outputs(run_id)
+        run_started = time.monotonic()
+        self.db.update_run(run_id, status="RUNNING", progress=0, updated_at=_now_iso())
+        try:
+            for index, node_id in enumerate(ordered):
+                # 暂停检查：用户暂停后调度器在节点边界停下，保持 PAUSED 等待续跑。
+                current = self.db.get_run(run_id)
+                if current is None or current["status"] == "PAUSED":
+                    logger.info("任务 %s 已暂停，停止在节点 %s 之前", run_id, node_id)
+                    return
+                # 断点续跑：跳过已产出结果的节点（其产物已作为输入可用）。
+                if node_id in outputs_by_node:
+                    continue
+                # 当前节点进度 = 已完成节点数 / 总节点数。
+                node_spec = next(item for item in definition.nodes if item.id == node_id)
+                self.db.update_run(
+                    run_id,
+                    current_node_id=node_id,
+                    progress=index / len(ordered),
+                    updated_at=_now_iso(),
+                )
+                # 解析节点声明的每个输入引用，缺任一输入即失败。
+                invoke_inputs: dict[str, str] = {}
+                for input_name, ref in node_spec.inputs.items():
+                    value = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node)
+                    if value is None:
+                        raise ValueError(f"missing input {input_name} for node {node_id}")
+                    invoke_inputs[input_name] = value
+
+                # 前端框选的 crop 等参数覆盖：按节点 ID 合并进节点参数。
+                node_params = dict(node_spec.params)
+                overrides = run.get("param_overrides") or {}
+                node_params.update(overrides.get(node_id, {}))
+
+                # 每个任务的每个节点使用独立产物目录，避免并发冲突。
+                node_started = time.monotonic()
+                output_dir = (
+                    self.storage_dir
+                    / "runs"
+                    / run_id
+                    / "steps"
+                    / node_id
+                )
+                response = registry.invoke(
+                    node_spec.node_type,
+                    InvokeRequest(
+                        run_id=run_id,
+                        node_instance_id="",
+                        inputs=invoke_inputs,
+                        params=node_params,
+                        output_dir=str(output_dir),
+                    ),
+                )
+                # 节点返回非 completed 即视为步骤失败。
+                if response.status != "completed":
+                    raise RuntimeError(response.error or f"node {node_id} failed")
+
+                # 记录节点输出，供后续节点引用和最终产物映射使用。
+                outputs_by_node[node_id] = {
+                    str(key): str(value) for key, value in response.outputs.items()
+                }
+                for key, uri in outputs_by_node[node_id].items():
+                    # 产物名带节点前缀，例如 asr.srt_uri，避免跨节点重名。
+                    artifact = {
+                        "run_id": run_id,
+                        "node_id": node_id,
+                        "name": f"{node_id}.{key}",
+                        "uri": uri,
+                        "mime_type": self._mime_type(uri),
+                        "size": self._file_size(uri),
+                    }
+                    self.db.create_artifact(artifact)
+                # 进度日志：节点序号/总数、耗时与任务累计运行时间（数据速度可观测）。
+                logger.info(
+                    "任务 %s 进度 %d/%d 节点: %s 耗时 %.1fs, 运行累计 %.1fs",
+                    run_id, index + 1, len(ordered), node_id,
+                    time.monotonic() - node_started,
+                    time.monotonic() - run_started,
+                )
+            # 处理 final_outputs，为用户端提供简洁的下载别名。
+            for alias, ref in definition.final_outputs.items():
+                resolved = self._resolve_ref(ref, run.get("input_uri"), outputs_by_node)
+                if resolved is not None:
+                    # 最终产物按 上传文件名.标识.时间戳 重命名，区分语言与版本。
+                    resolved = self._final_artifact_uri(resolved, run, definition, alias, ref)
+                    self.db.create_artifact(
+                        {
+                            "run_id": run_id,
+                            "node_id": ref.partition(".")[0],
+                            "name": alias,
+                            "uri": resolved,
+                            "mime_type": self._mime_type(resolved),
+                            "size": self._file_size(resolved),
+                        }
+                    )
+
+            # 全部节点成功后标记完成；期间被暂停则保持 PAUSED，等待续跑补做收尾。
+            if self.db.get_run(run_id)["status"] == "PAUSED":
+                logger.info("任务 %s 节点全部完成但已暂停，保持 PAUSED", run_id)
+                return
+            self.db.update_run(
+                run_id,
+                status="COMPLETED",
+                current_node_id=None,
+                progress=1.0,
+                updated_at=_now_iso(),
+            )
+        except Exception as exc:  # noqa: BLE001
+            # 任一步骤异常都结束任务并记录错误，等待用户重试。
+            self.db.update_run(
+                run_id,
+                status="FAILED",
+                error=str(exc),
+                updated_at=_now_iso(),
+            )
+
+    def _final_artifact_uri(
+        self,
+        resolved: str,
+        run: dict,
+        definition: WorkflowDefinition,
+        alias: str,
+        ref: str,
+    ) -> str:
+        """把最终产物重命名为 上传文件名.标识.时间戳 并返回新 URI。
+
+        标识优先取产出节点的 target_language 参数（如 zh-CN），否则回退为
+        产物别名；时间戳取当前时刻，用于区分同一上传文件的多次运行版本。
+        重命名在原地进行（同目录），不复制文件。
+        """
+        source = Path(resolved)
+        # 续跑等场景下源文件可能已被上次收尾重命名过：不再重命名，原样返回。
+        if not source.is_file():
+            return resolved
+        # 基础名来自上传文件名；无上传文件时退回通用名称 subtitle。
+        base = Path(run["input_uri"]).stem if run.get("input_uri") else "subtitle"
+        # 通过最终输出引用定位产出节点，取其语言参数作为标识。
+        node_id = ref.partition(".")[0]
+        node = next((item for item in definition.nodes if item.id == node_id), None)
+        tag = (node.params.get("target_language") if node else None) or alias
+        timestamp = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
+        new_path = source.with_name(f"{base}.{tag}.{timestamp}{source.suffix}")
+        source.rename(new_path)
+        return str(new_path)
+
+    @staticmethod
+    def _mime_type(uri: str) -> str:
+        """按扩展名推断产物 MIME 类型，未知类型使用通用二进制类型。"""
+        path = Path(uri)
+        suffix = path.suffix.lower()
+        return {
+            ".srt": "application/x-subrip",
+            ".ass": "text/plain",
+            ".wav": "audio/wav",
+            ".mp4": "video/mp4",
+            ".txt": "text/plain",
+        }.get(suffix, "application/octet-stream")
+
+    @staticmethod
+    def _file_size(uri: str) -> int:
+        """读取产物文件大小；文件缺失时按 0 处理。"""
+        try:
+            return Path(uri).stat().st_size
+        except OSError:
+            return 0