导出step文件合并为整体，方便后续修改

src/api/routes.py

@@ -7,22 +7,14 @@ import os
 from pathlib import Path
 
 from models.schemas import ProcessingStatus, create_task_info
-from core.stp_parser import STPParser
-from core.geometry_analyzer import GeometryAnalyzer
 from utils.file_handler import FileHandler
-from utils.html_generator import HTMLGenerator
 from services.storage_integration_rustfs import StorageIntegrationService
 from services.redis_task_manager import redis_task_manager
 from services.processing_service import processing_service
-from services.llm_service import llm_service
 from services.task_query_service import TaskQueryService
 from database.database import get_db_session
 from utils.logger import get_logger
 from sqlalchemy.ext.asyncio import AsyncSession
-from core.mold_generator import MoldCavityGenerator
-from core.aluminum_foam_mold import AluminumFoamMoldGenerator
-from core.mold_quality_inspector import AluminumFoamMoldQualityInspector
-from core.mesh_generator import MeshGenerator
 from core.cavity_layout_optimizer import CavityLayoutOptimizer
 from core.mold_system_designer import MoldSystemDesigner
 from core.side_action_designer import SideActionDesigner
@@ -36,16 +28,7 @@ logger = get_logger(__name__)
 
 router = APIRouter()
 
-stp_parser = STPParser()
-geometry_analyzer = GeometryAnalyzer()
 file_handler = FileHandler()
-html_generator = HTMLGenerator()
-mold_generator = MoldCavityGenerator(shrinkage_rate=0.005)
-# 铝泡沫模具生成器
-aluminum_foam_generator = AluminumFoamMoldGenerator(shrinkage_rate=0.015, draft_angle=3.0)
-# 铝泡沫模具质量检测器
-mold_quality_inspector = AluminumFoamMoldQualityInspector()
-mesh_generator = MeshGenerator(quality="medium")
 cavity_layout_optimizer = CavityLayoutOptimizer()
 mold_system_designer = MoldSystemDesigner()
 side_action_designer = SideActionDesigner()
@@ -56,8 +39,6 @@ edm_designer = EDMElectrodeDesigner()
 machining_simulator = MachiningSimulator()
 cad_exporter = CADExporter()
 
-tasks = {}
-
 
 @router.get("/health")
 @router.post("/health")
@@ -120,15 +101,15 @@ async def upload_stp(
         upload_time=str(datetime.now())
     )
     await redis_task_manager.set_task(task_id, task_info)
-    tasks[task_id] = task_info
 
     # 后台处理走统一编排服务，避免请求会话在后台失效
+    process_params = {"material": material}
     background_tasks.add_task(
         processing_service.process_file_with_storage,
         task_id,
         file_path,
         stp_file.id,
-        material,
+        process_params,
     )
     logger.info(f"[UPLOAD] 后台处理已调度: task_id={task_id}")
 
@@ -158,8 +139,6 @@ async def get_status(task_id: str, db_session: AsyncSession = Depends(get_db_ses
     try:
         task_view = await TaskQueryService.get_task_view(db_session, task_id)
         if task_view is None:
-            if task_id in tasks:
-                return tasks[task_id]
             raise HTTPException(404, "任务不存在")
         return task_view
     except HTTPException:
@@ -178,7 +157,6 @@ async def debug_tasks():
         "total_tasks": len(all_tasks),
         "tasks": all_tasks,
         "redis_connected": redis_task_manager.is_connected,
-        "memory_fallback_tasks": len(tasks),
     }
 
 
@@ -259,48 +237,6 @@ async def result_page(request: Request, task_id: str, db_session: AsyncSession =
     })
 
 
-async def process_file_with_storage(
-        task_id: str,
-        file_path: str,
-        stp_file_id: int,
-        db_session: AsyncSession,
-        material: str = "ABS"
-):
-    """处理文件的后台任务"""
-
-    storage_service = StorageIntegrationService()
-
-    try:
-        logger.info(f"开始处理文件并生成模具型腔: {file_path}")
-        
-        # 设置处理超时（5分钟）
-        import asyncio
-        timeout_seconds = 300  # 5分钟
-        
-        async def process_with_timeout():
-            # 处理逻辑将在下面添加
-            pass
-            
-        # 使用超时保护
-        try:
-            await asyncio.wait_for(process_file_core(storage_service, task_id, file_path, stp_file_id, db_session, material), timeout_seconds)
-        except asyncio.TimeoutError:
-            logger.error(f"处理超时: {task_id}")
-            raise Exception(f"处理超时，超过{timeout_seconds}秒未完成")
-            
-    except Exception as e:
-        logger.error(f"模具型腔生成失败: {e}")
-
-        await storage_service.update_stp_file_status(db_session, stp_file_id, "failed")
-        await storage_service.update_task_status(
-            db_session, task_id, "failed", error_message=str(e)
-        )
-
-        tasks[task_id]["status"] = ProcessingStatus.FAILED
-        tasks[task_id]["error"] = str(e)
-        tasks[task_id]["completed_at"] = str(datetime.now())
-
-
 # ==================== P3 新增 API ====================
 
 @router.post("/optimize-layout")
@@ -417,10 +353,13 @@ async def ai_parting_surface_detect(
     body = await request.json()
     task_id = body.get("task_id")
 
-    if not task_id or task_id not in tasks:
+    if not task_id:
+        raise HTTPException(404, "缺少 task_id")
+
+    task_data = await redis_task_manager.get_task(task_id)
+    if not task_data:
         raise HTTPException(404, "任务不存在")
 
-    task_data = tasks[task_id]
     geometry_data = task_data.get("geometry_data")
     if not geometry_data:
         raise HTTPException(400, "该任务尚未完成几何分析")
@@ -444,10 +383,13 @@ async def detect_undercuts(
     parting_direction = body.get("parting_direction", [0, 0, 1])
     mold_size = body.get("mold_size", {"length": 300, "width": 300, "height": 200})
 
-    if not task_id or task_id not in tasks:
+    if not task_id:
+        raise HTTPException(404, "缺少 task_id")
+
+    task_data = await redis_task_manager.get_task(task_id)
+    if not task_data:
         raise HTTPException(404, "任务不存在")
 
-    task_data = tasks[task_id]
     geometry_data = task_data.get("geometry_data")
     if not geometry_data:
         raise HTTPException(400, "该任务尚未完成几何分析")
@@ -570,16 +512,29 @@ async def export_mold_results(
     task_id = body.get("task_id")
     formats = body.get("formats", ["step", "stl"])
     components = body.get("components", ["cavity", "core"])
+    scheme_id = body.get("scheme_id")
 
-    if not task_id or task_id not in tasks:
+    if not task_id:
-        raise HTTPException(404, "任务不存在")
+        raise HTTPException(404, "缺少 task_id")
 
-    task_data = tasks[task_id]
+    cavity_shapes = processing_service.get_export_shapes(task_id, scheme_id=scheme_id)
-    cavity_shapes = task_data.get("cavity_shapes")
     if not cavity_shapes:
-        raise HTTPException(400, "该任务尚未完成模具生成或形状数据不可用")
+        task_data = await redis_task_manager.get_task(task_id)
+        if not task_data:
+            raise HTTPException(404, "任务不存在")
+        filename = task_data.get("filename", f"mold_{task_id}")
+        raise HTTPException(
+            400,
+            f"该任务的 OCC 形状数据已过期（仅保留 STEP 导出文件），"
+            f"请通过历史页面的下载链接获取已导出的 STEP 文件",
+        )
 
-    base_filename = Path(task_data.get("filename", f"mold_{task_id}")).stem
+    task_data = await redis_task_manager.get_task(task_id)
+    base_filename = (
+        Path(task_data.get("filename", f"mold_{task_id}")).stem
+        if task_data
+        else f"mold_{task_id}"
+    )
 
     result = cad_exporter.export_mold_results(
         cavity_data=cavity_shapes,
@@ -636,531 +591,4 @@ async def get_export_recommendations(
     return {"status": "success", "data": result}
 
 
-async def _save_analysis_metrics(session, stp_file_id, analysis_result):
-    """保存分析指标到数据库"""
-    from models.database import AnalysisMetrics
 
-    quality_metrics = analysis_result.get("quality_metrics", {})
-    analysis_summary = analysis_result.get("analysis_summary", "")
-    
-    # 创建分析指标记录
-    metrics = AnalysisMetrics(
-        stp_file_id=stp_file_id,
-        volume_utilization=quality_metrics.get("volume_utilization", 0),
-        topology_complexity=quality_metrics.get("topology_complexity", 0),
-        wall_uniformity=quality_metrics.get("wall_uniformity", 0),
-        analysis_summary=analysis_summary
-    )
-    
-    session.add(metrics)
-    await session.commit()
-    logger.info(f"分析指标保存成功: {metrics.id}")
-
-
-async def _save_verification_metrics(session, stp_file_id, verification_result):
-    """保存验证指标到数据库"""
-    from models.database import AnalysisMetrics
-    from sqlalchemy import select
-    
-    # 查找现有的分析指标记录
-    result = await session.execute(
-        select(AnalysisMetrics).where(AnalysisMetrics.stp_file_id == stp_file_id)
-    )
-    metrics = result.scalar_one_or_none()
-    
-    if metrics:
-        # 更新现有记录
-        metrics.verification_status = verification_result.get("status", "unknown")
-        comparison = verification_result.get("comparison", {})
-        volume_comparison = comparison.get("volume", {})
-        area_comparison = comparison.get("surface_area", {})
-        
-        metrics.verification_volume_diff = volume_comparison.get("difference_percent", 0)
-        metrics.verification_area_diff = area_comparison.get("difference_percent", 0)
-        metrics.verification_details = verification_result
-    else:
-        # 创建新记录
-        comparison = verification_result.get("comparison", {})
-        volume_comparison = comparison.get("volume", {})
-        area_comparison = comparison.get("surface_area", {})
-        
-        metrics = AnalysisMetrics(
-            stp_file_id=stp_file_id,
-            verification_status=verification_result.get("status", "unknown"),
-            verification_volume_diff=volume_comparison.get("difference_percent", 0),
-            verification_area_diff=area_comparison.get("difference_percent", 0),
-            verification_details=verification_result
-        )
-        session.add(metrics)
-    
-    await session.commit()
-    logger.info(f"验证指标保存成功: stp_file_id={stp_file_id}")
-
-
-async def process_file_core(
-        storage_service: StorageIntegrationService,
-        task_id: str,
-        file_path: str,
-        stp_file_id: int,
-        db_session: AsyncSession,
-        material: str = "ABS"
-):
-    """核心处理逻辑"""
-    
-    try:
-        logger.info(f"开始处理文件并生成模具型腔: {file_path}")
-
-        # 更新任务状态
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 20, "解析STP文件"
-        )
-
-        # 1. 解析STP文件
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 20, "解析STP文件"
-        )
-        
-        # 使用STPParser类进行真实解析
-        shape = stp_parser.load_step_file(Path(file_path))
-        geometry_data = stp_parser.analyze_geometry(shape)
-
-        # 2. 生成网格数据并持久化（详细 JSON 存 RustFS，摘要写 PostgreSQL）
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 30, "生成网格数据"
-        )
-
-        mesh_result = None
-        mesh_json = None
-        try:
-            mesh_result = mesh_generator.generate_mesh_from_shape(shape)
-
-            # mesh_generator 返回: vertices, faces, points, normals, point_count, vertex_count, face_count
-            vertices = mesh_result.get("vertices", [])
-            faces = mesh_result.get("faces", [])
-            points = mesh_result.get("points", [])
-            normals = mesh_result.get("normals", [])
-            point_count = mesh_result.get("point_count", 0)
-            vertex_count = mesh_result.get("vertex_count", 0)
-            face_count = mesh_result.get("face_count", 0)
-
-            if vertices and faces:
-                # 使用已计算的几何边界框，避免重复计算
-                bbox = geometry_data.get("bounding_box", {})
-
-                mesh_json = {
-                    "metadata": {
-                        "file_name": Path(file_path).name,
-                        "generated_at": datetime.now().isoformat(),
-                        "quality": "medium",
-                        "vertex_count": vertex_count,
-                        "face_count": face_count,
-                        "point_count": point_count,
-                    },
-                    "mesh": {
-                        "vertices": vertices,
-                        "faces": faces,
-                    },
-                    "pointcloud": {
-                        "points": points,
-                        "normals": normals,
-                        "count": point_count
-                    },
-                    "bounding_box": bbox,
-                }
-
-                await storage_service.save_mesh_data(
-                    db_session,
-                    stp_file_id=stp_file_id,
-                    mesh_json=mesh_json,
-                    quality="medium",
-                )
-                # 将简要网格摘要写入内存任务，便于前端展示汇总信息
-                tasks[task_id]["mesh_summary"] = {
-                    "vertex_count": vertex_count,
-                    "face_count": face_count,
-                    "point_count": point_count,
-                    "quality": "medium",
-                }
-        except Exception as mesh_err:
-            # 网格失败不影响整体流程，只记录日志
-            logger.warning(f"网格生成或保存失败，不影响主流程: {mesh_err}")
-
-        # 3. 生成模具型腔（使用真实的 MoldCavityGenerator）
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 40, "生成模具型腔"
-        )
-
-        # 材料属性（需在型腔生成前定义）
-        material_properties = {
-            "ABS": {"density": 1.05, "shrinkage": 0.005, "name": "ABS"},
-            "PP": {"density": 0.90, "shrinkage": 0.016, "name": "PP"},
-            "PE": {"density": 0.95, "shrinkage": 0.020, "name": "PE"},
-            "PC": {"density": 1.20, "shrinkage": 0.007, "name": "PC"},
-            "PA": {"density": 1.14, "shrinkage": 0.010, "name": "PA"},
-            "POM": {"density": 1.41, "shrinkage": 0.020, "name": "POM"},
-            "PMMA": {"density": 1.18, "shrinkage": 0.005, "name": "PMMA"},
-            "PBT": {"density": 1.31, "shrinkage": 0.015, "name": "PBT"},
-            "AlSi10Mg": {"density": 0.45, "shrinkage": 0.015, "name": "AlSi10Mg", "is_foam": True},
-            "AlSi12": {"density": 0.50, "shrinkage": 0.012, "name": "AlSi12", "is_foam": True},
-            "Pure Al Foam": {"density": 0.35, "shrinkage": 0.020, "name": "Pure Al Foam", "is_foam": True},
-            "AlSi7Mg": {"density": 0.40, "shrinkage": 0.018, "name": "AlSi7Mg", "is_foam": True},
-        }
-        
-        requested_material = material if material in material_properties else "ABS"
-        selected_material = material_properties.get(requested_material, material_properties["ABS"])
-        is_foam_material = selected_material.get("is_foam", False)
-
-        # 使用 MoldCavityGenerator 生成型腔数据
-        cavity_mesh_data = None
-        try:
-            if shape:
-                if is_foam_material:
-                    aluminum_foam_generator.set_material(selected_material["name"])
-                    cavity_result = aluminum_foam_generator.generate_mold_cavities(shape)
-                    cavity_mesh_data = aluminum_foam_generator.generate_detailed_cavity_json(cavity_result)
-                    logger.info(f"使用铝泡沫模具生成器: {selected_material['name']}")
-                else:
-                    cavity_result = mold_generator.generate_mold_cavities(shape)
-                    cavity_mesh_data = mold_generator.generate_detailed_cavity_json(cavity_result)
-                    logger.info(f"使用普通塑料模具生成器: {selected_material['name']}")
-                
-                if cavity_mesh_data:
-                    logger.info(f"型腔网格数据生成完成: {cavity_mesh_data.get('mold_cavities', {}).get('cavity', {}).get('vertex_count', 0)} 顶点")
-        except Exception as cavity_err:
-            logger.warning(f"型腔生成失败，使用简化数据: {cavity_err}")
-            import traceback
-            traceback.print_exc()
-            cavity_mesh_data = None
-
-        # 4. 生成详细JSON数据（使用计算值）
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 60, "生成型腔详细数据"
-        )
-
-        # 使用模具生成器计算各项参数
-        volume_mm3 = geometry_data.get("volume", 0)
-        surface_area_mm2 = geometry_data.get("surface_area", 0)
-        bbox = geometry_data.get("bounding_box", {})
-        bbox_dims = bbox.get("dimensions", [0, 0, 0])
-        
-        material_density = selected_material["density"]
-        shrinkage_rate = selected_material["shrinkage"]
-
-        # 计算产品重量
-        volume_cm3 = volume_mm3 / 1000
-        product_weight_g = volume_cm3 * material_density
-
-        # 计算投影面积（取X、Y方向）
-        if len(bbox_dims) >= 2:
-            projected_area_cm2 = (bbox_dims[0] * bbox_dims[1]) / 100
-        else:
-            projected_area_cm2 = 0
-
-        # 计算最优型腔数量
-        # 基于产品重量和投影面积计算
-        # 小产品（< 100g）可以多型腔，大产品（> 1000g）通常单型腔
-        if product_weight_g < 50:
-            cavity_count = 8  # 小产品，多型腔
-        elif product_weight_g < 100:
-            cavity_count = 4  # 中小产品
-        elif product_weight_g < 300:
-            cavity_count = 2  # 中等产品
-        elif product_weight_g < 1000:
-            cavity_count = 1  # 较大产品
-        else:
-            cavity_count = 1  # 大产品，单型腔
-        
-        # 根据投影面积调整型腔数量
-        # 如果单型腔投影面积超过 400 cm²，减少型腔数量
-        single_cavity_area = projected_area_cm2
-        if single_cavity_area > 400:
-            cavity_count = 1
-        elif single_cavity_area > 200 and cavity_count > 2:
-            cavity_count = 2
-
-        # 计算总投影面积（包括流道系统）
-        # 流道系统约占型腔投影面积的 15-25%
-        runner_ratio = 0.20
-        total_projected_area = single_cavity_area * cavity_count * (1 + runner_ratio)
-
-        # 计算夹紧力（总投影面积 × 注塑压力 / 1000 吨）
-        # 注塑压力根据材料选择：ABS 约 600-800 kg/cm²
-        injection_pressure = 700  # kg/cm²
-        clamping_force_ton = int(total_projected_area * injection_pressure / 1000)
-        
-        # 确保夹紧力在合理范围内
-        clamping_force_ton = max(50, min(clamping_force_ton, 3000))
-
-        # 计算壁厚范围
-        if surface_area_mm2 > 0 and volume_mm3 > 0:
-            avg_thickness_mm = (volume_mm3 / surface_area_mm2) * 0.6
-            wall_thickness_min = avg_thickness_mm * 0.7
-            wall_thickness_max = avg_thickness_mm * 1.3
-        else:
-            avg_thickness_mm = 2.5
-            wall_thickness_min = 2.0
-            wall_thickness_max = 3.0
-
-        # 计算复杂度评分
-        if surface_area_mm2 > 0 and volume_mm3 > 0:
-            complexity_score = min((avg_thickness_mm / 5.0), 1.0)
-        else:
-            complexity_score = 0.5
-
-        # 计算模具尺寸（基于型腔布局）
-        # 单型腔：产品尺寸 + 边距
-        # 多型腔：需要考虑型腔排列
-        cavity_spacing = 30  # 型腔间距 mm
-        edge_margin = 50  # 边缘余量 mm
-        
-        if cavity_count == 1:
-            mold_length = max(bbox_dims[0] if len(bbox_dims) > 0 else 120, 120) + 2 * edge_margin
-            mold_width = max(bbox_dims[1] if len(bbox_dims) > 1 else 100, 100) + 2 * edge_margin
-        elif cavity_count == 2:
-            # 2型腔：并排排列
-            mold_length = 2 * max(bbox_dims[0] if len(bbox_dims) > 0 else 120, 120) + cavity_spacing + 2 * edge_margin
-            mold_width = max(bbox_dims[1] if len(bbox_dims) > 1 else 100, 100) + 2 * edge_margin
-        elif cavity_count == 4:
-            # 4型腔：2x2 排列
-            mold_length = 2 * max(bbox_dims[0] if len(bbox_dims) > 0 else 120, 120) + cavity_spacing + 2 * edge_margin
-            mold_width = 2 * max(bbox_dims[1] if len(bbox_dims) > 1 else 100, 100) + cavity_spacing + 2 * edge_margin
-        else:
-            # 8型腔：2x4 排列
-            mold_length = 4 * max(bbox_dims[0] if len(bbox_dims) > 0 else 120, 120) + 3 * cavity_spacing + 2 * edge_margin
-            mold_width = 2 * max(bbox_dims[1] if len(bbox_dims) > 1 else 100, 100) + cavity_spacing + 2 * edge_margin
-        
-        mold_height = max(bbox_dims[2] if len(bbox_dims) > 2 else 60, 60) + 80  # 包含冷却系统
-
-        # 计算分型线长度（基于型腔布局）
-        if len(bbox_dims) >= 2:
-            single_parting_line = 2 * (bbox_dims[0] + bbox_dims[1])
-            parting_line_length = single_parting_line * cavity_count
-        else:
-            parting_line_length = 0
-
-        # 估算成型周期（基于体积和壁厚）
-        # 周期 = 冷却时间 + 注塑时间 + 开合模时间
-        cooling_time = (wall_thickness_max ** 2) * 4  # 冷却时间与壁厚平方成正比
-        injection_time = max(3, volume_cm3 / 100)  # 注塑时间
-        ejection_time = 3  # 顶出时间
-        cycle_time = cooling_time + injection_time + ejection_time + 5  # 开合模约5秒
-        
-        # 根据型腔数量调整周期（多型腔需要更长冷却时间）
-        if cavity_count > 1:
-            cycle_time = cycle_time * (1 + 0.1 * (cavity_count - 1))
-
-        detailed_cavity_json = {
-            "metadata": {
-                "file_name": Path(file_path).name,
-                "analysis_date": datetime.now().isoformat(),
-                "shrinkage_rate": shrinkage_rate,
-                "draft_angle": 2.0,
-                "selected_material": selected_material["name"]
-            },
-            "product_analysis": {
-                "volume": volume_mm3,
-                "surface_area": surface_area_mm2,
-                "bounding_box": bbox
-            },
-            "manufacturing_info": {
-                "recommended_material": selected_material["name"],
-                "material_density": f"{material_density} g/cm³",
-                "estimated_clamping_force": f"{clamping_force_ton} 吨",
-                "estimated_mold_size": {
-                    "length": int(mold_length),
-                    "width": int(mold_width),
-                    "height": int(mold_height)
-                },
-                "mold_material": "铝合金7075" if clamping_force_ton < 200 else "P20钢材",
-                "mold_hardness": "HB 150-170" if clamping_force_ton < 200 else "HRC 28-32",
-                "surface_finish": "Ra 0.8 μm",
-                "parting_line_length": f"{parting_line_length:.2f} mm",
-                "estimated_cycle_time": f"{int(cycle_time)} 秒",
-                "injection_pressure": f"{injection_pressure} kg/cm²"
-            },
-            "mold_cavities": {
-                "cavity_count": cavity_count,
-            }
-        }
-        
-        # 合并型腔网格数据（如果有）
-        if cavity_mesh_data and "mold_cavities" in cavity_mesh_data:
-            mold_cavities = cavity_mesh_data["mold_cavities"]
-            if "cavity" in mold_cavities:
-                detailed_cavity_json["mold_cavities"]["cavity"] = mold_cavities["cavity"]
-            if "core" in mold_cavities:
-                detailed_cavity_json["mold_cavities"]["core"] = mold_cavities["core"]
-            if "parting_surface" in mold_cavities:
-                detailed_cavity_json["mold_cavities"]["parting_surface"] = mold_cavities["parting_surface"]
-            logger.info(f"型腔网格数据已合并: cavity {mold_cavities.get('cavity', {}).get('vertex_count', 0)} 顶点")
-        
-        # 添加型腔关键信息
-        detailed_cavity_json["mold_cavities"]["cavity_key_info"] = {
-            "geometric_characteristics": {
-                "product_weight": f"{product_weight_g:.2f} g",
-                "wall_thickness_range": f"{wall_thickness_min:.2f} - {wall_thickness_max:.2f} mm",
-                "complexity_score": round(complexity_score, 2),
-                "product_volume": f"{volume_cm3:.2f} cm³",
-                "projected_area": f"{projected_area_cm2:.2f} cm²"
-            },
-            "quality_considerations": {
-                "potential_weld_lines": "center" if cavity_count > 1 else "minimal",
-                "sink_mark_areas": "thick_sections" if wall_thickness_max > 4 else "minimal",
-                "warpage_risk": "medium" if wall_thickness_max > 5 else "low"
-            }
-        }
-
-        # 5. 生成关键信息（模拟）
-        cavity_key_info = detailed_cavity_json["mold_cavities"]["cavity_key_info"]
-
-        # 6. 保存几何数据到数据库
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 70, "保存几何数据"
-        )
-
-        geometry_record = await storage_service.save_geometry_data(
-            db_session,
-            stp_file_id,
-            geometry_data,
-            geometry_data.get("analysis_method", "mold_cavity")
-        )
-
-        # 7. 保存模具型腔数据
-        await storage_service.save_mold_cavity_data(
-            db_session,
-            stp_file_id,
-            detailed_cavity_json
-        )
-
-        # 8. 生成HTML可视化（包含型腔信息和点云数据）
-        await storage_service.update_task_status(
-            db_session, task_id, "processing", 85, "生成可视化报告"
-        )
-
-        # 获取点云数据 - mesh_result 直接返回 points 和 normals
-        pointcloud_data = None
-        if mesh_result:
-            pointcloud_data = {
-                "points": mesh_result.get("points", []),
-                "normals": mesh_result.get("normals", []),
-                "point_count": mesh_result.get("point_count", 0)
-            }
-
-        html_file_path = html_generator.generate_and_save_visualization(
-            geometry_data,
-            Path(file_path).name,
-            cavity_data=detailed_cavity_json,
-            pointcloud_data=pointcloud_data
-        )
-
-        # 保存HTML文件信息
-        html_record = await storage_service.save_html_file(
-            db_session,
-            stp_file_id,
-            Path(html_file_path).name,
-            html_file_path
-        )
-
-        # 9. 分析模具设计
-        analysis_result = geometry_analyzer.analyze_mold_design(geometry_data, shape=shape)
-
-        # 9.5 保存完整的分析结果到数据库
-        if analysis_result:
-            await storage_service.save_features_and_recommendations(
-                db_session,
-                stp_file_id,
-                analysis_result.get("detected_features", []),
-                analysis_result.get("design_recommendations", [])
-            )
-            
-            # 保存质量指标和分析摘要到数据库
-            await _save_analysis_metrics(db_session, stp_file_id, analysis_result)
-
-        # 9.6 更新STP文件的分析摘要字段（用于快速查询）
-        await storage_service.update_stp_file_analysis_summary(
-            db_session,
-            stp_file_id,
-            volume=volume_mm3,
-            surface_area=surface_area_mm2,
-            product_weight=product_weight_g
-        )
-
-        # 9.7 FreeCAD 几何验证（可通过配置禁用）
-        verification_result = None
-        from config.settings import settings
-        
-        if settings.ENABLE_FREECAD_VERIFICATION:
-            await storage_service.update_task_status(
-                db_session, task_id, "processing", 90, "FreeCAD几何验证"
-            )
-            
-            try:
-                from services.verification_service import GeometryVerificationService
-                # 使用配置的超时时间
-                verification_svc = GeometryVerificationService(timeout=settings.FREECAD_VERIFICATION_TIMEOUT)
-                verification_result = await verification_svc.verify_stp_file(file_path)
-                
-                # 保存验证结果到数据库
-                if verification_result and analysis_result:
-                    await _save_verification_metrics(
-                        db_session, 
-                        stp_file_id, 
-                        verification_result
-                    )
-                
-                logger.info(f"FreeCAD验证完成: {verification_result.get('status', 'unknown') if verification_result else 'failed'}")
-            except Exception as ve:
-                logger.warning(f"FreeCAD验证失败（不影响主流程）: {ve}")
-                verification_result = {"status": "error", "error": str(ve)}
-        else:
-            logger.info("FreeCAD验证已禁用（设置 ENABLE_FREECAD_VERIFICATION=true 启用）")
-            verification_result = {"status": "disabled", "reason": "FreeCAD验证已禁用"}
-
-        # 9.8 LLM 增强分析
-        llm_report = None
-        if analysis_result:
-            side_action_ai = await llm_service.generate_side_action_analysis(
-                analysis_result, detailed_cavity_json
-            )
-            design_report = await llm_service.generate_design_report(
-                analysis_result, detailed_cavity_json
-            )
-            llm_report = llm_service.compose_llm_report(
-                design_report, side_action_ai
-            )
-
-        # 10. 完成处理
-        await storage_service.update_stp_file_status(db_session, stp_file_id, "completed")
-        await storage_service.update_task_status(
-            db_session, task_id, "completed", 100, "模具型腔生成完成"
-        )
-
-        # 更新内存任务状态
-        tasks[task_id]["geometry_data"] = geometry_data
-        tasks[task_id]["analysis_result"] = analysis_result
-        tasks[task_id]["cavity_data"] = detailed_cavity_json
-        tasks[task_id]["cavity_shapes"] = cavity_result
-        tasks[task_id]["key_info"] = detailed_cavity_json  # 传递完整数据给前端
-        tasks[task_id]["html_file"] = f"/html/{Path(html_file_path).name}"  # 只使用文件名
-        tasks[task_id]["verification"] = verification_result  # 添加验证结果
-        tasks[task_id]["llm_report"] = llm_report
-        tasks[task_id]["status"] = ProcessingStatus.COMPLETED
-        tasks[task_id]["completed_at"] = str(datetime.now())
-
-        # 调试日志
-        logger.info(f"模具型腔生成完成: {task_id}")
-        logger.info(f"key_info metadata: {detailed_cavity_json.get('metadata', {})}")
-        logger.info(f"key_info manufacturing_info: {detailed_cavity_json.get('manufacturing_info', {})}")
-        logger.info(f"key_info geometric_characteristics: {detailed_cavity_json.get('mold_cavities', {}).get('cavity_key_info', {}).get('geometric_characteristics', {})}")
-
-    except Exception as e:
-        logger.error(f"模具型腔生成失败: {e}")
-
-        await storage_service.update_stp_file_status(db_session, stp_file_id, "failed")
-        await storage_service.update_task_status(
-            db_session, task_id, "failed", error_message=str(e)
-        )
-
-        tasks[task_id]["status"] = ProcessingStatus.FAILED
-        tasks[task_id]["error"] = str(e)
-        tasks[task_id]["completed_at"] = str(datetime.now())

src/core/ai_mold_assistant.py

@@ -6,6 +6,7 @@ AI 分模辅助模型接口示例
 """
 from typing import Dict, Any, Optional
 import numpy as np
+from OCC.Core.TopoDS import TopoDS_Shape, TopoDS_Face
 
 
 class AIPartingSurfaceDetector:
@@ -40,7 +41,7 @@ class AIPartingSurfaceDetector:
         # self.model = torch.load(model_path)
         print(f"AI 模型加载：{model_path}")
     
-    def detect(self, product_shape: Any, analysis: Dict) -> Optional[Dict]:
+    def detect(self, product_shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
         """
         检测最优分型面
         
@@ -76,7 +77,7 @@ class AIPartingSurfaceDetector:
             "undercut_regions": []  # 倒扣区域
         }
     
-    def _preprocess_shape(self, shape: Any, analysis: Dict) -> Any:
+    def _preprocess_shape(self, shape: TopoDS_Shape, analysis: Dict) -> TopoDS_Shape:
         """
         预处理产品形状为 AI 模型输入
         
@@ -110,7 +111,7 @@ class AIDraftAnalyzer:
         """加载训练好的 AI 模型"""
         print(f"AI 拔模分析模型加载：{model_path}")
     
-    def analyze(self, product_shape: Any, parting_surface: Any, 
+    def analyze(self, product_shape: TopoDS_Shape, parting_surface: TopoDS_Face, 
                 base_draft_angle: float) -> Optional[Dict]:
         """
         分析拔模需求
@@ -156,7 +157,7 @@ class AICavityLayoutOptimizer:
         if model_path:
             self._load_model(model_path)
     
-    def optimize(self, product_shape: Any, cavity_count: int, 
+    def optimize(self, product_shape: TopoDS_Shape, cavity_count: int, 
                  mold_base_size: Dict) -> Optional[Dict]:
         """
         优化型腔布局

src/core/ai_parting_detector.py

@@ -22,6 +22,7 @@ GNN 模型：
 
 from typing import Dict, List, Any, Optional, Tuple
 import numpy as np
+from OCC.Core.TopoDS import TopoDS_Shape
 from utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -47,7 +48,7 @@ except ImportError:
 class ShapeGraphBuilder:
     """将 OCC 形状转换为图表示"""
 
-    def build_graph(self, shape: Any) -> Optional[Dict]:
+    def build_graph(self, shape: TopoDS_Shape) -> Optional[Dict]:
         """
         从 OCC 形状构建图数据
 
@@ -383,7 +384,7 @@ class AIPartingSurfaceDetectorV2:
             logger.error(f"GNN 模型加载失败: {e}")
             self.model = None
 
-    def detect(self, product_shape: Any, analysis: Dict) -> Optional[Dict]:
+    def detect(self, product_shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
         """
         检测最优分型面
 
@@ -407,7 +408,7 @@ class AIPartingSurfaceDetectorV2:
 
         return self._detect_with_geometry(product_shape, analysis)
 
-    def _detect_with_gnn(self, shape: Any, analysis: Dict) -> Optional[Dict]:
+    def _detect_with_gnn(self, shape: TopoDS_Shape, analysis: Dict) -> Optional[Dict]:
         """使用 GNN 模型检测分型面"""
         if not _TORCH_GEOMETRIC_AVAILABLE:
             return None
@@ -457,7 +458,7 @@ class AIPartingSurfaceDetectorV2:
             logger.warning(f"GNN 检测失败，回退到几何方法: {e}")
             return None
 
-    def _detect_with_geometry(self, shape: Any, analysis: Dict) -> Dict:
+    def _detect_with_geometry(self, shape: TopoDS_Shape, analysis: Dict) -> Dict:
         """几何方法回退：基于法向量统计的分型面检测"""
         try:
             graph_data = self.graph_builder.build_graph(shape)
@@ -504,7 +505,7 @@ class AIPartingSurfaceDetectorV2:
             "method": "fallback",
         }
 
-    def collect_training_sample(self, shape: Any, analysis: Dict,
+    def collect_training_sample(self, shape: TopoDS_Shape, analysis: Dict,
                                 ground_truth_normal: List[float],
                                 ground_truth_origin: List[float]) -> Optional[Dict]:
         """

src/core/aluminum_foam_mold.py

@@ -16,7 +16,7 @@ import numpy as np
 from OCC.Core.BRepBuilderAPI import BRepBuilderAPI_MakeFace
 from OCC.Core.BRepPrimAPI import BRepPrimAPI_MakeBox
 from OCC.Core.gp import gp_Pln, gp_Dir, gp_Pnt
-from OCC.Core.TopoDS import TopoDS_Face, topods
+from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Shape, topods
 from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
 from OCC.Core.TopExp import TopExp_Explorer
 from OCC.Core.TopAbs import TopAbs_FACE
@@ -128,7 +128,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
         else:
             logger.warning(f"未知材料 {material}")
 
-    def generate_mold_cavities(self, product_shape: Any) -> Dict[str, Any]:
+    def generate_mold_cavities(self, product_shape: TopoDS_Shape) -> Dict[str, Any]:
         """
         从产品的3D模型生成型腔和型芯
 
@@ -292,13 +292,13 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
 
     # ==================== 核心算法实现 ====================
 
-    def _analyze_product_geometry(self, shape: Any) -> Dict[str, Any]:
+    def _analyze_product_geometry(self, shape: TopoDS_Shape) -> Dict[str, Any]:
         """分析产品几何属性（扩展基类版本，增加法向量统计）"""
         result = super()._analyze_product_geometry(shape)
         result["normal_statistics"] = self._analyze_parting_direction(shape)
         return result
 
-    def _analyze_parting_direction(self, shape: Any) -> Dict[str, float]:
+    def _analyze_parting_direction(self, shape: TopoDS_Shape) -> Dict[str, float]:
         """分析产品法向量分布，按面积加权统计各轴方向强度"""
         stats = {"X": 0.0, "Y": 0.0, "Z": 0.0}
         explorer = TopExp_Explorer(shape, TopAbs_FACE)
@@ -328,11 +328,11 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
             for axis, value in stats.items()
         }
 
-    def _split_cavity_core(self, shape: Any, parting_surface: Any) -> Tuple[Any, Any]:
+    def _split_cavity_core(self, shape: TopoDS_Shape, parting_surface: TopoDS_Face) -> Tuple[TopoDS_Shape, TopoDS_Shape]:
         """分离型腔和型芯（铝泡沫使用更大余量）"""
         return super()._split_cavity_core(shape, parting_surface, margin=25)
 
-    def _detect_parting_surfaces(self, shape: Any, analysis: Dict) -> Dict[str, Any]:
+    def _detect_parting_surfaces(self, shape: TopoDS_Shape, analysis: Dict) -> Dict[str, Any]:
         """
         检测分型面（泡沫模具专用）
 
@@ -470,7 +470,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
 
         except Exception:
             return 50.0
-    def _generate_mold_block(self, cavity: Any, analysis: Dict) -> Any:
+    def _generate_mold_block(self, cavity: TopoDS_Shape, analysis: Dict) -> TopoDS_Shape:
         """生成完整的模具块（包含A/B板结构）"""
         try:
             bbox = analysis["bounding_box"]
@@ -493,7 +493,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
             logger.error(f"模具块生成失败: {e}")
             return cavity
 
-    def _extract_parting_surface_geometry(self, surface: Any) -> Dict[str, Any]:
+    def _extract_parting_surface_geometry(self, surface: TopoDS_Face) -> Dict[str, Any]:
         """提取分型面几何数据"""
         metadata = self._extract_plane_metadata(surface)
         return {
@@ -504,7 +504,7 @@ class AluminumFoamMoldGenerator(BaseMoldGenerator):
         }
 
     def _create_parting_surface_from_ai(self, ai_result: Dict, analysis: Dict,
-                                        shape: Any = None) -> Dict:
+                                        shape: Optional[TopoDS_Shape] = None) -> Dict:
         """从 AI 结果创建分型面"""
         origin = ai_result.get("origin", [0, 0, 0])
         normal = ai_result.get("normal", [0, 0, 1])

src/core/base_mold_generator.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Any, Tuple, Optional
+from typing import Dict, List, Any, Tuple, Optional, TYPE_CHECKING
 import math
 import numpy as np
 from OCC.Core.BRepOffsetAPI import BRepOffsetAPI_DraftAngle
@@ -6,7 +6,7 @@ from OCC.Core.BRepAlgoAPI import BRepAlgoAPI_Cut, BRepAlgoAPI_Section, BRepAlgoA
 from OCC.Core.BRepBuilderAPI import BRepBuilderAPI_MakeFace, BRepBuilderAPI_Transform
 from OCC.Core.BRepPrimAPI import BRepPrimAPI_MakeBox, BRepPrimAPI_MakeHalfSpace
 from OCC.Core.gp import gp_Pln, gp_Dir, gp_Pnt, gp_Trsf, gp_Ax2
-from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Compound, topods
+from OCC.Core.TopoDS import TopoDS_Shape, TopoDS_Face, TopoDS_Compound, topods
 from OCC.Core.BRep import BRep_Tool, BRep_Builder
 from OCC.Core.BRepMesh import BRepMesh_IncrementalMesh
 from OCC.Core.GProp import GProp_GProps
@@ -41,7 +41,7 @@ class BaseMoldGenerator:
         self.ai_draft_analyzer = draft_analyzer
         logger.info("AI 模型接口已设置")
 
-    def _apply_shrinkage_compensation(self, shape: Any) -> Any:
+    def _apply_shrinkage_compensation(self, shape: TopoDS_Shape) -> TopoDS_Shape:
         scale_factor = 1.0 + self.shrinkage_rate
         trsf = gp_Trsf()
         trsf.SetScale(gp_Pnt(0, 0, 0), scale_factor)
@@ -53,7 +53,7 @@ class BaseMoldGenerator:
             logger.warning(f"收缩率补偿失败: {e}")
             return shape
 
-    def _apply_draft_angles(self, shape: Any, parting_surface: Any) -> Any:
+    def _apply_draft_angles(self, shape: TopoDS_Shape, parting_surface: TopoDS_Face) -> TopoDS_Shape:
         try:
             draft_direction = self._get_draft_direction(parting_surface)
             if draft_direction is None:
@@ -76,7 +76,7 @@ class BaseMoldGenerator:
             logger.warning(f"拔模角处理失败，返回原始形状: {e}")
             return shape
 
-    def _get_draft_direction(self, parting_surface: Any) -> Optional[gp_Dir]:
+    def _get_draft_direction(self, parting_surface: TopoDS_Face) -> Optional[gp_Dir]:
         try:
             surface = BRepAdaptor_Surface(parting_surface)
             if surface.GetType() == 0:
@@ -85,7 +85,7 @@ class BaseMoldGenerator:
         except Exception:
             return gp_Dir(0, 0, 1)
 
-    def _find_draftable_faces(self, shape: Any, draft_direction: gp_Dir) -> List[Any]:
+    def _find_draftable_faces(self, shape: TopoDS_Shape, draft_direction: gp_Dir) -> List[TopoDS_Face]:
         draftable = []
         explorer = TopExp_Explorer(shape, TopAbs_FACE)
 
@@ -103,7 +103,7 @@ class BaseMoldGenerator:
 
         return draftable
 
-    def _get_face_normal(self, face: Any) -> Optional[gp_Dir]:
+    def _get_face_normal(self, face: TopoDS_Face) -> Optional[gp_Dir]:
         try:
             surface = BRepAdaptor_Surface(face)
             u = (surface.FirstUParameter() + surface.LastUParameter()) / 2
@@ -122,8 +122,8 @@ class BaseMoldGenerator:
         except Exception:
             return None
 
-    def _execute_draft(self, shape: Any, faces: List[Any],
+    def _execute_draft(self, shape: TopoDS_Shape, faces: List[TopoDS_Face],
-                       draft_direction: gp_Dir, draft_angle_rad: float) -> Any:
+                       draft_direction: gp_Dir, draft_angle_rad: float) -> TopoDS_Shape:
         try:
             draft = BRepOffsetAPI_DraftAngle(shape)
 
@@ -156,8 +156,8 @@ class BaseMoldGenerator:
             logger.warning(f"拔模执行失败: {e}")
             return shape
 
-    def _draft_faces_sequentially(self, shape: Any, faces: List[Any],
+    def _draft_faces_sequentially(self, shape: TopoDS_Shape, faces: List[TopoDS_Face],
-                                   draft_direction: gp_Dir, draft_angle_rad: float) -> Any:
+                                   draft_direction: gp_Dir, draft_angle_rad: float) -> TopoDS_Shape:
         current_shape = shape
         success_count = 0
 
@@ -190,7 +190,7 @@ class BaseMoldGenerator:
 
         return current_shape
 
-    def _analyze_product_geometry(self, shape: Any) -> Dict[str, Any]:
+    def _analyze_product_geometry(self, shape: TopoDS_Shape) -> Dict[str, Any]:
         try:
             props = GProp_GProps()
             brepgprop.VolumeProperties(shape, props)
@@ -224,7 +224,7 @@ class BaseMoldGenerator:
             logger.error(f"产品几何分析失败: {e}")
             raise
 
-    def _split_cavity_core(self, shape: Any, parting_surface: Any, margin: int = 20) -> Tuple[Any, Any]:
+    def _split_cavity_core(self, shape: TopoDS_Shape, parting_surface: TopoDS_Face, margin: int = 20) -> Tuple[TopoDS_Shape, TopoDS_Shape]:
         """
         分离型腔和型芯 — 完全嵌入 + 突出贴合方式。
 
@@ -274,7 +274,7 @@ class BaseMoldGenerator:
             return self._split_cavity_core_fallback(shape, None)
 
     @staticmethod
-    def _extract_parting_normal(parting_surface: Any) -> List[float]:
+    def _extract_parting_normal(parting_surface: TopoDS_Face) -> List[float]:
         """从分型面提取法向量"""
         try:
             surface = BRepAdaptor_Surface(parting_surface)
@@ -288,14 +288,14 @@ class BaseMoldGenerator:
 
     def _build_core_with_base(
         self,
-        shape: Any,
+        shape: TopoDS_Shape,
-        mold_block: Any,
+        mold_block: TopoDS_Shape,
         parting_plane: gp_Pln,
         mold_xmin: float, mold_ymin: float, mold_zmin: float,
         mold_xmax: float, mold_ymax: float, mold_zmax: float,
         xmin: float, ymin: float, zmin: float,
         xmax: float, ymax: float, zmax: float,
-    ) -> Any:
+    ) -> TopoDS_Shape:
         """
         构建带底座的型芯。
 
@@ -353,7 +353,7 @@ class BaseMoldGenerator:
         return self._build_core_compound(base_plate, shape)
 
     @staticmethod
-    def _build_core_compound(base_plate: Any, shape: Any) -> Any:
+    def _build_core_compound(base_plate: TopoDS_Shape, shape: TopoDS_Shape) -> TopoDS_Shape:
         """
         兜底方案：构建 TopoDS_Compound 包含底座平板 + 产品。
         即使布尔融合失败，底座也绝不会丢失。
@@ -366,7 +366,7 @@ class BaseMoldGenerator:
         logger.info("型芯 Compound 兜底构建 (底座+产品)")
         return compound
 
-    def _get_parting_plane(self, parting_surface: Any, shape: Any) -> Optional[gp_Pln]:
+    def _get_parting_plane(self, parting_surface: TopoDS_Face, shape: TopoDS_Shape) -> Optional[gp_Pln]:
         """从分型面提取平面方程"""
         try:
             surface = BRepAdaptor_Surface(parting_surface)
@@ -383,8 +383,8 @@ class BaseMoldGenerator:
             logger.warning(f"分型面平面提取失败: {e}")
             return None
 
-    def _split_mold_block_by_plane(self, mold_block: Any,
+    def _split_mold_block_by_plane(self, mold_block: TopoDS_Shape,
-                                    parting_plane: gp_Pln) -> Tuple[Any, Any]:
+                                    parting_plane: gp_Pln) -> Tuple[TopoDS_Shape, TopoDS_Shape]:
         """
         用分型面将模具块切分为A板（上模）和B板（下模）
 
@@ -439,8 +439,8 @@ class BaseMoldGenerator:
             logger.error(f"A/B板分离失败: {e}")
             return None, None
 
-    def _subtract_product_from_plate(self, plate: Any, product: Any,
+    def _subtract_product_from_plate(self, plate: TopoDS_Shape, product: TopoDS_Shape,
-                                      plate_name: str) -> Any:
+                                      plate_name: str) -> TopoDS_Shape:
         """从模板中减去产品形状，生成型腔或型芯"""
         try:
             cut_op = BRepAlgoAPI_Cut(plate, product)
@@ -455,8 +455,8 @@ class BaseMoldGenerator:
             logger.warning(f"{plate_name}减产品失败: {e}")
             return plate
 
-    def _split_cavity_core_fallback(self, shape: Any,
+    def _split_cavity_core_fallback(self, shape: TopoDS_Shape,
-                                     mold_block: Optional[Any] = None) -> Tuple[Any, Any]:
+                                     mold_block: Optional[TopoDS_Shape] = None) -> Tuple[TopoDS_Shape, TopoDS_Shape]:
         """
         分模回退方案：完全嵌入 + 突出贴合，用 Z 中心面做分型基准。
         """
@@ -507,7 +507,7 @@ class BaseMoldGenerator:
             except Exception:
                 return shape, shape
 
-    def detect_insert_regions(self, shape: Any, analysis: Dict,
+    def detect_insert_regions(self, shape: TopoDS_Shape, analysis: Dict,
                                depth_threshold: float = 30.0,
                                aspect_threshold: float = 3.0) -> List[Dict[str, Any]]:
         """
@@ -618,7 +618,7 @@ class BaseMoldGenerator:
 
         return inserts
 
-    def _extract_shape_geometry(self, shape: Any, shape_type: str) -> Dict[str, Any]:
+    def _extract_shape_geometry(self, shape: TopoDS_Shape, shape_type: str) -> Dict[str, Any]:
         try:
             mesh = BRepMesh_IncrementalMesh(shape, 0.1)
             mesh.Perform()
@@ -678,7 +678,7 @@ class BaseMoldGenerator:
                 "face_count": 0,
             }
 
-    def _extract_plane_metadata(self, surface: Any) -> Dict[str, Any]:
+    def _extract_plane_metadata(self, surface: TopoDS_Shape) -> Dict[str, Any]:
         """从分型面提取平面元数据（法向量、原点、边界）"""
         metadata = {
             "normal": [0.0, 0.0, 1.0],
@@ -743,7 +743,7 @@ class BaseMoldGenerator:
 
         return total_length
 
-    def _calculate_parting_line(self, shape: Any, parting_surface: Any) -> List[List[float]]:
+    def _calculate_parting_line(self, shape: TopoDS_Shape, parting_surface: TopoDS_Face) -> List[List[float]]:
         try:
             section = BRepAlgoAPI_Section(shape, parting_surface)
             section.Build()
@@ -783,7 +783,7 @@ class BaseMoldGenerator:
             logger.error(f"分型线计算失败: {e}")
             return self._simple_parting_line(shape)
 
-    def _simple_parting_line(self, shape: Any) -> List[List[float]]:
+    def _simple_parting_line(self, shape: TopoDS_Shape) -> List[List[float]]:
         try:
             bbox = Bnd_Box()
             brepbndlib.Add(shape, bbox)
@@ -800,8 +800,8 @@ class BaseMoldGenerator:
         except Exception:
             return [[-50, -50, 0], [50, -50, 0], [50, 50, 0], [-50, 50, 0], [-50, -50, 0]]
 
-    def extend_parting_surface(self, parting_surface: Any, shape: Any,
+    def extend_parting_surface(self, parting_surface: TopoDS_Face, shape: TopoDS_Shape,
-                                extension: float = 30.0) -> Any:
+                                extension: float = 30.0) -> TopoDS_Face:
         """
         将分型面延伸到模具块边界
 

src/core/cad_exporter.py

@@ -29,6 +29,7 @@ import os
 import re
 from typing import Dict, List, Any, Optional, Tuple
 from pathlib import Path
+from OCC.Core.TopoDS import TopoDS_Shape
 from utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -70,7 +71,7 @@ class CADExporter:
             relative = Path(os.path.basename(filepath))
         return relative.as_posix()
 
-    def export_step(self, shape: Any, filepath: str,
+    def export_step(self, shape: TopoDS_Shape, filepath: str,
                      schema: str = "AP214") -> bool:
         """
         导出 STEP 文件
@@ -118,7 +119,7 @@ class CADExporter:
             logger.error(f"STEP 导出失败: {e}")
             return False
 
-    def export_iges(self, shape: Any, filepath: str) -> bool:
+    def export_iges(self, shape: TopoDS_Shape, filepath: str) -> bool:
         """
         导出 IGES 文件
 
@@ -156,7 +157,7 @@ class CADExporter:
             logger.error(f"IGES 导出失败: {e}")
             return False
 
-    def export_stl(self, shape: Any, filepath: str,
+    def export_stl(self, shape: TopoDS_Shape, filepath: str,
                     ascii_mode: bool = True,
                     deflection: float = 0.1) -> bool:
         """
@@ -201,7 +202,7 @@ class CADExporter:
             logger.error(f"STL 导出失败: {e}")
             return False
 
-    def export_brep(self, shape: Any, filepath: str) -> bool:
+    def export_brep(self, shape: TopoDS_Shape, filepath: str) -> bool:
         """
         导出 BRep 文件（OpenCASCADE 原生格式）
 
@@ -240,18 +241,6 @@ class CADExporter:
                              components: List[str] = None,
                              task_id: Optional[str] = None,
                              scheme_id: Optional[str] = None) -> Dict[str, Any]:
-        """
-        批量导出模具设计结果
-
-        Args:
-            cavity_data: generate_mold_cavities() 的返回结果
-            base_filename: 基础文件名（不含扩展名）
-            formats: 导出格式列表 ["step", "iges", "stl", "brep"]
-            components: 导出组件列表 ["cavity", "core", "parting_surface", "all"]
-
-        Returns:
-            导出结果摘要
-        """
         if formats is None:
             formats = ["step", "stl"]
         if components is None:
@@ -279,7 +268,8 @@ class CADExporter:
             "parting_surface": ("parting_surface", "分型面"),
         }
 
-        shapes_to_export = []
+        shapes_to_export: List[Tuple[str, str, TopoDS_Shape]] = []
+        assembly_shapes: List[Tuple[TopoDS_Shape, str]] = []
 
         for comp in components:
             if comp == "all":
@@ -287,23 +277,45 @@ class CADExporter:
                     shape = cavity_data.get(data_key)
                     if shape is not None:
                         shapes_to_export.append((key, label, shape))
+                        assembly_shapes.append((shape, label))
                 break
             elif comp in shape_map:
                 data_key, label = shape_map[comp]
                 shape = cavity_data.get(data_key)
                 if shape is not None:
                     shapes_to_export.append((comp, label, shape))
+                    assembly_shapes.append((shape, label))
                 else:
                     results["errors"].append(f"{label}形状不可用")
 
+        # STEP: 所有组件合并为一个装配体文件
+        if "step" in formats and assembly_shapes:
+            filepath = os.path.join(export_dir, f"{base_filename}_mold.step")
+            success = self.export_assembly_step(assembly_shapes, filepath)
+            if success:
+                file_size = os.path.getsize(filepath)
+                relative_path = self.get_relative_path(filepath)
+                results["files"].append({
+                    "component": "assembly",
+                    "component_label": "模具装配体",
+                    "format": "step",
+                    "filepath": filepath,
+                    "relative_path": relative_path,
+                    "filename": os.path.basename(filepath),
+                    "size_bytes": file_size,
+                    "size_readable": self._format_file_size(file_size),
+                })
+            else:
+                results["errors"].append("装配体 STEP 导出失败")
+
+        # IGES / STL / BRep: 逐组件导出
+        non_assembly_formats = [f for f in formats if f != "step"]
         for comp_name, label, shape in shapes_to_export:
-            for fmt in formats:
+            for fmt in non_assembly_formats:
                 filepath = os.path.join(export_dir, f"{base_filename}_{comp_name}.{fmt}")
 
                 success = False
-                if fmt == "step":
+                if fmt == "iges":
-                    success = self.export_step(shape, filepath)
-                elif fmt == "iges":
                     success = self.export_iges(shape, filepath)
                 elif fmt == "stl":
                     success = self.export_stl(shape, filepath)
@@ -337,7 +349,7 @@ class CADExporter:
 
         return results
 
-    def export_assembly_step(self, shapes_with_names: List[Tuple[Any, str]],
+    def export_assembly_step(self, shapes_with_names: List[Tuple[TopoDS_Shape, str]],
                               filepath: str,
                               schema: str = "AP214") -> bool:
         """

src/core/geometry_analyzer.py

@@ -1,6 +1,7 @@
 from typing import Dict, List, Any, Optional
 import math
 import numpy as np
+from OCC.Core.TopoDS import TopoDS_Shape
 from models.schemas import (
     create_mold_feature,
     create_design_recommendation,
@@ -38,7 +39,7 @@ class GeometryAnalyzer:
     def analyze_mold_design(self, geometry_data: Dict[str, Any],
                             product_material: str = "ABS",
                             mold_material: str = "Aluminum",
-                            shape: Any = None
+                            shape: Optional[TopoDS_Shape] = None
                             ) -> Dict[str, Any]:
         """分析模具设计
 
@@ -72,7 +73,7 @@ class GeometryAnalyzer:
         )
 
     def _detect_features(self, geometry_data: Dict[str, Any],
-                         shape: Any = None) -> List[Dict[str, Any]]:
+                         shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
         """检测模具特征"""
         features = []
 
@@ -99,7 +100,7 @@ class GeometryAnalyzer:
         return features
 
     def _detect_wall_features(self, geometry_data: Dict[str, Any],
-                              shape: Any = None) -> List[Dict[str, Any]]:
+                              shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
         """检测壁厚特征"""
         features = []
 
@@ -230,7 +231,7 @@ class GeometryAnalyzer:
 
         return features
 
-    def _compute_precise_wall_thickness(self, shape: Any) -> Optional[Dict[str, Any]]:
+    def _compute_precise_wall_thickness(self, shape: TopoDS_Shape) -> Optional[Dict[str, Any]]:
         """使用 BRepExtrema_DistShapeShape 精确计算壁厚"""
         try:
             from OCC.Core.TopExp import TopExp_Explorer
@@ -319,7 +320,7 @@ class GeometryAnalyzer:
             return None
 
     def _detect_rib_features(self, geometry_data: Dict[str, Any],
-                             shape: Any = None) -> List[Dict[str, Any]]:
+                             shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
         """检测加强筋特征"""
         features = []
         topology = geometry_data.get("topology", {})
@@ -350,7 +351,7 @@ class GeometryAnalyzer:
         return features
 
     def _detect_boss_features(self, geometry_data: Dict[str, Any],
-                              shape: Any = None) -> List[Dict[str, Any]]:
+                              shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
         """检测BOSS柱特征"""
         features = []
         volume = geometry_data.get("volume", 0)
@@ -382,7 +383,7 @@ class GeometryAnalyzer:
         return features
 
     def _analyze_draft_angles(self, geometry_data: Dict[str, Any],
-                              shape: Any = None) -> List[Dict[str, Any]]:
+                              shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]:
         """分析拔模角度"""
         features = []
 
@@ -450,7 +451,7 @@ class GeometryAnalyzer:
 
         return features
 
-    def _compute_draft_angles_from_shape(self, shape: Any) -> Optional[Dict[str, Any]]:
+    def _compute_draft_angles_from_shape(self, shape: TopoDS_Shape) -> Optional[Dict[str, Any]]:
         """基于面法向量分析计算各面的拔模角度"""
         try:
             from OCC.Core.TopExp import TopExp_Explorer
@@ -516,7 +517,7 @@ class GeometryAnalyzer:
             logger.warning(f"拔模角度计算失败: {e}")
             return None
 
-    def _detect_curvature_features(self, shape: Any) -> List[Dict[str, Any]]:
+    def _detect_curvature_features(self, shape: TopoDS_Shape) -> List[Dict[str, Any]]:
         """检测高曲率区域（可能导致应力集中）"""
         features = []
         try:
@@ -604,7 +605,7 @@ class GeometryAnalyzer:
 
         return features
 
-    def _detect_fillet_features(self, shape: Any) -> List[Dict[str, Any]]:
+    def _detect_fillet_features(self, shape: TopoDS_Shape) -> List[Dict[str, Any]]:
         """检测圆角/倒角特征"""
         features = []
         try:

src/core/mesh_generator.py

@@ -8,6 +8,7 @@ from OCC.Core.BRepMesh import BRepMesh_IncrementalMesh
 from OCC.Core.TopExp import TopExp_Explorer
 from OCC.Core.TopAbs import TopAbs_FACE
 from OCC.Core.BRep import BRep_Tool
+from OCC.Core.TopoDS import TopoDS_Shape
 from OCC.Core.TopLoc import TopLoc_Location
 
 logger = logging.getLogger(__name__)
@@ -24,7 +25,7 @@ class MeshGenerator:
         }
         self.quality = self.quality_settings.get(quality, 0.3)
 
-    def generate_mesh_from_shape(self, shape, num_points: int = 20000) -> Dict:
+    def generate_mesh_from_shape(self, shape: TopoDS_Shape, num_points: int = 20000) -> Dict:
         """从PythonOCC形状生成点云数据"""
         try:
             mesh = BRepMesh_IncrementalMesh(shape, self.quality, False, 0.5, True)
@@ -121,7 +122,7 @@ class MeshGenerator:
             logger.error(traceback.format_exc())
             return self._create_sample_pointcloud()
 
-    def generate_multi_lod_mesh(self, shape) -> Dict:
+    def generate_multi_lod_mesh(self, shape: TopoDS_Shape) -> Dict:
         """生成多级LOD网格 - 一次OCC剖分，trimesh简化，避免重复计算
 
         返回结构:

src/core/mold_generator.py

@@ -2,7 +2,7 @@ from typing import Dict, List, Any, Tuple, Optional
 import numpy as np
 from OCC.Core.BRepBuilderAPI import BRepBuilderAPI_MakeFace
 from OCC.Core.gp import gp_Pln, gp_Dir, gp_Pnt
-from OCC.Core.TopoDS import TopoDS_Face, topods
+from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Shape, topods
 from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
 from OCC.Core.TopExp import TopExp_Explorer
 from OCC.Core.TopAbs import TopAbs_FACE
@@ -47,7 +47,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         else:
             logger.warning(f"未知材料 {material}, 使用默认密度 {self.material_density} g/cm³")
 
-    def generate_mold_cavities(self, product_shape: Any) -> Dict[str, Any]:
+    def generate_mold_cavities(self, product_shape: TopoDS_Shape) -> Dict[str, Any]:
         """
         从产品的3D模型生成型腔和型芯
 
@@ -196,7 +196,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
 
     # ==================== 内部方法 ====================
 
-    def _detect_parting_surface(self, shape: Any, analysis: Dict) -> Tuple[Any, List]:
+    def _detect_parting_surface(self, shape: TopoDS_Shape, analysis: Dict) -> Tuple[TopoDS_Face, List]:
         """
         检测分型面和分型线
 
@@ -219,7 +219,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         logger.info("使用简化方法检测分型面")
         return self._simple_parting_surface(shape, analysis)
 
-    def _detect_primary_parting(self, shape: Any, analysis: Dict) -> Dict[str, Any]:
+    def _detect_primary_parting(self, shape: TopoDS_Shape, analysis: Dict) -> Dict[str, Any]:
         """检测主分型面（AI优先 → 几何法向量 → 简化回退）"""
         if self.ai_parting_detector is not None:
             try:
@@ -283,7 +283,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         logger.info(f"转换得到 {len(regions)} 个兼容倒扣区域")
         return regions
 
-    def _analyze_face_normals(self, shape: Any) -> gp_Dir:
+    def _analyze_face_normals(self, shape: TopoDS_Shape) -> gp_Dir:
         """
         分析产品表面的法向量分布，找出最优分型方向
 
@@ -326,7 +326,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         else:
             return gp_Dir(0, 0, 1)
 
-    def _create_optimal_parting_plane(self, shape: Any, analysis: Dict,
+    def _create_optimal_parting_plane(self, shape: TopoDS_Shape, analysis: Dict,
                                       direction: gp_Dir) -> gp_Pln:
         """
         创建最优分型面
@@ -352,7 +352,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
 
         return parting_plane
 
-    def _simple_parting_surface(self, shape: Any, analysis: Dict) -> Tuple[Any, List]:
+    def _simple_parting_surface(self, shape: TopoDS_Shape, analysis: Dict) -> Tuple[TopoDS_Face, List]:
         """简化的分型面检测（回退方案）"""
         bbox = analysis["bounding_box"]
         center_z = bbox["center"][2]
@@ -372,7 +372,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         return parting_surface, parting_line
 
     def _create_parting_surface_from_ai(self, ai_result: Dict,
-                                        analysis: Dict, shape: Any = None) -> Tuple[Any, List]:
+                                        analysis: Dict, shape: Optional[TopoDS_Shape] = None) -> Tuple[TopoDS_Face, List]:
         """
         从 AI 模型结果创建分型面（预留接口）
 
@@ -405,7 +405,7 @@ class MoldCavityGenerator(BaseMoldGenerator):
         logger.info(f"从 AI 结果创建分型面：原点={origin}, 法向量={normal}")
         return parting_surface, parting_line
 
-    def _extract_parting_surface_geometry(self, surface: Any) -> Dict[str, Any]:
+    def _extract_parting_surface_geometry(self, surface: TopoDS_Face) -> Dict[str, Any]:
         """提取分型面几何数据"""
         metadata = self._extract_plane_metadata(surface)
 

src/core/multi_scheme_planner.py

@@ -6,7 +6,7 @@ from OCC.Core.GProp import GProp_GProps
 from OCC.Core.gp import gp_Dir, gp_Pln, gp_Pnt
 from OCC.Core.TopAbs import TopAbs_FACE
 from OCC.Core.TopExp import TopExp_Explorer
-from OCC.Core.TopoDS import TopoDS_Face, topods
+from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Shape, topods
 
 from core.mold_generator import MoldCavityGenerator
 from core.aluminum_foam_mold import AluminumFoamMoldGenerator
@@ -31,7 +31,7 @@ class MultiSchemeMoldPlanner:
 
     def generate_plan(
         self,
-        shape: Any,
+        shape: TopoDS_Shape,
         material: Dict[str, Any],
         is_foam_material: bool = False,
         max_schemes: int = 3,
@@ -92,7 +92,7 @@ class MultiSchemeMoldPlanner:
     def _build_scheme(
         self,
         generator: Any,
-        shape: Any,
+        shape: TopoDS_Shape,
         analysis: Dict[str, Any],
         candidate: Dict[str, Any],
         is_foam_material: bool,
@@ -209,10 +209,10 @@ class MultiSchemeMoldPlanner:
         generator: Any,
         analysis: Dict[str, Any],
         direction_vector: List[float],
-        shape: Any,
+        shape: TopoDS_Shape,
         offset_ratio: float = 0.0,
         opening_span_mm: Optional[float] = None,
-    ) -> Any:
+    ) -> TopoDS_Face:
         center = analysis.get("bounding_box", {}).get("center", [0, 0, 0])
         dims = analysis.get("bounding_box", {}).get("dimensions", [100, 100, 100])
         span = max(dims) * 1.5 + 30
@@ -272,7 +272,7 @@ class MultiSchemeMoldPlanner:
 
         return variants
 
-    def _collect_axis_normal_stats(self, generator: Any, shape: Any) -> Dict[str, float]:
+    def _collect_axis_normal_stats(self, generator: Any, shape: TopoDS_Shape) -> Dict[str, float]:
         """按坐标轴统计面法向分布强度，用于候选方向排序。"""
         stats = {"X": 0.0, "Y": 0.0, "Z": 0.0}
         explorer = TopExp_Explorer(shape, TopAbs_FACE)

src/core/side_action_designer.py

@@ -21,6 +21,7 @@
 from typing import Dict, List, Any, Optional, Tuple
 import math
 import numpy as np
+from OCC.Core.TopoDS import TopoDS_Shape, TopoDS_Face
 from utils.logger import get_logger
 
 logger = get_logger(__name__)
@@ -29,8 +30,8 @@ logger = get_logger(__name__)
 class UndercutDetector:
     """倒扣区域检测器"""
 
-    def detect_undercuts(self, shape: Any, parting_direction: List[float],
+    def detect_undercuts(self, shape: TopoDS_Shape, parting_direction: List[float],
-                         parting_surface: Any = None) -> Dict[str, Any]:
+                         parting_surface: Optional[TopoDS_Face] = None) -> Dict[str, Any]:
         """
         检测产品中的倒扣区域
 
@@ -443,8 +444,8 @@ class SideActionDesigner:
         self.slider_designer = SliderMechanismDesigner()
         self.lifter_designer = LifterMechanismDesigner()
 
-    def analyze_and_design(self, shape: Any, parting_direction: List[float],
+    def analyze_and_design(self, shape: TopoDS_Shape, parting_direction: List[float],
-                           mold_size: Dict, parting_surface: Any = None) -> Dict[str, Any]:
+                           mold_size: Dict, parting_surface: Optional[TopoDS_Face] = None) -> Dict[str, Any]:
         """
         综合分析倒扣并设计侧向分型机构
 

src/core/stp_parser.py

@@ -6,6 +6,7 @@ import json
 from utils.logger import get_logger
 from OCC.Core.GProp import GProp_GProps
 from OCC.Core.BRepGProp import brepgprop
+from OCC.Core.TopoDS import TopoDS_Shape
 
 logger = get_logger(__name__)
 
@@ -29,7 +30,7 @@ class STPParser:
 
 
 
-    def load_step_file(self, file_path: Path) -> Any:
+    def load_step_file(self, file_path: Path) -> TopoDS_Shape:
         """加载STP文件"""
         try:
             from OCC.Core.STEPControl import STEPControl_Reader
@@ -51,7 +52,7 @@ class STPParser:
             logger.error(f"STP解析失败: {e}")
             raise
 
-    def analyze_geometry(self, shape) -> Dict[str, Any]:
+    def analyze_geometry(self, shape: TopoDS_Shape) -> Dict[str, Any]:
         """分析几何属性"""
 
         try:
@@ -97,7 +98,7 @@ class STPParser:
             logger.error(f"几何分析失败: {e}")
             raise
 
-    def _compute_bounding_box(self, shape) -> Dict[str, Any]:
+    def _compute_bounding_box(self, shape: TopoDS_Shape) -> Dict[str, Any]:
         """计算边界框"""
         try:
             from OCC.Core.Bnd import Bnd_Box
@@ -125,7 +126,7 @@ class STPParser:
             logger.error(f"边界框计算失败: {e}")
             return self._default_bounding_box()
 
-    def _compute_volume(self, shape) -> float:
+    def _compute_volume(self, shape: TopoDS_Shape) -> float:
         """计算体积"""
         try:
             from OCC.Core.GProp import GProp_GProps
@@ -141,7 +142,7 @@ class STPParser:
             logger.error(f"体积计算失败: {e}")
             raise RuntimeError(f"体积计算失败: {e}") from e
 
-    def _compute_surface_area(self, shape) -> float:
+    def _compute_surface_area(self, shape: TopoDS_Shape) -> float:
         """计算表面积"""
         try:
             from OCC.Core.GProp import GProp_GProps
@@ -175,7 +176,7 @@ class STPParser:
             logger.error(f"表面积计算失败: {e}")
             raise RuntimeError(f"表面积计算失败: {e}") from e
 
-    def _compute_center_of_mass(self, shape) -> List[float]:
+    def _compute_center_of_mass(self, shape: TopoDS_Shape) -> List[float]:
         """计算质心"""
         try:
             from OCC.Core.GProp import GProp_GProps
@@ -194,7 +195,7 @@ class STPParser:
             except Exception:
                 raise RuntimeError(f"质心计算失败且边界框回退也失败: {e}") from e
 
-    def _compute_inertia_properties(self, shape) -> Dict[str, Any]:
+    def _compute_inertia_properties(self, shape: TopoDS_Shape) -> Dict[str, Any]:
         """计算惯性属性"""
         try:
             from OCC.Core.GProp import GProp_GProps
@@ -216,7 +217,7 @@ class STPParser:
             logger.error(f"惯性属性计算失败: {e}")
             return {}
 
-    def _analyze_topology(self, shape) -> Dict[str, int]:
+    def _analyze_topology(self, shape: TopoDS_Shape) -> Dict[str, int]:
         """分析拓扑"""
         try:
             from OCC.Core.TopExp import TopExp_Explorer