diff --git a/app/config.py b/app/config.py index cf1fddf..d811d90 100644 --- a/app/config.py +++ b/app/config.py @@ -144,6 +144,16 @@ class SoftwareConfig: return None return plugins.get(software_id) + def get_plugin_task_config(self, software_id: str, task_type: str) -> Optional[dict]: + plugin = self.get_plugin_config(software_id) + if not plugin: + return None + tasks = plugin.get("tasks", {}) + if not isinstance(tasks, dict): + return None + task_cfg = tasks.get(task_type) + return task_cfg if isinstance(task_cfg, dict) else None + settings = Settings() software_config = SoftwareConfig(settings.software_config_path) diff --git a/app/core/plugin_http_client.py b/app/core/plugin_http_client.py index 768c0fe..1cd49a5 100644 --- a/app/core/plugin_http_client.py +++ b/app/core/plugin_http_client.py @@ -18,7 +18,7 @@ class PluginHttpClient: def __init__(self, config_provider=software_config): self._config_provider = config_provider - async def submit_task(self, software_id: str, payload: Dict[str, Any]) -> Dict[str, Any]: + async def submit_task(self, software_id: str, task_type: str, payload: Dict[str, Any]) -> Dict[str, Any]: plugin_config = self._config_provider.get_plugin_config(software_id) if not plugin_config: raise PluginSubmitError(f"Plugin config for software '{software_id}' not found") @@ -26,19 +26,47 @@ class PluginHttpClient: base_url = plugin_config.get("base_url") submit_path = plugin_config.get("submit_path", "/api/plugin/tasks") timeout = int(plugin_config.get("request_timeout_sec", 10)) + task_config = self._config_provider.get_plugin_task_config(software_id, task_type) or {} + submit_path = task_config.get("path", submit_path) + body_mode = task_config.get("body_mode", "passthrough") if not base_url: raise PluginSubmitError(f"Plugin base_url for software '{software_id}' is missing") url = parse.urljoin(base_url.rstrip("/") + "/", submit_path.lstrip("/")) + request_payload = self._build_payload(payload=payload, body_mode=body_mode) try: - return await asyncio.to_thread(self._post_json, url, payload, timeout) + return await asyncio.to_thread(self._post_json, url, request_payload, timeout) except PluginSubmitError: raise except Exception as exc: raise PluginSubmitError(str(exc)) from exc + @staticmethod + def _build_payload(payload: Dict[str, Any], body_mode: str) -> Dict[str, Any]: + task_params = payload.get("task_params", {}) if isinstance(payload.get("task_params"), dict) else {} + model_path = payload.get("model_path") + callback = { + "execution_id": payload.get("execution_id"), + "batch_id": payload.get("batch_id"), + "item_id": payload.get("item_id"), + "callback_url": payload.get("callback_url"), + "attempt": payload.get("attempt"), + } + + if body_mode == "file_path": + return {"filePath": model_path, **task_params, **callback} + if body_mode == "task_params_only": + return {**task_params, **callback} + if body_mode == "project_open": + return {**task_params, **callback} + if body_mode == "passthrough": + return payload + + # Unknown modes fallback to passthrough for compatibility. + return payload + @staticmethod def _post_json(url: str, payload: Dict[str, Any], timeout: int) -> Dict[str, Any]: raw = json.dumps(payload).encode("utf-8") diff --git a/app/core/serial_batch_executor.py b/app/core/serial_batch_executor.py index 1bef78c..5a59d37 100644 --- a/app/core/serial_batch_executor.py +++ b/app/core/serial_batch_executor.py @@ -77,7 +77,11 @@ class SerialBatchExecutor: error_message = None try: - plugin_response = await self._plugin_client.submit_task(item.software_id, payload) + plugin_response = await self._plugin_client.submit_task( + item.software_id, + item.task_type, + payload, + ) await self._batch_manager.mark_item_waiting_callback(item_id, execution_id, plugin_response) timeout_sec = self._batch_manager.get_callback_timeout_sec(item.software_id) diff --git a/cad-batch-plan.md b/cad-batch-plan.md index a99c87a..20eba1b 100644 --- a/cad-batch-plan.md +++ b/cad-batch-plan.md @@ -1,102 +1,147 @@ -# 先落盘再实施：CAD 批处理串行执行框架计划 +# CAD 批处理串行执行框架（设计与开发进度） -## 简要总结 -先将本计划保存到项目根目录 `cad-batch-plan.md`，随后按“模型与执行器骨架 -> 插件回调闭环 -> WebSocket 接入 -> 测试验收”顺序实施。 -执行策略已锁定：全局严格 FIFO 串行、插件异步回调、失败重试后继续、首版内存态。 +## 1. 文档目的 +本文件用于同步两类信息： +1. 总体批处理设计（长期有效） +2. 当前代码开发进度（持续更新） -## 实施顺序（决策完成） -1. 文档落盘阶段 -将本计划原文写入 `cad-batch-plan.md`，作为实现基线文档。 +--- -2. 领域模型与配置阶段 -新增 `app/models/cad_batch.py`。 -扩展 `configs/software_config.yaml`：增加扩展名路由和插件地址配置。 -扩展 `app/config.py`：提供读取 `routing` 与 `plugins` 的方法。 +## 2. 总体设计（已定稿） -3. 核心执行引擎阶段 -新增 `app/core/cad_task_router.py`：扩展名归一化与 `software_id` 路由。 -新增 `app/core/plugin_http_client.py`：向插件提交任务 HTTP 调用。 -新增 `app/core/plugin_callback_registry.py`：`execution_id -> Future` 等待回调。 -新增 `app/core/serial_batch_executor.py`：全局单 worker + `asyncio.Queue` 串行执行。 -新增 `app/core/cad_batch_manager.py`：批次创建、入队、状态查询与聚合。 +### 2.1 目标 +- 前端提交批次任务（多模型、多步骤按条目表达） +- 后端按模型扩展名路由到对应 CAD 插件 +- 全局严格 FIFO 串行执行（跨软件也串行） +- 插件通过 HTTP 接口执行，结果通过回调闭环 +- 单任务失败后按重试策略处理，最终失败不阻塞后续任务 -4. 回调 API 阶段 -新增 `app/api/v1/plugin_callbacks.py`：接收插件异步回调。 -在 `app/main.py` 注册路由并在 `startup/shutdown` 启停执行器。 +### 2.2 执行策略 +- 队列：全局单队列 + 单 worker +- 顺序：严格按提交顺序 +- 状态机：`queued -> dispatching -> waiting_callback -> succeeded/failed` +- 失败策略：重试后失败，继续执行后续任务 +- 存储策略：首版内存态（重启不恢复） -5. WebSocket 接入阶段 -在 `app/api/v1/websocket.py` 增加批处理消息类型与处理分支。 -新增消息：提交批次、查询批次、查询单条任务。 -复用现有推送机制广播条目状态变化与批次汇总进度。 +### 2.3 架构模块 +- 模型层：`app/models/cad_batch.py` +- 路由层：`app/core/cad_task_router.py` +- 插件下发层：`app/core/plugin_http_client.py` +- 回调等待层：`app/core/plugin_callback_registry.py` +- 串行执行层：`app/core/serial_batch_executor.py` +- 业务管理层：`app/core/cad_batch_manager.py` +- 插件回调接口：`app/api/v1/plugin_callbacks.py` +- 前端通道（WebSocket）：`app/api/v1/websocket.py` -6. 测试与验收阶段 -新增 `tests/test_cad_task_router.py`。 -新增 `tests/test_serial_batch_executor.py`。 -新增 `tests/test_plugin_callback_api.py`。 -补充端到端场景测试（可用 mock 插件）。 +--- -## 公共接口与类型变更 -1. 配置结构新增 -`routing.extension_to_software: dict[str, str]`。 -`plugins.{software_id}.base_url/submit_path/request_timeout_sec/callback_timeout_sec/max_retries/retry_backoff_sec`。 +## 3. 公共接口设计 -2. 新增对外数据模型 -`BatchSubmitRequest`、`BatchSubmitItem`、`BatchJob`、`BatchItem`、`PluginCallbackPayload`。 -状态枚举：`BatchStatus`、`BatchItemStatus`。 +### 3.1 WebSocket 消息 +- `submit_batch_tasks`：提交批次 +- `get_batch_status`：查询批次 +- `get_batch_item_status`：查询条目 -3. 新增插件回调 HTTP 接口 -`POST /api/v1/plugin-callbacks/task-result`。 -核心字段：`execution_id`、`software_id`、`status`、`error_message`、`result`、`finished_at`、`token/signature`。 +### 3.2 WebSocket 推送事件 +- `batch_created` +- `batch_item_update` +- `batch_completed` -4. 新增 WebSocket 消息类型 -`submit_batch_tasks`、`get_batch_status`、`get_batch_item_status`。 -推送事件建议：`batch_created`、`batch_item_update`、`batch_completed`。 +### 3.3 插件回调 HTTP +- `POST /api/v1/plugin-callbacks/task-result` +- 核心字段： + - `execution_id` + - `software_id` + - `status` (`success|failed`) + - `error_message` + - `result` + - `finished_at` + - `token/signature`（当前实现为 token 校验） -## 核心执行规则 -1. 入队顺序 -严格按前端提交顺序入全局队列，不按 CAD 分组重排。 +--- -2. 单条任务流程 -`queued -> dispatching -> waiting_callback -> succeeded/failed`。 +## 4. 配置设计 -3. 失败策略 -网络错误、超时、插件失败均按配置重试；超过次数后标记失败并继续下一个任务。 +### 4.1 已支持配置 +- `routing.extension_to_software` +- `plugins.{software_id}.base_url` +- `plugins.{software_id}.request_timeout_sec` +- `plugins.{software_id}.callback_timeout_sec` +- `plugins.{software_id}.max_retries` +- `plugins.{software_id}.retry_backoff_sec` +- `plugins.{software_id}.callback_token` +- `plugins.{software_id}.tasks.{task_type}.path` +- `plugins.{software_id}.tasks.{task_type}.body_mode` -4. 幂等与重复回调 -同 `execution_id` 仅首次生效，重复回调只记录日志不改状态。 +### 4.2 body_mode 语义 +- `file_path`：发送 `filePath + task_params + callback元信息` +- `task_params_only`：发送 `task_params + callback元信息` +- `project_open`：发送 `task_params + callback元信息` +- `passthrough`：原样透传 -## 测试场景与验收标准 -1. 路由测试 -`.prt/.asm/.rvt/.prt.1` 路由正确，未知扩展名标记失败不阻塞队列。 +--- -2. 串行测试 -混合 CAD 任务执行中任意时刻仅 1 条任务处于运行态，顺序与提交一致。 +## 5. 当前开发进度（截至本次更新） -3. 回调闭环测试 -插件提交成功且回调成功时，任务状态可从等待态正确转成功态。 +### 5.1 已完成 +- [x] 计划文档落盘：`cad-batch-plan.md` +- [x] 批处理核心模型、执行器、管理器已实现 +- [x] 插件异步回调机制与幂等处理已实现 +- [x] 主应用 `startup/shutdown` 已接入批处理执行器 +- [x] WebSocket 提交/查询批次能力已接入 +- [x] 插件任务映射已支持按 `software_id + task_type` 路由 +- [x] 单元测试已覆盖路由/执行器/回调/API 映射 -4. 重试测试 -首轮超时、次轮成功时，重试计数与退避间隔符合配置。 +### 5.2 已集成的 CAD 插件资料 -5. 继续执行测试 -中间任务最终失败后，后续任务继续执行并可完成。 +#### Creo (localhost:12345) +- `open_model` -> `/api/model/open` (`file_path`) +- `shrinkwrap_shell` -> `/api/creo/shrinkwrap/shell` (`task_params_only`) +- `shell_analysis` -> `/api/analysis/shell-analysis` (`task_params_only`) +- `close_model` -> `/api/model/close` (`task_params_only`, 占位) -6. 批次聚合测试 -全成功为 `completed`；部分失败为 `completed_with_errors`；统计字段准确。 +#### Revit (localhost:9000) +- `open_model` -> `/api/open` (`file_path`) +- `shell_analyze` -> `/api/shell/analyze` (`task_params_only`) +- `shell_execute` -> `/api/shell/execute` (`task_params_only`) +- `close_model` -> `/api/close` (`task_params_only`, 占位) -7. 回调安全测试 -非法 token/signature 回调被拒绝，任务状态不被污染。 +#### PDMS (localhost:9001) +- `open_project` -> `/api/project/open` (`project_open`) +- `open_mdb` -> `/api/mdb/open` (`project_open`) +- `shrinkwrap_model` -> `/api/model/shrinkwrap` (`task_params_only`) +- `simplify_model` -> `/api/model/simplify` (`task_params_only`) +- `close_project` -> `/api/project/close` (`task_params_only`, 占位) -## 默认值与假设 -1. 单模型单步骤任务粒度。 -2. 插件地址来自静态 YAML 配置。 -3. 插件通过异步 HTTP 回调返回结果。 -4. 首版不做重启恢复，仅内存态。 -5. 离线或不可达插件按重试后失败并继续。 +### 5.3 当前测试状态 +- 已通过：`9 passed` +- 覆盖文件： + - `tests/test_cad_task_router.py` + - `tests/test_serial_batch_executor.py` + - `tests/test_plugin_callback_api.py` + - `tests/test_plugin_http_client_mapping.py` -## 交付完成定义 -1. 已存在并提交 `cad-batch-plan.md`。 -2. 核心新增模块可启动并可接收批次任务。 -3. 回调可驱动状态闭环。 -4. WebSocket 可提交与查询批次。 -5. 关键测试通过且无阻断级错误。 +--- + +## 6. 待补资料与后续工作 + +### 6.1 你后续补充后可继续对接 +1. 每个 `task_type` 的最终命名标准（是否沿用当前命名） +2. Revit 大任务轮询接口细节（如 `/api/task` 的请求/返回） +3. 关闭类接口是否真实可用（当前按占位映射） +4. 回调安全方案最终版（token 还是 signature，签名算法与字段） +5. 各 API 的完整错误码与失败语义（用于统一重试/终止策略） + +### 6.2 下一步建议 +- 基于真实插件联调一轮 `open -> 处理 -> close` +- 把占位关闭接口替换为确认可用的真实接口 +- 增加端到端集成测试（含回调超时、插件离线、重复回调场景） + +--- + +## 7. 验收标准（当前版本） +- 可提交批次并进入全局串行队列 +- 可按扩展名路由并向插件正确下发 +- 可通过回调驱动状态闭环 +- 单任务失败不会阻塞后续任务 +- 可通过 WebSocket 查询批次与条目状态 diff --git a/configs/software_config.yaml b/configs/software_config.yaml index 4b59c74..490af86 100644 --- a/configs/software_config.yaml +++ b/configs/software_config.yaml @@ -58,28 +58,70 @@ routing: plugins: creo: - base_url: "http://127.0.0.1:19001" + base_url: "http://localhost:12345" submit_path: "/api/plugin/tasks" request_timeout_sec: 10 callback_timeout_sec: 60 max_retries: 1 retry_backoff_sec: [1, 3] callback_token: "creo-callback-token" + tasks: + open_model: + path: "/api/model/open" + body_mode: "file_path" + shrinkwrap_shell: + path: "/api/creo/shrinkwrap/shell" + body_mode: "task_params_only" + shell_analysis: + path: "/api/analysis/shell-analysis" + body_mode: "task_params_only" + close_model: + path: "/api/model/close" + body_mode: "task_params_only" pdms: - base_url: "http://127.0.0.1:19002" + base_url: "http://localhost:9001" submit_path: "/api/plugin/tasks" request_timeout_sec: 10 callback_timeout_sec: 60 max_retries: 1 retry_backoff_sec: [1, 3] callback_token: "pdms-callback-token" + tasks: + open_project: + path: "/api/project/open" + body_mode: "project_open" + open_mdb: + path: "/api/mdb/open" + body_mode: "project_open" + shrinkwrap_model: + path: "/api/model/shrinkwrap" + body_mode: "task_params_only" + simplify_model: + path: "/api/model/simplify" + body_mode: "task_params_only" + close_project: + path: "/api/project/close" + body_mode: "task_params_only" revit: - base_url: "http://127.0.0.1:19003" + base_url: "http://localhost:9000" submit_path: "/api/plugin/tasks" request_timeout_sec: 10 callback_timeout_sec: 60 max_retries: 1 retry_backoff_sec: [1, 3] callback_token: "revit-callback-token" + tasks: + open_model: + path: "/api/open" + body_mode: "file_path" + shell_analyze: + path: "/api/shell/analyze" + body_mode: "task_params_only" + shell_execute: + path: "/api/shell/execute" + body_mode: "task_params_only" + close_model: + path: "/api/close" + body_mode: "task_params_only" diff --git a/tests/test_plugin_http_client_mapping.py b/tests/test_plugin_http_client_mapping.py new file mode 100644 index 0000000..34f9282 --- /dev/null +++ b/tests/test_plugin_http_client_mapping.py @@ -0,0 +1,73 @@ +import pytest + +from app.core.plugin_http_client import PluginHttpClient + + +@pytest.mark.asyncio +async def test_plugin_http_client_uses_creo_open_model_endpoint_and_payload(monkeypatch): + client = PluginHttpClient() + captured = {} + + def fake_post_json(url, payload, timeout): + captured["url"] = url + captured["payload"] = payload + captured["timeout"] = timeout + return {"status": "success"} + + monkeypatch.setattr(PluginHttpClient, "_post_json", staticmethod(fake_post_json)) + + response = await client.submit_task( + "creo", + "open_model", + { + "model_path": r"C:\\models\\part.prt", + "task_params": {}, + "execution_id": "exec-1", + "batch_id": "batch-1", + "item_id": "item-1", + "callback_url": "http://localhost/callback", + "attempt": 0, + }, + ) + + assert response["status"] == "success" + assert captured["url"].endswith("/api/model/open") + assert captured["payload"]["filePath"].endswith("part.prt") + assert captured["payload"]["execution_id"] == "exec-1" + + +@pytest.mark.asyncio +async def test_plugin_http_client_uses_pdms_open_project_payload(monkeypatch): + client = PluginHttpClient() + captured = {} + + def fake_post_json(url, payload, timeout): + captured["url"] = url + captured["payload"] = payload + return {"status": "success"} + + monkeypatch.setattr(PluginHttpClient, "_post_json", staticmethod(fake_post_json)) + + response = await client.submit_task( + "pdms", + "open_project", + { + "model_path": "", + "task_params": { + "projectName": "SAM", + "username": "SYSTEM", + "password": "XXXXXX", + "systemName": "CATA", + }, + "execution_id": "exec-2", + "batch_id": "batch-2", + "item_id": "item-2", + "callback_url": "http://localhost/callback", + "attempt": 0, + }, + ) + + assert response["status"] == "success" + assert captured["url"].endswith("/api/project/open") + assert captured["payload"]["projectName"] == "SAM" + assert captured["payload"]["execution_id"] == "exec-2" diff --git a/tests/test_serial_batch_executor.py b/tests/test_serial_batch_executor.py index 9bed126..88f0a28 100644 --- a/tests/test_serial_batch_executor.py +++ b/tests/test_serial_batch_executor.py @@ -22,7 +22,7 @@ class FakePluginHttpClient: self._registry = callback_registry self.calls = [] - async def submit_task(self, software_id: str, payload: dict) -> dict: + async def submit_task(self, software_id: str, task_type: str, payload: dict) -> dict: self.calls.append(payload) behavior = payload.get("task_params", {}).get("behavior", "success") attempt = payload.get("attempt", 0)