Doni/CadHubManage
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Implement batch processing for IFC to STP conversion and enhance plugin task handling

Browse Source
This commit is contained in:
sladro 2026-03-02 11:43:59 +08:00
parent 0a805c5acc
commit 77500b4708
11 changed files with 398 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

121
app/api/v1/websocket.py
View File

@ -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({

7
app/core/cad_batch_manager.py
View File

@ -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")

7
app/core/plugin_http_client.py
View File

@ -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

10
app/core/serial_batch_executor.py
View File

@ -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)

4
app/models/cad_batch.py
View File

@ -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

18
cad-batch-plan.md
View File

@ -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` 实际行为与容错策略
- 增加端到端集成测试（含回调超时、插件离线、重复回调场景）
---

6
configs/software_config.yaml
View File

@ -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"

77
tests/test_creo_close_model_sync_flow.py Normal file
View File

@ -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

72
tests/test_plugin_http_client_mapping.py
View File

@ -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"] == {}

90
tests/test_websocket_ifc_batch.py Normal file
View File

@ -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

2
命令.md
View File

@ -1,6 +1,6 @@
### 环境
- conda activate websocket311
备注：要先执行conda deactivate和deactivate推出虚虚拟环境
备注：要先执行conda deactivate和deactivate退出虚虚拟环境
### 打包
- python scripts\build_exe.py --clean