shudao-main/shudao-chat-py/utils/config.py

from pydantic_settings import BaseSettings
from typing import Optional
import yaml
from pathlib import Path


class AppConfig(BaseSettings):
    name: str = "shudao-chat-py"
    host: str = "0.0.0.0"
    port: int = 22000
    debug: bool = True


class DatabaseConfig(BaseSettings):
    user: str
    password: str
    host: str
    port: int
    database: str
    pool_size: int = 100
    max_overflow: int = 10
    pool_recycle: int = 3600


class DeepSeekConfig(BaseSettings):
    api_key: str
    api_url: str


class Qwen3Config(BaseSettings):
    api_url: str
    model: str


class IntentConfig(BaseSettings):
    api_url: str
    model: str


class YoloConfig(BaseSettings):
    base_url: str


class SearchConfig(BaseSettings):
    api_url: str
    heartbeat_url: str


class DifyConfig(BaseSettings):
    workflow_url: str
    workflow_id: str
    auth_token: str


class AuthConfig(BaseSettings):
    api_url: str


class OSSConfig(BaseSettings):
    access_key_id: str
    access_key_secret: str
    bucket: str
    endpoint: str
    parse_encrypt_key: str


class Settings:
    def __init__(self, config_path: str = "config.yaml"):
        # 获取项目根目录
        if not Path(config_path).is_absolute():
            # 相对于当前文件的路径
            current_dir = Path(__file__).parent.parent
            config_file = current_dir / config_path
        else:
            config_file = Path(config_path)
        
        if config_file.exists():
            with open(config_file, 'r', encoding='utf-8') as f:
                config_data = yaml.safe_load(f)
        else:
            raise FileNotFoundError(f"配置文件不存在: {config_file}")

        self.app = AppConfig(**config_data.get('app', {}))
        self.database = DatabaseConfig(**config_data.get('database', {}))
        self.deepseek = DeepSeekConfig(**config_data.get('deepseek', {}))
        self.qwen3 = Qwen3Config(**config_data.get('qwen3', {}))
        self.intent = IntentConfig(**config_data.get('intent', {}))
        self.yolo = YoloConfig(**config_data.get('yolo', {}))
        self.search = SearchConfig(**config_data.get('search', {}))
        self.dify = DifyConfig(**config_data.get('dify', {}))
        self.auth = AuthConfig(**config_data.get('auth', {}))
        self.oss = OSSConfig(**config_data.get('oss', {}))
        self.base_url = config_data.get('base_url', 'https://aqai.shudaodsj.com:22000')


settings = Settings()


def get_base_url() -> str:
    return settings.base_url


def get_proxy_url(original_url: str) -> str:
    """将原始URL转换为代理URL"""
    if not original_url:
        return ""
    return f"{settings.base_url}/apiv1/oss/parse?url={original_url}"