geMoldInsight/src/shared/config/settings.py

import os
import urllib.parse
from typing import Dict, Any, List
from dotenv import load_dotenv

load_dotenv()


class Settings:
    """配置管理器"""

    def __init__(self):
        self.HOST = os.getenv("HOST", "0.0.0.0")
        self.PORT = int(os.getenv("PORT", "8000"))
        self.DEBUG = os.getenv("DEBUG", "false").lower() == "true"
        self.SERVE_FRONTEND_STATIC = os.getenv("SERVE_FRONTEND_STATIC", "false").lower() == "true"

        self.UPLOAD_DIR = os.getenv("UPLOAD_DIR", "./uploads")
        self.MAX_FILE_SIZE = int(os.getenv("MAX_FILE_SIZE", "104857600"))
        self.ALLOWED_EXTENSIONS = os.getenv("ALLOWED_EXTENSIONS", ".stp,.step,.stp.gz")

        self.POINTCLOUD_SAMPLE_COUNT = int(os.getenv("POINTCLOUD_SAMPLE_COUNT", "10000"))
        self.MESH_QUALITY = os.getenv("MESH_QUALITY", "high")
        self.PARALLEL_PROCESSING = os.getenv("PARALLEL_PROCESSING", "true").lower() == "true"

        # RUSTFS_* 不给代码兜底默认值（含 MINIO_* 兼容别名）：
        # 缺失时由 rustfs_storage.connect 抛出明确配置错误，而不是拿占位口令连库
        self.RUSTFS_ENDPOINT = os.getenv("RUSTFS_ENDPOINT") or os.getenv("MINIO_ENDPOINT")
        self.RUSTFS_ACCESS_KEY = os.getenv("RUSTFS_ACCESS_KEY") or os.getenv("MINIO_ACCESS_KEY")
        self.RUSTFS_SECRET_KEY = os.getenv("RUSTFS_SECRET_KEY") or os.getenv("MINIO_SECRET_KEY")
        self.RUSTFS_TIMEOUT = int(os.getenv("RUSTFS_TIMEOUT", "30"))
        self.RUSTFS_PRESIGNED_URL_EXPIRES = int(os.getenv("RUSTFS_PRESIGNED_URL_EXPIRES", "3600"))

        # 数据库配置改为惰性校验：允许在无 DB 环境下 import 项目模块（测试/静态分析）
        self.DB_HOST = os.getenv("DB_HOST")
        self.DB_PORT = int(os.getenv("DB_PORT")) if os.getenv("DB_PORT") else None
        self.DB_NAME = os.getenv("DB_NAME")
        self.DB_USER = os.getenv("DB_USER")
        self.DB_PASSWORD = os.getenv("DB_PASSWORD")

        # 启动时是否自动执行 alembic 迁移（D12）：多副本同时启动会并发迁移，
        # 生产多副本应设 false，改由部署流程单点执行 alembic CLI 或本模块 __main__
        self.AUTO_MIGRATE = os.getenv("AUTO_MIGRATE", "true").lower() == "true"

        self.SECRET_KEY = os.getenv("SECRET_KEY")
        self.ALGORITHM = os.getenv("ALGORITHM", "HS256")
        self.ACCESS_TOKEN_EXPIRE_MINUTES = int(os.getenv("ACCESS_TOKEN_EXPIRE_MINUTES", "1440"))

        self.ADMIN_USERNAME = os.getenv("ADMIN_USERNAME", "admin")
        self.ADMIN_PASSWORD = os.getenv("ADMIN_PASSWORD")
        self.ADMIN_EMAIL = os.getenv("ADMIN_EMAIL", "admin@gemold.com")
        self.ADMIN_FULL_NAME = os.getenv("ADMIN_FULL_NAME", "系统管理员")

        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")
        self.REDIS_PORT = int(os.getenv("REDIS_PORT", "6379"))
        self.REDIS_PASSWORD = os.getenv("REDIS_PASSWORD", "")
        self.REDIS_DB = int(os.getenv("REDIS_DB", "0"))

        # CORS 白名单（逗号分隔，默认允许本机开发地址）
        self.CORS_ORIGINS = self._parse_cors_origins(
            os.getenv("CORS_ORIGINS", "")
        )

        # LLM 增强分析配置（可选）
        self.LLM_ENABLED = os.getenv("LLM_ENABLED", "false").lower() == "true"
        self.LLM_API_URL = os.getenv("LLM_API_URL", "https://api.openai.com/v1")
        self.LLM_API_KEY = os.getenv("LLM_API_KEY", "")
        self.LLM_MODEL = os.getenv("LLM_MODEL", "gpt-4o-mini")
        self.LLM_TIMEOUT = int(os.getenv("LLM_TIMEOUT", "60"))
        self.LLM_MAX_TOKENS = int(os.getenv("LLM_MAX_TOKENS", "2000"))

    @property
    def DATABASE_URL(self) -> str:
        missing_configs = []
        if not self.DB_HOST:
            missing_configs.append("DB_HOST")
        if not self.DB_PORT:
            missing_configs.append("DB_PORT")
        if not self.DB_NAME:
            missing_configs.append("DB_NAME")
        if not self.DB_USER:
            missing_configs.append("DB_USER")
        if self.DB_PASSWORD is None:
            missing_configs.append("DB_PASSWORD")

        if missing_configs:
            raise ValueError(
                f"数据库配置缺失，请在.env文件中设置: {', '.join(missing_configs)}"
            )

        safe_password = urllib.parse.quote((self.DB_PASSWORD or "").encode("utf-8"), safe="")
        return f"postgresql+asyncpg://{self.DB_USER}:{safe_password}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"

    @property
    def allowed_extensions_set(self) -> set:
        return set(ext.strip() for ext in self.ALLOWED_EXTENSIONS.split(","))

    @property
    def redis_url(self) -> str:
        """Redis 连接串（Celery broker/backend 使用；RedisTaskManager 走分参数连接，不经此处）"""
        if self.REDIS_PASSWORD:
            return f"redis://:{self.REDIS_PASSWORD}@{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"
        return f"redis://{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB}"

    @staticmethod
    def _parse_cors_origins(raw: str) -> List[str]:
        """解析 CORS_ORIGINS 环境变量，逗号分隔。
        为空时返回空列表（由 app_factory 决定是否降级为 ['*']）。
        """
        if not raw or not raw.strip():
            return []
        return [o.strip().rstrip("/") for o in raw.split(",") if o.strip()]


settings = Settings()