diff --git a/config/settings.py b/config/settings.py index 8faf82b..7e3377d 100644 --- a/config/settings.py +++ b/config/settings.py @@ -66,6 +66,8 @@ class Settings: self.ENABLE_FREECAD_VERIFICATION = os.getenv("ENABLE_FREECAD_VERIFICATION", "false").lower() == "true" self.FREECAD_VERIFICATION_TIMEOUT = int(os.getenv("FREECAD_VERIFICATION_TIMEOUT", "120")) + self.PROCESSING_TIMEOUT_BASE = int(os.getenv("PROCESSING_TIMEOUT_BASE", "300")) + self.PROCESSING_TIMEOUT_PER_MB = int(os.getenv("PROCESSING_TIMEOUT_PER_MB", "15")) # Redis self.REDIS_HOST = os.getenv("REDIS_HOST", "localhost") diff --git a/docker-compose.yml b/docker-compose.yml index 08b9b93..a53c38a 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -80,6 +80,34 @@ services: networks: - moldinsight_network + celery-worker: + build: . + container_name: moldinsight_celery + command: > + /bin/bash -c "source /opt/conda/etc/profile.d/conda.sh && conda activate moldinsight && cd /app/src && celery -A celery_app worker --concurrency=2 --loglevel=info" + volumes: + - ./uploads:/app/uploads + - ./html_output:/app/html_output + - ./logs:/app/logs + - ./.env:/app/.env:ro + environment: + - DB_HOST=postgres + - DB_PORT=5432 + - DB_NAME=${DB_NAME:-moldinsight} + - DB_USER=${DB_USER:-moldinsight_user} + - DB_PASSWORD=${DB_PASSWORD:-moldinsight_password} + - RUSTFS_ENDPOINT=http://minio:9000 + - RUSTFS_ACCESS_KEY=${MINIO_ACCESS_KEY:-minioadmin} + - RUSTFS_SECRET_KEY=${MINIO_SECRET_KEY:-minioadmin} + depends_on: + postgres: + condition: service_healthy + minio: + condition: service_healthy + restart: unless-stopped + networks: + - moldinsight_network + volumes: postgres_data: driver: local diff --git a/requirements.txt b/requirements.txt index dad4f68..56ed767 100644 --- a/requirements.txt +++ b/requirements.txt @@ -44,6 +44,7 @@ aiohttp>=3.13.4 # ============================================ # 消息队列 # ============================================ +celery[redis]>=5.3.0 kafka-python>=2.0.2 redis>=4.5.0 diff --git a/src/api/routes.py b/src/api/routes.py index 613ca44..f5af82f 100644 --- a/src/api/routes.py +++ b/src/api/routes.py @@ -1,5 +1,5 @@ # api/routes.py -from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Request, Depends +from fastapi import APIRouter, UploadFile, File, HTTPException, Request, Depends from typing import Optional, Dict, Any, List import uuid from datetime import datetime @@ -10,8 +10,8 @@ from models.schemas import ProcessingStatus, create_task_info from utils.file_handler import FileHandler from services.storage_integration_rustfs import StorageIntegrationService from services.redis_task_manager import redis_task_manager -from services.processing_service import processing_service from services.task_query_service import TaskQueryService +from celery_tasks import process_stp_task from database.database import get_db_session from utils.logger import get_logger from sqlalchemy.ext.asyncio import AsyncSession @@ -54,7 +54,6 @@ async def health(): @router.post("/upload") async def upload_stp( - background_tasks: BackgroundTasks, file: UploadFile = File(...), material: Optional[str] = "ABS", db_session: AsyncSession = Depends(get_db_session), @@ -104,13 +103,7 @@ async def upload_stp( # 后台处理走统一编排服务，避免请求会话在后台失效 process_params = {"material": material} - background_tasks.add_task( - processing_service.process_file_with_storage, - task_id, - file_path, - stp_file.id, - process_params, - ) + process_stp_task.delay(task_id, str(file_path), stp_file.id, process_params) logger.info(f"[UPLOAD] 后台处理已调度: task_id={task_id}") return { diff --git a/src/api/v1/upload_router.py b/src/api/v1/upload_router.py index cb66cd1..1b6888b 100644 --- a/src/api/v1/upload_router.py +++ b/src/api/v1/upload_router.py @@ -1,5 +1,5 @@ # api/v1/upload_router.py -from fastapi import APIRouter, UploadFile, File, HTTPException, BackgroundTasks, Depends, Form +from fastapi import APIRouter, UploadFile, File, HTTPException, Depends, Form import uuid from datetime import datetime from pathlib import Path @@ -12,8 +12,8 @@ from database.database import get_db_session from utils.logger import get_logger from sqlalchemy.ext.asyncio import AsyncSession from services.auth_service import get_current_active_user -from services.processing_service import processing_service from models.database import User +from celery_tasks import process_stp_task logger = get_logger(__name__) @@ -24,7 +24,6 @@ file_handler = FileHandler() @router.post("/upload") async def upload_stp( - background_tasks: BackgroundTasks, file: UploadFile = File(...), material: str = Form(...), draft_angle: float = Form(...), @@ -91,10 +90,7 @@ async def upload_stp( task_info["file_hash"] = file_meta["sha256"] await redis_task_manager.set_task(task_id, task_info) - background_tasks.add_task( - processing_service.process_file_with_storage, - task_id, file_path, stp_file.id, process_params - ) + process_stp_task.delay(task_id, str(file_path), stp_file.id, process_params) logger.info(f"[UPLOAD] 后台处理已调度: task_id={task_id}") return { diff --git a/src/celery_app.py b/src/celery_app.py new file mode 100644 index 0000000..0dfcd5c --- /dev/null +++ b/src/celery_app.py @@ -0,0 +1,34 @@ +import os +from dotenv import load_dotenv +from celery import Celery + +load_dotenv() + +redis_host = os.getenv("REDIS_HOST", "localhost") +redis_port = os.getenv("REDIS_PORT", "6379") +redis_password = os.getenv("REDIS_PASSWORD", "") +redis_db = os.getenv("REDIS_DB", "0") + +if redis_password: + broker_url = f"redis://:{redis_password}@{redis_host}:{redis_port}/{redis_db}" +else: + broker_url = f"redis://{redis_host}:{redis_port}/{redis_db}" + +app = Celery( + "moldinsight", + broker=broker_url, + backend=broker_url, + include=["celery_tasks"], +) + +app.conf.update( + task_serializer="json", + result_serializer="json", + accept_content=["json"], + task_acks_late=True, + worker_prefetch_multiplier=1, + task_track_started=True, + task_soft_time_limit=1800, + task_time_limit=2100, + broker_connection_retry_on_startup=True, +) diff --git a/src/celery_tasks.py b/src/celery_tasks.py new file mode 100644 index 0000000..975a2e6 --- /dev/null +++ b/src/celery_tasks.py @@ -0,0 +1,27 @@ +from celery_app import app +from services.processing_service import processing_service +from utils.logger import get_logger + +logger = get_logger(__name__) + + +@app.task(bind=True, max_retries=1, default_retry_delay=60) +def process_stp_task(self, task_id: str, file_path: str, stp_file_id: int, + process_params: dict): + """Celery 任务：异步处理 STP 文件生成模具型腔""" + import asyncio + + try: + logger.info(f"[celery] 开始处理: {task_id}") + asyncio.run(processing_service.process_file_with_storage( + task_id, file_path, stp_file_id, process_params + )) + logger.info(f"[celery] 处理完成: {task_id}") + return {"task_id": task_id, "status": "completed"} + except Exception as exc: + logger.error(f"[celery] 处理失败: {task_id}, error={exc}") + try: + self.retry(exc=exc) + except Exception: + pass + raise diff --git a/src/core/geometry_analyzer.py b/src/core/geometry_analyzer.py index 8fa2aed..0a5a1a0 100644 --- a/src/core/geometry_analyzer.py +++ b/src/core/geometry_analyzer.py @@ -1,5 +1,6 @@ from typing import Dict, List, Any, Optional import math +from concurrent.futures import ThreadPoolExecutor, as_completed import numpy as np from OCC.Core.TopoDS import TopoDS_Shape from models.schemas import ( @@ -74,27 +75,25 @@ class GeometryAnalyzer: def _detect_features(self, geometry_data: Dict[str, Any], shape: Optional[TopoDS_Shape] = None) -> List[Dict[str, Any]]: - """检测模具特征""" - features = [] + """检测模具特征 — 独立检测并行执行""" + features: List[Dict[str, Any]] = [] - wall_features = self._detect_wall_features(geometry_data, shape) - features.extend(wall_features) + with ThreadPoolExecutor(max_workers=4, thread_name_prefix="feat") as pool: + futures = { + pool.submit(self._detect_wall_features, geometry_data, shape): "wall", + pool.submit(self._detect_rib_features, geometry_data, shape): "rib", + pool.submit(self._detect_boss_features, geometry_data, shape): "boss", + pool.submit(self._analyze_draft_angles, geometry_data, shape): "draft", + } + if shape is not None: + futures[pool.submit(self._detect_curvature_features, shape)] = "curvature" + futures[pool.submit(self._detect_fillet_features, shape)] = "fillet" - rib_features = self._detect_rib_features(geometry_data, shape) - features.extend(rib_features) - - boss_features = self._detect_boss_features(geometry_data, shape) - features.extend(boss_features) - - draft_features = self._analyze_draft_angles(geometry_data, shape) - features.extend(draft_features) - - if shape is not None: - curvature_features = self._detect_curvature_features(shape) - features.extend(curvature_features) - - fillet_features = self._detect_fillet_features(shape) - features.extend(fillet_features) + for future in as_completed(futures): + try: + features.extend(future.result()) + except Exception: + pass logger.info(f"检测到 {len(features)} 个特征") return features diff --git a/src/services/processing_service.py b/src/services/processing_service.py index 0f5f5e7..4992d5a 100644 --- a/src/services/processing_service.py +++ b/src/services/processing_service.py @@ -65,8 +65,15 @@ class ProcessingService: try: logger.info(f"开始处理文件并生成模具型腔: {file_path}") - # 设置处理超时（5分钟） - timeout_seconds = 300 + from config.settings import settings + + file_size_bytes = Path(file_path).stat().st_size if Path(file_path).exists() else 0 + file_size_mb = max(file_size_bytes / (1024 * 1024), 1) + timeout_seconds = min( + max(settings.PROCESSING_TIMEOUT_BASE, int(file_size_mb * settings.PROCESSING_TIMEOUT_PER_MB)), + 1800, + ) + logger.info(f"处理超时设置为 {timeout_seconds}s (文件 {file_size_mb:.1f}MB)") try: await asyncio.wait_for( @@ -575,7 +582,7 @@ class ProcessingService: pointcloud_data: Optional[Dict[str, Any]] = None, lod_data: Optional[Dict[str, Any]] = None, ) -> Dict[str, Any]: - """为每个候选分模方案生成独立HTML预览链接。""" + """为候选分模方案生成轻量摘要链接（完整HTML仅最优方案按需生成）""" candidate_schemes = detailed_cavity_json.get("candidate_schemes", []) if not candidate_schemes: return detailed_cavity_json @@ -585,15 +592,14 @@ class ProcessingService: if not cavity_data: continue suffix = scheme.get("scheme_id") - html_path = self.html_generator.generate_and_save_visualization( - geometry_data, - stp_filename, - cavity_data=cavity_data, - pointcloud_data=pointcloud_data, - suffix=suffix, - lod_data=lod_data, + base_stem = Path(stp_filename).stem.replace(" ", "_") + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + summary_name = f"mold_{base_stem}_{suffix}_{ts}_summary.json" + summary_content = self.html_generator.generate_3d_viewer_summary( + geometry_data, cavity_data ) - scheme["html_file"] = f"/html/{Path(html_path).name}" + self.html_generator.save_data_file(summary_content, summary_name) + scheme["summary_file"] = f"/html/{summary_name}" best_scheme = CalculationService.get_best_scheme(detailed_cavity_json) if best_scheme: