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

重构路由

Browse Source
This commit is contained in:
chenjw28
2026-04-13 09:43:24 +08:00
parent d4f6017427
commit 8ad85bbd90
20 changed files with 1414 additions and 921 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env
+6
View File
@@ -43,6 +43,12 @@ SECRET_KEY=your-secret-key-change-in-production-min-32-chars
ALGORITHM=HS256 ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=1440 ACCESS_TOKEN_EXPIRE_MINUTES=1440
# Redis配置
REDIS_HOST=szcjw
REDIS_PORT=6379
REDIS_PASSWORD=Qqs1996*
REDIS_DB=0
# 管理员配置 # 管理员配置
ADMIN_USERNAME=cjw ADMIN_USERNAME=cjw
ADMIN_PASSWORD=Qqs1996* ADMIN_PASSWORD=Qqs1996*
src/api/routes.py
+12 -857
View File
@@ -1,863 +1,18 @@
# api/routes.py # api/routes.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Request, Depends """路由聚合注册 — 纯 thin controller，所有业务逻辑已迁移至 services 层"""
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 fastapi import APIRouter
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__) from api.v1.health_router import router as health_router
from api.v1.upload_router import router as upload_router
from api.v1.task_router import router as task_router
from api.v1.history_router import router as history_router
from api.v1.debug_router import router as debug_router
router = APIRouter() router = APIRouter()
stp_parser = STPParser() router.include_router(health_router, tags=["health"])
geometry_analyzer = GeometryAnalyzer() router.include_router(upload_router, tags=["upload"])
file_handler = FileHandler() router.include_router(task_router, tags=["tasks"])
html_generator = HTMLGenerator() router.include_router(history_router, tags=["history"])
mold_generator = MoldCavityGenerator(shrinkage_rate=0.005) router.include_router(debug_router, tags=["debug"])
# 铝泡沫模具生成器
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())
src/api/v1/__init__.py
+1
View File
@@ -0,0 +1 @@
# api/v1/__init__.py
src/api/v1/debug_router.py
+18
View File
@@ -0,0 +1,18 @@
# api/v1/debug_router.py
from fastapi import APIRouter
from services.redis_task_manager import redis_task_manager
router = APIRouter()
@router.get("/debug/tasks")
@router.post("/debug/tasks")
async def debug_tasks():
"""调试接口：查看所有任务"""
all_tasks = await redis_task_manager.get_all_tasks()
return {
"total_tasks": len(all_tasks),
"tasks": all_tasks,
"redis_connected": redis_task_manager.is_connected
}
src/api/v1/health_router.py
+18
View File
@@ -0,0 +1,18 @@
# api/v1/health_router.py
from fastapi import APIRouter
from services.redis_task_manager import redis_task_manager
router = APIRouter()
@router.get("/health")
@router.post("/health")
async def health():
task_count = await redis_task_manager.get_task_count()
return {
"status": "healthy",
"pythonocc": True,
"total_tasks": task_count,
"redis_connected": redis_task_manager.is_connected
}
src/api/v1/history_router.py
+37
View File
@@ -0,0 +1,37 @@
# api/v1/history_router.py
from fastapi import APIRouter, Depends
import urllib.parse
from services.storage_integration_rustfs import StorageIntegrationService
from database.database import get_db_session
from sqlalchemy.ext.asyncio import AsyncSession
router = APIRouter()
@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)):
"""获取指定文件名的所有上传记录（支持多上传历史）"""
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
src/api/v1/task_router.py
+79
View File
@@ -0,0 +1,79 @@
# api/v1/task_router.py
from fastapi import APIRouter, HTTPException, Request, Depends
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from services.task_query_service import TaskQueryService
from database.database import get_db_session
from utils.logger import get_logger
from models.database import ProcessingTask, STPFile
logger = get_logger(__name__)
router = APIRouter()
@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:
task_view = await TaskQueryService.get_task_view(db_session, task_id)
if task_view is None:
raise HTTPException(404, "任务不存在")
return task_view
except HTTPException:
raise
except Exception as e:
logger.error(f"获取任务状态失败: {e}")
raise HTTPException(500, f"获取任务状态失败: {str(e)}")
@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)):
"""结果详情页面"""
# 从数据库查询任务详情
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_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"
})
src/api/v1/upload_router.py
+85
View File
@@ -0,0 +1,85 @@
# api/v1/upload_router.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Depends
from typing import Optional
import uuid
from datetime import datetime
from pathlib import Path
from models.schemas import ProcessingStatus, create_task_info
from utils.file_handler import FileHandler
from services.storage_integration_rustfs import StorageIntegrationService
from services.redis_task_manager import redis_task_manager
from database.database import get_db_session
from utils.logger import get_logger
from sqlalchemy.ext.asyncio import AsyncSession
from services.auth_service import get_current_active_user
from services.processing_service import processing_service
from models.database import User
logger = get_logger(__name__)
router = APIRouter()
file_handler = FileHandler()
@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)
# 创建任务记录（存入 Redis）
task_info = 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())
)
await redis_task_manager.set_task(task_id, task_info)
# 后台处理（使用独立数据库会话，避免请求会话关闭问题）
background_tasks.add_task(
processing_service.process_file_with_storage,
task_id, file_path, stp_file.id, 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
}
}
src/core/aluminum_foam_mold.py
+12 -11
View File
@@ -461,7 +461,7 @@ class AluminumFoamMoldGenerator:
try: try:
parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
except: except Exception:
# 回退到默认平面 # 回退到默认平面
parting_plane = gp_Pln(gp_Pnt(0, 0, center[2]), gp_Dir(0, 0, 1)) parting_plane = gp_Pln(gp_Pnt(0, 0, center[2]), gp_Dir(0, 0, 1))
parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
@@ -488,7 +488,7 @@ class AluminumFoamMoldGenerator:
"direction": [1, 0, 0], "direction": [1, 0, 0],
"reason": "产品扁平，需要垂直分型" "reason": "产品扁平，需要垂直分型"
}) })
except: except Exception:
pass pass
return { return {
@@ -585,7 +585,7 @@ class AluminumFoamMoldGenerator:
param = first_param + i * step param = first_param + i * step
point = curve.Value(param) point = curve.Value(param)
edges.append([point.X(), point.Y(), point.Z()]) edges.append([point.X(), point.Y(), point.Z()])
except: except Exception:
pass pass
explorer.Next() explorer.Next()
@@ -615,8 +615,9 @@ class AluminumFoamMoldGenerator:
[xmin, ymax, center_z], [xmin, ymax, center_z],
[xmin, ymin, center_z] [xmin, ymin, center_z]
] ]
except: except Exception:
return [[0, 0, 0], [100, 0, 0], [100, 100, 0], [0, 100, 0], [0, 0, 0]] logger.warning("分型线简化计算失败，返回空列表")
return []
def _smooth_parting_line(self, parting_line: List[List[float]]) -> List[List[float]]: def _smooth_parting_line(self, parting_line: List[List[float]]) -> List[List[float]]:
""" """
@@ -678,7 +679,7 @@ class AluminumFoamMoldGenerator:
return 50.0 return 50.0
except: except Exception:
return 50.0 return 50.0
def _apply_shrinkage_compensation(self, shape: Any) -> Any: def _apply_shrinkage_compensation(self, shape: Any) -> Any:
@@ -850,7 +851,7 @@ class AluminumFoamMoldGenerator:
"origin": [0, 0, 0], "origin": [0, 0, 0],
"bounds": {"u_range": [-200, 200], "v_range": [-200, 200]} "bounds": {"u_range": [-200, 200], "v_range": [-200, 200]}
} }
except: except Exception:
return { return {
"type": "plane", "type": "plane",
"normal": [0, 0, 1], "normal": [0, 0, 1],
@@ -870,7 +871,7 @@ class AluminumFoamMoldGenerator:
try: try:
parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
except: except Exception:
parting_plane = gp_Pln(gp_Pnt(0, 0, 0), gp_Dir(0, 0, 1)) parting_plane = gp_Pln(gp_Pnt(0, 0, 0), gp_Dir(0, 0, 1))
parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face() parting_surface = BRepBuilderAPI_MakeFace(parting_plane).Face()
@@ -930,7 +931,7 @@ class AluminumFoamMoldGenerator:
avg_thickness = (volume / surface_area) * 0.6 avg_thickness = (volume / surface_area) * 0.6
return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm" return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm"
return "10.0 - 30.0 mm (铝泡沫典型)" return "无法估算 (缺少几何数据)"
def _calculate_complexity_score(self, analysis: Dict) -> float: def _calculate_complexity_score(self, analysis: Dict) -> float:
"""计算复杂度评分""" """计算复杂度评分"""
@@ -942,7 +943,7 @@ class AluminumFoamMoldGenerator:
complexity = min(thickness_ratio / 5.0, 10.0) complexity = min(thickness_ratio / 5.0, 10.0)
return round(complexity, 1) return round(complexity, 1)
return 5.0 return 0.0
def _estimate_cycle_time(self, analysis: Dict) -> str: def _estimate_cycle_time(self, analysis: Dict) -> str:
"""估算成型周期""" """估算成型周期"""
@@ -963,7 +964,7 @@ class AluminumFoamMoldGenerator:
def _assess_warpage_risk(self, analysis: Dict) -> str: def _assess_warpage_risk(self, analysis: Dict) -> str:
"""评估翘曲风险""" """评估翘曲风险"""
bbox = analysis.get("bounding_box", {}).get("dimensions", [1, 1, 1]) bbox = analysis.get("bounding_box", {}).get("dimensions", [0, 0, 0])
aspect_ratio = max(bbox) / min(bbox) aspect_ratio = max(bbox) / min(bbox)
if aspect_ratio > 5: if aspect_ratio > 5:
src/core/geometry_analyzer.py
+6 -6
View File
@@ -135,7 +135,7 @@ class GeometryAnalyzer:
elif volume > 0: elif volume > 0:
# 如果没有surface_area，基于边界框估算壁厚 # 如果没有surface_area，基于边界框估算壁厚
bbox = geometry_data.get("bounding_box", {}) bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100]) dimensions = bbox.get("dimensions", [0, 0, 0])
bbox_volume = dimensions[0] * dimensions[1] * dimensions[2] bbox_volume = dimensions[0] * dimensions[1] * dimensions[2]
if bbox_volume > 0: if bbox_volume > 0:
volume_efficiency = volume / bbox_volume volume_efficiency = volume / bbox_volume
@@ -144,7 +144,7 @@ class GeometryAnalyzer:
features.append(create_mold_feature( features.append(create_mold_feature(
feature_type="thin_wall", feature_type="thin_wall",
confidence=0.7, confidence=0.7,
location=bbox.get("center", [50, 50, 50]), location=bbox.get("center", [0, 0, 0]),
dimensions=[avg_thickness, avg_thickness, avg_thickness], dimensions=[avg_thickness, avg_thickness, avg_thickness],
parameters={"average_thickness": avg_thickness, "estimation_method": "bbox_based"}, parameters={"average_thickness": avg_thickness, "estimation_method": "bbox_based"},
recommendations=[ recommendations=[
@@ -187,14 +187,14 @@ class GeometryAnalyzer:
volume = geometry_data.get("volume", 0) volume = geometry_data.get("volume", 0)
bbox = geometry_data.get("bounding_box", {}) bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100]) dimensions = bbox.get("dimensions", [0, 0, 0])
volume_efficiency = volume / (dimensions[0] * dimensions[1] * dimensions[2]) volume_efficiency = volume / (dimensions[0] * dimensions[1] * dimensions[2]) if all(d > 0 for d in dimensions) else 0
if volume_efficiency < 0.3: if volume_efficiency < 0.3:
features.append(create_mold_feature( features.append(create_mold_feature(
feature_type="boss_feature", feature_type="boss_feature",
confidence=0.65, confidence=0.65,
location=bbox.get("center", [50, 50, 50]), location=bbox.get("center", [0, 0, 0]),
dimensions=[6.0, 12.0, 6.0], dimensions=[6.0, 12.0, 6.0],
parameters={"volume_efficiency": volume_efficiency}, parameters={"volume_efficiency": volume_efficiency},
recommendations=[ recommendations=[
@@ -293,7 +293,7 @@ class GeometryAnalyzer:
# 体积利用率 # 体积利用率
bbox = geometry_data.get("bounding_box", {}) bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100]) dimensions = bbox.get("dimensions", [0, 0, 0])
volume = geometry_data.get("volume", 0) volume = geometry_data.get("volume", 0)
bbox_volume = dimensions[0] * dimensions[1] * dimensions[2] bbox_volume = dimensions[0] * dimensions[1] * dimensions[2]
src/core/mold_generator.py
+17 -15
View File
@@ -471,13 +471,6 @@ class MoldCavityGenerator:
"bounds": bounds "bounds": bounds
} }
return {
"type": "plane",
"normal": [0, 0, 1],
"origin": [0, 0, 0],
"bounds": bounds
}
def _calculate_mold_size(self, analysis: Dict) -> Dict[str, float]: def _calculate_mold_size(self, analysis: Dict) -> Dict[str, float]:
"""估算模具尺寸""" """估算模具尺寸"""
product_bbox = analysis["bounding_box"]["dimensions"] product_bbox = analysis["bounding_box"]["dimensions"]
@@ -531,14 +524,14 @@ class MoldCavityGenerator:
return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm" return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm"
elif volume > 0: elif volume > 0:
# 如果没有surface_area，基于体积估算 # 如果没有surface_area，基于体积估算
bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [1, 1, 1]) bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [0, 0, 0])
bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2] bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2]
if bbox_volume > 0: if bbox_volume > 0:
efficiency = volume / bbox_volume efficiency = volume / bbox_volume
avg_thickness = (bbox_dims[0] + bbox_dims[1]) / 2 * efficiency avg_thickness = (bbox_dims[0] + bbox_dims[1]) / 2 * efficiency
return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm" return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm"
return "2.0 - 4.0 mm (默认)" return "无法估算 (缺少几何数据)"
def _calculate_complexity_score(self, analysis: Dict) -> float: def _calculate_complexity_score(self, analysis: Dict) -> float:
"""计算复杂度评分（0-10）""" """计算复杂度评分（0-10）"""
@@ -552,14 +545,14 @@ class MoldCavityGenerator:
return round(complexity, 1) return round(complexity, 1)
elif volume > 0: elif volume > 0:
# 如果没有surface_area，基于拓扑复杂度评分 # 如果没有surface_area，基于拓扑复杂度评分
bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [100, 100, 100]) bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [0, 0, 0])
bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2] bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2]
if bbox_volume > 0: if bbox_volume > 0:
volume_ratio = volume / bbox_volume volume_ratio = volume / bbox_volume
complexity = (1.0 - volume_ratio) * 10 complexity = (1.0 - volume_ratio) * 10
return round(min(max(complexity, 0), 10), 1) return round(min(max(complexity, 0), 10), 1)
return 5.0 return 0.0
def _estimate_cycle_time(self, analysis: Dict) -> str: def _estimate_cycle_time(self, analysis: Dict) -> str:
"""估算成型周期""" """估算成型周期"""
@@ -594,7 +587,7 @@ class MoldCavityGenerator:
def _assess_warpage_risk(self, analysis: Dict) -> str: def _assess_warpage_risk(self, analysis: Dict) -> str:
"""评估翘曲风险""" """评估翘曲风险"""
bbox = analysis.get("bounding_box", {}).get("dimensions", [1, 1, 1]) bbox = analysis.get("bounding_box", {}).get("dimensions", [0, 0, 0])
aspect_ratio = max(bbox) / min(bbox) aspect_ratio = max(bbox) / min(bbox)
if aspect_ratio > 5: if aspect_ratio > 5:
@@ -715,6 +708,15 @@ class MoldCavityGenerator:
使用 BRepAlgoAPI_Section 进行布尔运算求交 使用 BRepAlgoAPI_Section 进行布尔运算求交
""" """
# 获取实际边界框作为回退
try:
bbox_fallback = Bnd_Box()
brepbndlib.Add(shape, bbox_fallback)
xmin, ymin, zmin, xmax, ymax, zmax = bbox_fallback.Get()
fallback_bbox = {"bounding_box": {"min": [xmin, ymin, zmin], "max": [xmax, ymax, zmax], "center": [(xmin+xmax)/2, (ymin+ymax)/2, (zmin+zmax)/2]}}
except Exception:
fallback_bbox = {"bounding_box": {"min": [0, 0, 0], "max": [0, 0, 0], "center": [0, 0, 0]}}
try: try:
# 创建截面运算 # 创建截面运算
section = BRepAlgoAPI_Section(shape, parting_surface) section = BRepAlgoAPI_Section(shape, parting_surface)
@@ -724,7 +726,7 @@ class MoldCavityGenerator:
logger.warning("截面运算未完成，使用简化分型线") logger.warning("截面运算未完成，使用简化分型线")
return self._simple_parting_line( return self._simple_parting_line(
parting_surface, parting_surface,
{"bounding_box": {"min": [-50, -50, 0], "max": [50, 50, 100]}} fallback_bbox
) )
# 提取交线（边） # 提取交线（边）
@@ -755,7 +757,7 @@ class MoldCavityGenerator:
logger.warning("未找到交线，使用简化分型线") logger.warning("未找到交线，使用简化分型线")
return self._simple_parting_line( return self._simple_parting_line(
parting_surface, parting_surface,
{"bounding_box": {"min": [-50, -50, 0], "max": [50, 50, 100]}} fallback_bbox
) )
logger.info(f"计算得到 {len(edges)} 个分型线点") logger.info(f"计算得到 {len(edges)} 个分型线点")
@@ -765,7 +767,7 @@ class MoldCavityGenerator:
logger.error(f"分型线计算失败：{e}") logger.error(f"分型线计算失败：{e}")
return self._simple_parting_line( return self._simple_parting_line(
parting_surface, parting_surface,
{"bounding_box": {"min": [-50, -50, 0], "max": [50, 50, 100]}} fallback_bbox
) )
def _simple_parting_surface(self, analysis: Dict) -> Tuple[Any, List]: def _simple_parting_surface(self, analysis: Dict) -> Tuple[Any, List]:
src/core/stp_parser.py
+26 -30
View File
@@ -133,10 +133,13 @@ class STPParser:
props = GProp_GProps() props = GProp_GProps()
brepgprop.VolumeProperties(shape, props) brepgprop.VolumeProperties(shape, props)
return props.Mass() volume = props.Mass()
if volume <= 0:
raise ValueError("计算得到的体积为0或负数，形状可能无效")
return volume
except Exception as e: except Exception as e:
logger.error(f"体积计算失败: {e}") logger.error(f"体积计算失败: {e}")
return 1000000.0 raise RuntimeError(f"体积计算失败: {e}") from e
def _compute_surface_area(self, shape) -> float: def _compute_surface_area(self, shape) -> float:
"""计算表面积""" """计算表面积"""
@@ -147,7 +150,6 @@ class STPParser:
props = GProp_GProps() props = GProp_GProps()
brepgprop.SurfaceProperties(shape, props) brepgprop.SurfaceProperties(shape, props)
area = props.Mass() area = props.Mass()
logger.info(f"表面积计算成功: {area:.2f} mm²")
# 如果计算结果为0，使用备选估算方法 # 如果计算结果为0，使用备选估算方法
if area <= 0: if area <= 0:
@@ -155,18 +157,23 @@ class STPParser:
raise ValueError("Surface area is zero") raise ValueError("Surface area is zero")
return area return area
except Exception as e: except ValueError:
logger.error(f"表面积计算失败: {e}")
# 基于边界框估算表面积 # 基于边界框估算表面积
try: try:
bbox = self._compute_bounding_box(shape) bbox = self._compute_bounding_box(shape)
dims = bbox.get("dimensions", [100, 100, 100]) dims = bbox.get("dimensions", [0, 0, 0])
if any(d <= 0 for d in dims):
raise RuntimeError("边界框尺寸无效，无法估算表面积")
# 简化的估算公式：2*(lw + lh + wh) # 简化的估算公式：2*(lw + lh + wh)
estimated_area = 2 * (dims[0]*dims[1] + dims[0]*dims[2] + dims[1]*dims[2]) estimated_area = 2 * (dims[0]*dims[1] + dims[0]*dims[2] + dims[1]*dims[2])
logger.warning(f"使用边界框估算表面积: {estimated_area:.2f} mm²") logger.warning(f"使用边界框估算表面积: {estimated_area:.2f} mm²")
return estimated_area return estimated_area
except: except Exception as e:
return 60000.0 logger.error(f"表面积估算失败: {e}")
raise RuntimeError(f"表面积计算失败: {e}") from e
except Exception as e:
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) -> List[float]:
"""计算质心""" """计算质心"""
@@ -180,7 +187,12 @@ class STPParser:
return [float(center.X()), float(center.Y()), float(center.Z())] return [float(center.X()), float(center.Y()), float(center.Z())]
except Exception as e: except Exception as e:
logger.error(f"质心计算失败: {e}") logger.error(f"质心计算失败: {e}")
return [0.0, 0.0, 0.0] # 回退到边界框中心
try:
bbox = self._compute_bounding_box(shape)
return bbox.get("center", [0.0, 0.0, 0.0])
except Exception:
raise RuntimeError(f"质心计算失败且边界框回退也失败: {e}") from e
def _compute_inertia_properties(self, shape) -> Dict[str, Any]: def _compute_inertia_properties(self, shape) -> Dict[str, Any]:
"""计算惯性属性""" """计算惯性属性"""
@@ -227,30 +239,14 @@ class STPParser:
logger.error(f"拓扑分析失败: {e}") logger.error(f"拓扑分析失败: {e}")
raise raise
def _create_dummy_shape(self):
"""创建虚拟形状"""
return "dummy_shape"
def _simulate_analysis(self) -> Dict[str, Any]:
"""模拟分析结果"""
logger.info("使用模拟分析数据")
return {
"bounding_box": self._default_bounding_box(),
"volume": 1000000.0,
"surface_area": 60000.0,
"topology": {"faces": 6, "edges": 12, "vertices": 8},
"center_of_mass": [50.0, 50.0, 50.0],
"inertia_properties": {},
"analysis_method": "simulated"
}
def _default_bounding_box(self) -> Dict[str, Any]: def _default_bounding_box(self) -> Dict[str, Any]:
"""默认边界框""" """默认边界框（边界框计算失败时的回退值，标注为估算）"""
return { return {
"min": [0.0, 0.0, 0.0], "min": [0.0, 0.0, 0.0],
"max": [100.0, 100.0, 100.0], "max": [0.0, 0.0, 0.0],
"dimensions": [100.0, 100.0, 100.0], "dimensions": [0.0, 0.0, 0.0],
"center": [50.0, 50.0, 50.0] "center": [0.0, 0.0, 0.0],
"estimated": True
} }
def export_to_json(self, geometry_data: Dict[str, Any], output_path: Path) -> str: def export_to_json(self, geometry_data: Dict[str, Any], output_path: Path) -> str:
src/main.py
+22
View File
@@ -79,6 +79,28 @@ async def startup_event():
print(f"[FAIL] RustFS连接失败: {e}") print(f"[FAIL] RustFS连接失败: {e}")
print("[WARN] 文件上传功能将不可用，但其他功能正常") print("[WARN] 文件上传功能将不可用，但其他功能正常")
# 初始化Redis连接
try:
from services.redis_task_manager import redis_task_manager
await redis_task_manager.connect()
if redis_task_manager.is_connected:
print("[OK] Redis连接成功")
else:
print("[WARN] Redis连接失败，任务状态将使用内存回退")
except Exception as e:
print(f"[WARN] Redis初始化异常: {e}，任务状态将使用内存回退")
@app.on_event("shutdown")
async def shutdown_event():
"""应用关闭时清理资源"""
try:
from services.redis_task_manager import redis_task_manager
await redis_task_manager.disconnect()
print("[OK] Redis连接已断开")
except Exception as e:
print(f"[WARN] Redis断开异常: {e}")
# 创建必要目录 # 创建必要目录
UPLOAD_DIR = Path("uploads") UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True) UPLOAD_DIR.mkdir(exist_ok=True)
src/services/calculation_service.py
+240
View File
@@ -0,0 +1,240 @@
# services/calculation_service.py
"""模具工程参数计算服务 — 从 process_file_core 中抽取的纯计算逻辑"""
from typing import Dict, Any, List, Optional
from datetime import datetime
from pathlib import Path
class CalculationService:
"""将 process_file_core 中的工程计算逻辑抽取为独立服务，方便单测和复用"""
# ─── 基础计算 ───
@staticmethod
def calculate_product_weight(volume_mm3: float, density: float) -> float:
"""计算产品重量（克）"""
volume_cm3 = volume_mm3 / 1000
return volume_cm3 * density
@staticmethod
def calculate_projected_area(bbox_dims: List[float]) -> float:
"""计算投影面积（cm²），取 X、Y 方向"""
if len(bbox_dims) >= 2:
return (bbox_dims[0] * bbox_dims[1]) / 100
return 0.0
@staticmethod
def calculate_cavity_count(product_weight_g: float, projected_area_cm2: float) -> int:
"""
计算最优型腔数量
基于产品重量和投影面积：
- 小产品（< 50g）可以多型腔
- 大产品（> 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
else:
cavity_count = 1
# 根据投影面积调整
if projected_area_cm2 > 400:
cavity_count = 1
elif projected_area_cm2 > 200 and cavity_count > 2:
cavity_count = 2
return cavity_count
@staticmethod
def calculate_clamping_force(
projected_area_cm2: float,
cavity_count: int,
runner_ratio: float = 0.20,
injection_pressure: float = 700,
) -> int:
"""
计算所需夹紧力（吨）
runner_ratio: 流道系统占型腔投影面积比（0.15-0.25）
injection_pressure: 注塑压力 kg/cm²
"""
total_projected_area = projected_area_cm2 * cavity_count * (1 + runner_ratio)
clamping_force_ton = int(total_projected_area * injection_pressure / 1000)
return max(50, min(clamping_force_ton, 3000))
@staticmethod
def calculate_wall_thickness(volume_mm3: float, surface_area_mm2: float) -> Dict[str, float]:
"""计算壁厚范围"""
if surface_area_mm2 > 0 and volume_mm3 > 0:
avg = (volume_mm3 / surface_area_mm2) * 0.6
return {
"avg_thickness_mm": avg,
"wall_thickness_min": avg * 0.7,
"wall_thickness_max": avg * 1.3,
}
return {
"avg_thickness_mm": 2.5,
"wall_thickness_min": 2.0,
"wall_thickness_max": 3.0,
}
@staticmethod
def calculate_complexity(avg_thickness_mm: float) -> float:
"""计算复杂度评分（0~1）"""
return min((avg_thickness_mm / 5.0), 1.0) if avg_thickness_mm > 0 else 0.5
@staticmethod
def calculate_mold_size(
bbox_dims: List[float], cavity_count: int,
cavity_spacing: float = 30, edge_margin: float = 50,
) -> Dict[str, float]:
"""计算模具尺寸（长×宽×高），单位 mm"""
dim_x = max(bbox_dims[0] if len(bbox_dims) > 0 else 120, 120)
dim_y = max(bbox_dims[1] if len(bbox_dims) > 1 else 100, 100)
dim_z = max(bbox_dims[2] if len(bbox_dims) > 2 else 60, 60)
if cavity_count == 1:
length = dim_x + 2 * edge_margin
width = dim_y + 2 * edge_margin
elif cavity_count == 2:
length = 2 * dim_x + cavity_spacing + 2 * edge_margin
width = dim_y + 2 * edge_margin
elif cavity_count == 4:
length = 2 * dim_x + cavity_spacing + 2 * edge_margin
width = 2 * dim_y + cavity_spacing + 2 * edge_margin
else: # 8 型腔: 2x4
length = 4 * dim_x + 3 * cavity_spacing + 2 * edge_margin
width = 2 * dim_y + cavity_spacing + 2 * edge_margin
height = dim_z + 80 # 包含冷却系统
return {"length": length, "width": width, "height": height}
@staticmethod
def calculate_parting_line_length(bbox_dims: List[float], cavity_count: int) -> float:
"""计算分型线长度（mm）"""
if len(bbox_dims) >= 2:
return 2 * (bbox_dims[0] + bbox_dims[1]) * cavity_count
return 0.0
@staticmethod
def calculate_cycle_time(
wall_thickness_max: float, volume_cm3: float, cavity_count: int,
) -> int:
"""
估算成型周期（秒）
周期 = 冷却时间 + 注塑时间 + 顶出时间 + 开合模时间
"""
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
# 多型腔需要更长冷却时间
if cavity_count > 1:
cycle_time = cycle_time * (1 + 0.1 * (cavity_count - 1))
return int(cycle_time)
# ─── 组装方法 ───
@classmethod
def build_detailed_cavity_json(
cls,
geometry_data: Dict[str, Any],
material: Dict[str, Any],
file_path: str,
cavity_mesh_data: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""
组装完整的 detailed_cavity_json（整合以上所有计算结果）
Args:
geometry_data: STP 解析得到的几何数据
material: MaterialService.get_material() 返回的材料属性字典
file_path: STP 文件路径
cavity_mesh_data: 型腔网格数据（可选）
"""
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 = material["density"]
shrinkage_rate = material["shrinkage"]
# 各项计算
volume_cm3 = volume_mm3 / 1000
product_weight_g = cls.calculate_product_weight(volume_mm3, material_density)
projected_area_cm2 = cls.calculate_projected_area(bbox_dims)
cavity_count = cls.calculate_cavity_count(product_weight_g, projected_area_cm2)
clamping_force_ton = cls.calculate_clamping_force(projected_area_cm2, cavity_count)
wall = cls.calculate_wall_thickness(volume_mm3, surface_area_mm2)
complexity_score = cls.calculate_complexity(wall["avg_thickness_mm"])
mold_size = cls.calculate_mold_size(bbox_dims, cavity_count)
parting_line_length = cls.calculate_parting_line_length(bbox_dims, cavity_count)
cycle_time = cls.calculate_cycle_time(wall["wall_thickness_max"], volume_cm3, cavity_count)
injection_pressure = 700 # kg/cm²
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": material["name"],
},
"product_analysis": {
"volume": volume_mm3,
"surface_area": surface_area_mm2,
"bounding_box": bbox,
},
"manufacturing_info": {
"recommended_material": material["name"],
"material_density": f"{material_density} g/cm³",
"estimated_clamping_force": f"{clamping_force_ton} 吨",
"estimated_mold_size": {
"length": int(mold_size["length"]),
"width": int(mold_size["width"]),
"height": int(mold_size["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"{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"]
for key in ("cavity", "core", "parting_surface"):
if key in mold_cavities:
detailed_cavity_json["mold_cavities"][key] = mold_cavities[key]
# 添加型腔关键信息
detailed_cavity_json["mold_cavities"]["cavity_key_info"] = {
"geometric_characteristics": {
"product_weight": f"{product_weight_g:.2f} g",
"wall_thickness_range": f"{wall['wall_thickness_min']:.2f} - {wall['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["wall_thickness_max"] > 4 else "minimal",
"warpage_risk": "medium" if wall["wall_thickness_max"] > 5 else "low",
},
}
return detailed_cavity_json
src/services/material_service.py
+45
View File
@@ -0,0 +1,45 @@
# services/material_service.py
"""材料属性管理服务"""
from typing import Dict, Any, List, Optional
MATERIAL_PROPERTIES: Dict[str, Dict[str, Any]] = {
"ABS": {"density": 1.05, "shrinkage": 0.005, "name": "ABS", "is_foam": False},
"PP": {"density": 0.90, "shrinkage": 0.016, "name": "PP", "is_foam": False},
"PE": {"density": 0.95, "shrinkage": 0.020, "name": "PE", "is_foam": False},
"PC": {"density": 1.20, "shrinkage": 0.007, "name": "PC", "is_foam": False},
"PA": {"density": 1.14, "shrinkage": 0.010, "name": "PA", "is_foam": False},
"POM": {"density": 1.41, "shrinkage": 0.020, "name": "POM", "is_foam": False},
"PMMA": {"density": 1.18, "shrinkage": 0.005, "name": "PMMA", "is_foam": False},
"PBT": {"density": 1.31, "shrinkage": 0.015, "name": "PBT", "is_foam": False},
"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},
}
# 默认回退材料
_DEFAULT_MATERIAL = MATERIAL_PROPERTIES["ABS"]
class MaterialService:
"""材料属性管理服务 — 集中管理材料字典，便于扩展和单测"""
@staticmethod
def get_material(material_name: str) -> Dict[str, Any]:
"""获取材料属性，不存在则回退到 ABS"""
return MATERIAL_PROPERTIES.get(material_name, _DEFAULT_MATERIAL)
@staticmethod
def is_foam_material(material_name: str) -> bool:
return MATERIAL_PROPERTIES.get(material_name, _DEFAULT_MATERIAL).get("is_foam", False)
@staticmethod
def list_all_materials() -> List[str]:
return list(MATERIAL_PROPERTIES.keys())
@staticmethod
def resolve_material(requested: str) -> str:
"""解析请求的材料名，若不在字典中则回退为 ABS"""
return requested if requested in MATERIAL_PROPERTIES else "ABS"
src/services/processing_service.py
+437
View File
@@ -0,0 +1,437 @@
# 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 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()
# ─── 对外入口 ───
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)
cavity_mesh_data = 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_detailed_cavity_json(
geometry_data=geometry_data,
material=selected_material,
file_path=str(file_path),
cavity_mesh_data=cavity_mesh_data,
)
if cavity_mesh_data and "mold_cavities" in cavity_mesh_data:
mold_cavities = cavity_mesh_data["mold_cavities"]
logger.info(f"型腔网格数据已合并: cavity {mold_cavities.get('cavity', {}).get('vertex_count', 0)} 顶点")
# 5. 生成关键信息
cavity_key_info = detailed_cavity_json["mold_cavities"]["cavity_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. 保存模具型腔数据
await self.storage_service.save_mold_cavity_data(
db_session, stp_file_id, detailed_cavity_json
)
# 8. 生成HTML可视化
await self.storage_service.update_task_status(
db_session, task_id, "processing", 85, "生成可视化报告"
)
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 = self.html_generator.generate_and_save_visualization(
geometry_data,
Path(file_path).name,
cavity_data=detailed_cavity_json,
pointcloud_data=pointcloud_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,
"cavity_data": detailed_cavity_json,
"key_info": detailed_cavity_json,
"html_file": 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]]:
"""生成模具型腔数据"""
cavity_mesh_data = None
try:
if shape:
if is_foam_material:
self.aluminum_foam_generator.set_material(selected_material["name"])
cavity_result = self.aluminum_foam_generator.generate_mold_cavities(shape)
cavity_mesh_data = self.aluminum_foam_generator.generate_detailed_cavity_json(cavity_result)
logger.info(f"使用铝泡沫模具生成器: {selected_material['name']}")
else:
cavity_result = self.mold_generator.generate_mold_cavities(shape)
cavity_mesh_data = self.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}")
traceback.print_exc()
cavity_mesh_data = None
return cavity_mesh_data
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 _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()
src/services/redis_task_manager.py
+224
View File
@@ -0,0 +1,224 @@
# services/redis_task_manager.py
"""Redis 任务管理器 - 替代内存字典，支持 TTL 自动清理"""
import json
import os
from typing import Dict, Any, Optional
from datetime import datetime
import redis.asyncio as aioredis
from utils.logger import get_logger
logger = get_logger(__name__)
class RedisTaskManager:
"""基于 Redis 的任务状态管理"""
_instance: Optional["RedisTaskManager"] = None
def __init__(self):
self._redis: Optional[aioredis.Redis] = None
self._prefix = "moldinsight:task:"
self._ttl = 86400 * 7 # 任务默认保留 7 天
self._connected = False
@classmethod
def get_instance(cls) -> "RedisTaskManager":
if cls._instance is None:
cls._instance = RedisTaskManager()
return cls._instance
async def connect(self):
"""连接 Redis"""
if self._connected and self._redis:
return
host = os.getenv("REDIS_HOST", "szcjw")
port = int(os.getenv("REDIS_PORT", "6379"))
password = os.getenv("REDIS_PASSWORD", "")
db = int(os.getenv("REDIS_DB", "0"))
try:
self._redis = aioredis.Redis(
host=host,
port=port,
password=password if password else None,
db=db,
decode_responses=True,
socket_connect_timeout=5,
socket_timeout=5,
retry_on_timeout=True,
)
# 测试连接
await self._redis.ping()
self._connected = True
logger.info(f"Redis 连接成功: {host}:{port}")
except Exception as e:
logger.error(f"Redis 连接失败: {e}，任务状态将使用内存回退")
self._redis = None
self._connected = False
async def disconnect(self):
"""断开 Redis 连接"""
if self._redis:
await self._redis.aclose()
self._redis = None
self._connected = False
logger.info("Redis 连接已断开")
@property
def is_connected(self) -> bool:
return self._connected and self._redis is not None
# ---- 内存回退 ----
_fallback_tasks: Dict[str, Dict[str, Any]] = {}
def _fallback_set(self, task_id: str, data: Dict[str, Any]):
self._fallback_tasks[task_id] = data
def _fallback_get(self, task_id: str) -> Optional[Dict[str, Any]]:
return self._fallback_tasks.get(task_id)
def _fallback_delete(self, task_id: str):
self._fallback_tasks.pop(task_id, None)
def _fallback_all(self) -> Dict[str, Dict[str, Any]]:
return dict(self._fallback_tasks)
def _fallback_count(self) -> int:
return len(self._fallback_tasks)
# ---- 公共接口 ----
async def set_task(self, task_id: str, data: Dict[str, Any], ttl: Optional[int] = None):
"""设置任务数据"""
effective_ttl = ttl or self._ttl
# 确保数据可序列化
serializable = self._make_serializable(data)
if self.is_connected:
try:
key = f"{self._prefix}{task_id}"
await self._redis.setex(key, effective_ttl, json.dumps(serializable, ensure_ascii=False))
return
except Exception as e:
logger.warning(f"Redis 写入失败，回退到内存: {e}")
self._fallback_set(task_id, serializable)
async def get_task(self, task_id: str) -> Optional[Dict[str, Any]]:
"""获取任务数据"""
if self.is_connected:
try:
key = f"{self._prefix}{task_id}"
raw = await self._redis.get(key)
if raw:
return json.loads(raw)
return None
except Exception as e:
logger.warning(f"Redis 读取失败，回退到内存: {e}")
return self._fallback_get(task_id)
async def update_task(self, task_id: str, updates: Dict[str, Any]):
"""更新任务的部分字段"""
current = await self.get_task(task_id)
if current is None:
logger.warning(f"任务 {task_id} 不存在，无法更新")
return
current.update(self._make_serializable(updates))
await self.set_task(task_id, current)
async def delete_task(self, task_id: str):
"""删除任务"""
if self.is_connected:
try:
key = f"{self._prefix}{task_id}"
await self._redis.delete(key)
return
except Exception as e:
logger.warning(f"Redis 删除失败，回退到内存: {e}")
self._fallback_delete(task_id)
async def get_all_tasks(self) -> Dict[str, Dict[str, Any]]:
"""获取所有任务"""
if self.is_connected:
try:
pattern = f"{self._prefix}*"
keys = []
async for key in self._redis.scan_iter(match=pattern):
keys.append(key)
result = {}
for key in keys:
task_id = key.replace(self._prefix, "")
raw = await self._redis.get(key)
if raw:
result[task_id] = json.loads(raw)
return result
except Exception as e:
logger.warning(f"Redis 扫描失败，回退到内存: {e}")
return self._fallback_all()
async def get_task_count(self) -> int:
"""获取任务总数"""
if self.is_connected:
try:
pattern = f"{self._prefix}*"
count = 0
async for _ in self._redis.scan_iter(match=pattern):
count += 1
return count
except Exception as e:
logger.warning(f"Redis 计数失败，回退到内存: {e}")
return self._fallback_count()
async def cleanup_old_tasks(self, max_age_seconds: int = 86400 * 7):
"""清理过期任务（Redis 由 TTL 自动管理，内存回退需手动清理）"""
now = datetime.now()
to_delete = []
for task_id, task in self._fallback_tasks.items():
completed_at = task.get("completed_at")
if completed_at:
try:
completed_dt = datetime.fromisoformat(completed_at)
if (now - completed_dt).total_seconds() > max_age_seconds:
to_delete.append(task_id)
except (ValueError, TypeError):
pass
for task_id in to_delete:
del self._fallback_tasks[task_id]
if to_delete:
logger.info(f"清理了 {len(to_delete)} 个过期内存任务")
# ---- 工具方法 ----
@staticmethod
def _make_serializable(obj: Any) -> Any:
"""确保对象可 JSON 序列化"""
if isinstance(obj, dict):
return {k: RedisTaskManager._make_serializable(v) for k, v in obj.items()}
if isinstance(obj, (list, tuple)):
return [RedisTaskManager._make_serializable(v) for v in obj]
if isinstance(obj, datetime):
return obj.isoformat()
if hasattr(obj, "value"):
# Enum 类型
return obj.value
if isinstance(obj, (int, float, str, bool, type(None))):
return obj
return str(obj)
# 全局单例
redis_task_manager = RedisTaskManager.get_instance()
src/services/task_query_service.py
+127
View File
@@ -0,0 +1,127 @@
# services/task_query_service.py
"""任务状态查询服务 — 从 task_router.py 中的持久化任务组装逻辑抽取"""
from typing import Optional, Dict, Any, List
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from services.storage_integration_rustfs import StorageIntegrationService
from services.redis_task_manager import redis_task_manager
from models.database import ProcessingTask, STPFile, MeshData
from utils.logger import get_logger
logger = get_logger(__name__)
class TaskQueryService:
"""任务状态查询与视图组装"""
@staticmethod
async def get_task_view(db_session: AsyncSession, task_id: str) -> Optional[Dict[str, Any]]:
"""
获取任务视图 — 优先返回 Redis 缓存，否则从 PostgreSQL + RustFS 组装
Returns:
任务视图字典，如果任务不存在返回 None
"""
# 1. Redis/内存任务（进行中的任务）
task = await redis_task_manager.get_task(task_id)
if task:
logger.info(f"返回缓存任务状态：{task_id} - {task.get('status')}")
return task
# 2. 持久化任务（已完成/失败，或服务重启后的任务）
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:
return None
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
geometry_json = TaskQueryService._extract_geometry_json(file_with_data)
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 = await TaskQueryService._get_mesh_summary(db_session, stp_file.id)
# 构造与内存任务兼容的任务视图
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
@staticmethod
def _extract_geometry_json(file_with_data: dict) -> Optional[Dict[str, Any]]:
"""从 file_with_data 中提取 geometry_json"""
if not file_with_data.get("geometry_data"):
return None
geo_raw = file_with_data["geometry_data"]
if isinstance(geo_raw, dict):
if "geometry_data" in geo_raw:
return geo_raw["geometry_data"]
return geo_raw
return None
@staticmethod
async def _get_mesh_summary(db_session: AsyncSession, stp_file_id: int) -> Optional[Dict[str, Any]]:
"""从数据库查询网格摘要"""
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:
return {
"vertex_count": mesh_record.vertex_count,
"face_count": mesh_record.face_count,
"point_count": mesh_record.point_count,
"quality": mesh_record.quality,
}
return None
src/services/verification_service.py
+1 -1
View File
@@ -91,7 +91,7 @@ class GeometryVerificationService:
if 'org.freecad.FreeCAD' in result.stdout: if 'org.freecad.FreeCAD' in result.stdout:
cmd = ['flatpak', 'run', 'org.freecad.FreeCAD', str(self.verification_script), str(Path(stp_path_str).absolute())] cmd = ['flatpak', 'run', 'org.freecad.FreeCAD', str(self.verification_script), str(Path(stp_path_str).absolute())]
logger.info("找到 FreeCAD Flatpak 版本") logger.info("找到 FreeCAD Flatpak 版本")
except: except Exception:
pass pass
# 如果 which 找不到，尝试直接检查常见路径 # 如果 which 找不到，尝试直接检查常见路径
src/storage/rustfs_storage.py
+1 -1
View File
@@ -350,7 +350,7 @@ class RustFSManager:
try: try:
await self.get_file_info(file_type, object_key) await self.get_file_info(file_type, object_key)
return True return True
except: except Exception:
return False return False
async def get_storage_stats(self) -> Dict[str, Any]: async def get_storage_stats(self) -> Dict[str, Any]: