# services/processing_service.py """STP 文件处理流程编排器 — 协调解析、网格生成、型腔生成、保存、验证""" import asyncio import traceback from datetime import datetime from pathlib import Path from typing import Optional, Dict, Any from sqlalchemy.ext.asyncio import AsyncSession from core.stp_parser import STPParser from core.geometry_analyzer import GeometryAnalyzer 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.multi_scheme_planner import MultiSchemeMoldPlanner from services.storage_integration_rustfs import StorageIntegrationService from services.redis_task_manager import redis_task_manager from services.material_service import MaterialService from services.calculation_service import CalculationService from models.schemas import ProcessingStatus from database.database import db_manager from utils.html_generator import HTMLGenerator from utils.logger import get_logger logger = get_logger(__name__) class ProcessingService: """核心处理流程编排 — 协调 STP 解析、网格、型腔、计算、保存、验证""" def __init__(self): self.stp_parser = STPParser() self.geometry_analyzer = GeometryAnalyzer() self.mold_generator = MoldCavityGenerator(shrinkage_rate=0.005) self.aluminum_foam_generator = AluminumFoamMoldGenerator(shrinkage_rate=0.015, draft_angle=3.0) self.mold_quality_inspector = AluminumFoamMoldQualityInspector() self.mesh_generator = MeshGenerator(quality="medium") self.html_generator = HTMLGenerator() self.storage_service = StorageIntegrationService() self.multi_scheme_planner = MultiSchemeMoldPlanner() # ─── 对外入口 ─── async def process_file_with_storage( self, task_id: str, file_path: str, stp_file_id: int, material: str = "ABS", ): """处理文件的后台任务 — 使用独立数据库会话""" # 创建独立的数据库会话，避免请求范围会话关闭 async with db_manager.session() as db_session: try: logger.info(f"开始处理文件并生成模具型腔: {file_path}") # 设置处理超时（5分钟） timeout_seconds = 300 try: await asyncio.wait_for( self.process_file_core( 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 self.storage_service.update_stp_file_status(db_session, stp_file_id, "failed") await self.storage_service.update_task_status( db_session, task_id, "failed", error_message=str(e) ) # 安全更新 Redis 任务状态 task = await redis_task_manager.get_task(task_id) if task: await redis_task_manager.update_task(task_id, { "status": ProcessingStatus.FAILED, "error": str(e), "completed_at": str(datetime.now()), }) async def process_file_core( self, task_id: str, file_path: str, stp_file_id: int, db_session: AsyncSession, material: str = "ABS", ): """核心处理逻辑""" try: logger.info(f"开始处理文件并生成模具型腔: {file_path}") # 1. 解析STP文件 await self.storage_service.update_task_status( db_session, task_id, "processing", 20, "解析STP文件" ) shape = self.stp_parser.load_step_file(Path(file_path)) geometry_data = self.stp_parser.analyze_geometry(shape) # 2. 生成网格数据并持久化 await self.storage_service.update_task_status( db_session, task_id, "processing", 30, "生成网格数据" ) mesh_result = await self._step_generate_mesh( shape, geometry_data, file_path, db_session, stp_file_id, task_id ) # 3. 生成模具型腔 await self.storage_service.update_task_status( db_session, task_id, "processing", 40, "生成模具型腔" ) # 材料属性 — 通过 MaterialService 集中管理 requested_material = MaterialService.resolve_material(material) selected_material = MaterialService.get_material(requested_material) is_foam_material = MaterialService.is_foam_material(requested_material) plan_result = await self._step_generate_cavity( shape, selected_material, is_foam_material ) # 4. 生成详细JSON数据 — 委托 CalculationService await self.storage_service.update_task_status( db_session, task_id, "processing", 60, "生成型腔详细数据" ) detailed_cavity_json = CalculationService.build_plan_result( geometry_data=geometry_data, material=selected_material, file_path=str(file_path), plan_result=plan_result, ) best_scheme = CalculationService.get_best_scheme(detailed_cavity_json) best_cavity_data = best_scheme.get("cavity_data", {}) if best_scheme else {} best_key_info = best_scheme.get("key_info", {}) if best_scheme else {} if best_cavity_data.get("mold_cavities"): cavity_geometry = best_cavity_data["mold_cavities"].get("cavity", {}) logger.info( f"推荐方案型腔数据已合并: cavity {cavity_geometry.get('vertex_count', 0)} 顶点" ) # 5. 生成关键信息 cavity_key_info = best_key_info # 6. 保存几何数据到数据库 await self.storage_service.update_task_status( db_session, task_id, "processing", 70, "保存几何数据" ) await self.storage_service.save_geometry_data( db_session, stp_file_id, geometry_data, geometry_data.get("analysis_method", "mold_cavity"), ) # 7. 生成HTML可视化 await self.storage_service.update_task_status( db_session, task_id, "processing", 85, "生成可视化报告" ) pointcloud_data = None lod_data = None if mesh_result: pointcloud_data = { "points": mesh_result.get("points", []), "normals": mesh_result.get("normals", []), "vertices": mesh_result.get("vertices", []), "faces": mesh_result.get("faces", []), "point_count": mesh_result.get("point_count", 0), "vertex_count": mesh_result.get("vertex_count", 0), "face_count": mesh_result.get("face_count", 0), } # 生成多级LOD数据（用于前端按距离切换精度） try: lod_result = self.mesh_generator.generate_multi_lod_mesh(shape) if lod_result and lod_result.get("lods"): lod_data = lod_result logger.info(f"LOD数据生成成功: {len(lod_result['lods'])} 级 (面数: {[lod_result['lods'][k]['face_count'] for k in sorted(lod_result['lods'].keys())]})") except Exception as lod_err: logger.warning(f"LOD数据生成失败，使用单级精度: {lod_err}") detailed_cavity_json = await self._attach_scheme_previews( detailed_cavity_json=detailed_cavity_json, geometry_data=geometry_data, stp_filename=Path(file_path).name, pointcloud_data=pointcloud_data, lod_data=lod_data, ) best_scheme = CalculationService.get_best_scheme(detailed_cavity_json) best_cavity_data = best_scheme.get("cavity_data", {}) if best_scheme else best_cavity_data best_key_info = best_scheme.get("key_info", {}) if best_scheme else best_key_info # 8. 保存模具型腔数据（包含方案级预览链接） await self.storage_service.save_mold_cavity_data( db_session, stp_file_id, detailed_cavity_json ) html_file_path = self.html_generator.generate_and_save_visualization( geometry_data, Path(file_path).name, cavity_data=best_cavity_data, pointcloud_data=pointcloud_data, lod_data=lod_data, ) await self.storage_service.save_html_file( db_session, stp_file_id, Path(html_file_path).name, html_file_path, ) # 9. 分析模具设计 analysis_result = self.geometry_analyzer.analyze_mold_design(geometry_data) if analysis_result: await self.storage_service.save_features_and_recommendations( db_session, stp_file_id, analysis_result.get("detected_features", []), analysis_result.get("design_recommendations", []), ) await self._save_analysis_metrics(db_session, stp_file_id, analysis_result) # 9.6 更新STP文件的分析摘要字段 await self.storage_service.update_stp_file_analysis_summary( db_session, stp_file_id, volume=geometry_data.get("volume", 0), surface_area=geometry_data.get("surface_area", 0), product_weight=CalculationService.calculate_product_weight( geometry_data.get("volume", 0), selected_material["density"] ), ) # 9.7 FreeCAD 几何验证 verification_result = await self._step_verify( file_path, db_session, task_id, stp_file_id, analysis_result ) # 10. 完成处理 await self.storage_service.update_stp_file_status(db_session, stp_file_id, "completed") await self.storage_service.update_task_status( db_session, task_id, "completed", 100, "模具型腔生成完成" ) # 更新任务缓存状态 await redis_task_manager.update_task(task_id, { "geometry_data": geometry_data, "analysis_result": analysis_result, "plan_result": detailed_cavity_json, "candidate_schemes": detailed_cavity_json.get("candidate_schemes", []), "best_scheme_id": detailed_cavity_json.get("best_scheme_id"), "cavity_data": best_cavity_data, "key_info": best_key_info, "html_file": best_scheme.get("html_file", f"/html/{Path(html_file_path).name}") if best_scheme else f"/html/{Path(html_file_path).name}", "verification": verification_result, "status": ProcessingStatus.COMPLETED, "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 self.storage_service.update_stp_file_status(db_session, stp_file_id, "failed") await self.storage_service.update_task_status( db_session, task_id, "failed", error_message=str(e) ) task = await redis_task_manager.get_task(task_id) if task: await redis_task_manager.update_task(task_id, { "status": ProcessingStatus.FAILED, "error": str(e), "completed_at": str(datetime.now()), }) # ─── 内部步骤 ─── async def _step_generate_mesh( self, shape, geometry_data: dict, file_path: str, db_session: AsyncSession, stp_file_id: int, task_id: str, ) -> Optional[Dict[str, Any]]: """生成网格数据并持久化，失败不影响主流程""" mesh_result = None try: mesh_result = self.mesh_generator.generate_mesh_from_shape(shape) 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 self.storage_service.save_mesh_data( db_session, stp_file_id=stp_file_id, mesh_json=mesh_json, quality="medium", ) await redis_task_manager.update_task(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}") return mesh_result async def _step_generate_cavity( self, shape, selected_material: dict, is_foam_material: bool, ) -> Optional[Dict[str, Any]]: """生成多方案分模结果""" plan_result = None try: if shape: plan_result = self.multi_scheme_planner.generate_plan( shape=shape, material=selected_material, is_foam_material=is_foam_material, ) logger.info( f"多方案分模完成: 生成 {len(plan_result.get('candidate_schemes', []))} 套方案" ) except Exception as cavity_err: logger.warning(f"多方案分模失败，使用简化数据: {cavity_err}") traceback.print_exc() plan_result = None return plan_result async def _step_verify( self, file_path: str, db_session: AsyncSession, task_id: str, stp_file_id: int, analysis_result: Optional[dict], ) -> Optional[Dict[str, Any]]: """FreeCAD 几何验证（可通过配置禁用）""" from config.settings import settings if not settings.ENABLE_FREECAD_VERIFICATION: logger.info("FreeCAD验证已禁用（设置 ENABLE_FREECAD_VERIFICATION=true 启用）") return {"status": "disabled", "reason": "FreeCAD验证已禁用"} await self.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 self._save_verification_metrics(db_session, stp_file_id, verification_result) logger.info(f"FreeCAD验证完成: {verification_result.get('status', 'unknown') if verification_result else 'failed'}") return verification_result except Exception as ve: logger.warning(f"FreeCAD验证失败（不影响主流程）: {ve}") return {"status": "error", "error": str(ve)} async def _attach_scheme_previews( self, detailed_cavity_json: Dict[str, Any], geometry_data: Dict[str, Any], stp_filename: str, pointcloud_data: Optional[Dict[str, Any]] = None, lod_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: """为每个候选分模方案生成独立HTML预览链接。""" candidate_schemes = detailed_cavity_json.get("candidate_schemes", []) if not candidate_schemes: return detailed_cavity_json for scheme in candidate_schemes: cavity_data = scheme.get("cavity_data") if not cavity_data: continue suffix = scheme.get("scheme_id") html_path = self.html_generator.generate_and_save_visualization( geometry_data, stp_filename, cavity_data=cavity_data, pointcloud_data=pointcloud_data, suffix=suffix, lod_data=lod_data, ) scheme["html_file"] = f"/html/{Path(html_path).name}" best_scheme = CalculationService.get_best_scheme(detailed_cavity_json) if best_scheme: detailed_cavity_json["html_file"] = best_scheme.get("html_file") return detailed_cavity_json # ─── 指标持久化 ─── async def _save_analysis_metrics(self, session: AsyncSession, stp_file_id: int, analysis_result: dict): """保存分析指标到数据库""" 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(self, session: AsyncSession, stp_file_id: int, verification_result: dict): """保存验证指标到数据库""" 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() comparison = verification_result.get("comparison", {}) volume_comparison = comparison.get("volume", {}) area_comparison = comparison.get("surface_area", {}) if metrics: metrics.verification_status = verification_result.get("status", "unknown") 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: 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}") # 模块级单例，供路由层直接使用 processing_service = ProcessingService()