From 77500b4708411329dc0f2802ef9f89ef1cccf31f Mon Sep 17 00:00:00 2001 From: sladro Date: Mon, 2 Mar 2026 11:43:59 +0800 Subject: [PATCH] feat: Implement batch processing for IFC to STP conversion and enhance plugin task handling --- app/api/v1/websocket.py | 121 +++++++++++++++++++++-- app/core/cad_batch_manager.py | 7 ++ app/core/plugin_http_client.py | 7 ++ app/core/serial_batch_executor.py | 10 ++ app/models/cad_batch.py | 4 +- cad-batch-plan.md | 18 +++- configs/software_config.yaml | 6 +- tests/test_creo_close_model_sync_flow.py | 77 +++++++++++++++ tests/test_plugin_http_client_mapping.py | 72 ++++++++++++++ tests/test_websocket_ifc_batch.py | 90 +++++++++++++++++ 命令.md | 2 +- 11 files changed, 398 insertions(+), 16 deletions(-) create mode 100644 tests/test_creo_close_model_sync_flow.py create mode 100644 tests/test_websocket_ifc_batch.py diff --git a/app/api/v1/websocket.py b/app/api/v1/websocket.py index a2ecd1b..057c69f 100644 --- a/app/api/v1/websocket.py +++ b/app/api/v1/websocket.py @@ -2,11 +2,13 @@ WebSocket API路由 提供WebSocket连接端点 """ +import asyncio +import json +import logging +import uuid + from fastapi import APIRouter, WebSocket, WebSocketDisconnect, Depends, Query from app.core.websocket_manager import websocket_manager, MessageType -import json -import uuid -import logging logger = logging.getLogger(__name__) @@ -38,6 +40,7 @@ class WSMessageType: SET_FILE_EXTENSIONS = "set_file_extensions" GET_FILE_CONFIG = "get_file_config" CONVERT_IFC_TO_STP = "convert_ifc_to_stp" + CONVERT_IFC_TO_STP_BATCH = "convert_ifc_to_stp_batch" SUBMIT_BATCH_TASKS = "submit_batch_tasks" GET_BATCH_STATUS = "get_batch_status" GET_BATCH_ITEM_STATUS = "get_batch_item_status" @@ -141,6 +144,8 @@ async def handle_client_message(message: dict, client_id: str, user_id: str): - get_file_config: 获取文件配置（基础路径和扩展名） - set_base_path: 设置CAD文件基础路径 - set_file_extensions: 设置文件扩展名配置 + - convert_ifc_to_stp: IFC 转 STP（单文件） + - convert_ifc_to_stp_batch: IFC 转 STP（批量） """ from app.core.software_manager import software_manager from app.core.log_manager import log_manager @@ -1010,7 +1015,7 @@ async def handle_client_message(message: dict, client_id: str, user_id: str): if not ifc_path or not stp_path: await websocket_manager.send_personal_message({ "type": MessageType.ERROR, - "message": "缂哄皯鍙傛暟: ifc_path 鎴? stp_path", + "message": "缺少参数: ifc_path 或 stp_path", "data": { "ifc_path": ifc_path, "stp_path": stp_path @@ -1022,17 +1027,18 @@ async def handle_client_message(message: dict, client_id: str, user_id: str): try: from app.core.ifc2stp_converter import Ifc2StpConverter - result = Ifc2StpConverter().convert(ifc_path=ifc_path, stp_path=stp_path) + converter = Ifc2StpConverter() + result = await asyncio.to_thread(converter.convert, ifc_path, stp_path) await websocket_manager.send_personal_message({ "type": MessageType.INFO, - "message": "IFC杞琒TP鎴愬姛", + "message": "IFC 转 STP 成功", "data": result, "timestamp": websocket_manager._get_timestamp() }, client_id) except Exception as e: await websocket_manager.send_personal_message({ "type": MessageType.ERROR, - "message": f"IFC杞琒TP澶辫触: {str(e)}", + "message": f"IFC 转 STP 失败: {str(e)}", "data": { "ifc_path": ifc_path, "stp_path": stp_path @@ -1040,6 +1046,107 @@ async def handle_client_message(message: dict, client_id: str, user_id: str): "timestamp": websocket_manager._get_timestamp() }, client_id) + elif message_type == WSMessageType.CONVERT_IFC_TO_STP_BATCH: + items = message.get("items") + continue_on_error = bool(message.get("continue_on_error", True)) + + if not isinstance(items, list) or len(items) == 0: + await websocket_manager.send_personal_message({ + "type": MessageType.ERROR, + "message": "缺少参数: items（非空数组）", + "data": { + "items": items + }, + "timestamp": websocket_manager._get_timestamp() + }, client_id) + return + + normalized_items = [] + invalid_items = [] + for index, item in enumerate(items): + if not isinstance(item, dict): + invalid_items.append({ + "index": index, + "error": "item 必须是对象", + "item": item + }) + continue + + ifc_path = item.get("ifc_path") + stp_path = item.get("stp_path") + if not ifc_path or not stp_path: + invalid_items.append({ + "index": index, + "error": "缺少 ifc_path 或 stp_path", + "item": item + }) + continue + + normalized_items.append({ + "index": index, + "ifc_path": ifc_path, + "stp_path": stp_path + }) + + if invalid_items: + await websocket_manager.send_personal_message({ + "type": MessageType.ERROR, + "message": "批量参数校验失败", + "data": { + "invalid_count": len(invalid_items), + "invalid_items": invalid_items + }, + "timestamp": websocket_manager._get_timestamp() + }, client_id) + return + + from app.core.ifc2stp_converter import Ifc2StpConverter + + converter = Ifc2StpConverter() + results = [] + success_count = 0 + failed_count = 0 + + for item in normalized_items: + ifc_path = item["ifc_path"] + stp_path = item["stp_path"] + try: + result = await asyncio.to_thread(converter.convert, ifc_path, stp_path) + success_count += 1 + results.append({ + "index": item["index"], + "status": "success", + "ifc_path": ifc_path, + "stp_path": stp_path, + "result": result + }) + except Exception as e: + failed_count += 1 + results.append({ + "index": item["index"], + "status": "failed", + "ifc_path": ifc_path, + "stp_path": stp_path, + "error": str(e) + }) + if not continue_on_error: + break + + response_type = MessageType.INFO if failed_count == 0 else MessageType.ERROR + await websocket_manager.send_personal_message({ + "type": response_type, + "message": "IFC 批量转 STP 完成" if failed_count == 0 else "IFC 批量转 STP 完成（部分失败）", + "data": { + "total_count": len(results), + "requested_count": len(normalized_items), + "success_count": success_count, + "failed_count": failed_count, + "continue_on_error": continue_on_error, + "results": results + }, + "timestamp": websocket_manager._get_timestamp() + }, client_id) + else: # 未知消息类型 await websocket_manager.send_personal_message({ diff --git a/app/core/cad_batch_manager.py b/app/core/cad_batch_manager.py index 49f90b4..3639db2 100644 --- a/app/core/cad_batch_manager.py +++ b/app/core/cad_batch_manager.py @@ -175,6 +175,13 @@ class CadBatchManager: except (TypeError, ValueError): return 0 + def get_task_completion_mode(self, software_id: str, task_type: str) -> str: + task_cfg = software_config.get_plugin_task_config(software_id, task_type) or {} + mode = task_cfg.get("completion_mode", "callback") + if mode not in {"callback", "sync"}: + return "callback" + return mode + def validate_callback_token(self, software_id: str, token: Optional[str]) -> bool: plugin_cfg = software_config.get_plugin_config(software_id) or {} expected = plugin_cfg.get("callback_token") diff --git a/app/core/plugin_http_client.py b/app/core/plugin_http_client.py index 1cd49a5..8c9d37a 100644 --- a/app/core/plugin_http_client.py +++ b/app/core/plugin_http_client.py @@ -61,6 +61,13 @@ class PluginHttpClient: return {**task_params, **callback} if body_mode == "project_open": return {**task_params, **callback} + if body_mode == "creo_close_model": + return { + "software_type": "creo", + "force_close": task_params.get("force_close") is True, + } + if body_mode == "empty": + return {} if body_mode == "passthrough": return payload diff --git a/app/core/serial_batch_executor.py b/app/core/serial_batch_executor.py index 5a59d37..bc1fc8d 100644 --- a/app/core/serial_batch_executor.py +++ b/app/core/serial_batch_executor.py @@ -82,6 +82,16 @@ class SerialBatchExecutor: item.task_type, payload, ) + completion_mode = self._batch_manager.get_task_completion_mode(item.software_id, item.task_type) + + if completion_mode == "sync": + success_flag = bool(plugin_response.get("success")) + if success_flag: + await self._batch_manager.mark_item_succeeded(item_id, plugin_response) + return + error_message = plugin_response.get("error") or "Plugin sync task failed" + raise PluginSubmitError(error_message) + 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/app/models/cad_batch.py b/app/models/cad_batch.py index 0d28ab0..abe982b 100644 --- a/app/models/cad_batch.py +++ b/app/models/cad_batch.py @@ -4,7 +4,7 @@ from datetime import datetime from enum import Enum from typing import Any, Dict, List, Optional -from pydantic import BaseModel, Field +from pydantic import BaseModel, ConfigDict, Field class BatchStatus(str, Enum): @@ -29,6 +29,7 @@ class PluginResultStatus(str, Enum): class BatchSubmitItem(BaseModel): + model_config = ConfigDict(protected_namespaces=()) model_path: str task_type: str task_params: Dict[str, Any] = Field(default_factory=dict) @@ -41,6 +42,7 @@ class BatchSubmitRequest(BaseModel): class BatchItem(BaseModel): + model_config = ConfigDict(protected_namespaces=()) id: str batch_id: str sequence: int diff --git a/cad-batch-plan.md b/cad-batch-plan.md index 20eba1b..a9c3726 100644 --- a/cad-batch-plan.md +++ b/cad-batch-plan.md @@ -72,13 +72,20 @@ - `plugins.{software_id}.callback_token` - `plugins.{software_id}.tasks.{task_type}.path` - `plugins.{software_id}.tasks.{task_type}.body_mode` +- `plugins.{software_id}.tasks.{task_type}.completion_mode` ### 4.2 body_mode 语义 - `file_path`：发送 `filePath + task_params + callback元信息` - `task_params_only`：发送 `task_params + callback元信息` - `project_open`：发送 `task_params + callback元信息` +- `creo_close_model`：发送 `{ software_type: "creo", force_close: boolean }` +- `empty`：发送空请求体 `{}` - `passthrough`：原样透传 +### 4.3 completion_mode 语义 +- `callback`：下发后进入 `waiting_callback`，等待插件回调判定结果 +- `sync`：以接口同步返回结果直接判定成功/失败，不等待回调 + --- ## 5. 当前开发进度（截至本次更新） @@ -98,13 +105,13 @@ - `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`, 占位) +- `close_model` -> `/api/model/close` (`creo_close_model`, `sync`, 已接入) #### 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`, 占位) +- `close_model` -> `/api/close` (`empty`, `sync`, 已接入) #### PDMS (localhost:9001) - `open_project` -> `/api/project/open` (`project_open`) @@ -114,12 +121,13 @@ - `close_project` -> `/api/project/close` (`task_params_only`, 占位) ### 5.3 当前测试状态 -- 已通过：`9 passed` +- 已通过：`13 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` + - `tests/test_creo_close_model_sync_flow.py` --- @@ -128,13 +136,13 @@ ### 6.1 你后续补充后可继续对接 1. 每个 `task_type` 的最终命名标准（是否沿用当前命名） 2. Revit 大任务轮询接口细节（如 `/api/task` 的请求/返回） -3. 关闭类接口是否真实可用（当前按占位映射） +3. PDMS `close_project` 是否真实可用（当前仍按占位映射） 4. 回调安全方案最终版（token 还是 signature，签名算法与字段） 5. 各 API 的完整错误码与失败语义（用于统一重试/终止策略） ### 6.2 下一步建议 - 基于真实插件联调一轮 `open -> 处理 -> close` -- 把占位关闭接口替换为确认可用的真实接口 +- 优先确认 PDMS `close_project` 实际行为与容错策略 - 增加端到端集成测试（含回调超时、插件离线、重复回调场景） --- diff --git a/configs/software_config.yaml b/configs/software_config.yaml index 490af86..d81130c 100644 --- a/configs/software_config.yaml +++ b/configs/software_config.yaml @@ -77,7 +77,8 @@ plugins: body_mode: "task_params_only" close_model: path: "/api/model/close" - body_mode: "task_params_only" + body_mode: "creo_close_model" + completion_mode: "sync" pdms: base_url: "http://localhost:9001" @@ -124,4 +125,5 @@ plugins: body_mode: "task_params_only" close_model: path: "/api/close" - body_mode: "task_params_only" + body_mode: "empty" + completion_mode: "sync" diff --git a/tests/test_creo_close_model_sync_flow.py b/tests/test_creo_close_model_sync_flow.py new file mode 100644 index 0000000..1252984 --- /dev/null +++ b/tests/test_creo_close_model_sync_flow.py @@ -0,0 +1,77 @@ +import asyncio +from copy import deepcopy + +import pytest + +from app.config import software_config +from app.core.cad_batch_manager import CadBatchManager +from app.core.cad_task_router import CadTaskRouter +from app.core.plugin_callback_registry import PluginCallbackRegistry +from app.models.cad_batch import BatchStatus, BatchSubmitItem, BatchSubmitRequest + + +class SyncClosePluginClient: + def __init__(self): + self.calls = [] + + async def submit_task(self, software_id: str, task_type: str, payload: dict) -> dict: + self.calls.append((software_id, task_type, payload)) + if task_type == "close_model": + return { + "success": True, + "data": { + "model_name": "abc.prt", + "was_modified": False, + "close_time": "2026-03-01T12:34:56Z", + }, + "error": None, + } + return {"success": False, "error": "unsupported task in this test"} + + +@pytest.mark.asyncio +async def test_creo_close_model_sync_completion(): + backup = deepcopy(software_config.load_config()) + try: + software_config._config["plugins"]["creo"]["tasks"]["close_model"]["completion_mode"] = "sync" + + manager = CadBatchManager( + task_router=CadTaskRouter({".prt": "creo"}), + plugin_client=SyncClosePluginClient(), + callback_registry=PluginCallbackRegistry(), + ) + await manager.start() + + req = BatchSubmitRequest( + items=[ + BatchSubmitItem( + model_path="x.prt", + task_type="close_model", + task_params={"force_close": True}, + ) + ] + ) + + try: + batch = await manager.create_batch(req, submitter_id="tester") + + start = asyncio.get_event_loop().time() + final_batch = None + while asyncio.get_event_loop().time() - start < 2: + final_batch = await manager.get_batch(batch.id) + if final_batch and final_batch.status in {BatchStatus.COMPLETED, BatchStatus.COMPLETED_WITH_ERRORS, BatchStatus.FAILED}: + break + await asyncio.sleep(0.02) + + assert final_batch is not None + assert final_batch.status == BatchStatus.COMPLETED + + items = await manager.get_batch_items(batch.id) + assert len(items) == 1 + assert items[0].status.value == "succeeded" + assert items[0].result.get("success") is True + assert items[0].result.get("data", {}).get("model_name") == "abc.prt" + finally: + await manager.stop() + finally: + software_config._config = backup diff --git a/tests/test_plugin_http_client_mapping.py b/tests/test_plugin_http_client_mapping.py index 34f9282..a1fa8e4 100644 --- a/tests/test_plugin_http_client_mapping.py +++ b/tests/test_plugin_http_client_mapping.py @@ -71,3 +71,75 @@ async def test_plugin_http_client_uses_pdms_open_project_payload(monkeypatch): assert captured["url"].endswith("/api/project/open") assert captured["payload"]["projectName"] == "SAM" assert captured["payload"]["execution_id"] == "exec-2" + + +@pytest.mark.asyncio +async def test_plugin_http_client_uses_creo_close_model_payload(monkeypatch): + client = PluginHttpClient() + captured = {} + + def fake_post_json(url, payload, timeout): + captured["url"] = url + captured["payload"] = payload + return {"success": True, "data": {"model_name": "x.prt"}, "error": None} + + monkeypatch.setattr(PluginHttpClient, "_post_json", staticmethod(fake_post_json)) + + response = await client.submit_task( + "creo", + "close_model", + { + "model_path": r"C:\\models\\part.prt", + "task_params": {"force_close": True, "software_type": "wrong_value_ignored"}, + "execution_id": "exec-close", + "batch_id": "batch-close", + "item_id": "item-close", + "callback_url": "http://localhost/callback", + "attempt": 0, + }, + ) + + assert response["success"] is True + assert captured["url"].endswith("/api/model/close") + assert captured["payload"] == {"software_type": "creo", "force_close": True} + + +@pytest.mark.asyncio +async def test_plugin_http_client_uses_revit_close_model_empty_payload(monkeypatch): + client = PluginHttpClient() + captured = {} + + def fake_post_json(url, payload, timeout): + captured["url"] = url + captured["payload"] = payload + return { + "success": True, + "code": 200, + "message": "文件关闭成功", + "data": { + "result": "文件关闭成功", + "fileName": "xxx.rvt", + "filePath": r"C:\\path\\xxx.rvt", + }, + "timestamp": "2026-03-01T08:00:00Z", + } + + monkeypatch.setattr(PluginHttpClient, "_post_json", staticmethod(fake_post_json)) + + response = await client.submit_task( + "revit", + "close_model", + { + "model_path": r"C:\\models\\a.rvt", + "task_params": {"ignored": True}, + "execution_id": "exec-revit-close", + "batch_id": "batch-revit-close", + "item_id": "item-revit-close", + "callback_url": "http://localhost/callback", + "attempt": 0, + }, + ) + + assert response["success"] is True + assert captured["url"].endswith("/api/close") + assert captured["payload"] == {} diff --git a/tests/test_websocket_ifc_batch.py b/tests/test_websocket_ifc_batch.py new file mode 100644 index 0000000..7c83a23 --- /dev/null +++ b/tests/test_websocket_ifc_batch.py @@ -0,0 +1,90 @@ +import sys +import types + +import pytest + +from app.api.v1.websocket import WSMessageType, handle_client_message +from app.core.websocket_manager import MessageType +from app.api.v1 import websocket as websocket_module + + +@pytest.mark.asyncio +async def test_convert_ifc_to_stp_batch_success(monkeypatch): + sent_messages = [] + convert_calls = [] + + async def fake_send(message, client_id): + sent_messages.append((message, client_id)) + + monkeypatch.setattr(websocket_module.websocket_manager, "send_personal_message", fake_send) + monkeypatch.setattr(websocket_module.websocket_manager, "_get_timestamp", lambda: "2026-03-01T00:00:00") + + class FakeConverter: + def convert(self, ifc_path: str, stp_path: str): + convert_calls.append((ifc_path, stp_path)) + return { + "ifc_path": ifc_path, + "stp_path": stp_path, + "duration_ms": 1, + "products_total": 10, + "products_success": 10, + "products_failed": 0, + } + + fake_converter_module = types.SimpleNamespace(Ifc2StpConverter=FakeConverter) + monkeypatch.setitem(sys.modules, "app.core.ifc2stp_converter", fake_converter_module) + + await handle_client_message( + { + "type": WSMessageType.CONVERT_IFC_TO_STP_BATCH, + "items": [ + {"ifc_path": "D:/cad/a.ifc", "stp_path": "D:/cad/a.stp"}, + {"ifc_path": "D:/cad/b.ifc", "stp_path": "D:/cad/b.stp"}, + ], + }, + client_id="test-client", + user_id="test-user", + ) + + assert convert_calls == [ + ("D:/cad/a.ifc", "D:/cad/a.stp"), + ("D:/cad/b.ifc", "D:/cad/b.stp"), + ] + assert len(sent_messages) == 1 + payload, target_client_id = sent_messages[0] + assert target_client_id == "test-client" + assert payload["type"] == MessageType.INFO + assert payload["data"]["requested_count"] == 2 + assert payload["data"]["success_count"] == 2 + assert payload["data"]["failed_count"] == 0 + assert [item["status"] for item in payload["data"]["results"]] == ["success", "success"] + + +@pytest.mark.asyncio +async def test_convert_ifc_to_stp_batch_invalid_items(monkeypatch): + sent_messages = [] + + async def fake_send(message, client_id): + sent_messages.append((message, client_id)) + + monkeypatch.setattr(websocket_module.websocket_manager, "send_personal_message", fake_send) + monkeypatch.setattr(websocket_module.websocket_manager, "_get_timestamp", lambda: "2026-03-01T00:00:00") + + await handle_client_message( + { + "type": WSMessageType.CONVERT_IFC_TO_STP_BATCH, + "items": [ + {"ifc_path": "D:/cad/a.ifc"}, + "invalid-item", + ], + }, + client_id="test-client", + user_id="test-user", + ) + + assert len(sent_messages) == 1 + payload, target_client_id = sent_messages[0] + assert target_client_id == "test-client" + assert payload["type"] == MessageType.ERROR + assert payload["message"] == "批量参数校验失败" + assert payload["data"]["invalid_count"] == 2 diff --git a/命令.md b/命令.md index f8dc06b..be62f59 100644 --- a/命令.md +++ b/命令.md @@ -1,6 +1,6 @@ ### 环境 - conda activate websocket311 -备注：要先执行conda deactivate和deactivate推出虚虚拟环境 +备注：要先执行conda deactivate和deactivate退出虚虚拟环境 ### 打包 - python scripts\build_exe.py --clean