This commit is contained in:
cjw
2026-02-11 22:40:35 +08:00
parent 96000f3f11
commit 7b09eb3d89
1094 changed files with 242874 additions and 1 deletions
+1
View File
@@ -0,0 +1 @@
# API 模块
+327
View File
@@ -0,0 +1,327 @@
# api/routes.py
from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Request, Depends
from typing import Optional
import uuid
from datetime import datetime
from pathlib import Path
from models.schemas import ProcessingStatus, create_task_info
from core.stp_parser import STPParser
from core.geometry_analyzer import GeometryAnalyzer
from utils.file_handler import FileHandler
from utils.html_generator import HTMLGenerator
from services.storage_integration_rustfs import StorageIntegrationService
from database.database import get_db_session
from utils.logger import get_logger
from sqlalchemy.ext.asyncio import AsyncSession
from core.mold_generator import MoldCavityGenerator
logger = get_logger(__name__)
router = APIRouter()
# 服务实例
stp_parser = STPParser()
geometry_analyzer = GeometryAnalyzer()
file_handler = FileHandler()
html_generator = HTMLGenerator()
# 初始化模具生成器(可配置不同材料的收缩率)
mold_generator = MoldCavityGenerator(shrinkage_rate=0.005) # ABS材料
# 内存中的任务存储
tasks = {}
@router.get("/")
async def read_root(request: Request):
"""主页面"""
from fastapi.templating import Jinja2Templates
import os
# 使用绝对路径确保模板目录正确
templates_dir = os.path.join(os.path.dirname(os.path.dirname(os.path.dirname(__file__))), "templates")
templates = Jinja2Templates(directory=templates_dir)
return templates.TemplateResponse("index.html", {
"request": request,
"pythonocc_available": True,
"version": "3.0.0"
})
@router.get("/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(...),
db_session: AsyncSession = Depends(get_db_session)
):
"""上传STP文件并存储到数据库"""
if not file.filename.lower().endswith(('.stp', '.step')):
raise HTTPException(400, "只支持STP/STEP文件")
task_id = str(uuid.uuid4())
# 保存文件
file_path = await file_handler.save_uploaded_file(file)
content = await file.read()
# 创建存储集成服务实例
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
)
# 创建处理任务记录
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=len(content),
upload_time=str(datetime.now())
)
# 后台处理(包含数据库存储)
background_tasks.add_task(process_file_with_storage, task_id, file_path, stp_file.id, db_session)
return {
"task_id": task_id,
"status": "processing",
"message": "文件上传成功,开始处理并存储到数据库",
"file_info": {
"filename": file.filename,
"size": len(content),
"pythonocc_available": True,
"database_file_id": stp_file.id
}
}
@router.get("/status/{task_id}")
async def get_status(task_id: str):
"""获取任务状态"""
if task_id not in tasks:
raise HTTPException(404, "任务不存在")
task = tasks[task_id]
logger.info(f"返回任务状态: {task_id} - {task['status']}")
return task
@router.get("/debug/tasks")
async def debug_tasks():
"""调试接口:查看所有任务"""
return {
"total_tasks": len(tasks),
"tasks": tasks
}
async def process_file_with_storage(
task_id: str,
file_path: str,
stp_file_id: int,
db_session: AsyncSession
):
"""处理文件的后台任务"""
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), 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())
return
async def process_file_core(
storage_service: StorageIntegrationService,
task_id: str,
file_path: str,
stp_file_id: int,
db_session: AsyncSession
):
"""核心处理逻辑"""
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. 生成模具型腔(模拟数据)
await storage_service.update_task_status(
db_session, task_id, "processing", 40, "生成模具型腔(模拟)"
)
cavity_data = {
"cavity_count": 1,
"cavity_dimensions": {"length": 100, "width": 80, "height": 50},
"runner_system": "cold_runner",
"gating_type": "edge_gate"
}
# 3. 生成详细JSON数据(模拟)
await storage_service.update_task_status(
db_session, task_id, "processing", 60, "生成型腔详细数据(模拟)"
)
detailed_cavity_json = {
"metadata": {
"file_name": Path(file_path).name,
"analysis_date": datetime.now().isoformat(),
"shrinkage_rate": 0.005,
"draft_angle": 2.0
},
"product_analysis": {
"volume": geometry_data["volume"],
"surface_area": geometry_data["surface_area"],
"bounding_box": geometry_data["bounding_box"]
},
"manufacturing_info": {
"recommended_material": "ABS",
"estimated_clamping_force": "150 吨",
"estimated_mold_size": {
"length": 120,
"width": 100,
"height": 60
}
},
"mold_cavities": {
"cavity_count": 1,
"cavity_key_info": {
"geometric_characteristics": {
"product_weight": "1.2 g",
"wall_thickness_range": "1.5-3.0 mm",
"complexity_score": 0.7
},
"quality_considerations": {
"potential_weld_lines": "center",
"sink_mark_areas": "thick_sections",
"warpage_risk": "low"
}
}
}
}
# 4. 生成关键信息(模拟)
cavity_key_info = detailed_cavity_json["mold_cavities"]["cavity_key_info"]
# 5. 保存几何数据到数据库
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")
)
# 6. 保存模具型腔数据
await storage_service.save_mold_cavity_data(
db_session,
stp_file_id,
detailed_cavity_json
)
# 7. 生成HTML可视化(包含型腔信息)
await storage_service.update_task_status(
db_session, task_id, "processing", 85, "生成可视化报告"
)
html_file_path = html_generator.generate_and_save_visualization(
geometry_data,
Path(file_path).name
)
# 保存HTML文件信息
html_record = await storage_service.save_html_file(
db_session,
stp_file_id,
Path(html_file_path).name,
html_file_path
)
# 8. 分析模具设计
analysis_result = geometry_analyzer.analyze_mold_design(geometry_data)
# 9. 完成处理
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"] = cavity_key_info
tasks[task_id]["status"] = ProcessingStatus.COMPLETED
tasks[task_id]["completed_at"] = str(datetime.now())
logger.info(f"模具型腔生成完成: {task_id}")
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())
+1
View File
@@ -0,0 +1 @@
# Core 模块
+340
View File
@@ -0,0 +1,340 @@
# core/geometry_analyzer.py
from typing import Dict, List, Any
# import features # 暂时注释掉,避免导入错误
import numpy as np
from models.schemas import (
create_mold_feature,
create_design_recommendation,
create_analysis_result
)
from utils.logger import get_logger
logger = get_logger(__name__)
class GeometryAnalyzer:
"""几何分析器 - 简化版"""
def __init__(self):
self.feature_thresholds = {
"thin_wall": 2.0,
"thick_wall": 8.0,
"small_feature": 5.0,
"large_feature": 1000.0,
"high_complexity": 50,
}
self.product_materials = {
"ABS": {"shrinkage": 0.005, "min_wall": 1.2},
"PP": {"shrinkage": 0.016, "min_wall": 1.0},
"PC": {"shrinkage": 0.007, "min_wall": 1.5},
}
self.mold_materials = {
"Aluminum": {"thermal_conductivity": 200, "hardness": "HB80", "cost": "low"},
"P20_Steel": {"thermal_conductivity": 30, "hardness": "HRC30", "cost": "medium"},
"H13_Steel": {"thermal_conductivity": 25, "hardness": "HRC48", "cost": "high"}
}
def analyze_mold_design(self, geometry_data: Dict[str, Any],
product_material: str = "ABS",
mold_material: str = "Aluminum"
) -> Dict[str, Any]:
"""分析模具设计"""
logger.info("开始模具设计分析")
# 检测特征
features = self._detect_features(geometry_data)
# 使用产品材料属性
product_props = self.product_materials.get(product_material, {})
shrinkage = product_props.get("shrinkage", 0.005)
# 使用模具材料属性
mold_props = self.mold_materials.get(mold_material, {})
thermal_cond = mold_props.get("thermal_conductivity", 200)
# 生成设计建议
recommendations = self._generate_recommendations(
geometry_data, features, product_material
)
# 计算质量指标
quality_metrics = self._calculate_quality_metrics(geometry_data, features)
# 生成分析摘要
analysis_summary = self._generate_analysis_summary(geometry_data, features, recommendations)
return create_analysis_result(
geometry_data=geometry_data,
detected_features=features,
design_recommendations=recommendations,
quality_metrics=quality_metrics,
analysis_summary=analysis_summary
)
def _detect_features(self, geometry_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""检测模具特征"""
features = []
# 壁厚分析
wall_features = self._detect_wall_features(geometry_data)
features.extend(wall_features)
# 加强筋检测
rib_features = self._detect_rib_features(geometry_data)
features.extend(rib_features)
# BOSS柱检测
boss_features = self._detect_boss_features(geometry_data)
features.extend(boss_features)
# 拔模角度分析
draft_features = self._analyze_draft_angles(geometry_data)
features.extend(draft_features)
logger.info(f"检测到 {len(features)} 个特征")
return features
def _detect_wall_features(self, geometry_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""检测壁厚特征"""
features = []
volume = geometry_data.get("volume", 0)
surface_area = geometry_data.get("surface_area", 0)
if volume > 0 and surface_area > 0:
avg_thickness = (volume / surface_area) * 0.6
if avg_thickness < self.feature_thresholds["thin_wall"]:
features.append(create_mold_feature(
feature_type="thin_wall",
confidence=0.85,
location=geometry_data.get("center_of_mass", [0, 0, 0]),
dimensions=[avg_thickness, avg_thickness, avg_thickness],
parameters={"average_thickness": avg_thickness},
recommendations=[
f"平均壁厚 {avg_thickness:.2f}mm 过薄,建议增加到 {self.feature_thresholds['thin_wall']}mm 以上",
"考虑增加加强筋以提高结构强度",
"检查注塑填充是否充分"
]
))
elif avg_thickness > self.feature_thresholds["thick_wall"]:
features.append(create_mold_feature(
feature_type="thick_wall",
confidence=0.75,
location=geometry_data.get("center_of_mass", [0, 0, 0]),
dimensions=[avg_thickness, avg_thickness, avg_thickness],
parameters={"average_thickness": avg_thickness},
recommendations=[
f"平均壁厚 {avg_thickness:.2f}mm 过厚,可能产生缩痕",
"考虑减薄壁厚或增加加强筋",
"优化冷却系统设计"
]
))
elif volume > 0:
# 如果没有surface_area,基于边界框估算壁厚
bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100])
bbox_volume = dimensions[0] * dimensions[1] * dimensions[2]
if bbox_volume > 0:
volume_efficiency = volume / bbox_volume
avg_thickness = (dimensions[0] + dimensions[1]) / 2 * volume_efficiency
if avg_thickness < self.feature_thresholds["thin_wall"]:
features.append(create_mold_feature(
feature_type="thin_wall",
confidence=0.7,
location=bbox.get("center", [50, 50, 50]),
dimensions=[avg_thickness, avg_thickness, avg_thickness],
parameters={"average_thickness": avg_thickness, "estimation_method": "bbox_based"},
recommendations=[
f"估算平均壁厚 {avg_thickness:.2f}mm 过薄,建议检查表面积数据",
"考虑增加加强筋以提高结构强度"
]
))
return features
def _detect_rib_features(self, geometry_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""检测加强筋特征"""
features = []
topology = geometry_data.get("topology", {})
face_count = topology.get("faces", 0)
edge_count = topology.get("edges", 0)
complexity_ratio = edge_count / max(face_count, 1)
if complexity_ratio > 3.0:
features.append(create_mold_feature(
feature_type="rib_structure",
confidence=0.7,
location=geometry_data.get("center_of_mass", [0, 0, 0]),
dimensions=[2.0, 8.0, 2.0],
parameters={"complexity_ratio": complexity_ratio},
recommendations=[
"检测到可能的加强筋结构",
"建议加强筋厚度为壁厚的50-80%",
"加强筋高度不超过壁厚的3倍",
"加强筋根部增加圆角避免应力集中"
]
))
return features
def _detect_boss_features(self, geometry_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""检测BOSS柱特征"""
features = []
volume = geometry_data.get("volume", 0)
bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100])
volume_efficiency = volume / (dimensions[0] * dimensions[1] * dimensions[2])
if volume_efficiency < 0.3:
features.append(create_mold_feature(
feature_type="boss_feature",
confidence=0.65,
location=bbox.get("center", [50, 50, 50]),
dimensions=[6.0, 12.0, 6.0],
parameters={"volume_efficiency": volume_efficiency},
recommendations=[
"检测到可能的BOSS柱结构",
"建议BOSS柱外径为螺钉直径的2-2.5倍",
"BOSS柱高度不超过直径的2倍",
"增加拔模角度1-2度",
"根部增加圆角R0.5-R1.0"
]
))
return features
def _analyze_draft_angles(self, geometry_data: Dict[str, Any]) -> List[Dict[str, Any]]:
"""分析拔模角度"""
features = []
features.append(create_mold_feature(
feature_type="draft_angle",
confidence=0.8,
location=geometry_data.get("center_of_mass", [0, 0, 0]),
dimensions=[1.0, 2.0, 1.0],
parameters={"recommended_angle": 2.0},
recommendations=[
"建议所有垂直面添加1-2度拔模角度",
"纹理表面需要3-5度拔模角度",
"深腔结构需要更大的拔模角度"
]
))
return features
def _generate_recommendations(self, geometry_data: Dict[str, Any],
features: List[Dict[str, Any]],
material: str) -> List[Dict[str, Any]]:
"""生成设计建议"""
recommendations = []
# 壁厚建议
wall_rec = self._get_wall_thickness_recommendation(geometry_data, material)
if wall_rec:
recommendations.append(wall_rec)
# 拔模角度建议
recommendations.append(create_design_recommendation(
rec_type="draft_angle",
priority="high",
description="添加拔模角度",
parameters={"min_angle": 1.0, "preferred_angle": 2.0},
reason="确保顺利脱模"
))
# 基于检测到的特征生成建议
for feature in features:
if feature["feature_type"] == "thin_wall":
rec = create_design_recommendation(
rec_type="wall_thickness",
priority="high",
description="增加壁厚",
parameters={
"current": feature["parameters"]["average_thickness"],
"recommended": self.feature_thresholds["thin_wall"]
},
reason="壁厚不足影响结构强度"
)
recommendations.append(rec)
return recommendations
def _get_wall_thickness_recommendation(self, geometry_data: Dict[str, Any],
material: str) -> Dict[str, Any]:
"""获取壁厚建议"""
volume = geometry_data.get("volume", 0)
surface_area = geometry_data.get("surface_area", 0)
if volume > 0 and surface_area > 0:
avg_thickness = (volume / surface_area) * 0.6
material_props = self.product_materials.get(material, self.product_materials["ABS"])
min_wall = material_props["min_wall"]
if avg_thickness < min_wall:
return create_design_recommendation(
rec_type="wall_thickness",
priority="high",
description=f"增加壁厚至{min_wall}mm以上",
parameters={"current": avg_thickness, "recommended": min_wall},
reason=f"{material}材料最小壁厚要求"
)
return None
def _calculate_quality_metrics(self, geometry_data: Dict[str, Any],
features: List[Dict[str, Any]]) -> Dict[str, float]:
"""计算质量指标"""
metrics = {}
# 体积利用率
bbox = geometry_data.get("bounding_box", {})
dimensions = bbox.get("dimensions", [100, 100, 100])
volume = geometry_data.get("volume", 0)
bbox_volume = dimensions[0] * dimensions[1] * dimensions[2]
metrics["volume_utilization"] = volume / bbox_volume if bbox_volume > 0 else 0
# 拓扑复杂度
topology = geometry_data.get("topology", {})
face_count = topology.get("faces", 0)
metrics["topology_complexity"] = face_count / 100.0
# 壁厚均匀性评分
surface_area = geometry_data.get("surface_area", 0)
if volume > 0 and surface_area > 0:
thickness_ratio = (volume / surface_area) * 0.6
ideal_thickness = 3.0
metrics["wall_uniformity"] = 1.0 - abs(thickness_ratio - ideal_thickness) / ideal_thickness
elif volume > 0 and bbox_volume > 0:
# 如果没有surface_area,基于体积利用率估算
metrics["wall_uniformity"] = max(0.5, metrics["volume_utilization"])
else:
metrics["wall_uniformity"] = 0.5
return metrics
def _generate_analysis_summary(self, geometry_data: Dict[str, Any],
features: List[Dict[str, Any]],
recommendations: List[Dict[str, Any]]) -> str:
"""生成分析摘要"""
volume = geometry_data.get("volume", 0)
high_priority_recs = len([r for r in recommendations if r["priority"] == "high"])
summary_parts = []
if volume > 0:
summary_parts.append(f"模型体积: {volume / 1000:.1f} cm³")
if features:
feature_types = set(f["feature_type"] for f in features)
summary_parts.append(f"检测到 {len(feature_types)} 类特征")
if high_priority_recs > 0:
summary_parts.append(f"有 {high_priority_recs} 个高优先级建议")
return " | ".join(summary_parts) if summary_parts else "分析完成"
+102
View File
@@ -0,0 +1,102 @@
# src/core/mesh_generator.py
import logging
import numpy as np
from typing import Dict, List, Optional
import pyvista as pv
import trimesh
from OCC.Core.BRepMesh import BRepMesh_IncrementalMesh
logger = logging.getLogger(__name__)
class MeshGenerator:
"""网格生成器 - 使用PyVista和Trimesh"""
def __init__(self, quality: str = "medium"):
self.quality_settings = {
"low": 0.5,
"medium": 0.1,
"high": 0.01
}
self.quality = self.quality_settings.get(quality, 0.1)
def generate_mesh_from_shape(self, shape, num_points: int = 10000) -> Dict:
"""从形状生成网格数据"""
try:
# 方法1: 使用PythonOCC生成网格
occ_mesh = self._generate_occ_mesh(shape)
# 方法2: 转换为PyVista网格
pv_mesh = self._convert_to_pyvista(occ_mesh)
# 方法3: 转换为Trimesh网格
tri_mesh = self._convert_to_trimesh(pv_mesh)
# 生成点云
pointcloud = self._generate_pointcloud(tri_mesh, num_points)
return {
"pyvista_mesh": pv_mesh,
"trimesh_mesh": tri_mesh,
"pointcloud": pointcloud
}
except Exception as e:
logger.error(f"网格生成失败: {e}")
raise
def _generate_occ_mesh(self, shape) -> any:
"""使用PythonOCC生成网格"""
mesh = BRepMesh_IncrementalMesh(shape, self.quality)
mesh.Perform()
return mesh
def _convert_to_pyvista(self, occ_mesh) -> pv.PolyData:
"""转换为PyVista网格"""
# 这里需要从OCC网格中提取顶点和面数据
# 简化实现 - 实际需要遍历OCC网格数据结构
try:
# 创建示例网格数据
cube = pv.Cube()
return cube
except Exception as e:
logger.warning(f"PyVista转换失败,使用备用方法: {e}")
return self._create_sample_mesh()
def _convert_to_trimesh(self, pv_mesh) -> trimesh.Trimesh:
"""转换为Trimesh网格"""
try:
# 从PyVista转换
vertices = pv_mesh.points
faces = pv_mesh.faces.reshape(-1, 4)[:, 1:4] # 假设三角形网格
return trimesh.Trimesh(vertices=vertices, faces=faces)
except Exception as e:
logger.warning(f"Trimesh转换失败: {e}")
return self._create_sample_trimesh()
def _generate_pointcloud(self, mesh: trimesh.Trimesh, num_points: int) -> Dict:
"""从网格生成点云"""
try:
# 均匀采样点云
points, face_indices = trimesh.sample.sample_surface(mesh, num_points)
# 计算法向量
normals = mesh.face_normals[face_indices]
return {
"points": points.tolist(),
"normals": normals.tolist(),
"count": len(points)
}
except Exception as e:
logger.error(f"点云生成失败: {e}")
raise
def _create_sample_mesh(self) -> pv.PolyData:
"""创建示例网格(备用)"""
return pv.Cube()
def _create_sample_trimesh(self) -> trimesh.Trimesh:
"""创建示例Trimesh(备用)"""
return trimesh.creation.box([100, 80, 50])
+523
View File
@@ -0,0 +1,523 @@
# src/core/mold_generator.py
from pathlib import Path
from typing import Dict, List, Any, Tuple, Optional
import numpy as np
from OCC.Core.BRepOffsetAPI import BRepOffsetAPI_MakeThickSolid
from OCC.Core.BRepAlgoAPI import BRepAlgoAPI_Cut, BRepAlgoAPI_Fuse
from OCC.Core.BRepBuilderAPI import BRepBuilderAPI_MakeFace, BRepBuilderAPI_Transform
from OCC.Core.Geom import Geom_Plane
from OCC.Core.gp import gp_Pln, gp_Dir, gp_Pnt, gp_Vec, gp_Trsf
from OCC.Core.TopTools import TopTools_ListOfShape
from OCC.Core.TopoDS import TopoDS_Face, TopoDS_Shape
from OCC.Core.BRep import BRep_Tool
from OCC.Core.BRepMesh import BRepMesh_IncrementalMesh
from OCC.Core.BRepExtrema import BRepExtrema_DistShapeShape
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
from models.schemas import create_mold_cavity_data, create_mold_key_info
from utils.logger import get_logger
logger = get_logger(__name__)
class MoldCavityGenerator:
"""模具型腔生成器 - 基于产品模型生成Cavity和Core"""
def __init__(self, shrinkage_rate: float = 0.005, draft_angle: float = 2.0):
"""
初始化模具生成器
Args:
shrinkage_rate: 收缩率(默认0.5% for ABS)
draft_angle: 拔模角(默认2度)
"""
self.shrinkage_rate = shrinkage_rate
self.draft_angle = draft_angle # 度
# 分型面检测参数
self.parting_line_tolerance = 0.1
self.max_draft_angle = 5.0
def generate_mold_cavities(self, product_shape: Any) -> Dict[str, Any]:
"""
从产品的3D模型生成型腔和型芯
Returns:
{
"cavity": cavity_shape, # 型腔(产品外部)
"core": core_shape, # 型芯(产品内部)
"parting_surface": parting_surface, # 分型面
"parting_line": parting_line # 分型线
}
"""
logger.info("开始生成模具型腔...")
try:
# Step 1: 分析产品几何
analysis = self._analyze_product_geometry(product_shape)
# Step 2: 检测分型面和分型线
parting_surface, parting_line = self._detect_parting_surface(
product_shape, analysis
)
# Step 3: 应用收缩率补偿
scaled_shape = self._apply_shrinkage_compensation(product_shape)
# Step 4: 添加拔模角
drafted_shape = self._apply_draft_angles(scaled_shape, parting_surface)
# Step 5: 分离型腔和型芯
cavity, core = self._split_cavity_core(drafted_shape, parting_surface)
logger.info("模具型腔生成完成")
return {
"cavity": cavity,
"core": core,
"parting_surface": parting_surface,
"parting_line": parting_line,
"analysis": analysis
}
except Exception as e:
logger.error(f"模具型腔生成失败: {e}")
raise
def generate_detailed_cavity_json(self, cavity_data: Dict) -> Dict[str, Any]:
"""
生成详细的型腔三维JSON数据
Returns:
包含完整几何信息的JSON结构
"""
cavity = cavity_data["cavity"]
core = cavity_data["core"]
parting_surface = cavity_data["parting_surface"]
analysis = cavity_data["analysis"]
# 提取型腔几何数据
cavity_geometry = self._extract_shape_geometry(cavity, "cavity")
core_geometry = self._extract_shape_geometry(core, "core")
# 提取分型面数据
parting_geometry = self._extract_parting_surface_geometry(
parting_surface
)
detailed_json = {
"metadata": {
"version": "2.0",
"generated_at": str(np.datetime64('now')),
"shrinkage_rate": self.shrinkage_rate,
"draft_angle": self.draft_angle,
"unit": "mm"
},
"product_analysis": {
"bounding_box": analysis.get("bounding_box", {}), # 使用get方法
"volume": analysis.get("volume", 0), # 使用get方法
"surface_area": analysis.get("surface_area", 0), # 使用get方法
"center_of_mass": analysis.get("center_of_mass", [0, 0, 0]) # 使用get方法
},
"mold_cavities": {
"cavity": cavity_geometry,
"core": core_geometry
},
"parting_surface": parting_geometry,
"manufacturing_info": {
"estimated_mold_size": self._calculate_mold_size(analysis),
"estimated_clamping_force": self._calculate_clamping_force(analysis),
"recommended_material": self._get_recommended_material()
}
}
return detailed_json
def generate_cavity_key_info(self, cavity_data: Dict) -> Dict[str, Any]:
"""
生成模具型腔的关键信息
Returns:
关键参数摘要
"""
analysis = cavity_data["analysis"]
key_info = {
"mold_parameters": {
"shrinkage_rate": f"{self.shrinkage_rate * 100:.2f}%",
"draft_angle": f"{self.draft_angle}°",
"parting_line_length": self._calculate_parting_line_length(
cavity_data["parting_line"]
),
"cavity_depth": analysis.get("bounding_box", {}).get("dimensions", [0, 0, 0])[2]
},
"geometric_characteristics": {
"product_volume": f"{analysis.get('volume', 0) / 1000:.2f} cm³",
"product_weight": self._calculate_product_weight(analysis),
"wall_thickness_range": self._estimate_wall_thickness(analysis),
"complexity_score": self._calculate_complexity_score(analysis)
},
"manufacturing_requirements": {
"cavity_material": "Aluminum Alloy 7075",
"hardness": "HRC 30-35",
"surface_finish": "SPI A2",
"estimated_cycle_time": self._estimate_cycle_time(analysis),
"recommended_injection_pressure": "80-120 MPa"
},
"quality_considerations": {
"potential_weld_lines": self._identify_weld_line_risk(analysis),
"sink_mark_areas": self._identify_sink_mark_risk(analysis),
"warpage_risk": self._assess_warpage_risk(analysis)
}
}
return key_info
# ==================== 内部方法 ====================
def _analyze_product_geometry(self, shape: Any) -> Dict[str, Any]:
"""分析产品几何属性"""
# 计算体积属性
volume_props = GProp_GProps()
brepgprop.VolumeProperties(shape, volume_props)
# 计算表面积属性
surface_props = GProp_GProps()
brepgprop.SurfaceProperties(shape, surface_props)
# 计算边界框
from OCC.Core.Bnd import Bnd_Box
from OCC.Core.BRepBndLib import brepbndlib
bbox = Bnd_Box()
brepbndlib.Add(shape, bbox)
xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
return {
"volume": volume_props.Mass(),
"surface_area": surface_props.Mass(),
"center_of_mass": [
volume_props.CentreOfMass().X(),
volume_props.CentreOfMass().Y(),
volume_props.CentreOfMass().Z()
],
"bounding_box": {
"min": [xmin, ymin, zmin],
"max": [xmax, ymax, zmax],
"dimensions": [xmax - xmin, ymax - ymin, zmax - zmin],
"center": [(xmin + xmax) / 2, (ymin + ymax) / 2, (zmin + zmax) / 2]
},
"inertia_matrix": self._get_inertia_matrix(volume_props)
}
def _detect_parting_surface(self, shape: Any, analysis: Dict) -> Tuple[Any, List]:
"""检测分型面和分型线"""
# 简化的分型面检测:基于Z方向的最高点和最低点
bbox = analysis["bounding_box"]
center_z = bbox["center"][2]
# 创建分型面(XY平面)
parting_plane = gp_Pln(
gp_Pnt(0, 0, center_z),
gp_Dir(0, 0, 1)
)
parting_surface = BRepBuilderAPI_MakeFace(
parting_plane,
bbox["min"][0] - 10, bbox["max"][0] + 10,
bbox["min"][1] - 10, bbox["max"][1] + 10
).Face()
# 分型线(简化)
parting_line = [
[bbox["min"][0], bbox["min"][1], center_z],
[bbox["max"][0], bbox["min"][1], center_z],
[bbox["max"][0], bbox["max"][1], center_z],
[bbox["min"][0], bbox["max"][1], center_z],
[bbox["min"][0], bbox["min"][1], center_z]
]
return parting_surface, parting_line
def _apply_shrinkage_compensation(self, shape: Any) -> Any:
"""应用收缩率补偿(放大模型)"""
scale_factor = 1.0 + self.shrinkage_rate
# 创建缩放变换
trsf = gp_Trsf()
trsf.SetScale(gp_Pnt(0, 0, 0), scale_factor)
from OCC.Core.BRepBuilderAPI import BRepBuilderAPI_Transform
scaled_shape = BRepBuilderAPI_Transform(shape, trsf, True).Shape()
return scaled_shape
def _apply_draft_angles(self, shape: Any, parting_surface: Any) -> Any:
"""添加拔模角(简化实现)"""
# 实际实现需要复杂的拔模面处理
# 这里返回原始形状(假设已在CAD中处理)
logger.warning("拔模角处理为简化实现,建议在设计阶段处理")
return shape
def _split_cavity_core(self, shape: Any, parting_surface: Any) -> Tuple[Any, Any]:
"""分离型腔和型芯"""
try:
# 使用分型面切割产品
# 上半部分为型腔(Cavity)
# 下半部分为型芯(Core)
# 这里需要实现BRepAlgoAPI_Section或类似的切割操作
# 简化:返回相同的形状(实际需实现切割逻辑)
return shape, shape # (cavity, core)
except Exception as e:
logger.error(f"型腔分离失败: {e}")
return shape, shape
def _extract_shape_geometry(self, shape: Any, shape_type: str) -> Dict[str, Any]:
"""提取形状几何数据为JSON格式"""
try:
# 网格化
mesh = BRepMesh_IncrementalMesh(shape, 0.1)
mesh.Perform()
# 提取顶点和面
from OCC.Core.TopExp import TopExp_Explorer
from OCC.Core.TopAbs import TopAbs_FACE
from OCC.Core.BRep import BRep_Tool
from OCC.Core.Poly import Poly_Triangulation
from OCC.Core.TopLoc import TopLoc_Location
vertices = []
faces = []
explorer = TopExp_Explorer(shape, TopAbs_FACE)
vertex_index = 0
while explorer.More():
# 使用 explorer.Current() 直接获取面
face = explorer.Current()
location = TopLoc_Location()
triangulation = BRep_Tool.Triangulation(face, location)
if triangulation:
# 提取顶点
nb_nodes = triangulation.NbNodes()
for i in range(1, nb_nodes + 1):
node = triangulation.Node(i)
# 应用位置变换
transformed = node.Transformed(location.Transformation())
vertices.extend([
float(transformed.X()),
float(transformed.Y()),
float(transformed.Z())
])
# 提取三角形面
nb_triangles = triangulation.NbTriangles()
for i in range(1, nb_triangles + 1):
triangle = triangulation.Triangle(i)
# 三角形顶点索引需要加上之前的顶点数量
idx1 = triangle.Value(1) + vertex_index - 1
idx2 = triangle.Value(2) + vertex_index - 1
idx3 = triangle.Value(3) + vertex_index - 1
faces.extend([int(idx1), int(idx2), int(idx3)])
vertex_index += nb_nodes
explorer.Next()
vertex_count = len(vertices) // 3
face_count = len(faces) // 3
return {
"type": shape_type,
"vertices": vertices,
"faces": faces,
"vertex_count": vertex_count,
"face_count": face_count,
"triangulation": "BRepMesh三角化"
}
except Exception as e:
logger.error(f"{shape_type}几何提取失败: {e}")
return {
"type": shape_type,
"vertices": [],
"faces": [],
"vertex_count": 0,
"face_count": 0,
"triangulation": f"提取失败: {str(e)}"
}
def _extract_parting_surface_geometry(self, surface: Any) -> Dict[str, Any]:
"""提取分型面几何数据"""
# 尝试从surface获取边界信息,失败则使用默认值
try:
from OCC.Core.BRepAdaptor import BRepAdaptor_Surface
adaptor = BRepAdaptor_Surface(surface)
u_min, u_max = adaptor.FirstUParameter(), adaptor.LastUParameter()
v_min, v_max = adaptor.FirstVParameter(), adaptor.LastVParameter()
bounds = {
"u_range": [float(u_min), float(u_max)],
"v_range": [float(v_min), float(v_max)]
}
except Exception as e:
logger.warning(f"分型面边界提取失败,使用默认值: {e}")
bounds = {
"u_range": [-200, 200],
"v_range": [-200, 200]
}
# 分型面是水平面,法向量为 [0, 0, 1],原点在 Z 轴中心
return {
"type": "plane",
"normal": [0, 0, 1],
"origin": [0, 0, 0],
"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]:
"""估算模具尺寸"""
product_bbox = analysis["bounding_box"]["dimensions"]
# 模具通常比产品大20-50mm
margin = 30 # mm
return {
"length": product_bbox[0] + 2 * margin,
"width": product_bbox[1] + 2 * margin,
"height": product_bbox[2] + 2 * margin + 100, # 增加100mm用于模架
"margin": margin
}
def _calculate_clamping_force(self, analysis: Dict) -> str:
"""估算锁模力"""
volume_cm3 = analysis.get("volume", 0) / 1000 # mm³ → cm³
# 经验公式: 锁模力 ≈ 投影面积 × 压力 × 安全系数
# 简化估算
if volume_cm3 < 10:
return "50-100 吨"
elif volume_cm3 < 100:
return "150-300 吨"
elif volume_cm3 < 500:
return "400-600 吨"
else:
return "800+ 吨"
def _get_recommended_material(self) -> str:
"""推荐模具材料 - 铝模具专用"""
return "Aluminum Alloy 7075 (铝合金模具)"
def _calculate_product_weight(self, analysis: Dict) -> str:
"""计算产品重量(泡沫材料,密度约0.1 g/cm³)"""
volume_cm3 = analysis.get("volume", 0) / 1000
weight_g = volume_cm3 * 0.1 # EPP泡沫密度约0.1 g/cm³
return f"{weight_g:.2f} g"
def _estimate_wall_thickness(self, analysis: Dict) -> str:
"""估算壁厚范围"""
volume = analysis.get("volume", 0)
surface_area = analysis.get("surface_area", 0)
if surface_area > 0 and volume > 0:
avg_thickness = (volume / surface_area) * 0.6
return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm"
elif volume > 0:
# 如果没有surface_area,基于体积估算
bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [1, 1, 1])
bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2]
if bbox_volume > 0:
efficiency = volume / bbox_volume
avg_thickness = (bbox_dims[0] + bbox_dims[1]) / 2 * efficiency
return f"{avg_thickness * 0.7:.2f} - {avg_thickness * 1.3:.2f} mm"
return "2.0 - 4.0 mm (默认)"
def _calculate_complexity_score(self, analysis: Dict) -> float:
"""计算复杂度评分(0-10)"""
# 基于体积、表面积比、边界框等
volume = analysis.get("volume", 0)
surface_area = analysis.get("surface_area", 0)
if surface_area > 0 and volume > 0:
thickness_ratio = (volume / surface_area) * 0.6
complexity = min(thickness_ratio / 5.0, 10.0)
return round(complexity, 1)
elif volume > 0:
# 如果没有surface_area,基于拓扑复杂度评分
bbox_dims = analysis.get("bounding_box", {}).get("dimensions", [100, 100, 100])
bbox_volume = bbox_dims[0] * bbox_dims[1] * bbox_dims[2]
if bbox_volume > 0:
volume_ratio = volume / bbox_volume
complexity = (1.0 - volume_ratio) * 10
return round(min(max(complexity, 0), 10), 1)
return 5.0
def _estimate_cycle_time(self, analysis: Dict) -> str:
"""估算成型周期"""
volume_cm3 = analysis.get("volume", 0) / 1000
if volume_cm3 < 10:
return "15-25 秒"
elif volume_cm3 < 50:
return "25-40 秒"
elif volume_cm3 < 200:
return "40-60 秒"
else:
return "60-90 秒"
def _identify_weld_line_risk(self, analysis: Dict) -> str:
"""识别熔接痕风险"""
# 基于几何复杂度判断
complexity = self._calculate_complexity_score(analysis)
if complexity > 7:
return "高 - 建议优化浇口位置"
elif complexity > 4:
return "中 - 需仿真验证"
else:
return "低"
def _identify_sink_mark_risk(self, analysis: Dict) -> str:
"""识别缩痕风险"""
thickness = self._estimate_wall_thickness(analysis)
# 简化的风险评估
return "中 - 建议壁厚均匀性检查"
def _assess_warpage_risk(self, analysis: Dict) -> str:
"""评估翘曲风险"""
bbox = analysis.get("bounding_box", {}).get("dimensions", [1, 1, 1])
aspect_ratio = max(bbox) / min(bbox)
if aspect_ratio > 5:
return "高 - 建议增加加强筋"
elif aspect_ratio > 3:
return "中 - 需优化冷却"
else:
return "低"
def _get_inertia_matrix(self, props: GProp_GProps) -> List[List[float]]:
"""获取惯性矩阵"""
inertia = props.MatrixOfInertia()
return [
[inertia.Value(1, 1), inertia.Value(1, 2), inertia.Value(1, 3)],
[inertia.Value(2, 1), inertia.Value(2, 2), inertia.Value(2, 3)],
[inertia.Value(3, 1), inertia.Value(3, 2), inertia.Value(3, 3)]
]
def _calculate_parting_line_length(self, parting_line: List) -> float:
"""计算分型线长度"""
# 简化的长度计算
return 250.0 # mm
+292
View File
@@ -0,0 +1,292 @@
# core/stp_parser.py
from pathlib import Path
from typing import Dict, Any, Optional, List
import numpy as np
import json
from utils.logger import get_logger
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
logger = get_logger(__name__)
class STPParser:
"""STP文件解析器"""
def __init__(self):
# 强制要求PythonOCC必须可用
self._verify_occ_availability()
def _verify_occ_availability(self):
"""验证PythonOCC是否可用,不可用则抛出异常"""
try:
from OCC.Core.STEPControl import STEPControl_Reader
from OCC.Core.IFSelect import IFSelect_RetDone
logger.info("PythonOCC验证通过")
except ImportError as e:
logger.error("PythonOCC不可用,服务无法运行")
raise RuntimeError("PythonOCC未安装,请安装PythonOCC后再运行服务") from e
def load_step_file(self, file_path: Path) -> Any:
"""加载STP文件"""
try:
from OCC.Core.STEPControl import STEPControl_Reader
from OCC.Core.IFSelect import IFSelect_RetDone
logger.info(f"加载STP文件: {file_path}")
reader = STEPControl_Reader()
status = reader.ReadFile(str(file_path))
if status == IFSelect_RetDone:
reader.TransferRoots()
shape = reader.OneShape()
logger.info("STP文件加载成功")
return shape
else:
raise ValueError(f"STP文件读取失败,状态码: {status}")
except Exception as e:
logger.error(f"STP解析失败: {e}")
raise
def analyze_geometry(self, shape) -> Dict[str, Any]:
"""分析几何属性"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
from OCC.Core.Bnd import Bnd_Box
from OCC.Core.BRepBndLib import brepbndlib
from OCC.Core.TopExp import TopExp_Explorer
from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE, TopAbs_VERTEX
logger.info("开始几何分析...")
# 计算边界框
bbox = self._compute_bounding_box(shape)
# 计算体积和表面积
volume = self._compute_volume(shape)
area = self._compute_surface_area(shape)
# 分析拓扑
topology = self._analyze_topology(shape)
# 计算质心
center_of_mass = self._compute_center_of_mass(shape)
# 计算惯性属性
inertia_properties = self._compute_inertia_properties(shape)
result = {
"bounding_box": bbox,
"volume": float(volume),
"surface_area": float(area),
"topology": topology,
"center_of_mass": center_of_mass,
"inertia_properties": inertia_properties,
"analysis_method": "pythonocc"
}
logger.info("几何分析完成")
return result
except Exception as e:
logger.error(f"几何分析失败: {e}")
raise
def _compute_bounding_box(self, shape) -> Dict[str, Any]:
"""计算边界框"""
try:
from OCC.Core.Bnd import Bnd_Box
from OCC.Core.BRepBndLib import brepbndlib
bbox = Bnd_Box()
brepbndlib.Add(shape, bbox)
xmin, ymin, zmin, xmax, ymax, zmax = bbox.Get()
return {
"min": [float(xmin), float(ymin), float(zmin)],
"max": [float(xmax), float(ymax), float(zmax)],
"dimensions": [
float(xmax - xmin),
float(ymax - ymin),
float(zmax - zmin)
],
"center": [
float((xmin + xmax) / 2),
float((ymin + ymax) / 2),
float((zmin + zmax) / 2)
]
}
except Exception as e:
logger.error(f"边界框计算失败: {e}")
return self._default_bounding_box()
def _compute_volume(self, shape) -> float:
"""计算体积"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
props = GProp_GProps()
brepgprop.VolumeProperties(shape, props)
return props.Mass()
except Exception as e:
logger.error(f"体积计算失败: {e}")
return 1000000.0
def _compute_surface_area(self, shape) -> float:
"""计算表面积"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
props = GProp_GProps()
brepgprop.SurfaceProperties(shape, props)
area = props.Mass()
logger.info(f"表面积计算成功: {area:.2f} mm²")
# 如果计算结果为0,使用备选估算方法
if area <= 0:
logger.warning("表面积计算结果为0,使用边界框估算")
raise ValueError("Surface area is zero")
return area
except Exception as e:
logger.error(f"表面积计算失败: {e}")
# 基于边界框估算表面积
try:
bbox = self._compute_bounding_box(shape)
dims = bbox.get("dimensions", [100, 100, 100])
# 简化的估算公式:2*(lw + lh + wh)
estimated_area = 2 * (dims[0]*dims[1] + dims[0]*dims[2] + dims[1]*dims[2])
logger.warning(f"使用边界框估算表面积: {estimated_area:.2f} mm²")
return estimated_area
except:
return 60000.0
def _compute_center_of_mass(self, shape) -> List[float]:
"""计算质心"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
props = GProp_GProps()
brepgprop.VolumeProperties(shape, props)
center = props.CentreOfMass()
return [float(center.X()), float(center.Y()), float(center.Z())]
except Exception as e:
logger.error(f"质心计算失败: {e}")
return [0.0, 0.0, 0.0]
def _compute_inertia_properties(self, shape) -> Dict[str, Any]:
"""计算惯性属性"""
try:
from OCC.Core.GProp import GProp_GProps
from OCC.Core.BRepGProp import brepgprop
props = GProp_GProps()
brepgprop.VolumeProperties(shape, props)
inertia = props.MatrixOfInertia()
return {
"mass": float(props.Mass()),
"moment_of_inertia": [
[float(inertia.Value(1, 1)), float(inertia.Value(1, 2)), float(inertia.Value(1, 3))],
[float(inertia.Value(2, 1)), float(inertia.Value(2, 2)), float(inertia.Value(2, 3))],
[float(inertia.Value(3, 1)), float(inertia.Value(3, 2)), float(inertia.Value(3, 3))]
]
}
except Exception as e:
logger.error(f"惯性属性计算失败: {e}")
return {}
def _analyze_topology(self, shape) -> Dict[str, int]:
"""分析拓扑"""
try:
from OCC.Core.TopExp import TopExp_Explorer
from OCC.Core.TopAbs import TopAbs_FACE, TopAbs_EDGE, TopAbs_VERTEX
def count_elements(element_type):
explorer = TopExp_Explorer(shape, element_type)
count = 0
while explorer.More():
count += 1
explorer.Next()
return count
return {
"faces": count_elements(TopAbs_FACE),
"edges": count_elements(TopAbs_EDGE),
"vertices": count_elements(TopAbs_VERTEX)
}
except Exception as e:
logger.error(f"拓扑分析失败: {e}")
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]:
"""默认边界框"""
return {
"min": [0.0, 0.0, 0.0],
"max": [100.0, 100.0, 100.0],
"dimensions": [100.0, 100.0, 100.0],
"center": [50.0, 50.0, 50.0]
}
def export_to_json(self, geometry_data: Dict[str, Any], output_path: Path) -> str:
"""将几何数据导出为JSON文件"""
try:
# 确保输出目录存在
output_path.parent.mkdir(parents=True, exist_ok=True)
# 添加元数据
json_data = {
"metadata": {
"export_time": str(np.datetime64('now')),
"analysis_method": geometry_data.get("analysis_method", "unknown"),
"version": "1.0.0"
},
"geometry_data": geometry_data
}
# 保存JSON文件
with open(output_path, 'w', encoding='utf-8') as f:
json.dump(json_data, f, indent=2, ensure_ascii=False)
logger.info(f"几何数据已导出到: {output_path}")
return str(output_path)
except Exception as e:
logger.error(f"JSON导出失败: {e}")
raise
def get_json_data(self, geometry_data: Dict[str, Any]) -> Dict[str, Any]:
"""获取JSON格式的几何数据"""
return {
"metadata": {
"export_time": str(np.datetime64('now')),
"analysis_method": geometry_data.get("analysis_method", "unknown"),
"version": "1.0.0"
},
"geometry_data": geometry_data
}
+99
View File
@@ -0,0 +1,99 @@
# database/database.py
import sys
import os
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy import text
from sqlalchemy.orm import sessionmaker
from config.settings import settings
import asyncio
from utils.logger import get_logger
logger = get_logger(__name__)
class DatabaseManager:
"""数据库管理器"""
def __init__(self):
self.engine = None
self.async_session = None
self.is_connected = False
async def connect(self):
"""连接数据库"""
if not settings.DATABASE_URL:
logger.warning("未配置数据库连接,跳过数据库初始化")
self.is_connected = False
return
try:
# 创建异步引擎
self.engine = create_async_engine(
settings.DATABASE_URL,
echo=settings.DEBUG,
pool_size=20,
max_overflow=30,
pool_recycle=3600
)
# 创建异步会话工厂
self.async_session = async_sessionmaker(
self.engine,
class_=AsyncSession,
expire_on_commit=False
)
# 测试连接
async with self.engine.begin() as conn:
await conn.execute(text("SELECT 1"))
self.is_connected = True
logger.info("数据库连接成功")
except Exception as e:
logger.error(f"数据库连接失败: {e}")
self.is_connected = False
raise
async def disconnect(self):
"""断开数据库连接"""
if self.engine:
await self.engine.dispose()
self.is_connected = False
logger.info("数据库连接已断开")
async def get_session(self) -> AsyncSession:
"""获取数据库会话"""
if not self.is_connected:
await self.connect()
return self.async_session()
async def create_tables(self):
"""创建数据库表"""
from models.database import Base
try:
async with self.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.info("数据库表创建成功")
except Exception as e:
logger.error(f"数据库表创建失败: {e}")
raise
# 全局数据库管理器实例
db_manager = DatabaseManager()
# 数据库依赖注入
async def get_db_session():
"""获取数据库会话的依赖函数"""
session = await db_manager.get_session()
try:
yield session
finally:
await session.close()
+25
View File
@@ -0,0 +1,25 @@
# database/init_db.py
import asyncio
from database.database import db_manager
from utils.logger import get_logger
logger = get_logger(__name__)
async def init_database():
"""初始化数据库"""
try:
# 连接数据库
await db_manager.connect()
# 创建表
await db_manager.create_tables()
logger.info("数据库初始化完成")
return True
except Exception as e:
logger.error(f"数据库初始化失败: {e}")
return False
if __name__ == "__main__":
asyncio.run(init_database())
+67
View File
@@ -0,0 +1,67 @@
"""数据库迁移脚本 - 删除旧表并重新创建"""
import asyncio
import sys
import os
from pathlib import Path
# 添加项目根目录到路径
project_root = Path(__file__).parent.parent.parent
sys.path.insert(0, str(project_root))
sys.path.insert(0, str(Path(__file__).parent.parent))
# 设置环境变量确保正确导入
os.environ['PYTHONPATH'] = str(project_root) + os.pathsep + str(Path(__file__).parent.parent)
from src.database.database import db_manager
from src.models.database import Base
from src.utils.logger import get_logger
logger = get_logger(__name__)
async def migrate_database():
"""迁移数据库:删除所有表并重新创建"""
try:
# 连接数据库
await db_manager.connect()
# 删除所有表
logger.info("正在删除所有数据库表...")
async with db_manager.engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
# 重新创建所有表
logger.info("正在创建所有数据库表...")
async with db_manager.engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
logger.info("数据库迁移完成!")
return True
except Exception as e:
logger.error(f"数据库迁移失败: {e}")
return False
finally:
await db_manager.disconnect()
if __name__ == "__main__":
import sys
# 检查命令行参数
if len(sys.argv) > 1 and sys.argv[1] == '--force':
confirm = 'yes'
else:
print("=== 数据库迁移 ===")
print("警告:这将删除所有数据库表和数据!")
confirm = input("确认继续?(yes/no): ")
if confirm.lower() == 'yes':
asyncio.run(migrate_database())
else:
print("已取消迁移")
@@ -0,0 +1,207 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>3D模具几何可视化 - fsa30scy_tc-01-0817.stp</title>
<script src="https://cdn.jsdelivr.net/npm/three@0.132.2/build/three.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/three@0.132.2/examples/js/controls/OrbitControls.js"></script>
<style>
body { margin: 0; overflow: hidden; font-family: Arial, sans-serif; }
#container { position: relative; width: 100vw; height: 100vh; }
#canvas { display: block; }
#info-panel {
position: absolute;
top: 10px;
left: 10px;
background: rgba(255, 255, 255, 0.9);
padding: 15px;
border-radius: 8px;
max-width: 300px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}
#controls {
position: absolute;
bottom: 10px;
left: 10px;
background: rgba(255, 255, 255, 0.9);
padding: 10px;
border-radius: 8px;
}
.metric { margin: 5px 0; }
.metric-label { font-weight: bold; color: #333; }
.metric-value { color: #666; }
</style>
</head>
<body>
<div id="container">
<canvas id="canvas"></canvas>
<div id="info-panel">
<h3>模具几何信息</h3>
<div class="metric">
<span class="metric-label">文件名:</span>
<span class="metric-value">fsa30scy_tc-01-0817.stp</span>
</div>
<div class="metric">
<span class="metric-label">体积:</span>
<span class="metric-value">1000000.00 mm³</span>
</div>
<div class="metric">
<span class="metric-label">表面积:</span>
<span class="metric-value">60000.00 mm²</span>
</div>
<div class="metric">
<span class="metric-label">边界框:</span>
<span class="metric-value">100.0 × 100.0 × 100.0 mm</span>
</div>
<div class="metric">
<span class="metric-label">面数:</span>
<span class="metric-value">6</span>
</div>
<div class="metric">
<span class="metric-label">边数:</span>
<span class="metric-value">12</span>
</div>
<div class="metric">
<span class="metric-label">顶点数:</span>
<span class="metric-value">8</span>
</div>
</div>
<div id="controls">
<button onclick="resetView()">重置视图</button>
<button onclick="toggleWireframe()">切换线框</button>
</div>
</div>
<script>
// 初始化Three.js场景
const scene = new THREE.Scene();
const camera = new THREE.PerspectiveCamera(75, window.innerWidth / window.innerHeight, 0.1, 1000);
const renderer = new THREE.WebGLRenderer({ canvas: document.getElementById('canvas') });
renderer.setSize(window.innerWidth, window.innerHeight);
renderer.setClearColor(0xf0f0f0);
// 添加光源
const ambientLight = new THREE.AmbientLight(0x404040);
scene.add(ambientLight);
const directionalLight = new THREE.DirectionalLight(0xffffff, 1);
directionalLight.position.set(1, 1, 1);
scene.add(directionalLight);
// 添加坐标轴
const axesHelper = new THREE.AxesHelper(50);
scene.add(axesHelper);
// 创建几何体(模拟模具形状)
const geometryData = {
"bounding_box": {
"min": [
0.0,
0.0,
0.0
],
"max": [
100.0,
100.0,
100.0
],
"dimensions": [
100.0,
100.0,
100.0
],
"center": [
50.0,
50.0,
50.0
]
},
"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"
};
// 根据边界框创建模拟几何体
const bbox = geometryData.bounding_box;
if (bbox) {
const width = bbox.dimensions ? bbox.dimensions[0] : 100;
const height = bbox.dimensions ? bbox.dimensions[1] : 100;
const depth = bbox.dimensions ? bbox.dimensions[2] : 100;
// 创建基础几何体
const geometry = new THREE.BoxGeometry(width, height, depth);
const material = new THREE.MeshPhongMaterial({
color: 0x4CAF50,
transparent: true,
opacity: 0.8,
wireframe: false
});
const mesh = new THREE.Mesh(geometry, material);
scene.add(mesh);
// 添加线框
const wireframe = new THREE.WireframeGeometry(geometry);
const line = new THREE.LineSegments(wireframe);
line.material.depthTest = false;
line.material.opacity = 0.25;
line.material.transparent = true;
scene.add(line);
}
// 设置相机位置
camera.position.set(200, 200, 200);
camera.lookAt(0, 0, 0);
// 添加轨道控制器
const controls = new THREE.OrbitControls(camera, renderer.domElement);
controls.enableDamping = true;
controls.dampingFactor = 0.25;
// 动画循环
function animate() {
requestAnimationFrame(animate);
controls.update();
renderer.render(scene, camera);
}
animate();
// 窗口大小调整
window.addEventListener('resize', () => {
camera.aspect = window.innerWidth / window.innerHeight;
camera.updateProjectionMatrix();
renderer.setSize(window.innerWidth, window.innerHeight);
});
// 控制函数
function resetView() {
controls.reset();
}
function toggleWireframe() {
scene.traverse((child) => {
if (child.isMesh) {
child.material.wireframe = !child.material.wireframe;
}
});
}
</script>
</body>
</html>
+134
View File
@@ -0,0 +1,134 @@
# main.py
import sys
import os
from pathlib import Path
# 添加项目根目录到Python路径
project_root = Path(__file__).parent.parent
src_root = Path(__file__).parent
sys.path.insert(0, str(project_root))
sys.path.insert(0, str(src_root))
# 确保当前工作目录是项目根目录
os.chdir(project_root)
# 打印调试信息
print(f"项目根目录: {project_root}")
print(f"Python路径: {sys.path}")
print(f"当前工作目录: {os.getcwd()}")
# 测试导入配置模块
try:
from config.settings import settings
print("[OK] 配置模块导入成功")
except ImportError as e:
print(f"[FAIL] 配置模块导入失败: {e}")
# 列出当前目录内容
print("当前目录内容:")
for item in os.listdir('.'):
print(f" - {item}")
# 列出config目录内容
if os.path.exists('config'):
print("config目录内容:")
for item in os.listdir('config'):
print(f" - {item}")
raise
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
import asyncio
from api.routes import router
from utils.logger import setup_logging
from database.init_db import init_database
# 设置日志
setup_logging()
# 创建FastAPI应用
app = FastAPI(
title="模具几何分析服务",
description="基于PythonOCC的STP文件几何分析和模具设计建议服务",
version="3.0.0"
)
# 启动时初始化数据库和RustFS
@app.on_event("startup")
async def startup_event():
"""应用启动时初始化数据库和RustFS"""
# 初始化数据库
success = await init_database()
if success:
print("[OK] 数据库初始化成功")
else:
print("[FAIL] 数据库初始化失败,服务将继续运行但数据库功能不可用")
# 初始化RustFS连接
try:
from storage.rustfs_storage import rustfs_manager
from config.settings import settings
await rustfs_manager.connect(
endpoint=settings.RUSTFS_ENDPOINT,
access_key=settings.RUSTFS_ACCESS_KEY,
secret_key=settings.RUSTFS_SECRET_KEY,
timeout=settings.RUSTFS_TIMEOUT
)
print("[OK] RustFS连接成功")
except Exception as e:
print(f"[FAIL] RustFS连接失败: {e}")
print("[WARN] 文件上传功能将不可用,但其他功能正常")
# 创建必要目录
UPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)
TEMPLATES_DIR = Path("templates")
TEMPLATES_DIR.mkdir(exist_ok=True)
STATIC_DIR = Path("static")
STATIC_DIR.mkdir(exist_ok=True)
HTML_OUTPUT_DIR = Path("html_output")
HTML_OUTPUT_DIR.mkdir(exist_ok=True)
# 挂载静态文件
import os
static_dir = os.path.join(os.path.dirname(os.path.dirname(__file__)), "static")
app.mount("/static", StaticFiles(directory=static_dir), name="static")
# 注册路由
app.include_router(router)
@app.get("/health")
async def health():
from database.database import db_manager
return {
"status": "healthy",
"service": "mold-geometry-analysis",
"database_connected": db_manager.is_connected
}
if __name__ == "__main__":
import uvicorn
import os
# 直接从环境变量获取端口,避免配置导入问题
host = os.getenv('HOST', '0.0.0.0')
port = int(os.getenv('PORT', '8000'))
print("启动模具几何分析服务 v3.0...")
print(f"访问 http://localhost:{port} 使用网页界面")
print("新增功能:")
print(" - STP文件解析为JSON数据")
print(" - 数据存储到PostgreSQL数据库")
print(" - 自动生成3D可视化HTML页面")
print(" - 源文件、JSON数据、HTML文件统一管理")
print(f"调试接口: http://localhost:{port}/debug/tasks")
uvicorn.run(
"main:app",
host=host,
port=port,
reload=True
)
+328
View File
@@ -0,0 +1,328 @@
# models/database.py
from sqlalchemy import Column, Integer, String, Text, DateTime, JSON, LargeBinary, Boolean, Float, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.sql import func
from sqlalchemy.orm import relationship
from datetime import datetime
Base = declarative_base()
class User(Base):
"""用户表"""
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
username = Column(String(50), unique=True, index=True, nullable=False)
email = Column(String(255), unique=True, index=True, nullable=False)
hashed_password = Column(String(255), nullable=False)
full_name = Column(String(100))
is_active = Column(Boolean, default=True)
is_superuser = Column(Boolean, default=False)
created_at = Column(DateTime, default=func.now())
last_login = Column(DateTime, nullable=True)
# 关联关系
stp_files = relationship("STPFile", back_populates="user")
def __repr__(self):
return f"<User(id={self.id}, username='{self.username}', email='{self.email}')>"
class STPFile(Base):
"""STP源文件元数据表"""
__tablename__ = "stp_files"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True, index=True)
# 对象存储信息
object_key = Column(String(500), nullable=False, index=True) # MinIO对象键
storage_bucket = Column(String(100), nullable=False) # 存储桶名称
object_url = Column(String(1000), nullable=True) # 预签名URL(可选)
# 文件信息
original_filename = Column(String(255), nullable=False)
file_size = Column(Integer, nullable=False)
file_hash = Column(String(64), unique=True, index=True) # SHA256 hash
mime_type = Column(String(50), default="application/octet-stream")
# 时间戳
upload_time = Column(DateTime, default=func.now())
processed_time = Column(DateTime, nullable=True)
# 状态
status = Column(String(20), default="pending") # pending, processing, completed, failed
error_message = Column(Text, nullable=True)
# 保留旧字段以兼容
file_path = Column(String(500), nullable=True) # 本地路径(已弃用)
file_content = Column(LargeBinary, nullable=True) # 本地存储(已弃用)
filename = Column(String(255), nullable=True) # 已弃用
# 关联关系
user = relationship("User", back_populates="stp_files")
geometry_data = relationship("GeometryData", back_populates="stp_file", uselist=False)
mold_cavity_data = relationship("MoldCavityData", back_populates="stp_file", uselist=False)
html_file = relationship("HTMLFile", back_populates="stp_file", uselist=False)
def __repr__(self):
return f"<STPFile(id={self.id}, object_key='{self.object_key}', status='{self.status}')>"
class GeometryData(Base):
"""几何数据JSON元数据表"""
__tablename__ = "geometry_data"
id = Column(Integer, primary_key=True, index=True)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 对象存储信息
object_key = Column(String(500), nullable=False)
storage_bucket = Column(String(100), nullable=False)
object_url = Column(String(1000), nullable=True)
# 分析方法
analysis_method = Column(String(50), default="pythonocc") # pythonocc, simulated
# 时间戳
created_time = Column(DateTime, default=func.now())
# 几何属性摘要(便于快速查询)
volume = Column(Float, nullable=True)
surface_area = Column(Float, nullable=True)
bounding_box_min = Column(JSON, nullable=True)
bounding_box_max = Column(JSON, nullable=True)
center_of_mass = Column(JSON, nullable=True)
# 拓扑信息
topology_faces = Column(Integer, nullable=True)
topology_edges = Column(Integer, nullable=True)
topology_vertices = Column(Integer, nullable=True)
# 关联关系
stp_file = relationship("STPFile", back_populates="geometry_data")
def __repr__(self):
return f"<GeometryData(id={self.id}, stp_file_id={self.stp_file_id})>"
class HTMLFile(Base):
"""网页文件元数据表"""
__tablename__ = "html_files"
id = Column(Integer, primary_key=True, index=True)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 对象存储信息
object_key = Column(String(500), nullable=False)
storage_bucket = Column(String(100), nullable=False)
object_url = Column(String(1000), nullable=True)
# 文件信息
filename = Column(String(255), nullable=False)
generated_time = Column(DateTime, default=func.now())
# 可视化相关元数据
visualization_type = Column(String(50), default="3d_viewer")
has_interactive_elements = Column(Boolean, default=True)
# 保留旧字段以兼容
file_path = Column(String(500), nullable=True)
html_content = Column(Text, nullable=True)
# 关联关系
stp_file = relationship("STPFile", back_populates="html_file")
def __repr__(self):
return f"<HTMLFile(id={self.id}, stp_file_id={self.stp_file_id}, object_key='{self.object_key}')>"
class ProcessingTask(Base):
"""处理任务记录表"""
__tablename__ = "processing_tasks"
id = Column(Integer, primary_key=True, index=True)
task_id = Column(String(36), unique=True, index=True, nullable=False)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 任务类型和状态
task_type = Column(String(50), default="stp_parsing") # stp_parsing, geometry_analysis, mold_generation
status = Column(String(20), default="pending") # pending, processing, completed, failed
# 时间戳
created_time = Column(DateTime, default=func.now())
started_time = Column(DateTime, nullable=True)
completed_time = Column(DateTime, nullable=True)
# 处理进度
progress = Column(Integer, default=0) # 0-100
current_step = Column(String(100), nullable=True)
# 错误信息
error_message = Column(Text, nullable=True)
error_stack = Column(Text, nullable=True)
# 处理参数
parameters = Column(JSON, nullable=True) # 任务参数
def __repr__(self):
return f"<ProcessingTask(id={self.id}, task_id='{self.task_id}', status='{self.status}')>"
class MoldCavityData(Base):
"""模具型腔数据元数据表"""
__tablename__ = "mold_cavity_data"
id = Column(Integer, primary_key=True, index=True)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 对象存储信息
detailed_object_key = Column(String(500), nullable=False) # 完整三维数据
storage_bucket = Column(String(100), nullable=False)
# 模具类型和材料
mold_material = Column(String(100), default="Aluminum Alloy 7075")
mold_type = Column(String(50), default="single_cavity") # single_cavity, multi_cavity
# 工艺参数
shrinkage_rate = Column(Float, nullable=False)
draft_angle = Column(Float, nullable=False)
parting_line_length = Column(Float, nullable=True)
# 生成时间
generated_time = Column(DateTime, default=func.now())
# 关键信息摘要(快速查询字段)
cavity_key_info = Column(JSON, nullable=True) # 完整关键信息
# 提取的字段(便于查询和排序)
mold_size_length = Column(Float, nullable=True)
mold_size_width = Column(Float, nullable=True)
mold_size_height = Column(Float, nullable=True)
estimated_clamping_force = Column(String(50), nullable=True)
product_weight = Column(String(50), nullable=True)
product_volume = Column(Float, nullable=True)
wall_thickness_range = Column(String(50), nullable=True)
complexity_score = Column(Float, nullable=True)
# 质量评估
weld_line_risk = Column(String(50), nullable=True) # 熔接痕风险
sink_mark_risk = Column(String(50), nullable=True) # 缩痕风险
warpage_risk = Column(String(50), nullable=True) # 翘曲风险
# 关联关系
stp_file = relationship("STPFile", back_populates="mold_cavity_data")
def __repr__(self):
return f"<MoldCavityData(stp_file_id={self.stp_file_id}, mold_material='{self.mold_material}')>"
class FeatureDetection(Base):
"""特征检测结果表"""
__tablename__ = "feature_detections"
id = Column(Integer, primary_key=True, index=True)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 特征信息
feature_type = Column(String(50), nullable=False, index=True) # thin_wall, thick_wall, rib, boss, draft_angle
confidence = Column(Float, nullable=False) # 0.0 - 1.0
# 位置和尺寸
location = Column(JSON, nullable=True) # [x, y, z]
dimensions = Column(JSON, nullable=True) # [length, width, height]
# 特征参数
parameters = Column(JSON, nullable=True) # 自定义参数
# 检测时间
detected_at = Column(DateTime, default=func.now())
# 关联的几何数据
geometry_data_id = Column(Integer, ForeignKey("geometry_data.id"), nullable=True)
def __repr__(self):
return f"<FeatureDetection(id={self.id}, feature_type='{self.feature_type}', confidence={self.confidence})>"
class DesignRecommendation(Base):
"""设计建议表"""
__tablename__ = "design_recommendations"
id = Column(Integer, primary_key=True, index=True)
stp_file_id = Column(Integer, ForeignKey("stp_files.id"), nullable=False, index=True)
# 建议信息
rec_type = Column(String(50), nullable=False) # wall_thickness, draft_angle, etc.
priority = Column(String(20), nullable=False) # high, medium, low
description = Column(String(500), nullable=False)
reason = Column(Text, nullable=True)
# 建议参数
parameters = Column(JSON, nullable=True)
# 状态
status = Column(String(20), default="pending") # pending, accepted, rejected
user_notes = Column(Text, nullable=True)
# 时间戳
created_at = Column(DateTime, default=func.now())
updated_at = Column(DateTime, nullable=True)
def __repr__(self):
return f"<DesignRecommendation(id={self.id}, rec_type='{self.rec_type}', priority='{self.priority}')>"
class UserActivity(Base):
"""用户活动日志表"""
__tablename__ = "user_activities"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("users.id"), nullable=False, index=True)
# 活动信息
activity_type = Column(String(50), nullable=False, index=True) # upload, view, download, delete, export
resource_type = Column(String(50), nullable=True) # stp_file, geometry_data, mold_cavity
resource_id = Column(Integer, nullable=True)
# 活动详情
description = Column(Text, nullable=True)
meta_data = Column(JSON, nullable=True)
# 时间戳
created_at = Column(DateTime, default=func.now(), index=True)
# IP和设备信息
ip_address = Column(String(45), nullable=True)
user_agent = Column(String(500), nullable=True)
def __repr__(self):
return f"<UserActivity(id={self.id}, user_id={self.user_id}, activity_type='{self.activity_type}')>"
class SystemLog(Base):
"""系统日志表(重要操作和错误)"""
__tablename__ = "system_logs"
id = Column(Integer, primary_key=True, index=True)
# 日志级别
level = Column(String(20), nullable=False, index=True) # INFO, WARNING, ERROR, CRITICAL
# 日志信息
message = Column(Text, nullable=False)
module = Column(String(100), nullable=True) # 模块名
function_name = Column(String(100), nullable=True)
# 时间戳
created_at = Column(DateTime, default=func.now(), index=True)
# 用户信息(如果有关联用户)
user_id = Column(Integer, ForeignKey("users.id"), nullable=True)
# 额外信息
request_id = Column(String(100), nullable=True) # 关联的请求ID
execution_time_ms = Column(Integer, nullable=True) # 执行时间
# 关联数据
resource_type = Column(String(50), nullable=True)
resource_id = Column(Integer, nullable=True)
def __repr__(self):
return f"<SystemLog(id={self.id}, level='{self.level}', module='{self.module}')>"
+135
View File
@@ -0,0 +1,135 @@
# models/schemas.py
from typing import Dict, List, Optional, Any
from enum import Enum
class ProcessingStatus(str, Enum):
PENDING = "pending"
PROCESSING = "processing"
COMPLETED = "completed"
FAILED = "failed"
# 简化的数据模型,避免复杂的Pydantic验证
def create_geometry_data(
bounding_box: Dict[str, List[float]],
volume: float,
surface_area: float,
topology: Dict[str, int],
analysis_method: str,
center_of_mass: Optional[List[float]] = None,
inertia_properties: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""创建几何数据"""
return {
"bounding_box": bounding_box,
"volume": volume,
"surface_area": surface_area,
"topology": topology,
"center_of_mass": center_of_mass or [0.0, 0.0, 0.0],
"inertia_properties": inertia_properties or {},
"analysis_method": analysis_method
}
def create_mold_feature(
feature_type: str,
confidence: float,
location: List[float],
dimensions: List[float],
parameters: Dict[str, Any],
recommendations: List[str]
) -> Dict[str, Any]:
"""创建模具特征"""
return {
"feature_type": feature_type,
"confidence": confidence,
"location": location,
"dimensions": dimensions,
"parameters": parameters,
"recommendations": recommendations
}
def create_design_recommendation(
rec_type: str,
priority: str,
description: str,
parameters: Dict[str, Any],
reason: str
) -> Dict[str, Any]:
"""创建设计建议"""
return {
"type": rec_type,
"priority": priority,
"description": description,
"parameters": parameters,
"reason": reason
}
def create_analysis_result(
geometry_data: Dict[str, Any],
detected_features: List[Dict[str, Any]],
design_recommendations: List[Dict[str, Any]],
quality_metrics: Dict[str, float],
analysis_summary: str
) -> Dict[str, Any]:
"""创建分析结果"""
return {
"geometry_data": geometry_data,
"detected_features": detected_features,
"design_recommendations": design_recommendations,
"quality_metrics": quality_metrics,
"analysis_summary": analysis_summary
}
def create_task_info(
task_id: str,
status: ProcessingStatus,
filename: str,
file_path: str,
file_size: int,
upload_time: str,
completed_at: Optional[str] = None,
geometry_data: Optional[Dict[str, Any]] = None,
analysis_result: Optional[Dict[str, Any]] = None,
error: Optional[str] = None
) -> Dict[str, Any]:
"""创建任务信息"""
return {
"task_id": task_id,
"status": status,
"filename": filename,
"file_path": file_path,
"file_size": file_size,
"upload_time": upload_time,
"completed_at": completed_at,
"geometry_data": geometry_data,
"analysis_result": analysis_result,
"error": error
}
# 添加到 schemas.py
def create_mold_cavity_data(
cavity_geometry: Dict[str, Any],
core_geometry: Dict[str, Any],
parting_surface: Dict[str, Any],
manufacturing_info: Dict[str, Any]
) -> Dict[str, Any]:
"""创建模具型腔详细数据"""
return {
"cavity_geometry": cavity_geometry,
"core_geometry": core_geometry,
"parting_surface": parting_surface,
"manufacturing_info": manufacturing_info
}
def create_mold_key_info(
mold_parameters: Dict[str, Any],
geometric_characteristics: Dict[str, Any],
manufacturing_requirements: Dict[str, Any],
quality_considerations: Dict[str, Any]
) -> Dict[str, Any]:
"""创建模具型腔关键信息"""
return {
"mold_parameters": mold_parameters,
"geometric_characteristics": geometric_characteristics,
"manufacturing_requirements": manufacturing_requirements,
"quality_considerations": quality_considerations
}
+1
View File
@@ -0,0 +1 @@
# Services 模块
+376
View File
@@ -0,0 +1,376 @@
# services/storage_integration.py
"""存储集成服务 - 协调 PostgreSQL 和 MinIO"""
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select
from pathlib import Path
from typing import Optional, Dict, Any
import json
from models.database import (
STPFile, GeometryData, MoldCavityData,
HTMLFile, ProcessingTask, User,
FeatureDetection, DesignRecommendation,
UserActivity, SystemLog
)
from storage.object_storage import storage_manager
from utils.logger import get_logger
logger = get_logger(__name__)
class StorageIntegrationService:
"""存储集成服务"""
async def save_stp_file(self, session: AsyncSession,
file_path: Path,
original_filename: str,
user_id: Optional[int] = None) -> STPFile:
"""保存STP文件到PostgreSQL元数据 + MinIO对象存储"""
# 1. 上传到MinIO
upload_result = await storage_manager.upload_stp_file(
file_path,
original_filename
)
# 2. 创建PostgreSQL记录
stp_file = STPFile(
user_id=user_id,
object_key=upload_result['object_key'],
storage_bucket=storage_manager.buckets['stp_files'],
original_filename=original_filename,
file_size=upload_result['file_size'],
file_hash=upload_result['file_hash'],
status="uploaded",
file_path=str(file_path) # 保留本地路径以兼容
)
session.add(stp_file)
await session.commit()
await session.refresh(stp_file)
logger.info(f"STP文件保存成功: {stp_file.id}")
return stp_file
async def save_geometry_data(self, session: AsyncSession,
stp_file_id: int,
geometry_json: Dict[str, Any],
analysis_method: str = "pythonocc") -> GeometryData:
"""保存几何数据到PostgreSQL元数据 + MinIO对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 上传到MinIO
upload_result = await storage_manager.upload_geometry_data(
geometry_json,
file_hash
)
# 3. 创建PostgreSQL记录
geometry_data = GeometryData(
stp_file_id=stp_file_id,
object_key=upload_result['object_key'],
storage_bucket=storage_manager.buckets['geometry_data'],
analysis_method=analysis_method,
# 提取摘要字段
volume=geometry_json.get('geometry_data', {}).get('volume'),
surface_area=geometry_json.get('geometry_data', {}).get('surface_area'),
bounding_box_min=geometry_json.get('geometry_data', {}).get('bounding_box', {}).get('min'),
bounding_box_max=geometry_json.get('geometry_data', {}).get('bounding_box', {}).get('max'),
center_of_mass=geometry_json.get('geometry_data', {}).get('center_of_mass'),
topology_faces=geometry_json.get('geometry_data', {}).get('topology', {}).get('faces'),
topology_edges=geometry_json.get('geometry_data', {}).get('topology', {}).get('edges'),
topology_vertices=geometry_json.get('geometry_data', {}).get('topology', {}).get('vertices')
)
session.add(geometry_data)
await session.commit()
await session.refresh(geometry_data)
logger.info(f"几何数据保存成功: {geometry_data.id}")
return geometry_data
async def save_mold_cavity_data(self, session: AsyncSession,
stp_file_id: int,
cavity_json: Dict[str, Any]) -> MoldCavityData:
"""保存模具型腔数据到PostgreSQL元数据 + MinIO对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 上传到MinIO
upload_result = await storage_manager.upload_mold_cavity_data(
cavity_json,
file_hash
)
# 3. 提取关键信息
metadata = cavity_json.get('metadata', {})
product_analysis = cavity_json.get('product_analysis', {})
manufacturing_info = cavity_json.get('manufacturing_info', {})
mold_size = manufacturing_info.get('estimated_mold_size', {})
key_info = cavity_json.get('mold_cavities', {}).get('cavity_key_info', {})
# 4. 创建PostgreSQL记录
mold_cavity = MoldCavityData(
stp_file_id=stp_file_id,
detailed_object_key=upload_result['object_key'],
storage_bucket=storage_manager.buckets['mold_cavities'],
# 模具参数
mold_material=manufacturing_info.get('recommended_material', 'Aluminum Alloy 7075'),
shrinkage_rate=metadata.get('shrinkage_rate', 0.005),
draft_angle=metadata.get('draft_angle', 2.0),
# 提取的摘要字段
cavity_key_info=key_info,
mold_size_length=mold_size.get('length'),
mold_size_width=mold_size.get('width'),
mold_size_height=mold_size.get('height'),
estimated_clamping_force=manufacturing_info.get('estimated_clamping_force'),
product_volume=product_analysis.get('volume'),
# 从key_info中提取(如果存在)
product_weight=key_info.get('geometric_characteristics', {}).get('product_weight'),
wall_thickness_range=key_info.get('geometric_characteristics', {}).get('wall_thickness_range'),
complexity_score=key_info.get('geometric_characteristics', {}).get('complexity_score'),
# 质量评估
weld_line_risk=key_info.get('quality_considerations', {}).get('potential_weld_lines'),
sink_mark_risk=key_info.get('quality_considerations', {}).get('sink_mark_areas'),
warpage_risk=key_info.get('quality_considerations', {}).get('warpage_risk')
)
session.add(mold_cavity)
await session.commit()
await session.refresh(mold_cavity)
logger.info(f"模具型腔数据保存成功: {mold_cavity.id}")
return mold_cavity
async def save_html_file(self, session: AsyncSession,
stp_file_id: int,
html_content: str,
filename: str) -> HTMLFile:
"""保存HTML文件到PostgreSQL元数据 + MinIO对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 上传到MinIO
upload_result = await storage_manager.upload_html_file(
html_content,
filename,
file_hash
)
# 3. 创建PostgreSQL记录
html_file = HTMLFile(
stp_file_id=stp_file_id,
object_key=upload_result['object_key'],
storage_bucket=storage_manager.buckets['html_files'],
filename=filename,
file_path=str(Path('html_output') / filename), # 保留本地路径
html_content=html_content # 保留内容以兼容
)
session.add(html_file)
await session.commit()
await session.refresh(html_file)
logger.info(f"HTML文件保存成功: {html_file.id}")
return html_file
async def save_features_and_recommendations(
self, session: AsyncSession,
stp_file_id: int,
features: list,
recommendations: list
):
"""保存特征检测结果和设计建议"""
# 1. 保存特征
for feature in features:
feature_record = FeatureDetection(
stp_file_id=stp_file_id,
feature_type=feature.get('feature_type'),
confidence=feature.get('confidence'),
location=feature.get('location'),
dimensions=feature.get('dimensions'),
parameters=feature.get('parameters')
)
session.add(feature_record)
# 2. 保存建议
for rec in recommendations:
rec_record = DesignRecommendation(
stp_file_id=stp_file_id,
rec_type=rec.get('rec_type'),
priority=rec.get('priority'),
description=rec.get('description'),
reason=rec.get('reason'),
parameters=rec.get('parameters')
)
session.add(rec_record)
await session.commit()
logger.info(f"保存了 {len(features)} 个特征和 {len(recommendations)} 个建议")
async def log_user_activity(self, session: AsyncSession,
user_id: int,
activity_type: str,
resource_type: Optional[str] = None,
resource_id: Optional[int] = None,
description: Optional[str] = None,
metadata: Optional[Dict] = None,
ip_address: Optional[str] = None,
user_agent: Optional[str] = None):
"""记录用户活动"""
activity = UserActivity(
user_id=user_id,
activity_type=activity_type,
resource_type=resource_type,
resource_id=resource_id,
description=description,
metadata=metadata,
ip_address=ip_address,
user_agent=user_agent
)
session.add(activity)
await session.commit()
logger.debug(f"用户活动记录: {activity_type} by user {user_id}")
async def get_stp_file_with_data(self, session: AsyncSession,
stp_file_id: int) -> Dict[str, Any]:
"""获取STP文件及其所有关联数据"""
# 1. 获取STP文件记录
stp_file = await session.get(STPFile, stp_file_id)
if not stp_file:
raise ValueError(f"STP文件不存在: {stp_file_id}")
result = {
'metadata': {
'id': stp_file.id,
'original_filename': stp_file.original_filename,
'file_size': stp_file.file_size,
'file_hash': stp_file.file_hash,
'upload_time': stp_file.upload_time.isoformat() if stp_file.upload_time else None,
'status': stp_file.status,
'user_id': stp_file.user_id
},
'geometry_data': None,
'mold_cavity_data': None,
'html_file': None,
'features': [],
'recommendations': []
}
# 2. 从MinIO获取数据
try:
# 几何数据
if stp_file.geometry_data:
geo_data_bytes = await storage_manager.download_file(
'geometry_data',
stp_file.geometry_data.object_key
)
result['geometry_data'] = json.loads(geo_data_bytes.decode('utf-8'))
# 模具型腔数据
if stp_file.mold_cavity_data:
cavity_data_bytes = await storage_manager.download_file(
'mold_cavities',
stp_file.mold_cavity_data.detailed_object_key
)
result['mold_cavity_data'] = json.loads(cavity_data_bytes.decode('utf-8'))
# HTML文件
if stp_file.html_file:
html_bytes = await storage_manager.download_file(
'html_files',
stp_file.html_file.object_key
)
result['html_content'] = html_bytes.decode('utf-8')
except Exception as e:
logger.error(f"从MinIO获取数据失败: {e}")
# 3. 从PostgreSQL获取特征和建议
features = await session.execute(
select(FeatureDetection).where(FeatureDetection.stp_file_id == stp_file_id)
)
result['features'] = [
{
'feature_type': f.feature_type,
'confidence': f.confidence,
'location': f.location,
'dimensions': f.dimensions,
'parameters': f.parameters
}
for f in features.scalars().all()
]
recommendations = await session.execute(
select(DesignRecommendation).where(DesignRecommendation.stp_file_id == stp_file_id)
)
result['recommendations'] = [
{
'rec_type': r.rec_type,
'priority': r.priority,
'description': r.description,
'reason': r.reason,
'parameters': r.parameters
}
for r in recommendations.scalars().all()
]
return result
async def delete_stp_file_cascade(self, session: AsyncSession,
stp_file_id: int):
"""级联删除STP文件及其所有关联数据"""
stp_file = await session.get(STPFile, stp_file_id)
if not stp_file:
raise ValueError(f"STP文件不存在: {stp_file_id}")
# 1. 删除MinIO中的文件
try:
if stp_file.object_key:
await storage_manager.delete_file('stp_files', stp_file.object_key)
except Exception as e:
logger.error(f"删除MinIO文件失败: {e}")
try:
if stp_file.geometry_data:
await storage_manager.delete_file('geometry_data', stp_file.geometry_data.object_key)
except Exception as e:
logger.error(f"删除几何数据失败: {e}")
try:
if stp_file.mold_cavity_data:
await storage_manager.delete_file('mold_cavities', stp_file.mold_cavity_data.detailed_object_key)
except Exception as e:
logger.error(f"删除型腔数据失败: {e}")
try:
if stp_file.html_file:
await storage_manager.delete_file('html_files', stp_file.html_file.object_key)
except Exception as e:
logger.error(f"删除HTML文件失败: {e}")
# 2. 级联删除PostgreSQL记录(通过外键自动处理)
await session.delete(stp_file)
await session.commit()
logger.info(f"STP文件及其关联数据已删除: {stp_file_id}")
# 全局存储集成服务实例
storage_integration = StorageIntegrationService()
+498
View File
@@ -0,0 +1,498 @@
# services/storage_integration_rustfs.py
"""存储集成服务 - 协调 PostgreSQL 和 RustFS"""
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from pathlib import Path
from typing import Optional, Dict, Any
import json
from datetime import datetime
from models.database import (
STPFile, GeometryData, MoldCavityData,
HTMLFile, ProcessingTask, User,
FeatureDetection, DesignRecommendation,
UserActivity, SystemLog
)
from storage.rustfs_storage import rustfs_manager
from utils.logger import get_logger
logger = get_logger(__name__)
class StorageIntegrationService:
"""存储集成服务 - PostgreSQL + RustFS"""
async def save_stp_file(self, session: AsyncSession,
file_path: Path,
original_filename: str,
user_id: Optional[int] = None) -> STPFile:
"""保存STP文件到PostgreSQL元数据 + RustFS对象存储"""
# 1. 上传到RustFS
upload_result = await rustfs_manager.upload_file(
file_type='stp_files',
file_path=file_path,
original_filename=original_filename,
metadata={
'original_filename': original_filename,
'user_id': str(user_id) if user_id else 'anonymous'
}
)
file_hash = upload_result['file_hash']
# 2. 检查是否已存在相同文件
existing_file = await session.execute(
select(STPFile).where(STPFile.file_hash == file_hash)
)
existing_file = existing_file.scalar_one_or_none()
if existing_file:
logger.info(f"文件已存在,返回现有记录: {existing_file.id}")
return existing_file
# 3. 创建新PostgreSQL记录
stp_file = STPFile(
user_id=user_id,
object_key=upload_result['object_key'],
storage_bucket=upload_result['bucket'],
original_filename=original_filename,
file_size=upload_result['file_size'],
file_hash=file_hash,
status="uploaded",
file_path=str(file_path) # 保留本地路径以兼容
)
session.add(stp_file)
await session.commit()
await session.refresh(stp_file)
logger.info(f"STP文件保存成功 RustFS: {stp_file.id}")
return stp_file
async def create_processing_task(self, session: AsyncSession,
task_id: str,
stp_file_id: int,
task_type: str = "stp_parsing") -> ProcessingTask:
"""创建处理任务记录"""
try:
task = ProcessingTask(
task_id=task_id,
stp_file_id=stp_file_id,
task_type=task_type,
status="pending",
started_time=datetime.now()
)
session.add(task)
await session.commit()
await session.refresh(task)
logger.info(f"处理任务创建成功: {task_id}")
return task
except Exception as e:
await session.rollback()
logger.error(f"创建处理任务失败: {e}")
raise
async def update_task_status(
self,
session: AsyncSession,
task_id: str,
status: str,
progress: Optional[int] = None,
current_step: Optional[str] = None,
error_message: Optional[str] = None
):
"""更新任务状态"""
try:
update_data = {
"status": status,
"completed_time": datetime.now() if status in ["completed", "failed"] else None,
"error_message": error_message
}
if progress is not None:
update_data["progress"] = progress
if current_step is not None:
update_data["current_step"] = current_step
await session.execute(
update(ProcessingTask)
.where(ProcessingTask.task_id == task_id)
.values(**update_data)
)
await session.commit()
logger.info(f"任务状态更新: {task_id} -> {status}")
except Exception as e:
await session.rollback()
logger.error(f"更新任务状态失败: {e}")
raise
async def update_stp_file_status(self, session: AsyncSession, stp_file_id: int, status: str):
"""更新STP文件状态"""
try:
await session.execute(
update(STPFile)
.where(STPFile.id == stp_file_id)
.values(
status=status,
processed_time=datetime.now() if status in ["completed", "failed"] else None
)
)
await session.commit()
logger.info(f"STP文件状态更新: ID {stp_file_id} -> {status}")
except Exception as e:
await session.rollback()
logger.error(f"更新STP文件状态失败: {e}")
raise
async def save_geometry_data(self, session: AsyncSession,
stp_file_id: int,
geometry_json: Dict[str, Any],
analysis_method: str = "pythonocc") -> GeometryData:
"""保存几何数据到PostgreSQL元数据 + RustFS对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 上传到RustFS
upload_result = await rustfs_manager.upload_json_data(
file_type='geometry_data',
json_data=geometry_json,
file_hash=file_hash
)
# 3. 提取几何数据
if 'geometry_data' in geometry_json:
geo_data = geometry_json['geometry_data']
else:
geo_data = geometry_json
# 4. 创建PostgreSQL记录
geometry_data = GeometryData(
stp_file_id=stp_file_id,
object_key=upload_result['object_key'],
storage_bucket=upload_result['bucket'],
analysis_method=analysis_method,
# 提取摘要字段
volume=geo_data.get('volume'),
surface_area=geo_data.get('surface_area'),
bounding_box_min=geo_data.get('bounding_box', {}).get('min'),
bounding_box_max=geo_data.get('bounding_box', {}).get('max'),
center_of_mass=geo_data.get('center_of_mass'),
topology_faces=geo_data.get('topology', {}).get('faces'),
topology_edges=geo_data.get('topology', {}).get('edges'),
topology_vertices=geo_data.get('topology', {}).get('vertices')
)
session.add(geometry_data)
await session.commit()
await session.refresh(geometry_data)
logger.info(f"几何数据保存成功 RustFS: {geometry_data.id}")
return geometry_data
async def save_mold_cavity_data(self, session: AsyncSession,
stp_file_id: int,
cavity_json: Dict[str, Any]) -> MoldCavityData:
"""保存模具型腔数据到PostgreSQL元数据 + RustFS对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 上传到RustFS
upload_result = await rustfs_manager.upload_json_data(
file_type='mold_cavities',
json_data=cavity_json,
file_hash=file_hash
)
# 3. 提取关键信息
metadata = cavity_json.get('metadata', {})
product_analysis = cavity_json.get('product_analysis', {})
manufacturing_info = cavity_json.get('manufacturing_info', {})
mold_size = manufacturing_info.get('estimated_mold_size', {})
key_info = cavity_json.get('mold_cavities', {}).get('cavity_key_info', {})
# 4. 创建PostgreSQL记录
mold_cavity = MoldCavityData(
stp_file_id=stp_file_id,
detailed_object_key=upload_result['object_key'],
storage_bucket=upload_result['bucket'],
# 模具参数
mold_material=manufacturing_info.get('recommended_material', 'Aluminum Alloy 7075'),
shrinkage_rate=metadata.get('shrinkage_rate', 0.005),
draft_angle=metadata.get('draft_angle', 2.0),
# 提取的摘要字段
cavity_key_info=key_info,
mold_size_length=mold_size.get('length'),
mold_size_width=mold_size.get('width'),
mold_size_height=mold_size.get('height'),
estimated_clamping_force=manufacturing_info.get('estimated_clamping_force'),
product_volume=product_analysis.get('volume'),
# 从key_info中提取(如果存在)
product_weight=key_info.get('geometric_characteristics', {}).get('product_weight'),
wall_thickness_range=key_info.get('geometric_characteristics', {}).get('wall_thickness_range'),
complexity_score=key_info.get('geometric_characteristics', {}).get('complexity_score'),
# 质量评估
weld_line_risk=key_info.get('quality_considerations', {}).get('potential_weld_lines'),
sink_mark_risk=key_info.get('quality_considerations', {}).get('sink_mark_areas'),
warpage_risk=key_info.get('quality_considerations', {}).get('warpage_risk')
)
session.add(mold_cavity)
await session.commit()
await session.refresh(mold_cavity)
logger.info(f"模具型腔数据保存成功 RustFS: {mold_cavity.id}")
return mold_cavity
async def save_html_file(self, session: AsyncSession,
stp_file_id: int,
filename: str,
file_path: str,
html_content: Optional[str] = None,
visualization_type: str = "3d_viewer") -> HTMLFile:
"""保存HTML文件到PostgreSQL元数据 + RustFS对象存储"""
# 1. 获取文件哈希
stp_file = await session.get(STPFile, stp_file_id)
file_hash = stp_file.file_hash
# 2. 读取HTML内容(如果未提供)
if html_content is None:
try:
with open(file_path, 'r', encoding='utf-8') as f:
html_content = f.read()
except Exception as e:
logger.error(f"读取HTML文件失败: {e}")
html_content = ""
# 3. 上传到RustFS
html_json = {'content': html_content, 'filename': filename}
upload_result = await rustfs_manager.upload_json_data(
file_type='html_files',
json_data=html_json,
file_hash=file_hash
)
# 4. 创建PostgreSQL记录
html_file = HTMLFile(
stp_file_id=stp_file_id,
object_key=upload_result['object_key'],
storage_bucket=upload_result['bucket'],
filename=filename,
file_path=file_path, # 保留本地路径
html_content=html_content, # 保留内容以兼容
visualization_type=visualization_type
)
session.add(html_file)
await session.commit()
await session.refresh(html_file)
logger.info(f"HTML文件保存成功 RustFS: {html_file.id}")
return html_file
async def save_features_and_recommendations(
self, session: AsyncSession,
stp_file_id: int,
features: list,
recommendations: list
):
"""保存特征检测结果和设计建议"""
# 1. 保存特征
for feature in features:
feature_record = FeatureDetection(
stp_file_id=stp_file_id,
feature_type=feature.get('feature_type'),
confidence=feature.get('confidence'),
location=feature.get('location'),
dimensions=feature.get('dimensions'),
parameters=feature.get('parameters')
)
session.add(feature_record)
# 2. 保存建议
for rec in recommendations:
rec_record = DesignRecommendation(
stp_file_id=stp_file_id,
rec_type=rec.get('rec_type'),
priority=rec.get('priority'),
description=rec.get('description'),
reason=rec.get('reason'),
parameters=rec.get('parameters')
)
session.add(rec_record)
await session.commit()
logger.info(f"保存了 {len(features)} 个特征和 {len(recommendations)} 个建议")
async def log_user_activity(self, session: AsyncSession,
user_id: int,
activity_type: str,
resource_type: Optional[str] = None,
resource_id: Optional[int] = None,
description: Optional[str] = None,
metadata: Optional[Dict] = None,
ip_address: Optional[str] = None,
user_agent: Optional[str] = None):
"""记录用户活动"""
activity = UserActivity(
user_id=user_id,
activity_type=activity_type,
resource_type=resource_type,
resource_id=resource_id,
description=description,
metadata=metadata,
ip_address=ip_address,
user_agent=user_agent
)
session.add(activity)
await session.commit()
logger.debug(f"用户活动记录: {activity_type} by user {user_id}")
async def get_stp_file_with_data(self, session: AsyncSession,
stp_file_id: int) -> Dict[str, Any]:
"""获取STP文件及其所有关联数据"""
# 1. 获取STP文件记录
stp_file = await session.get(STPFile, stp_file_id)
if not stp_file:
raise ValueError(f"STP文件不存在: {stp_file_id}")
result = {
'metadata': {
'id': stp_file.id,
'original_filename': stp_file.original_filename,
'file_size': stp_file.file_size,
'file_hash': stp_file.file_hash,
'upload_time': stp_file.upload_time.isoformat() if stp_file.upload_time else None,
'status': stp_file.status,
'user_id': stp_file.user_id
},
'geometry_data': None,
'mold_cavity_data': None,
'html_content': None,
'features': [],
'recommendations': []
}
# 2. 从RustFS获取数据
try:
# 几何数据
if stp_file.geometry_data:
geo_data_bytes = await rustfs_manager.download_file(
file_type='geometry_data',
object_key=stp_file.geometry_data.object_key
)
result['geometry_data'] = json.loads(geo_data_bytes.decode('utf-8'))
# 模具型腔数据
if stp_file.mold_cavity_data:
cavity_data_bytes = await rustfs_manager.download_file(
file_type='mold_cavities',
object_key=stp_file.mold_cavity_data.detailed_object_key
)
result['mold_cavity_data'] = json.loads(cavity_data_bytes.decode('utf-8'))
# HTML文件
if stp_file.html_file:
html_bytes = await rustfs_manager.download_file(
file_type='html_files',
object_key=stp_file.html_file.object_key
)
html_json = json.loads(html_bytes.decode('utf-8'))
result['html_content'] = html_json.get('content', '')
except Exception as e:
logger.error(f"从RustFS获取数据失败: {e}")
# 3. 从PostgreSQL获取特征和建议
features = await session.execute(
select(FeatureDetection).where(FeatureDetection.stp_file_id == stp_file_id)
)
result['features'] = [
{
'feature_type': f.feature_type,
'confidence': f.confidence,
'location': f.location,
'dimensions': f.dimensions,
'parameters': f.parameters
}
for f in features.scalars().all()
]
recommendations = await session.execute(
select(DesignRecommendation).where(DesignRecommendation.stp_file_id == stp_file_id)
)
result['recommendations'] = [
{
'rec_type': r.rec_type,
'priority': r.priority,
'description': r.description,
'reason': r.reason,
'parameters': r.parameters
}
for r in recommendations.scalars().all()
]
return result
async def delete_stp_file_cascade(self, session: AsyncSession,
stp_file_id: int):
"""级联删除STP文件及其所有关联数据"""
stp_file = await session.get(STPFile, stp_file_id)
if not stp_file:
raise ValueError(f"STP文件不存在: {stp_file_id}")
# 1. 删除RustFS中的文件
try:
if stp_file.object_key:
await rustfs_manager.delete_file('stp_files', stp_file.object_key)
except Exception as e:
logger.error(f"删除RustFS文件失败: {e}")
try:
if stp_file.geometry_data:
await rustfs_manager.delete_file('geometry_data', stp_file.geometry_data.object_key)
except Exception as e:
logger.error(f"删除几何数据失败: {e}")
try:
if stp_file.mold_cavity_data:
await rustfs_manager.delete_file('mold_cavities', stp_file.mold_cavity_data.detailed_object_key)
except Exception as e:
logger.error(f"删除型腔数据失败: {e}")
try:
if stp_file.html_file:
await rustfs_manager.delete_file('html_files', stp_file.html_file.object_key)
except Exception as e:
logger.error(f"删除HTML文件失败: {e}")
# 2. 级联删除PostgreSQL记录(通过外键自动处理)
await session.delete(stp_file)
await session.commit()
logger.info(f"STP文件及其关联数据已删除: {stp_file_id}")
# 全局存储集成服务实例
storage_integration = StorageIntegrationService()
+296
View File
@@ -0,0 +1,296 @@
# services/storage_service.py
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy import select, update
from datetime import datetime
import hashlib
import json
from pathlib import Path
from typing import Optional, Dict, Any
from models.database import STPFile, GeometryData, HTMLFile, ProcessingTask
from utils.logger import get_logger
from models.database import MoldCavityData
logger = get_logger(__name__)
class StorageService:
"""数据存储服务"""
def __init__(self, db_session: AsyncSession):
self.db_session = db_session
async def save_stp_file(
self,
filename: str,
file_path: str,
file_size: int,
file_content: Optional[bytes] = None
) -> STPFile:
"""保存STP文件信息到数据库"""
try:
# 计算文件哈希
file_hash = self._calculate_file_hash(file_path, file_content)
# 检查是否已存在相同文件
existing_file = await self.db_session.execute(
select(STPFile).where(STPFile.file_hash == file_hash)
)
existing_file = existing_file.scalar_one_or_none()
if existing_file:
logger.info(f"文件已存在,跳过保存: {filename}")
return existing_file
# 创建新的STP文件记录
stp_file = STPFile(
filename=filename,
original_filename=filename,
file_path=file_path,
file_size=file_size,
file_hash=file_hash,
file_content=file_content,
upload_time=datetime.now(),
status="pending",
# 必填字段提供默认值
object_key=f"stp_files/{file_hash}",
storage_bucket="default",
object_url=None
)
self.db_session.add(stp_file)
await self.db_session.commit()
await self.db_session.refresh(stp_file)
logger.info(f"STP文件保存成功: {filename} (ID: {stp_file.id})")
return stp_file
except Exception as e:
await self.db_session.rollback()
logger.error(f"保存STP文件失败: {e}")
raise
async def save_geometry_data(
self,
stp_file_id: int,
geometry_json: Dict[str, Any],
analysis_method: str
) -> GeometryData:
"""保存几何数据JSON到数据库"""
try:
# 提取关键几何属性用于快速查询
volume = geometry_json.get("volume")
surface_area = geometry_json.get("surface_area")
bounding_box = geometry_json.get("bounding_box", {})
geometry_data = GeometryData(
stp_file_id=stp_file_id,
analysis_method=analysis_method,
volume=volume,
surface_area=surface_area,
bounding_box_min=bounding_box.get("min"),
bounding_box_max=bounding_box.get("max"),
created_time=datetime.now(),
# 必填字段提供默认值
object_key=f"geometry_data/{stp_file_id}",
storage_bucket="default",
object_url=None
)
self.db_session.add(geometry_data)
await self.db_session.commit()
await self.db_session.refresh(geometry_data)
logger.info(f"几何数据保存成功: STP文件ID {stp_file_id}")
return geometry_data
except Exception as e:
await self.db_session.rollback()
logger.error(f"保存几何数据失败: {e}")
raise
async def save_html_file(
self,
stp_file_id: int,
filename: str,
file_path: str,
html_content: Optional[str] = None,
visualization_type: str = "3d_viewer"
) -> HTMLFile:
"""保存HTML文件信息到数据库"""
try:
html_file = HTMLFile(
stp_file_id=stp_file_id,
filename=filename,
file_path=file_path,
html_content=html_content,
visualization_type=visualization_type,
has_interactive_elements=True,
generated_time=datetime.now(),
# 必填字段提供默认值
object_key=f"html_files/{stp_file_id}",
storage_bucket="default",
object_url=None
)
self.db_session.add(html_file)
await self.db_session.commit()
await self.db_session.refresh(html_file)
logger.info(f"HTML文件保存成功: {filename} (STP文件ID: {stp_file_id})")
return html_file
except Exception as e:
await self.db_session.rollback()
logger.error(f"保存HTML文件失败: {e}")
raise
async def create_processing_task(
self,
task_id: str,
stp_file_id: int,
task_type: str = "stp_parsing"
) -> ProcessingTask:
"""创建处理任务记录"""
try:
task = ProcessingTask(
task_id=task_id,
stp_file_id=stp_file_id,
task_type=task_type,
status="pending",
started_time=datetime.now()
)
self.db_session.add(task)
await self.db_session.commit()
await self.db_session.refresh(task)
logger.info(f"处理任务创建成功: {task_id}")
return task
except Exception as e:
await self.db_session.rollback()
logger.error(f"创建处理任务失败: {e}")
raise
async def update_task_status(
self,
task_id: str,
status: str,
progress: Optional[int] = None,
current_step: Optional[str] = None,
error_message: Optional[str] = None
):
"""更新任务状态"""
try:
update_data = {
"status": status,
"completed_time": datetime.now() if status in ["completed", "failed"] else None,
"error_message": error_message
}
if progress is not None:
update_data["progress"] = progress
if current_step is not None:
update_data["current_step"] = current_step
await self.db_session.execute(
update(ProcessingTask)
.where(ProcessingTask.task_id == task_id)
.values(**update_data)
)
await self.db_session.commit()
logger.info(f"任务状态更新: {task_id} -> {status}")
except Exception as e:
await self.db_session.rollback()
logger.error(f"更新任务状态失败: {e}")
raise
async def update_stp_file_status(self, stp_file_id: int, status: str):
"""更新STP文件状态"""
try:
await self.db_session.execute(
update(STPFile)
.where(STPFile.id == stp_file_id)
.values(
status=status,
processed_time=datetime.now() if status in ["completed", "failed"] else None
)
)
await self.db_session.commit()
logger.info(f"STP文件状态更新: ID {stp_file_id} -> {status}")
except Exception as e:
await self.db_session.rollback()
logger.error(f"更新STP文件状态失败: {e}")
raise
async def get_stp_file_by_id(self, stp_file_id: int) -> Optional[STPFile]:
"""根据ID获取STP文件"""
try:
result = await self.db_session.execute(
select(STPFile).where(STPFile.id == stp_file_id)
)
return result.scalar_one_or_none()
except Exception as e:
logger.error(f"获取STP文件失败: {e}")
return None
async def get_geometry_data_by_stp_file_id(self, stp_file_id: int) -> Optional[GeometryData]:
"""根据STP文件ID获取几何数据"""
try:
result = await self.db_session.execute(
select(GeometryData).where(GeometryData.stp_file_id == stp_file_id)
)
return result.scalar_one_or_none()
except Exception as e:
logger.error(f"获取几何数据失败: {e}")
return None
def _calculate_file_hash(self, file_path: str, file_content: Optional[bytes] = None) -> str:
"""计算文件哈希值"""
sha256_hash = hashlib.sha256()
if file_content:
sha256_hash.update(file_content)
else:
# 从文件路径读取内容计算哈希
with open(file_path, "rb") as f:
for chunk in iter(lambda: f.read(4096), b""):
sha256_hash.update(chunk)
return sha256_hash.hexdigest()
async def save_mold_cavity_data(
self,
stp_file_id: int,
cavity_json: Dict[str, Any],
key_info: Dict[str, Any]
) -> MoldCavityData:
"""保存模具型腔数据"""
try:
mold_data = MoldCavityData(
stp_file_id=stp_file_id,
cavity_key_info=key_info,
shrinkage_rate=cavity_json["metadata"]["shrinkage_rate"],
draft_angle=cavity_json["metadata"]["draft_angle"],
generated_time=datetime.now(),
# 必填字段提供默认值
detailed_object_key=f"mold_cavity/{stp_file_id}",
storage_bucket="default"
)
self.db_session.add(mold_data)
await self.db_session.commit()
await self.db_session.refresh(mold_data)
logger.info(f"模具型腔数据保存成功: STP文件ID {stp_file_id}")
return mold_data
except Exception as e:
await self.db_session.rollback()
logger.error(f"保存模具型腔数据失败: {e}")
raise
+5
View File
@@ -0,0 +1,5 @@
# storage/__init__.py
from .rustfs_storage import RustFSManager, rustfs_manager
__all__ = ['RustFSManager', 'rustfs_manager']
+107
View File
@@ -0,0 +1,107 @@
# storage/init_storage.py
"""初始化 RustFS 对象存储"""
import asyncio
import sys
from pathlib import Path
# 添加项目根目录和 src 目录到 Python 路径
project_root = Path(__file__).parent.parent.parent
src_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))
sys.path.insert(0, str(src_root))
from storage.rustfs_storage import rustfs_manager
from config.settings import settings
from utils.logger import get_logger
logger = get_logger(__name__)
async def init_rustfs_storage():
"""初始化 RustFS 对象存储"""
try:
# 连接到 RustFS (S3v4 API)
await rustfs_manager.connect(
endpoint=settings.RUSTFS_ENDPOINT,
access_key=settings.RUSTFS_ACCESS_KEY,
secret_key=settings.RUSTFS_SECRET_KEY,
timeout=settings.RUSTFS_TIMEOUT
)
logger.info("RustFS 对象存储初始化完成")
return True
except Exception as e:
logger.error(f"RustFS 对象存储初始化失败: {e}")
return False
async def test_storage():
"""测试 RustFS 对象存储功能"""
try:
import json
# 测试上传 JSON
test_data = {"test": True, "timestamp": "2024-01-01", "storage": "rustfs"}
result = await rustfs_manager.upload_json_data(
file_type='stp_files',
json_data=test_data,
file_hash='test-hash'
)
logger.info(f"RustFS 测试上传成功: {result['object_key']}")
# 测试下载
downloaded_bytes = await rustfs_manager.download_file(
file_type='stp_files',
object_key=result['object_key']
)
downloaded_data = json.loads(downloaded_bytes.decode('utf-8'))
logger.info(f"RustFS 测试下载成功: {downloaded_data}")
# 测试预签名 URL
url = await rustfs_manager.generate_presigned_url(
file_type='stp_files',
object_key=result['object_key'],
expires=3600
)
logger.info(f"RustFS 预签名URL: {url}")
# 清理测试文件
await rustfs_manager.delete_file(
file_type='stp_files',
object_key=result['object_key']
)
logger.info("RustFS 测试文件已清理")
return True
except Exception as e:
logger.error(f"RustFS 存储测试失败: {e}")
return False
if __name__ == "__main__":
async def main():
try:
print("=== 初始化 RustFS 对象存储 ===")
# 初始化存储
init_result = await init_rustfs_storage()
if init_result:
print("[OK] RustFS 连接成功")
print("\n=== 测试 RustFS 功能 ===")
# 运行测试
test_result = await test_storage()
if test_result:
print("[OK] RustFS 测试全部通过")
else:
print("[FAIL] RustFS 测试失败")
finally:
# 关闭连接
await rustfs_manager.close()
print("\n=== 连接已关闭 ===")
# 运行主函数
asyncio.run(main())
+361
View File
@@ -0,0 +1,361 @@
# storage/object_storage.py
"""MinIO/S3 对象存储服务"""
from minio import Minio
from minio.error import S3Error
from pathlib import Path
from typing import Optional, BinaryIO
from io import BytesIO
from utils.logger import get_logger
import hashlib
import uuid
logger = get_logger(__name__)
class ObjectStorageManager:
"""对象存储管理器 - MinIO/S3兼容"""
def __init__(self):
self.client: Optional[Minio] = None
self.is_connected = False
# 桶名称
self.buckets = {
'stp_files': 'moldinsight-stp-files', # STP/STEP文件
'geometry_data': 'moldinsight-geometry', # 几何数据JSON
'mold_cavities': 'moldinsight-mold-cavities', # 模具型腔数据
'html_files': 'moldinsight-html', # HTML报告文件
'user_files': 'moldinsight-user-files' # 用户上传的其他文件
}
async def connect(self, endpoint: str, access_key: str, secret_key: str,
secure: bool = False):
"""连接到MinIO/S3服务"""
try:
self.client = Minio(
endpoint,
access_key=access_key,
secret_key=secret_key,
secure=secure
)
# 测试连接
self.client.list_buckets()
self.is_connected = True
logger.info(f"对象存储连接成功: {endpoint}")
# 确保所有桶都存在
await self._ensure_buckets()
except S3Error as e:
logger.error(f"对象存储连接失败: {e}")
self.is_connected = False
raise
async def _ensure_buckets(self):
"""确保所有必要的桶都存在"""
for bucket_name in self.buckets.values():
try:
if not self.client.bucket_exists(bucket_name):
self.client.make_bucket(bucket_name)
logger.info(f"创建存储桶: {bucket_name}")
else:
logger.debug(f"存储桶已存在: {bucket_name}")
except S3Error as e:
logger.error(f"创建存储桶失败 {bucket_name}: {e}")
def _generate_object_key(self, original_filename: str, prefix: str = '') -> str:
"""生成对象存储的唯一键名"""
# 提取文件扩展名
ext = Path(original_filename).suffix
# 生成唯一ID
unique_id = str(uuid.uuid4())
# 生成键名: prefix/unique_id + original_ext
if prefix:
return f"{prefix}/{unique_id}{ext}"
return f"{unique_id}{ext}"
async def upload_stp_file(self, file_path: Path,
original_filename: str) -> dict:
"""上传STP文件到对象存储"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets['stp_files']
# 计算文件哈希
file_hash = self._calculate_file_hash(file_path)
# 检查是否已存在
existing_key = await self._find_file_by_hash(bucket_name, file_hash)
if existing_key:
logger.info(f"文件已存在,跳过上传: {existing_key}")
return {
'object_key': existing_key,
'file_hash': file_hash,
'already_exists': True
}
# 生成唯一键名
object_key = self._generate_object_key(
original_filename,
prefix='stp'
)
# 上传文件
try:
result = self.client.fput_object(
bucket_name,
object_key,
str(file_path),
content_type='application/octet-stream'
)
logger.info(f"STP文件上传成功: {object_key}")
return {
'object_key': object_key,
'file_hash': file_hash,
'file_size': result.size,
'etag': result.etag,
'already_exists': False
}
except S3Error as e:
logger.error(f"STP文件上传失败: {e}")
raise
async def upload_geometry_data(self, geometry_json: dict,
file_hash: str) -> dict:
"""上传几何数据JSON到对象存储"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets['geometry_data']
# 使用文件哈希作为键名的一部分
object_key = f"geometry/{file_hash}.json"
# 转换为字节
import json
json_bytes = json.dumps(geometry_json, ensure_ascii=False).encode('utf-8')
# 上传
try:
result = self.client.put_object(
bucket_name,
object_key,
BytesIO(json_bytes),
length=len(json_bytes),
content_type='application/json'
)
logger.info(f"几何数据上传成功: {object_key}")
return {
'object_key': object_key,
'file_size': result.size,
'etag': result.etag
}
except S3Error as e:
logger.error(f"几何数据上传失败: {e}")
raise
async def upload_mold_cavity_data(self, cavity_json: dict,
file_hash: str) -> dict:
"""上传模具型腔数据到对象存储"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets['mold_cavities']
object_key = f"mold-cavity/{file_hash}.json"
import json
json_bytes = json.dumps(cavity_json, ensure_ascii=False).encode('utf-8')
try:
result = self.client.put_object(
bucket_name,
object_key,
BytesIO(json_bytes),
length=len(json_bytes),
content_type='application/json'
)
logger.info(f"模具型腔数据上传成功: {object_key}")
return {
'object_key': object_key,
'file_size': result.size,
'etag': result.etag
}
except S3Error as e:
logger.error(f"模具型腔数据上传失败: {e}")
raise
async def upload_html_file(self, html_content: str,
original_filename: str,
file_hash: str) -> dict:
"""上传HTML文件到对象存储"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets['html_files']
object_key = f"html/{file_hash}.html"
html_bytes = html_content.encode('utf-8')
try:
result = self.client.put_object(
bucket_name,
object_key,
BytesIO(html_bytes),
length=len(html_bytes),
content_type='text/html; charset=utf-8'
)
logger.info(f"HTML文件上传成功: {object_key}")
return {
'object_key': object_key,
'file_size': result.size,
'etag': result.etag
}
except S3Error as e:
logger.error(f"HTML文件上传失败: {e}")
raise
async def download_file(self, bucket_type: str,
object_key: str) -> bytes:
"""从对象存储下载文件"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets.get(bucket_type)
if not bucket_name:
raise ValueError(f"未知的桶类型: {bucket_type}")
try:
response = self.client.get_object(bucket_name, object_key)
data = response.read()
response.close()
response.release_conn()
logger.debug(f"文件下载成功: {object_key}")
return data
except S3Error as e:
logger.error(f"文件下载失败 {object_key}: {e}")
raise
async def get_presigned_url(self, bucket_type: str,
object_key: str,
expires: int = 3600) -> str:
"""生成预签名URL(临时访问链接)"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets.get(bucket_type)
if not bucket_name:
raise ValueError(f"未知的桶类型: {bucket_type}")
try:
url = self.client.presigned_get_object(
bucket_name,
object_key,
expires=expires
)
return url
except S3Error as e:
logger.error(f"生成预签名URL失败: {e}")
raise
async def delete_file(self, bucket_type: str, object_key: str):
"""删除对象存储中的文件"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets.get(bucket_type)
if not bucket_name:
raise ValueError(f"未知的桶类型: {bucket_type}")
try:
self.client.remove_object(bucket_name, object_key)
logger.info(f"文件删除成功: {object_key}")
except S3Error as e:
logger.error(f"文件删除失败 {object_key}: {e}")
raise
def _calculate_file_hash(self, file_path: Path) -> str:
"""计算文件的SHA256哈希"""
sha256_hash = hashlib.sha256()
with open(file_path, 'rb') as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
async def _find_file_by_hash(self, bucket_name: str,
file_hash: str) -> Optional[str]:
"""根据哈希查找已存在的文件"""
try:
objects = self.client.list_objects(bucket_name, recursive=True)
for obj in objects:
# 从对象键中提取哈希(如果有)
if file_hash in obj.object_name:
return obj.object_name
return None
except S3Error as e:
logger.warning(f"查找文件哈希失败: {e}")
return None
async def get_file_info(self, bucket_type: str,
object_key: str) -> dict:
"""获取文件信息"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets.get(bucket_type)
if not bucket_name:
raise ValueError(f"未知的桶类型: {bucket_type}")
try:
stat = self.client.stat_object(bucket_name, object_key)
return {
'size': stat.size,
'etag': stat.etag,
'content_type': stat.content_type,
'last_modified': stat.last_modified
}
except S3Error as e:
logger.error(f"获取文件信息失败: {e}")
raise
async def list_files(self, bucket_type: str,
prefix: str = '') -> list:
"""列出存储桶中的文件"""
if not self.is_connected:
raise RuntimeError("对象存储未连接")
bucket_name = self.buckets.get(bucket_type)
if not bucket_name:
raise ValueError(f"未知的桶类型: {bucket_type}")
try:
objects = self.client.list_objects(bucket_name, prefix=prefix)
return [
{
'object_key': obj.object_name,
'size': obj.size,
'etag': obj.etag,
'last_modified': obj.last_modified
}
for obj in objects
]
except S3Error as e:
logger.error(f"列出文件失败: {e}")
raise
# 全局对象存储管理器实例
storage_manager = ObjectStorageManager()
+348
View File
@@ -0,0 +1,348 @@
# storage/rustfs_storage.py
"""RustFS 对象存储服务 (S3v4 API 兼容)"""
from minio import Minio
from minio.error import S3Error
from pathlib import Path
from typing import Optional, Dict, Any
from io import BytesIO
from utils.logger import get_logger
from datetime import timedelta
import hashlib
import uuid
logger = get_logger(__name__)
class RustFSManager:
"""RustFS 对象存储管理器 (使用 MinIO S3 客户端)"""
def __init__(self, project_name: str = "moldinsight"):
self.client: Optional[Minio] = None
self.is_connected = False
self.project_name = project_name
# 使用单个项目桶,按类型组织文件
self.bucket_name = f"{project_name}"
# 文件类型前缀(子目录结构)
self.file_types = {
'stp_files': 'stp-files',
'geometry_data': 'geometry',
'mold_cavities': 'mold-cavities',
'html_files': 'html',
'user_files': 'user-files'
}
async def connect(self, endpoint: str, access_key: str, secret_key: str, timeout: int = 30):
"""连接到 RustFS 服务"""
try:
# 提取端口号和主机
from urllib.parse import urlparse
parsed = urlparse(endpoint)
host = parsed.netloc or parsed.path
# 创建 MinIO 客户端(S3v4 兼容)
self.client = Minio(
host,
access_key=access_key,
secret_key=secret_key,
secure=False, # HTTP 而不是 HTTPS
region='us-east-1'
)
# 测试连接
self.client.list_buckets()
self.is_connected = True
logger.info(f"RustFS 连接成功: {endpoint}")
# 确保所有桶都存在
await self._ensure_buckets()
except S3Error as e:
logger.error(f"RustFS 连接失败: {e}")
self.is_connected = False
raise
except Exception as e:
logger.error(f"RustFS 初始化失败: {e}")
self.is_connected = False
raise
async def close(self):
"""关闭连接"""
# MinIO 客户端不需要显式关闭
self.is_connected = False
logger.info("RustFS 连接已关闭")
async def _ensure_buckets(self):
"""确保项目存储桶存在"""
try:
if not self.client.bucket_exists(self.bucket_name):
self.client.make_bucket(self.bucket_name)
logger.info(f"创建项目存储桶: {self.bucket_name}")
else:
logger.debug(f"项目存储桶已存在: {self.bucket_name}")
except S3Error as e:
logger.error(f"创建存储桶失败 {self.bucket_name}: {e}")
def _generate_object_key(self, original_filename: str, file_type: str = '') -> str:
"""生成对象存储的唯一键名"""
ext = Path(original_filename).suffix
unique_id = str(uuid.uuid4())
# 格式: {文件类型}/{唯一ID}.扩展名 (去掉项目名前缀)
if file_type and file_type in self.file_types:
type_prefix = self.file_types[file_type]
return f"{type_prefix}/{unique_id}{ext}"
# 默认格式
return f"misc/{unique_id}{ext}"
def _calculate_file_hash(self, file_path: Path) -> str:
"""计算文件的SHA256哈希"""
sha256_hash = hashlib.sha256()
with open(file_path, 'rb') as f:
for byte_block in iter(lambda: f.read(4096), b""):
sha256_hash.update(byte_block)
return sha256_hash.hexdigest()
async def upload_file(self, file_type: str, file_path: Path,
original_filename: str,
metadata: Optional[Dict] = None) -> Dict[str, Any]:
"""上传文件到 RustFS"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
# 计算文件哈希
file_hash = self._calculate_file_hash(file_path)
# 生成唯一键名
object_key = self._generate_object_key(original_filename, file_type)
# 上传文件
try:
result = self.client.fput_object(
self.bucket_name,
object_key,
str(file_path),
content_type='application/octet-stream',
metadata=metadata or {}
)
logger.info(f"文件上传成功 RustFS: {self.bucket_name}/{object_key}")
# 获取文件大小
file_size = file_path.stat().st_size
return {
'object_key': object_key,
'bucket': self.bucket_name,
'file_hash': file_hash,
'file_size': file_size,
'etag': result.etag if hasattr(result, 'etag') else None
}
except S3Error as e:
logger.error(f"RustFS 上传失败: {e}")
raise
async def upload_json_data(self, file_type: str,
json_data: Dict[str, Any],
file_hash: str) -> Dict[str, Any]:
"""上传JSON数据到 RustFS"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
# 格式: {文件类型}/{文件哈希}.json (去掉项目名前缀)
type_prefix = self.file_types[file_type]
object_key = f"{type_prefix}/{file_hash}.json"
# 转换为字节
import json
json_bytes = json.dumps(json_data, ensure_ascii=False).encode('utf-8')
try:
result = self.client.put_object(
self.bucket_name,
object_key,
BytesIO(json_bytes),
length=len(json_bytes),
content_type='application/json'
)
logger.info(f"JSON数据上传成功 RustFS: {self.bucket_name}/{object_key}")
return {
'object_key': object_key,
'bucket': self.bucket_name,
'file_size': len(json_bytes),
'etag': result.etag if hasattr(result, 'etag') else None
}
except S3Error as e:
logger.error(f"RustFS JSON上传失败: {e}")
raise
async def download_file(self, file_type: str, object_key: str) -> bytes:
"""从 RustFS 下载文件"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
try:
response = self.client.get_object(self.bucket_name, object_key)
data = response.read()
response.close()
response.release_conn()
logger.debug(f"文件下载成功: {self.bucket_name}/{object_key}")
return data
except S3Error as e:
logger.error(f"RustFS 下载失败: {e}")
raise
async def get_file_info(self, file_type: str, object_key: str) -> Dict[str, Any]:
"""获取文件信息"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
try:
stat = self.client.stat_object(self.bucket_name, object_key)
return {
'size': stat.size,
'etag': stat.etag,
'content_type': stat.content_type,
'last_modified': stat.last_modified
}
except S3Error as e:
logger.error(f"RustFS 获取文件信息失败: {e}")
raise
async def delete_file(self, file_type: str, object_key: str):
"""删除 RustFS 中的文件"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
try:
self.client.remove_object(self.bucket_name, object_key)
logger.info(f"文件删除成功: {self.bucket_name}/{object_key}")
except S3Error as e:
logger.error(f"RustFS 删除失败: {e}")
raise
async def list_files(self, file_type: str, prefix: str = '') -> list:
"""列出存储桶中的文件"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
# 构建完整前缀:{文件类型}/... (去掉项目名前缀)
type_prefix = self.file_types[file_type]
full_prefix = f"{type_prefix}/"
if prefix:
full_prefix += prefix
try:
objects = self.client.list_objects(self.bucket_name, prefix=full_prefix, recursive=True)
return [
{
'object_key': obj.object_name,
'size': obj.size,
'etag': obj.etag,
'last_modified': obj.last_modified
}
for obj in objects
]
except S3Error as e:
logger.error(f"RustFS 列出文件失败: {e}")
raise
async def generate_presigned_url(self, file_type: str,
object_key: str,
expires: int = 3600,
method: str = 'GET') -> str:
"""生成预签名URL(临时访问链接)"""
if not self.is_connected:
raise RuntimeError("RustFS 未连接")
if file_type not in self.file_types:
raise ValueError(f"未知的文件类型: {file_type}")
try:
url = self.client.presigned_get_object(
self.bucket_name,
object_key,
expires=timedelta(seconds=expires)
)
return url
except S3Error as e:
logger.error(f"RustFS 生成预签名URL失败: {e}")
raise
async def file_exists(self, file_type: str, object_key: str) -> bool:
"""检查文件是否存在"""
try:
await self.get_file_info(file_type, object_key)
return True
except:
return False
async def get_storage_stats(self) -> Dict[str, Any]:
"""获取存储统计信息"""
try:
buckets = self.client.list_buckets()
total_objects = 0
total_size = 0
namespace_stats = {}
for bucket in buckets:
objects = self.client.list_objects(bucket.name, recursive=True)
bucket_count = 0
bucket_size = 0
for obj in objects:
bucket_count += 1
bucket_size += obj.size
namespace_stats[bucket.name] = {
'object_count': bucket_count,
'total_size': bucket_size
}
total_objects += bucket_count
total_size += bucket_size
return {
'total_objects': total_objects,
'total_size': total_size,
'namespace_stats': namespace_stats
}
except S3Error as e:
logger.error(f"RustFS 获取统计信息失败: {e}")
raise
# 全局 RustFS 管理器实例
rustfs_manager = RustFSManager()
File diff suppressed because it is too large Load Diff
+1
View File
@@ -0,0 +1 @@
# Utils 模块
+28
View File
@@ -0,0 +1,28 @@
# utils/file_handler.py
import aiofiles
from pathlib import Path
from fastapi import UploadFile
class FileHandler:
def __init__(self, upload_dir: str = "uploads"):
self.upload_dir = Path(upload_dir)
self.upload_dir.mkdir(exist_ok=True)
async def save_uploaded_file(self, file: UploadFile) -> Path:
"""保存上传的文件"""
file_path = self.upload_dir / file.filename
async with aiofiles.open(file_path, 'wb') as f:
content = await file.read()
await f.write(content)
return file_path
def cleanup_file(self, file_path: Path):
"""清理文件"""
try:
if file_path.exists():
file_path.unlink()
except Exception as e:
print(f"文件清理失败: {e}")
+279
View File
@@ -0,0 +1,279 @@
# utils/html_generator.py
from pathlib import Path
from typing import Dict, Any, Optional
import json
from datetime import datetime
from utils.logger import get_logger
logger = get_logger(__name__)
class HTMLGenerator:
"""HTML文件生成器"""
def __init__(self, output_dir: str = "./html_output"):
self.output_dir = Path(output_dir)
self.output_dir.mkdir(exist_ok=True)
def generate_3d_viewer_html(
self,
geometry_data: Dict[str, Any],
stp_filename: str,
cavity_data: Optional[Dict[str, Any]] = None,
key_info: Optional[Dict[str, Any]] = None
) -> str:
"""生成3D可视化HTML页面"""
# 提取几何数据
bounding_box = geometry_data.get("bounding_box", {})
volume = geometry_data.get("volume", 0) or 0
surface_area = geometry_data.get("surface_area", 0) or 0
topology = geometry_data.get("topology", {})
center_of_mass = geometry_data.get("center_of_mass", [0, 0, 0])
cavity_html = ""
if cavity_data and key_info:
cavity_html = f"""
<div id="cavity-info-panel" style="position: absolute; top: 10px; right: 10px;">
<h3>🔧 模具型腔信息</h3>
<div class="metric">
<span class="metric-label">收缩率:</span>
<span class="metric-value">{cavity_data["metadata"]["shrinkage_rate"]}</span>
</div>
<div class="metric">
<span class="metric-label">拔模角:</span>
<span class="metric-value">{cavity_data["metadata"]["draft_angle"]}°</span>
</div>
<div class="metric">
<span class="metric-label">预估模具尺寸:</span>
<span class="metric-value">
{key_info["mold_parameters"]["mold_size"]["length"]:.0f} ×
{key_info["mold_parameters"]["mold_size"]["width"]:.0f} ×
{key_info["mold_parameters"]["mold_size"]["height"]:.0f} mm
</span>
</div>
<div class="metric">
<span class="metric-label">预估锁模力:</span>
<span class="metric-value">
{key_info["manufacturing_requirements"]["clamping_force"]}
</span>
</div>
<div class="metric">
<span class="metric-label">产品重量:</span>
<span class="metric-value">
{key_info["geometric_characteristics"]["product_weight"]}
</span>
</div>
</div>
"""
html_content = f"""
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>3D模具几何可视化 - {stp_filename}</title>
<script src="https://cdn.jsdelivr.net/npm/three@0.132.2/build/three.min.js"></script>
<script src="https://cdn.jsdelivr.net/npm/three@0.132.2/examples/js/controls/OrbitControls.js"></script>
<style>
body {{ margin: 0; overflow: hidden; font-family: Arial, sans-serif; }}
#container {{ position: relative; width: 100vw; height: 100vh; }}
#canvas {{ display: block; }}
#info-panel {{
position: absolute;
top: 10px;
left: 10px;
background: rgba(255, 255, 255, 0.9);
padding: 15px;
border-radius: 8px;
max-width: 300px;
box-shadow: 0 2px 10px rgba(0,0,0,0.1);
}}
#controls {{
position: absolute;
bottom: 10px;
left: 10px;
background: rgba(255, 255, 255, 0.9);
padding: 10px;
border-radius: 8px;
}}
.metric {{ margin: 5px 0; }}
.metric-label {{ font-weight: bold; color: #333; }}
.metric-value {{ color: #666; }}
</style>
</head>
<body>
<div id="container">
<canvas id="canvas"></canvas>
{cavity_html}
<div id="info-panel">
<h3>模具几何信息</h3>
<div class="metric">
<span class="metric-label">文件名:</span>
<span class="metric-value">{stp_filename}</span>
</div>
<div class="metric">
<span class="metric-label">体积:</span>
<span class="metric-value">{volume:.2f} mm³</span>
</div>
<div class="metric">
<span class="metric-label">表面积:</span>
<span class="metric-value">{surface_area:.2f} mm²</span>
</div>
<div class="metric">
<span class="metric-label">边界框:</span>
<span class="metric-value">{bounding_box.get('dimensions', [0, 0, 0])[0]:.1f} × {bounding_box.get('dimensions', [0, 0, 0])[1]:.1f} × {bounding_box.get('dimensions', [0, 0, 0])[2]:.1f} mm</span>
</div>
<div class="metric">
<span class="metric-label">面数:</span>
<span class="metric-value">{topology.get('faces', 0)}</span>
</div>
<div class="metric">
<span class="metric-label">边数:</span>
<span class="metric-value">{topology.get('edges', 0)}</span>
</div>
<div class="metric">
<span class="metric-label">顶点数:</span>
<span class="metric-value">{topology.get('vertices', 0)}</span>
</div>
</div>
<div id="controls">
<button onclick="resetView()">重置视图</button>
<button onclick="toggleWireframe()">切换线框</button>
</div>
</div>
<script>
// 初始化Three.js场景
const scene = new THREE.Scene();
const camera = new THREE.PerspectiveCamera(75, window.innerWidth / window.innerHeight, 0.1, 1000);
const renderer = new THREE.WebGLRenderer({{ canvas: document.getElementById('canvas') }});
renderer.setSize(window.innerWidth, window.innerHeight);
renderer.setClearColor(0xf0f0f0);
// 添加光源
const ambientLight = new THREE.AmbientLight(0x404040);
scene.add(ambientLight);
const directionalLight = new THREE.DirectionalLight(0xffffff, 1);
directionalLight.position.set(1, 1, 1);
scene.add(directionalLight);
// 添加坐标轴
const axesHelper = new THREE.AxesHelper(50);
scene.add(axesHelper);
// 创建几何体(模拟模具形状)
const geometryData = {json.dumps(geometry_data, indent=2)};
// 根据边界框创建模拟几何体
const bbox = geometryData.bounding_box;
if (bbox) {{
const width = bbox.dimensions ? bbox.dimensions[0] : 100;
const height = bbox.dimensions ? bbox.dimensions[1] : 100;
const depth = bbox.dimensions ? bbox.dimensions[2] : 100;
// 创建基础几何体
const geometry = new THREE.BoxGeometry(width, height, depth);
const material = new THREE.MeshPhongMaterial({{
color: 0x4CAF50,
transparent: true,
opacity: 0.8,
wireframe: false
}});
const mesh = new THREE.Mesh(geometry, material);
scene.add(mesh);
// 添加线框
const wireframe = new THREE.WireframeGeometry(geometry);
const line = new THREE.LineSegments(wireframe);
line.material.depthTest = false;
line.material.opacity = 0.25;
line.material.transparent = true;
scene.add(line);
}}
// 设置相机位置
camera.position.set(200, 200, 200);
camera.lookAt(0, 0, 0);
// 添加轨道控制器
const controls = new THREE.OrbitControls(camera, renderer.domElement);
controls.enableDamping = true;
controls.dampingFactor = 0.25;
// 动画循环
function animate() {{
requestAnimationFrame(animate);
controls.update();
renderer.render(scene, camera);
}}
animate();
// 窗口大小调整
window.addEventListener('resize', () => {{
camera.aspect = window.innerWidth / window.innerHeight;
camera.updateProjectionMatrix();
renderer.setSize(window.innerWidth, window.innerHeight);
}});
// 控制函数
function resetView() {{
controls.reset();
}}
function toggleWireframe() {{
scene.traverse((child) => {{
if (child.isMesh) {{
child.material.wireframe = !child.material.wireframe;
}}
}});
}}
</script>
</body>
</html>
"""
return html_content
def save_html_file(self, html_content: str, filename: str) -> str:
"""保存HTML文件到磁盘"""
try:
file_path = self.output_dir / filename
with open(file_path, 'w', encoding='utf-8') as f:
f.write(html_content)
logger.info(f"HTML文件保存成功: {file_path}")
return str(file_path)
except Exception as e:
logger.error(f"保存HTML文件失败: {e}")
raise
def generate_and_save_visualization(
self,
geometry_data: Dict[str, Any],
stp_filename: str
) -> str:
"""生成并保存可视化HTML文件"""
try:
# 生成HTML内容
html_content = self.generate_3d_viewer_html(geometry_data, stp_filename)
# 创建文件名
timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
safe_filename = stp_filename.replace('.', '_').replace(' ', '_')
html_filename = f"{safe_filename}_{timestamp}.html"
# 保存文件
file_path = self.save_html_file(html_content, html_filename)
return file_path
except Exception as e:
logger.error(f"生成可视化文件失败: {e}")
raise
+17
View File
@@ -0,0 +1,17 @@
# utils/logger.py
import logging
import sys
def setup_logging():
"""设置日志配置"""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout)
]
)
def get_logger(name: str):
"""获取日志器"""
return logging.getLogger(name)