cjw/geMoldInsight
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

x

Browse Source
This commit is contained in:
chenjw28
2026-05-11 11:35:15 +08:00
parent 5e1d475a22
commit a68b9ca57f
22 changed files with 432 additions and 11341 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/services/llm_service.py
+334
View File
@@ -0,0 +1,334 @@
"""
LLM 增强分析服务
提供两个核心能力：
1. generate_design_report — 将分析 JSON 转换为结构化评审报告
2. recommend_parting_direction — 基于几何 + 制造约束推荐最优分型方向
适配层：OpenAI 兼容 API（支持 OpenAI / DeepSeek / vLLM / Ollama 等）
未配置 LLM 时静默降级，不影响主流程。
"""
import json
import re
from typing import Optional, Dict, Any, List
import httpx
from config.settings import settings
from utils.logger import get_logger
logger = get_logger(__name__)
_DESIGN_REPORT_SYSTEM = """你是一位资深注塑模具设计工程师，拥有 20 年模具 DFM 评审经验。
请根据提供的模具分析数据，生成一份专业的模具设计评审报告。
要求：
1. 使用中文
2. 按 "问题摘要 → 关键风险 → 分模方案推荐 → 制造可行性 → 修改建议" 结构组织
3. 技术术语准确（如：锁模力、投影面积、分型面、滑块、斜顶、拔模角、缩痕、熔接痕）
4. 每个建议标注优先级（高/中/低）和预计工时
5. 报告末尾给出一个总体评分（1-10分）
6. 如果数据不足以判断某项，明确标注"数据不足，需人工确认"
直接输出 Markdown 格式报告，不要输出 JSON。"""
_DESIGN_REPORT_USER = """请根据以下模具分析数据生成评审报告：
## 产品信息
- 文件：{filename}
- 材料：{material}
- 体积：{volume}
- 表面积：{surface_area}
- 边界框：{bbox}
## 检测特征
{features}
## 质量指标
{quality_metrics}
## 分模方案
{schemes}
## 制造参数
- 推荐模具材料：{mold_material}
- 推荐模具硬度：{mold_hardness}
- 预估锁模力：{clamping_force}
- 模具尺寸（长×宽×高）：{mold_size}
- 预估成型周期：{cycle_time}
- 拔模角：{draft_angle}
- 收缩率：{shrinkage_rate}
## 原始设计建议
{recommendations}"""
_PARTING_SYSTEM = """你是一位注塑模具分模专家。
根据产品几何特征和多个候选分模方向的评分数据，推荐最优分模方向。
输出要求：严格输出 JSON，不要输出其他内容。
JSON 格式：
{
"recommended_axis": "Z",
"confidence": 0.85,
"reasoning": "详细的中文推理过程...",
"risk_notes": ["风险1", "风险2"],
"rankings": [{"axis":"Z","rank":1,"score":92,"note":"..."},{"axis":"X","rank":2,"score":78,"note":"..."}]
}"""
_PARTING_USER = """请评估以下候选分模方向并推荐最优方案：
产品几何：
- 边界框 (mm)：{bbox}
- 面法向分布：{normal_stats}
- 惯性矩：{inertia}
约束条件：
- 材料：{material}
- 型腔数：{cavity_count}
- 最大锁模力 (吨)：{max_clamping_force}
- 泡沫材料：{is_foam}
候选方案：
{schemes}
请综合评估制造可行性、成本和风险，给出推荐。"""
class LLMService:
"""LLM 增强分析服务（单例）"""
def __init__(self):
self._enabled = settings.LLM_ENABLED
self._api_url = settings.LLM_API_URL.rstrip("/")
self._api_key = settings.LLM_API_KEY
self._model = settings.LLM_MODEL
self._timeout = settings.LLM_TIMEOUT
self._max_tokens = settings.LLM_MAX_TOKENS
if self._enabled:
logger.info(
"LLM 增强分析已启用: model=%s endpoint=%s",
self._model, self._api_url,
)
else:
logger.info("LLM 增强分析未启用（设置 LLM_ENABLED=true 启用）")
async def generate_design_report(
self,
analysis_result: Dict[str, Any],
detailed_cavity_json: Optional[Dict[str, Any]] = None,
) -> Optional[str]:
"""生成模具设计评审报告 (Markdown)"""
if not self._enabled:
return None
try:
prompt = self._build_design_report_prompt(analysis_result, detailed_cavity_json)
response = await self._chat(
system=_DESIGN_REPORT_SYSTEM,
user=prompt,
max_tokens=self._max_tokens,
)
if response:
logger.info("LLM 设计报告生成成功 (%d 字符)", len(response))
return response
except Exception as e:
logger.warning("LLM 设计报告生成失败（不影响主流程）: %s", e)
return None
async def recommend_parting_direction(
self,
geometry_data: Dict[str, Any],
candidate_schemes: List[Dict[str, Any]],
material: Dict[str, Any],
cavity_count: int = 1,
) -> Optional[Dict[str, Any]]:
"""推荐最优分型方向"""
if not self._enabled:
return None
try:
prompt = self._build_parting_prompt(
geometry_data, candidate_schemes, material, cavity_count,
)
response = await self._chat(
system=_PARTING_SYSTEM,
user=prompt,
max_tokens=min(self._max_tokens, 1200),
expect_json=True,
)
if response:
result = self._parse_json_response(response)
if result:
logger.info(
"LLM 分型推荐: %s (置信度 %.2f)",
result.get("recommended_axis", "?"),
result.get("confidence", 0),
)
return result
return None
except Exception as e:
logger.warning("LLM 分型推荐失败（不影响主流程）: %s", e)
return None
def _build_design_report_prompt(
self,
analysis_result: Dict[str, Any],
detailed_cavity_json: Optional[Dict[str, Any]],
) -> str:
detected_features = analysis_result.get("detected_features", [])
quality_metrics = analysis_result.get("quality_metrics", {})
recommendations = analysis_result.get("design_recommendations", [])
feature_text = json.dumps(detected_features, ensure_ascii=False, indent=2)
if len(feature_text) > 4000:
feature_text = feature_text[:4000] + "\n... (已截断)"
schemes_text = ""
if detailed_cavity_json:
schemes = detailed_cavity_json.get("candidate_schemes", [])
if schemes:
schemes_text = json.dumps(
[
{
"scheme_id": s.get("scheme_id"),
"rank": s.get("rank"),
"title": s.get("title"),
"score": s.get("score"),
"confidence_score": s.get("confidence_score"),
"summary": s.get("summary"),
"parting_axis": s.get("parting", {}).get("axis"),
"mold_structure_type": s.get("mold_structure_type"),
"dfm_violations": s.get("dfm_violations", []),
}
for s in schemes
],
ensure_ascii=False,
indent=2,
)
best_scheme = (
detailed_cavity_json.get("candidate_schemes", [{}])[0]
if detailed_cavity_json
else {}
)
cavity_data = best_scheme.get("cavity_data", {}) if isinstance(best_scheme, dict) else {}
mfg_info = cavity_data.get("manufacturing_info", {})
metadata = cavity_data.get("metadata", {})
return _DESIGN_REPORT_USER.format(
filename=metadata.get("file_name", "unknown.stp"),
material=metadata.get("selected_material", "ABS"),
volume=f"{analysis_result.get('geometry_data', {}).get('volume', 0):.1f} mm³",
surface_area=f"{analysis_result.get('geometry_data', {}).get('surface_area', 0):.1f} mm²",
bbox=json.dumps(analysis_result.get("geometry_data", {}).get("bounding_box", {}), ensure_ascii=False),
features=feature_text or "无特征检测数据",
quality_metrics=json.dumps(quality_metrics, ensure_ascii=False, indent=2),
schemes=schemes_text or "无分模方案数据",
mold_material=mfg_info.get("mold_material", "自动选择"),
mold_hardness=mfg_info.get("mold_hardness", "自动选择"),
clamping_force=mfg_info.get("estimated_clamping_force", "自动计算"),
mold_size=json.dumps(mfg_info.get("estimated_mold_size", {}), ensure_ascii=False),
cycle_time=mfg_info.get("estimated_cycle_time", "自动计算"),
draft_angle=f"{metadata.get('draft_angle', 2.0)}°",
shrinkage_rate=f"{metadata.get('shrinkage_rate', 0.0) * 100:.2f}%"
if isinstance(metadata.get("shrinkage_rate"), (int, float))
else "自动计算",
recommendations=json.dumps(recommendations, ensure_ascii=False, indent=2) if recommendations else "无",
)
def _build_parting_prompt(
self,
geometry_data: Dict[str, Any],
candidate_schemes: List[Dict[str, Any]],
material: Dict[str, Any],
cavity_count: int,
) -> str:
bbox = geometry_data.get("bounding_box", {})
axis_normal_stats = geometry_data.get("axis_normal_stats", {})
inertia = geometry_data.get("inertia_matrix", [])
inertia_diag = [
inertia[i][i] if i < len(inertia) and i < len(inertia[i]) else 0.0
for i in range(3)
]
schemes_text = json.dumps(
[
{
"axis": s.get("parting", {}).get("axis") or s.get("axis"),
"score": s.get("score"),
"confidence_score": s.get("confidence_score"),
"summary": s.get("summary"),
"mold_structure_type": s.get("mold_structure_type"),
"core_required": s.get("core_required"),
"dfm_violations": s.get("dfm_violations", []),
"undercut_regions_count": len(s.get("undercut_regions", [])),
"score_breakdown": s.get("score_breakdown", {}),
}
for s in candidate_schemes
],
ensure_ascii=False,
indent=2,
)
return _PARTING_USER.format(
bbox=json.dumps(bbox, ensure_ascii=False),
normal_stats=json.dumps(axis_normal_stats, ensure_ascii=False),
inertia=json.dumps(inertia_diag, ensure_ascii=False),
material=material.get("name", "ABS"),
cavity_count=cavity_count,
max_clamping_force="3000 吨（最大）",
is_foam="是" if material.get("is_foam") else "否",
schemes=schemes_text,
)
async def _chat(
self,
system: str,
user: str,
max_tokens: int = 2000,
expect_json: bool = False,
temperature: float = 0.3,
) -> Optional[str]:
url = f"{self._api_url}/chat/completions"
headers = {
"Authorization": f"Bearer {self._api_key}",
"Content-Type": "application/json",
}
payload = {
"model": self._model,
"messages": [
{"role": "system", "content": system},
{"role": "user", "content": user},
],
"max_tokens": max_tokens,
"temperature": temperature,
}
if expect_json:
payload["response_format"] = {"type": "json_object"}
async with httpx.AsyncClient(timeout=self._timeout) as client:
resp = await client.post(url, json=payload, headers=headers)
resp.raise_for_status()
data = resp.json()
content = data["choices"][0]["message"]["content"]
return content.strip() if content else None
@staticmethod
def _parse_json_response(raw: str) -> Optional[Dict[str, Any]]:
try:
return json.loads(raw)
except json.JSONDecodeError:
match = re.search(r"\{[\s\S]*\}", raw)
if match:
try:
return json.loads(match.group())
except json.JSONDecodeError:
pass
logger.warning("LLM JSON 解析失败: %s...", raw[:200])
return None
llm_service = LLMService()
src/services/processing_service.py
-505
View File
@@ -1,505 +0,0 @@
# 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()