# api/routes.py from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Request, Depends from typing import Optional, Dict, Any, List import uuid from datetime import datetime 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 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 services.auth_service import get_current_active_user from models.database import User 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") tasks = {} @router.get("/health") @router.post("/health") async def health(): return { "status": "healthy", "pythonocc": True, "total_tasks": len(tasks) } @router.post("/upload") async def upload_stp( background_tasks: BackgroundTasks, file: UploadFile = File(...), material: Optional[str] = "ABS", db_session: AsyncSession = Depends(get_db_session), current_user: User = Depends(get_current_active_user) ): """上传STP文件并存储到数据库""" if not file.filename.lower().endswith(('.stp', '.step')): raise HTTPException(400, "只支持STP/STEP文件") task_id = str(uuid.uuid4()) # 保存文件 file_path, file_size = await file_handler.save_uploaded_file(file) # 创建存储集成服务实例 storage_service = StorageIntegrationService() # 保存STP文件到RustFS + PostgreSQL stp_file = await storage_service.save_stp_file( session=db_session, file_path=file_path, original_filename=file.filename, user_id=current_user.id ) # 创建处理任务记录 await storage_service.create_processing_task(db_session, task_id, stp_file.id) # 创建内存任务记录 tasks[task_id] = create_task_info( task_id=task_id, status=ProcessingStatus.PROCESSING, filename=file.filename, file_path=str(file_path), file_size=file_size, upload_time=str(datetime.now()) ) # 后台处理（包含数据库存储） background_tasks.add_task(process_file_with_storage, task_id, file_path, stp_file.id, db_session, material) return { "task_id": task_id, "status": "processing", "message": "文件上传成功，开始处理并存储到数据库", "file_info": { "filename": file.filename, "size": file_size, "pythonocc_available": True, "database_file_id": stp_file.id } } @router.get("/status/{task_id}") @router.post("/status/{task_id}") async def get_status(task_id: str, db_session: AsyncSession = Depends(get_db_session)): """ 获取任务状态 优先返回内存中的任务信息； 如果内存中不存在，则从 PostgreSQL + RustFS 组装一个持久化的任务视图， 结构与内存任务保持尽量一致，便于前端集中展示总结性信息。 """ try: # 1. 内存任务（进行中的任务） if task_id in tasks: task = tasks[task_id] logger.info(f"返回内存任务状态：{task_id} - {task['status']}") logger.info(f"内存任务 analysis_result: {task.get('analysis_result', 'None')}") logger.info(f"内存任务 html_file: {task.get('html_file', 'None')}") return task # 2. 持久化任务（已完成/失败，或服务重启后的任务） from sqlalchemy import select from models.database import ProcessingTask, STPFile, GeometryData, MeshData, MoldCavityData storage_service = StorageIntegrationService() # 查询任务和文件元数据 result = await db_session.execute( select(ProcessingTask, STPFile) .join(STPFile, ProcessingTask.stp_file_id == STPFile.id) .where(ProcessingTask.task_id == task_id) ) row = result.first() if not row: raise HTTPException(404, "任务不存在") processing_task, stp_file = row # 从 RustFS 取几何 / 型腔 / 网格详细 JSON（小量数据，便于前端展示汇总） try: file_with_data = await storage_service.get_stp_file_with_data( db_session, stp_file_id=stp_file.id ) except Exception as e: logger.error(f"获取文件数据失败: {e}") file_with_data = {} geometry_json: Optional[Dict[str, Any]] = None if file_with_data.get("geometry_data"): geo_raw = file_with_data["geometry_data"] if isinstance(geo_raw, dict): if "geometry_data" in geo_raw: geometry_json = geo_raw["geometry_data"] else: geometry_json = geo_raw cavity_json: Optional[Dict[str, Any]] = file_with_data.get("mold_cavity_data") features_json: List[Dict[str, Any]] = file_with_data.get("features", []) recommendations_json: List[Dict[str, Any]] = file_with_data.get("recommendations", []) # 组装网格摘要 mesh_summary = None mesh_record = await db_session.execute( select(MeshData).where(MeshData.stp_file_id == stp_file.id) ) mesh_record = mesh_record.scalar_one_or_none() if mesh_record: mesh_summary = { "vertex_count": mesh_record.vertex_count, "face_count": mesh_record.face_count, "point_count": mesh_record.point_count, "quality": mesh_record.quality, } # 构造与内存任务兼容的任务视图 task_view = { "task_id": processing_task.task_id, "status": processing_task.status, "filename": stp_file.original_filename if stp_file else "", "file_path": stp_file.file_path or "", "file_size": stp_file.file_size if stp_file else 0, "upload_time": processing_task.created_time.isoformat() if processing_task.created_time else "", "completed_at": processing_task.completed_time.isoformat() if processing_task.completed_time else "", "geometry_data": geometry_json, "key_info": cavity_json, "cavity_data": cavity_json, "mesh_summary": mesh_summary, "analysis_result": { "geometry_data": geometry_json, "detected_features": features_json, "design_recommendations": recommendations_json, "quality_metrics": { "volume_utilization": file_with_data.get("analysis_metrics", {}).get("volume_utilization", 0), "topology_complexity": file_with_data.get("analysis_metrics", {}).get("topology_complexity", 0), "wall_uniformity": file_with_data.get("analysis_metrics", {}).get("wall_uniformity", 0) }, "analysis_summary": file_with_data.get("analysis_metrics", {}).get("analysis_summary", "分析完成") } if geometry_json or features_json or recommendations_json else None, "error": processing_task.error_message or stp_file.error_message or None, } logger.info(f"返回持久化任务状态: {task_id} - {processing_task.status}") return task_view except HTTPException: raise except Exception as e: logger.error(f"获取任务状态失败: {e}") raise HTTPException(500, f"获取任务状态失败: {str(e)}") @router.get("/debug/tasks") @router.post("/debug/tasks") async def debug_tasks(): """调试接口：查看所有任务""" return { "total_tasks": len(tasks), "tasks": tasks } @router.get("/history") @router.post("/history") async def get_file_history(db_session: AsyncSession = Depends(get_db_session)): """获取按文件名分组的文件历史记录（支持多上传）""" storage_service = StorageIntegrationService() file_groups = await storage_service.get_all_file_groups(db_session) return { "total_files": len(file_groups), "files": file_groups } @router.get("/history/{filename}") @router.post("/history/{filename}") async def get_file_records(filename: str, db_session: AsyncSession = Depends(get_db_session)): """获取指定文件名的所有上传记录（支持多上传历史）""" import urllib.parse decoded_filename = urllib.parse.unquote(filename) storage_service = StorageIntegrationService() file_records = await storage_service.get_file_history_by_filename( db_session, decoded_filename ) return file_records @router.get("/result/{task_id}") @router.post("/result/{task_id}") async def result_page(request: Request, task_id: str, db_session: AsyncSession = Depends(get_db_session)): """结果详情页面""" from sqlalchemy import select from models.database import ProcessingTask, STPFile, GeometryData, MoldCavityData, HTMLFile # 从数据库查询任务详情 result = await db_session.execute( select(ProcessingTask, STPFile) .join(STPFile, ProcessingTask.stp_file_id == STPFile.id) .where(ProcessingTask.task_id == task_id) ) task_record = result.first() if not task_record: raise HTTPException(404, "任务不存在") task, stp_file = task_record # 构建任务详情数据（先只包含基本数据） task_data = { "task_id": task.task_id, "filename": stp_file.original_filename if stp_file else "", "file_size": stp_file.file_size if stp_file else 0, "status": task.status, "progress": task.progress, "current_step": task.current_step, "created_at": task.created_time.isoformat() if task.created_time else "", "completed_at": task.completed_time.isoformat() if task.completed_time else "", "error": task.error_message if task.error_message else "" } from fastapi.templating import Jinja2Templates import os # 简化路径配置，直接使用当前工作目录下的templates文件夹 templates_dir = os.path.join(os.getcwd(), "templates") templates = Jinja2Templates(directory=templates_dir) return templates.TemplateResponse("result.html", { "request": request, "task": task_data, "pythonocc_available": True, "version": "3.0.0" }) 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()) 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) # 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验证已禁用"} # 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]["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]["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())