# services/processing_service.py """STP 文件处理流程编排器 — 协调解析、网格生成、型腔生成、保存、验证""" import asyncio import os import shutil import tempfile import time from datetime import datetime from pathlib import Path from typing import Optional, Dict, Any, List, Tuple from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from moldinsight.core.cad_exporter import CADExporter from moldinsight.services.occ_process_pool import OccProcessPool from moldinsight.services.task_storage_service import TaskStorageService from moldinsight.services.analysis_storage_service import AnalysisStorageService from moldinsight.storage.rustfs_storage import rustfs_manager from shared.services.redis_task_manager import redis_task_manager from moldinsight.services.material_service import MaterialService from moldinsight.services.calculation_service import CalculationService from moldinsight.services.llm_service import llm_service from shared.models.schemas import ProcessingStatus from moldinsight.models import STPFile from shared.database.database import db_manager from shared.utils.html_generator import HTMLGenerator from shared.utils.logger import get_logger logger = get_logger(__name__) class ProcessingService: """核心处理流程编排 — 协调 STP 解析、网格、型腔、计算、保存、验证""" def __init__(self): # 批次 3 按职责拆分：任务/文件生命周期 与 分析结果数据（原 StorageIntegrationService） self.task_storage = TaskStorageService() self.analysis_storage = AnalysisStorageService() # cad_exporter 仅用于导出文件条目构建/输出目录（export_artifacts 路径语义）； # 真正的 OCC 形状导出在子进程内完成（见 core/occ_worker.py） self.cad_exporter = CADExporter() # D11：HTMLGenerator 不再持有常驻实例，可视化产物统一写任务内临时目录后 # 上传 RustFS 报告键（见 process_file_core） # OCC 方案 B：所有几何操作（解析/布尔/三角化/倒扣/转换）经常驻 OCC 进程池， # 进程边界干净回收超时/崩溃——替代原线程级单通道 executor（见 OCC_THROUGHPUT.md） self._occ_pool = OccProcessPool() # ─── 对外入口 ─── async def run_occ(self, op_name: str, payload: Dict[str, Any], timeout: float = 600) -> Any: """在常驻 OCC 工作进程中执行同步几何操作（方案 B）。 OCC 非线程安全，所有几何计算统一经本入口在独立进程中串行执行； 超时/进程崩溃由进程池 terminate + 换新补位，干净回收（见 OCC_THROUGHPUT.md）。 op_name 须已在 occ_worker._OPS 注册；payload 只含文件路径 + 普通字典。 """ return await self._occ_pool.run(op_name, payload, timeout=timeout) async def _materialize_source_file(self, stp_file: STPFile) -> Tuple[Path, Optional[Path]]: """把待处理文件落到本地磁盘，返回 (本地路径, 临时目录或 None)。 RustFS 为主存储：处理方按 object_key 下载到任务专属临时目录 （文件名保留原始名——下游产物命名依赖 Path(file_path).name）。 RustFS 不可用或对象缺失时回退 STPFile.file_path 记录的节点本地路径 （依赖 compose 共享卷，属过渡方案）；两者皆不可用则抛错置任务失败。 """ original_name = Path(stp_file.original_filename or "model.stp").name or "model.stp" if stp_file.object_key: temp_dir = Path(tempfile.mkdtemp(prefix=f"moldinsight_{stp_file.id}_")) try: data = await rustfs_manager.download_file( file_type="stp_files", object_key=stp_file.object_key ) target = temp_dir / original_name target.write_bytes(data) return target, temp_dir except Exception as exc: shutil.rmtree(temp_dir, ignore_errors=True) logger.warning( f"RustFS 源文件下载失败 (object_key={stp_file.object_key})，" f"回退节点本地路径: {exc}" ) local = Path(stp_file.file_path) if stp_file.file_path else None if local and local.exists(): return local, None raise RuntimeError( f"源文件不可用：RustFS 对象 {stp_file.object_key!r} 下载失败，" f"且节点本地路径不存在: {stp_file.file_path!r}" ) async def process_file_with_storage( self, task_id: str, stp_file_id: int, process_params: Optional[Dict[str, Any]] = None, ): """处理文件的后台任务 — 使用独立数据库会话 分派入参只带 stp_file_id（D6）：源文件由本方法按 PG 元数据中的 object_key 从 RustFS 获取，不再依赖分派方传入节点本地路径 （API 与 Celery worker 容器文件系统不互通）。 """ # 创建独立的数据库会话，避免请求范围会话关闭 async with db_manager.session() as db_session: temp_dir: Optional[Path] = None try: result = await db_session.execute( select(STPFile).where(STPFile.id == stp_file_id) ) stp_file = result.scalar_one_or_none() if not stp_file: raise RuntimeError(f"STPFile 记录不存在: stp_file_id={stp_file_id}") source_path, temp_dir = await self._materialize_source_file(stp_file) file_path = str(source_path) logger.info(f"开始处理文件并生成模具型腔: {file_path}") from shared.config.settings import settings file_size_bytes = source_path.stat().st_size file_size_mb = max(file_size_bytes / (1024 * 1024), 1) timeout_seconds = min( max(settings.PROCESSING_TIMEOUT_BASE, int(file_size_mb * settings.PROCESSING_TIMEOUT_PER_MB)), 1800, ) logger.info(f"处理超时设置为 {timeout_seconds}s (文件 {file_size_mb:.1f}MB)") try: await asyncio.wait_for( self.process_file_core( task_id, file_path, stp_file_id, db_session, process_params ), timeout_seconds, ) except asyncio.TimeoutError: logger.error(f"处理超时: {task_id}") # OCC 进程无法取消：整体重建进程池，干净杀掉可能卡死的 OCC 操作 await self._occ_pool.recover() raise Exception(f"处理超时，超过{timeout_seconds}秒未完成") except Exception as e: logger.error(f"模具型腔生成失败: {e}") # D9：先丢弃未提交的数据本体，失败状态单独提交， # 避免 failed 更新把半成品 flush 数据一起带上 await db_session.rollback() await self.task_storage.update_stp_file_status(db_session, stp_file_id, "failed") await self.task_storage.update_task_status( db_session, task_id, "failed", error_message=str(e) ) # 安全更新 Redis 任务状态（Redis 仅热缓存，写失败不影响 PG 事实） 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()), }) finally: # 任务专属临时目录必须清理，长期运行不允许残留下载副本 if temp_dir: shutil.rmtree(temp_dir, ignore_errors=True) async def process_file_core( self, task_id: str, file_path: str, stp_file_id: int, db_session: AsyncSession, process_params: Optional[Dict[str, Any]] = None, ): """核心处理逻辑""" try: logger.info(f"开始处理文件并生成模具型腔: {file_path}") process_params = self._normalize_process_params(process_params) stage_timings: Dict[str, float] = {} # 1. 解析STP文件 await self.task_storage.update_task_status( db_session, task_id, "processing", 20, "解析STP文件" ) stage_started = time.perf_counter() # 方案 B：解析 + 几何分析在 OCC 子进程内一步完成（形状不跨进程） geometry_data = await self.run_occ( "parse_stp", {"stp_path": file_path}, timeout=timeout_seconds ) stage_timings["parse_stp"] = round(time.perf_counter() - stage_started, 3) # 2. 生成网格数据并持久化 await self.task_storage.update_task_status( db_session, task_id, "processing", 30, "生成网格数据" ) stage_started = time.perf_counter() mesh_result = await self._step_generate_mesh( geometry_data, file_path, db_session, stp_file_id, task_id, timeout=timeout_seconds, ) stage_timings["generate_mesh"] = round(time.perf_counter() - stage_started, 3) # 3. 生成模具型腔 await self.task_storage.update_task_status( db_session, task_id, "processing", 40, "生成模具型腔" ) # 材料属性 — 通过 MaterialService 集中管理 requested_material = MaterialService.resolve_material(process_params["material"]) selected_material = dict(MaterialService.get_material(requested_material)) selected_material["shrinkage"] = process_params["shrinkage_rate"] / 100.0 is_foam_material = MaterialService.is_foam_material(requested_material) stage_started = time.perf_counter() plan_result, export_artifacts = await self._step_generate_cavity( db_session, file_path, selected_material, is_foam_material, process_params, task_id, timeout=timeout_seconds, ) stage_timings["generate_cavity"] = round(time.perf_counter() - stage_started, 3) # 方案 B：各方案形状的持久化 STEP 已由子进程导出并返回 manifest（export_artifacts）， # 主进程不再持有 TopoDS 引用（无跨进程传输） # 4. 生成详细JSON数据 — 委托 CalculationService await self.task_storage.update_task_status( db_session, task_id, "processing", 60, "生成型腔详细数据" ) stage_started = time.perf_counter() detailed_cavity_json = CalculationService.build_plan_result( geometry_data=geometry_data, material=selected_material, file_path=str(file_path), plan_result=plan_result, ) stage_timings["build_plan_result"] = round(time.perf_counter() - stage_started, 3) 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.task_storage.update_task_status( db_session, task_id, "processing", 70, "保存几何数据" ) stage_started = time.perf_counter() await self.analysis_storage.save_geometry_data( db_session, stp_file_id, geometry_data, geometry_data.get("analysis_method", "mold_cavity"), ) # 阶段 A 提交（D9）：几何 + 网格原子落库——解析后的确定成果， # 后续型腔失败任务标 failed 时这些数据仍完整保留 await db_session.commit() # 7. 生成HTML可视化 await self.task_storage.update_task_status( db_session, task_id, "processing", 85, "生成可视化报告" ) pointcloud_data = None lod_data = None if mesh_result: lod0 = mesh_result.get("lods", {}).get("0", {}) pointcloud_data = { "points": mesh_result.get("points", []), "normals": mesh_result.get("normals", []), "vertices": lod0.get("vertices", []), "faces": lod0.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), } if mesh_result and mesh_result.get("lods"): lods = mesh_result["lods"] lod_data = mesh_result logger.info(f"LOD数据复用成功: {len(lods)} 级 (面数: {[lods[k]['face_count'] for k in sorted(lods.keys())]})") # D11：可视化产物先写任务内临时目录，再统一上传 RustFS 报告键 # （html/reports/{filename}），不再落节点本地 html_output—— # API 与 worker 容器文件系统不互通，本地盘从来不是可依赖的读取来源 html_out_dir = Path(tempfile.mkdtemp(prefix="moldinsight_html_")) try: html_generator = HTMLGenerator(output_dir=str(html_out_dir)) 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, html_generator=html_generator, ) 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.analysis_storage.save_mold_cavity_data( db_session, stp_file_id, detailed_cavity_json ) html_file_path = 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._upload_report_artifacts(Path(html_file_path)) await self.analysis_storage.save_html_file( db_session, stp_file_id, Path(html_file_path).name, html_file_path, ) finally: shutil.rmtree(html_out_dir, ignore_errors=True) stage_timings["persist_artifacts"] = round(time.perf_counter() - stage_started, 3) # 9. 分析模具设计 stage_started = time.perf_counter() analysis_result = await self.run_occ( "analyze_mold_design", { "geometry_data": geometry_data, "product_material": requested_material, "stp_path": file_path, }, timeout=timeout_seconds, ) if analysis_result: await self.analysis_storage.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) stage_timings["analyze_design"] = round(time.perf_counter() - stage_started, 3) # 9.6 更新STP文件的分析摘要字段 await self.task_storage.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.65 阶段 B 提交（D9）：型腔 / HTML / 特征 / 指标 / 摘要 / 验证指标 # 作为完整结果包原子落库——置 completed 前必须全部就位， # 期间任一步失败回滚后任务标 failed，不会出现"completed 但数据残缺" await db_session.commit() # 9.7 FreeCAD 几何验证 stage_started = time.perf_counter() verification_result = await self._step_verify( file_path, db_session, task_id, stp_file_id, analysis_result ) stage_timings["verify_geometry"] = round(time.perf_counter() - stage_started, 3) # 9.8 LLM 增强分析 llm_report = None stage_started = time.perf_counter() 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 ) stage_timings["generate_llm_report"] = round(time.perf_counter() - stage_started, 3) # 10. 完成处理——先 flush 任务参数，完成状态提交时一并原子落库（D9） await self.task_storage.update_task_parameters( db_session, task_id, { "stage_timings": stage_timings, "material": requested_material, "verification": verification_result, "llm_report": llm_report, "export_artifacts": export_artifacts, **process_params, }, ) await self.task_storage.update_stp_file_status(db_session, stp_file_id, "completed") await self.task_storage.update_task_status( db_session, task_id, "completed", 100, "模具型腔生成完成" ) # 更新任务缓存状态（仅保留轻量摘要，完整数据由PG+RustFS持久化； # 完成态视图由 TaskQueryService 从 PG+RustFS 组装，Redis 不再存 # geometry_data / analysis_result 等 MB 级大对象） await redis_task_manager.update_task(task_id, { "status": ProcessingStatus.COMPLETED, "completed_at": str(datetime.now()), "key_info": best_key_info, "best_scheme_id": detailed_cavity_json.get("best_scheme_id"), "material": requested_material, "parameters": process_params, "stage_timings": stage_timings, "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, "llm_report": llm_report, "export_artifacts": export_artifacts, }) 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}") # D9：先丢弃未提交的数据本体再置失败（同外层说明） await db_session.rollback() await self.task_storage.update_stp_file_status(db_session, stp_file_id, "failed") await self.task_storage.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, geometry_data: dict, file_path: str, db_session: AsyncSession, stp_file_id: int, task_id: str, timeout: float = 600, ) -> Optional[Dict[str, Any]]: """生成多级LOD网格并持久化（方案 B：子进程内一次 OCC 剖分），失败不影响主流程""" mesh_result = None try: mesh_result = await self.run_occ( "generate_mesh", {"stp_path": file_path}, timeout=timeout ) lod0 = mesh_result.get("lods", {}).get("0", {}) vertices = lod0.get("vertices", []) faces = lod0.get("faces", []) points = mesh_result.get("points", []) normals = mesh_result.get("normals", []) point_count = mesh_result.get("point_count", 0) vertex_count = lod0.get("vertex_count", mesh_result.get("vertex_count", 0)) face_count = lod0.get("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.analysis_storage.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, db_session: AsyncSession, file_path: str, selected_material: dict, is_foam_material: bool, process_params: Dict[str, Any], task_id: str, timeout: float = 600, ) -> Tuple[Dict[str, Any], Optional[Dict[str, Any]]]: """生成多方案分模结果（方案 B：子进程内完成分模 + 方案形状 STEP 导出）。 D8：型腔是任务的核心产出，生成失败必须让任务 failed—— 异常直接向编排层传播。返回 (plan_result, export_manifest)。 D17 Human-in-Loop 闭环：解析同指纹历史 hints（list of {scheme_axis, weight, sample_count, ...}）， 装进 OCC worker payload，让子进程内的 planner/candidate_generator/scheme_scorer 加成。 hints 解析失败不阻塞主流程（logger.warning 后视为空），保证已有任务不退化。 """ # D17：拉取同指纹老师傅经验（写入即消费） experience_hints: List[Dict[str, Any]] = [] try: from moldinsight.services.experience_feedback_service import ( experience_feedback_service, ) experience_hints = await experience_feedback_service.resolve_for_process_params( session=db_session, task_id=task_id, process_params=process_params or {}, ) if experience_hints: logger.info( f"D17 Human-in-Loop：注入 {len(experience_hints)} 条经验" f"到 task={task_id} 的分模方案" ) except Exception as hints_err: logger.warning( f"D17 hints 解析失败，回退到无 hints 模式: {hints_err}" ) experience_hints = [] result = await self.run_occ( "generate_cavity", { "stp_path": file_path, "task_id": task_id, "material": selected_material, "is_foam_material": is_foam_material, "process_params": process_params, "export_out_dir": os.path.abspath(self.cad_exporter.output_dir), "experience_hints": experience_hints, # D17 payload 通道 }, timeout=timeout, ) plan_result = result["plan_result"] logger.info( f"多方案分模完成: 生成 {len(plan_result.get('candidate_schemes', []))} 套方案" ) return plan_result, result["export_manifest"] async def regenerate_export_from_persisted( self, task_id: str, scheme_id: str, formats: Optional[List[str]], components: List[str], base_filename: str, scheme_files: List[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: """持久化 STEP 缓存失效（如服务重启）后，从持久化 STEP 重建导出文件。 方案 B 后主进程不再持有 TopoDS 形状（导出持久化发生在子进程）， 内存缓存路径已随旧 _cache_export_shapes/get_export_shapes 一并删除； 本方法读回单组件 STEP 现场转换缺失格式，用户无需重新分析。 所有组件均不可用时返回 None。 """ format_list = list(dict.fromkeys(formats or ["step", "stl"])) step_files = { f.get("component"): f for f in scheme_files or [] if f.get("format") == "step" and f.get("component") } if not step_files: return None files: List[Dict[str, Any]] = [] errors: List[str] = [] if "step" in format_list: assembly = step_files.get("assembly") if assembly: files.append(assembly) else: errors.append("模具装配体 (step) 不可用") for comp in components: comp_file = step_files.get(comp) if comp_file is None: errors.append(f"组件 {comp} 的持久化 STEP 不可用") continue for fmt in format_list: if fmt == "step": files.append(comp_file) continue step_path = os.path.join( self.cad_exporter.output_dir, str(comp_file.get("relative_path") or "").replace("/", os.sep), ) out_path = os.path.join( os.path.dirname(step_path), f"{base_filename}_{comp}.{fmt}" ) ok = await self.run_occ( "convert_component_step", {"step_path": str(step_path), "out_path": str(out_path), "fmt": fmt}, ) if ok: files.append( self.cad_exporter.build_file_entry(comp, fmt, out_path) ) else: errors.append(f"组件 {comp} ({fmt}) 转换失败") if not files: return None return { "base_filename": base_filename, "task_id": task_id, "scheme_id": scheme_id, "files": files, "errors": errors, "total_files": len(files), "total_errors": len(errors), "source": "regenerated", } @staticmethod def _normalize_process_params(process_params: Optional[Dict[str, Any]]) -> Dict[str, Any]: payload = dict(process_params or {}) return { "material": MaterialService.resolve_material(str(payload.get("material", "ABS"))), "draft_angle": float(payload.get("draft_angle", 2.0)), "shrinkage_rate": float(payload.get("shrinkage_rate", 0.5)), "parting_precision": float(payload.get("parting_precision", 0.1)), "cavity_match": int(payload.get("cavity_match", 95)), } 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 shared.config.settings import settings if not settings.ENABLE_FREECAD_VERIFICATION: logger.info("FreeCAD验证已禁用（设置 ENABLE_FREECAD_VERIFICATION=true 启用）") return {"status": "disabled", "reason": "FreeCAD验证已禁用"} await self.task_storage.update_task_status( db_session, task_id, "processing", 90, "FreeCAD几何验证" ) try: from moldinsight.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, html_generator: HTMLGenerator = None, ) -> Dict[str, Any]: """为候选分模方案生成轻量摘要链接（完整HTML仅最优方案按需生成）""" if html_generator is None: raise ValueError( "html_generator 不能为空（D11：摘要产物统一经任务临时目录上传 RustFS）" ) 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") base_stem = Path(stp_filename).stem.replace(" ", "_") ts = datetime.now().strftime("%Y%m%d_%H%M%S") summary_name = f"mold_{base_stem}_{suffix}_{ts}_summary.json" summary_content = html_generator.generate_3d_viewer_summary( geometry_data, cavity_data ) # D11：摘要 JSON 写任务临时目录后直传 RustFS 报告键，不落本地 html_output summary_path = html_generator.save_data_file(summary_content, summary_name) await rustfs_manager.upload_report_artifact( summary_name, Path(summary_path).read_bytes(), content_type="application/json", ) scheme["summary_file"] = f"/html/{summary_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 _upload_report_artifacts(self, html_file_path: Path): """D11：任务临时目录中的可视化产物（.html / _summary.json / _data.json） 上传 RustFS 报告键（html/reports/{filename}）。 HTML 内嵌相对 DATA_URL/SUMMARY_URL 引用两个 JSON， 三者必须同键前缀可达（读侧 /html/{filename} 直取）。 """ stem = html_file_path.with_suffix("") artifacts = [ (html_file_path, "text/html; charset=utf-8"), (stem.with_name(stem.name + "_summary.json"), "application/json"), (stem.with_name(stem.name + "_data.json"), "application/json"), ] for path, content_type in artifacts: if not path.exists(): raise RuntimeError(f"可视化产物缺失: {path.name}") await rustfs_manager.upload_report_artifact( path.name, path.read_bytes(), content_type=content_type ) # ─── 指标持久化 ─── async def _save_analysis_metrics(self, session: AsyncSession, stp_file_id: int, analysis_result: dict): """保存分析指标到数据库""" from moldinsight.models 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) # D9：flush 不 commit，随结果包（阶段 B）由编排层统一提交 await session.flush() logger.info(f"分析指标保存成功: {metrics.id}") async def _save_verification_metrics(self, session: AsyncSession, stp_file_id: int, verification_result: dict): """保存验证指标到数据库""" from moldinsight.models 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) # D9：flush 不 commit，随结果包（阶段 B）由编排层统一提交 await session.flush() logger.info(f"验证指标保存成功: stp_file_id={stp_file_id}") # 模块级单例，供路由层直接使用 processing_service = ProcessingService()