"""核心计算模块的惰性装载器(原 advanced_router._get_cached_import,D1 拆分时上提共用)。 - 惰性导入:避免路由模块级加载核心包(含 OCC 重模块)的导入开销与循环依赖 - 装载失败返回 None 且不缓存失败(与原实现一致,端点统一 503「服务不可用」) - 实例缓存:设计/加工模块为纯 Python 计算(构造后无 self 突变,方法仅读入参), 可安全地被 asyncio.to_thread 并发调用;OCC 相关的 side_action_designer 经 processing_service.run_occ 的常驻 OCC 进程池使用(方案 B,见 docs/topics/performance/OCC_THROUGHPUT.md) """ import threading from typing import Optional from shared.utils.logger import get_logger logger = get_logger(__name__) _lock = threading.Lock() _instances: dict = {} _LOADERS = { "side_action_designer": ("moldinsight.core.side_action_designer", "SideActionDesigner"), "cavity_layout_optimizer": ("moldinsight.core.cavity_layout_optimizer", "CavityLayoutOptimizer"), "mold_system_designer": ("moldinsight.core.mold_system_designer", "MoldSystemDesigner"), "mold_cam_designer": ("moldinsight.core.mold_cam", "MoldCAMDesigner"), "collision_detector": ("moldinsight.core.mold_machining", "CollisionDetector"), "toolpath_optimizer": ("moldinsight.core.mold_machining", "ToolpathOptimizer"), "edm_designer": ("moldinsight.core.mold_machining", "EDMElectrodeDesigner"), "machining_simulator": ("moldinsight.core.mold_machining", "MachiningSimulator"), } def get_core_module(key: str): if key in _instances: return _instances[key] if key not in _LOADERS: return None with _lock: if key in _instances: return _instances[key] module_path, class_name = _LOADERS[key] try: module = __import__(module_path, fromlist=[class_name]) instance = getattr(module, class_name)() except Exception as e: logger.warning(f"核心模块 {key} 加载失败: {e}") return None _instances[key] = instance return instance