From 439b63c81bbd4163e85d270c7477cd865fd1c6cf Mon Sep 17 00:00:00 2001 From: cat-shark Date: Thu, 13 Aug 2026 22:09:55 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E4=B8=BA=E5=85=A8=E9=83=A8=E4=BB=A3?= =?UTF-8?q?=E7=A0=81=E8=A1=A5=E5=85=85=E4=B8=AD=E6=96=87=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E5=B9=B6=E5=8A=A0=E5=85=A5=20AGENTS=20=E6=B3=A8=E9=87=8A?= =?UTF-8?q?=E8=A7=84=E8=8C=83?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 7 ++++ pyproject.toml | 5 +++ src/wov_sdk/__init__.py | 6 ++++ src/wov_sdk/models.py | 77 +++++++++++++++++++++++++++++++++++++++++ src/wov_sdk/server.py | 25 +++++++++++++ tests/test_models.py | 16 +++++++++ tests/test_server.py | 14 ++++++++ 7 files changed, 150 insertions(+) diff --git a/AGENTS.md b/AGENTS.md index 2e671c9..4a1286a 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -19,3 +19,10 @@ - Python 环境统一使用 uv 管理，不直接使用 pip 修改依赖。 - 测试必须达到 100% 行覆盖率，且只能通过调用真实代码路径覆盖。 + +## 代码注释规范 + +- 本仓库所有源码（Python、TOML 等支持注释的文件）必须配有详细中文注释，说明模块职责、协议模型字段语义、类与函数的作用以及关键逻辑，确保后续维护人员可以快速理解代码工作原理。 +- 新增或修改代码时，必须同步补充或更新对应注释；不得删除已有注释。 +- 测试代码同样必须配有中文注释，说明每条测试验证的行为。 +- JSON 数据文件不支持注释，字段语义以本仓库 `models.py` 中的模型注释为准；修改 JSON 字段时须同步更新文档。 diff --git a/pyproject.toml b/pyproject.toml index 997581e..6caae0e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -1,19 +1,24 @@ +# WOV SDK 项目配置：使用 uv 管理环境与依赖。 [project] name = "wov-sdk" version = "0.1.0" description = "WOV node protocol and SDK" requires-python = ">=3.11" +# 开发依赖放在 dev 组，由 uv sync 默认安装。 [dependency-groups] dev = ["pytest", "pytest-cov"] +# pytest 配置：只扫描 tests 目录，并强制 100% 行覆盖率。 [tool.pytest.ini_options] testpaths = ["tests"] addopts = "--cov=wov_sdk --cov-report=term-missing --cov-fail-under=100" +# 作为可发布包处理，便于节点仓库通过路径依赖引用。 [tool.uv] package = true +# 使用 src 布局，源码位于 src/wov_sdk。 [tool.setuptools] package-dir = {"" = "src"} diff --git a/src/wov_sdk/__init__.py b/src/wov_sdk/__init__.py index fc72086..52c909b 100644 --- a/src/wov_sdk/__init__.py +++ b/src/wov_sdk/__init__.py @@ -1,3 +1,8 @@ +"""WOV SDK 公共导出入口。 + +节点仓库和 wov-api 统一从 wov_sdk 导入协议模型，而无需关心具体模块路径。 +""" + from wov_sdk.models import ( HealthResponse, InvokeRequest, @@ -9,6 +14,7 @@ from wov_sdk.models import ( WorkflowNode, ) +# 对外稳定的公共 API 清单；新增模型时必须同步追加到这里。 __all__ = [ "HealthResponse", "InvokeRequest", diff --git a/src/wov_sdk/models.py b/src/wov_sdk/models.py index 4d85a50..0166ca0 100644 --- a/src/wov_sdk/models.py +++ b/src/wov_sdk/models.py @@ -1,3 +1,10 @@ +"""WOV 节点协议核心数据模型。 + +本模块定义节点 Manifest、调用请求/响应、健康检查、进度事件以及工作流 DAG 的 +通用数据结构。所有节点仓库和 wov-api 都依赖这些类，字段语义必须长期保持稳定， +新增能力时只能向后兼容地扩展字段。 +""" + from __future__ import annotations import json @@ -6,32 +13,54 @@ from typing import Any def _require_non_empty(value: str, name: str) -> None: + """校验必填字符串字段，空字符串或纯空白字符串都会被拒绝。""" if not value or not value.strip(): raise ValueError(f"{name} must not be empty") @dataclass class NodeManifest: + """节点注册清单：描述节点能力、启动命令、生命周期与资源参数。 + + 该清单由节点仓库根目录的 node.manifest.json 提供，也通过管理 API 注册到 + wov-api 的 SQLite 节点注册表中。NodeManager 依据它启动和回收节点进程。 + """ + + # 节点稳定唯一 ID，例如 faster-whisper；注册后不可随意更改。 id: str + # 展示名称，仅用于管理后台等界面。 name: str + # 节点版本号，与节点仓库 Git tag 保持一致。 version: str + # 能力标识，工作流通过 node_type 引用能力，而不是直接绑定具体仓库。 capability: str + # 启动命令；首项为 python 时 NodeManager 会替换为节点仓库 .venv 解释器。 command: list[str] + # 节点代码所在子仓库目录，相对 WOV 工作区根目录。 repo_dir: str = "." + # 启动节点进程时额外注入的环境变量。 env: dict[str, str] = field(default_factory=dict) + # 输入字段 schema，当前主要用于文档展示，后续可用于运行时校验。 input_schema: dict[str, Any] = field(default_factory=dict) + # 输出字段 schema，用于描述节点产物的名称与类型。 output_schema: dict[str, Any] = field(default_factory=dict) + # 单个实例允许并发承载的调用数；1 表示严格串行复用。 max_concurrency: int = 1 + # 空闲回收 TTL（秒）；0 表示调用结束即可被回收。 idle_ttl_seconds: int = 300 + # 节点进程启动后等待就绪输出的超时（秒）。 health_timeout_seconds: int = 10 + # 常驻标记；开启后即使空闲超过 TTL 也不会被回收。 keep_warm: bool = False def validate(self) -> None: + """校验 manifest 必填字段与数值边界，非法配置抛出 ValueError。""" _require_non_empty(self.id, "id") _require_non_empty(self.name, "name") _require_non_empty(self.version, "version") _require_non_empty(self.capability, "capability") _require_non_empty(self.repo_dir, "repo_dir") + # 命令不能为空，否则节点进程无法启动。 if not self.command: raise ValueError("command must not be empty") if self.max_concurrency < 1: @@ -42,6 +71,7 @@ class NodeManifest: raise ValueError("health_timeout_seconds must be >= 1") def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "id": self.id, "name": self.name, @@ -60,16 +90,19 @@ class NodeManifest: @classmethod def from_dict(cls, data: dict[str, Any]) -> "NodeManifest": + """从注册 API 或 JSON 文件解析出的字典恢复 manifest。""" return cls( id=str(data["id"]), name=str(data["name"]), version=str(data["version"]), capability=str(data["capability"]), + # command 可能缺省，解析时提供空列表兜底。 command=[str(item) for item in data.get("command", [])], repo_dir=str(data.get("repo_dir", ".")), env={str(k): str(v) for k, v in data.get("env", {}).items()}, input_schema=dict(data.get("input_schema", {})), output_schema=dict(data.get("output_schema", {})), + # 数值字段缺省时使用与 dataclass 一致的默认值。 max_concurrency=int(data.get("max_concurrency", 1)), idle_ttl_seconds=int(data.get("idle_ttl_seconds", 300)), health_timeout_seconds=int(data.get("health_timeout_seconds", 10)), @@ -78,6 +111,7 @@ class NodeManifest: @classmethod def load(cls, path: str) -> "NodeManifest": + """从磁盘上的 node.manifest.json 加载并校验 manifest。""" with open(path, "r", encoding="utf-8") as f: manifest = cls.from_dict(json.load(f)) manifest.validate() @@ -86,13 +120,21 @@ class NodeManifest: @dataclass class InvokeRequest: + """节点调用请求：由调度器或管理后台发送给节点 HTTP 服务。""" + + # 工作流运行 ID，用于追踪一次完整执行。 run_id: str + # 实际承载本次调用的节点实例 ID，由 NodeManager 回填。 node_instance_id: str + # 输入产物映射，key 为输入名，value 为产物 URI 或直接文本。 inputs: dict[str, Any] = field(default_factory=dict) + # 节点运行参数，例如采样率、语言、模型路径等。 params: dict[str, Any] = field(default_factory=dict) + # 节点产物输出目录。 output_dir: str = "." def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "run_id": self.run_id, "node_instance_id": self.node_instance_id, @@ -103,6 +145,7 @@ class InvokeRequest: @classmethod def from_dict(cls, data: dict[str, Any]) -> "InvokeRequest": + """从 HTTP 请求 JSON 解析调用请求。""" return cls( run_id=str(data["run_id"]), node_instance_id=str(data["node_instance_id"]), @@ -114,11 +157,17 @@ class InvokeRequest: @dataclass class InvokeResponse: + """节点调用响应：completed 表示成功，failed 表示执行失败。""" + + # 执行状态，固定为 completed / failed。 status: str + # 输出产物映射，key 为输出名，value 为产物 URI。 outputs: dict[str, Any] = field(default_factory=dict) + # 失败原因，仅在 failed 时有意义。 error: str | None = None def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "status": self.status, "outputs": self.outputs, @@ -127,7 +176,9 @@ class InvokeResponse: @classmethod def from_dict(cls, data: dict[str, Any]) -> "InvokeResponse": + """从 HTTP 响应 JSON 解析调用结果。""" return cls( + # 缺省按失败处理，避免未知状态被误判为成功。 status=str(data.get("status", "failed")), outputs=dict(data.get("outputs", {})), error=data.get("error"), @@ -136,11 +187,14 @@ class InvokeResponse: @dataclass class HealthResponse: + """节点健康检查响应：节点进程就绪后返回 ok 与自身标识。""" + status: str node_id: str version: str def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "status": self.status, "node_id": self.node_id, @@ -150,12 +204,15 @@ class HealthResponse: @dataclass class ProgressEvent: + """进度事件：预留用于节点向调度器上报执行进度。""" + run_id: str node_id: str progress: float message: str | None = None def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "run_id": self.run_id, "node_id": self.node_id, @@ -166,12 +223,19 @@ class ProgressEvent: @dataclass class WorkflowNode: + """工作流中的一个节点：声明节点类型、参数和输入引用。""" + + # 节点在 DAG 内的唯一 ID，例如 extract、asr。 id: str + # 引用的节点能力，例如 ffmpeg-extract、faster-whisper。 node_type: str + # 传递给节点 invoke 的 params。 params: dict[str, Any] = field(default_factory=dict) + # 输入引用，value 形如 "前序节点ID.输出名" 或 "input.入口字段"。 inputs: dict[str, str] = field(default_factory=dict) def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "id": self.id, "node_type": self.node_type, @@ -181,6 +245,7 @@ class WorkflowNode: @classmethod def from_dict(cls, data: dict[str, Any]) -> "WorkflowNode": + """从工作流定义 JSON 解析节点。""" return cls( id=str(data["id"]), node_type=str(data["node_type"]), @@ -191,30 +256,40 @@ class WorkflowNode: @dataclass class WorkflowEdge: + """工作流有向边：from_node 的输出流向 to_node 的输入。""" + from_node: str to_node: str def to_dict(self) -> dict[str, Any]: + """转换为 JSON 时使用 from/to 短字段名。""" return {"from": self.from_node, "to": self.to_node} @classmethod def from_dict(cls, data: dict[str, Any]) -> "WorkflowEdge": + """从工作流定义 JSON 解析边。""" return cls(from_node=str(data["from"]), to_node=str(data["to"])) @dataclass class WorkflowDefinition: + """工作流 DAG 定义：包含节点列表、依赖边和输入输出映射。""" + name: str version: int nodes: list[WorkflowNode] = field(default_factory=list) edges: list[WorkflowEdge] = field(default_factory=list) + # 用户上传入口与入口字段名的映射，例如 {"video_uri": "file"}。 entry_inputs: dict[str, Any] = field(default_factory=dict) + # 最终对外暴露的产物别名映射，例如 {"ass": "ass.ass_uri"}。 final_outputs: dict[str, Any] = field(default_factory=dict) def validate(self) -> None: + """校验 DAG 基本约束：名称、版本、节点 ID 唯一、边引用有效。""" _require_non_empty(self.name, "name") if self.version < 1: raise ValueError("version must be >= 1") + # 节点 ID 集合用于检查重复和边引用。 node_ids = {node.id for node in self.nodes} if len(node_ids) != len(self.nodes): raise ValueError("workflow node ids must be unique") @@ -223,6 +298,7 @@ class WorkflowDefinition: raise ValueError(f"edge references unknown node: {edge}") def to_dict(self) -> dict[str, Any]: + """转换为可 JSON 序列化的普通字典。""" return { "name": self.name, "version": self.version, @@ -234,6 +310,7 @@ class WorkflowDefinition: @classmethod def from_dict(cls, data: dict[str, Any]) -> "WorkflowDefinition": + """从工作流定义 JSON 解析 DAG。""" return cls( name=str(data["name"]), version=int(data.get("version", 1)), diff --git a/src/wov_sdk/server.py b/src/wov_sdk/server.py index 52da9f6..9952552 100644 --- a/src/wov_sdk/server.py +++ b/src/wov_sdk/server.py @@ -1,3 +1,10 @@ +"""标准节点 HTTP 服务端。 + +每个节点仓库通过 run_node 启动自己的 HTTP 服务，对外提供 GET /health 与 +POST /invoke 两个协议端点。wov-api 的 NodeManager 通过读取启动时打印的 +WOV_NODE_READY 行获得端口，再据此进行健康检查和调用。 +""" + from __future__ import annotations import json @@ -6,12 +13,15 @@ from typing import Callable from wov_sdk.models import HealthResponse, InvokeRequest, InvokeResponse, NodeManifest +# 节点业务处理函数签名：接收 InvokeRequest，返回 InvokeResponse。 InvokeHandler = Callable[[InvokeRequest], InvokeResponse] def _send_json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None: + """以 UTF-8 JSON 形式向客户端写出 HTTP 响应。""" body = json.dumps(payload, ensure_ascii=False).encode("utf-8") handler.send_response(status) + # 显式声明 JSON 字符集，保证中文内容在旧客户端下也能正确解析。 handler.send_header("Content-Type", "application/json; charset=utf-8") handler.send_header("Content-Length", str(len(body))) handler.end_headers() @@ -19,7 +29,11 @@ def _send_json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> N class NodeHTTPServer(ThreadingHTTPServer): + """线程化节点 HTTP 服务器，保存 manifest 和业务处理函数。""" + + # 每个请求使用独立线程，避免单个慢请求阻塞其他调用。 daemon_threads = True + # 允许地址复用，方便节点快速重启。 allow_reuse_address = True def __init__( @@ -28,18 +42,23 @@ class NodeHTTPServer(ThreadingHTTPServer): manifest: NodeManifest, handler: InvokeHandler, ) -> None: + """绑定节点清单与调用处理器，并使用内部请求处理器提供服务。""" self.manifest = manifest self.invoke_handler = handler super().__init__(server_address, _NodeRequestHandler) class _NodeRequestHandler(BaseHTTPRequestHandler): + """节点协议请求处理器：只实现 /health 与 /invoke 两个端点。""" + server: NodeHTTPServer def log_message(self, format: str, *args) -> None: # noqa: A002 + """静默访问日志，避免测试和日志系统产生噪声。""" return def do_GET(self) -> None: + """处理健康检查请求，返回节点 ID 与版本。""" if self.path != "/health": _send_json(self, 404, {"status": "error", "message": "not found"}) return @@ -51,16 +70,19 @@ class _NodeRequestHandler(BaseHTTPRequestHandler): _send_json(self, 200, response.to_dict()) def do_POST(self) -> None: + """处理调用请求，业务异常统一转换为 failed 响应。""" if self.path != "/invoke": _send_json(self, 404, {"status": "error", "message": "not found"}) return try: + # 按 Content-Length 精确读取请求体，避免长连接下的读悬挂。 length = int(self.headers.get("Content-Length", "0")) body = json.loads(self.rfile.read(length).decode("utf-8")) request = InvokeRequest.from_dict(body) response = self.server.invoke_handler(request) _send_json(self, 200, response.to_dict()) except Exception as exc: # noqa: BLE001 + # 任何处理异常都不能让节点进程崩溃，统一返回失败结果。 _send_json( self, 500, @@ -74,6 +96,7 @@ def create_node_server( host: str = "127.0.0.1", port: int = 0, ) -> NodeHTTPServer: + """创建节点服务器；manifest 校验通过后才允许启动。""" manifest.validate() return NodeHTTPServer((host, port), manifest, invoke_handler) @@ -84,7 +107,9 @@ def run_node( host: str = "127.0.0.1", port: int = 0, ) -> None: + """启动节点服务并阻塞运行，直到进程被外部终止。""" server = create_node_server(manifest, invoke_handler, host=host, port=port) + # 节点管理器从标准输出解析该行，因此必须 flush。 print(f"WOV_NODE_READY port={server.server_address[1]}", flush=True) try: server.serve_forever() diff --git a/tests/test_models.py b/tests/test_models.py index b515e56..3214685 100644 --- a/tests/test_models.py +++ b/tests/test_models.py @@ -1,3 +1,9 @@ +"""wov_sdk.models 的单元测试。 + +测试覆盖所有数据模型的 JSON 往返序列化、字段校验和 manifest 文件加载， +确保协议模型的稳定性。 +""" + import json import pytest @@ -15,6 +21,7 @@ from wov_sdk.models import ( def valid_manifest() -> NodeManifest: + """构造一个覆盖全部字段的合法 NodeManifest，供测试复用。""" return NodeManifest( id="echo", name="Echo", @@ -33,6 +40,7 @@ def valid_manifest() -> NodeManifest: def test_manifest_round_trip() -> None: + """验证 manifest 经过 to_dict/from_dict 后保持原值。""" manifest = valid_manifest() restored = NodeManifest.from_dict(manifest.to_dict()) assert restored == manifest @@ -53,6 +61,7 @@ def test_manifest_round_trip() -> None: ], ) def test_manifest_validation(field: str, value: object) -> None: + """验证必填字段为空或数值越界时抛出 ValueError。""" manifest = valid_manifest() setattr(manifest, field, value) with pytest.raises(ValueError): @@ -60,6 +69,7 @@ def test_manifest_validation(field: str, value: object) -> None: def test_manifest_load(tmp_path) -> None: + """验证 NodeManifest.load 能从 JSON 文件读取并校验。""" path = tmp_path / "node.manifest.json" path.write_text(json.dumps(valid_manifest().to_dict()), encoding="utf-8") loaded = NodeManifest.load(str(path)) @@ -67,6 +77,7 @@ def test_manifest_load(tmp_path) -> None: def test_invoke_request_round_trip() -> None: + """验证 InvokeRequest 的 JSON 往返序列化。""" request = InvokeRequest( run_id="run_1", node_instance_id="ni_1", @@ -79,12 +90,14 @@ def test_invoke_request_round_trip() -> None: def test_invoke_response_round_trip() -> None: + """验证 InvokeResponse 的 JSON 往返序列化。""" response = InvokeResponse(status="completed", outputs={"text": "hello"}) restored = InvokeResponse.from_dict(response.to_dict()) assert restored == response def test_health_and_progress_serialization() -> None: + """验证健康检查和进度事件模型的字典输出。""" health = HealthResponse(status="ok", node_id="echo", version="1.0.0") assert health.to_dict() == { "status": "ok", @@ -102,6 +115,7 @@ def test_health_and_progress_serialization() -> None: def test_workflow_node_and_edge_round_trip() -> None: + """验证工作流节点与边的 JSON 往返序列化。""" node = WorkflowNode( id="asr", node_type="faster-whisper", @@ -117,6 +131,7 @@ def test_workflow_node_and_edge_round_trip() -> None: def test_workflow_definition_round_trip_and_validation() -> None: + """验证完整 DAG 定义可往返序列化并通过校验。""" definition = WorkflowDefinition( name="demo", version=1, @@ -134,6 +149,7 @@ def test_workflow_definition_round_trip_and_validation() -> None: def test_workflow_definition_invalid() -> None: + """验证非法 DAG（空名、版本为 0、重复节点、未知边）被拒绝。""" with pytest.raises(ValueError): WorkflowDefinition(name="", version=1).validate() diff --git a/tests/test_server.py b/tests/test_server.py index 8709787..a2b7e36 100644 --- a/tests/test_server.py +++ b/tests/test_server.py @@ -1,3 +1,9 @@ +"""wov_sdk.server 的协议测试。 + +通过真实 HTTP 请求验证 /health 与 /invoke 端点的正常、404 和 500 分支， +并验证 run_node 的启动输出格式。 +""" + import json import threading import urllib.error @@ -10,6 +16,7 @@ from wov_sdk.server import NodeHTTPServer, create_node_server, run_node def valid_manifest() -> NodeManifest: + """构造一个合法的最小 Echo 节点 manifest。""" return NodeManifest( id="echo", name="Echo", @@ -20,6 +27,7 @@ def valid_manifest() -> NodeManifest: def test_create_node_server_validates_manifest() -> None: + """验证空 manifest 在创建服务器前就会被校验拒绝。""" with pytest.raises(ValueError): create_node_server( NodeManifest( @@ -34,6 +42,7 @@ def test_create_node_server_validates_manifest() -> None: def _post(server: NodeHTTPServer, path: str, body: dict | None = None) -> tuple[int, dict]: + """向节点服务器发送 POST 请求并返回状态码与 JSON 响应。""" data = json.dumps(body or {}).encode("utf-8") request = urllib.request.Request( f"http://127.0.0.1:{server.server_address[1]}{path}", @@ -49,6 +58,7 @@ def _post(server: NodeHTTPServer, path: str, body: dict | None = None) -> tuple[ def _get(server: NodeHTTPServer, path: str) -> tuple[int, dict]: + """向节点服务器发送 GET 请求并返回状态码与 JSON 响应。""" try: with urllib.request.urlopen( f"http://127.0.0.1:{server.server_address[1]}{path}", timeout=5 @@ -59,6 +69,7 @@ def _get(server: NodeHTTPServer, path: str) -> tuple[int, dict]: def test_http_protocol() -> None: + """验证健康检查、正常调用、404 与非法请求体等协议分支。""" def handler(request: InvokeRequest) -> InvokeResponse: return InvokeResponse( status="completed", @@ -66,6 +77,7 @@ def test_http_protocol() -> None: ) server = create_node_server(valid_manifest(), handler) + # 在独立线程中运行服务器，主线程执行 HTTP 客户端断言。 thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() try: @@ -103,6 +115,7 @@ def test_http_protocol() -> None: def test_http_protocol_handler_error() -> None: + """验证业务处理函数抛异常时返回 500 与 failed 状态。""" def failing_handler(request: InvokeRequest) -> InvokeResponse: raise RuntimeError("boom") @@ -130,6 +143,7 @@ def test_http_protocol_handler_error() -> None: def test_run_node(monkeypatch, capsys) -> None: + """验证 run_node 打印 WOV_NODE_READY 端口行并进入服务循环。""" called = [] def fake_serve_forever(server: NodeHTTPServer) -> None: