feat: Enhance plugin task submission with dynamic task configuration and payload handling

2026-03-01 17:50:20 +08:00 · 2026-03-01 17:50:20 +08:00 · 0a805c5acc
commit 0a805c5acc
parent b19ef1467a
7 changed files with 287 additions and 85 deletions
--- 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)
--- 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")
--- 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)
--- 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 查询批次与条目状态
--- 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"
--- a/tests/test_plugin_http_client_mapping.py
+++ 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"
--- 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)