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

from pydantic_settings import BaseSettings
import yaml
from pathlib import Path


class AppConfig(BaseSettings):
    name: str = "shudao-chat-py"
    host: str = "0.0.0.0"
    port: int = 22001
    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 AIChatConfig(BaseSettings):
    api_url: str = "http://127.0.0.1:28002/api/v1"
    timeout: int = 600


class SpeechProcurementConfig(BaseSettings):
    required_cloud_products: list[str] = [
        "asr_sentence_recognition",
        "asr_flash_file_recognition",
        "tts_text_to_voice",
        "tts_long_text",
    ]
    required_credentials: list[str] = [
        "app_id",
        "secret_id",
        "secret_key",
    ]
    optional_cloud_products: list[str] = [
        "cos",
    ]
    notes: list[str] = [
        "sentence_recognition_for_audio_within_60s_and_3mb",
        "flash_file_recognition_for_longer_audio_fallback",
        "text_to_voice_for_default_tts",
        "long_text_tts_for_content_over_150_chinese_chars",
    ]


class SpeechTencentConfig(BaseSettings):
    app_id: str = ""
    secret_id: str = ""
    secret_key: str = ""
    region: str = "ap-guangzhou"
    asr_endpoint: str = "asr.tencentcloudapi.com"
    tts_endpoint: str = "tts.tencentcloudapi.com"


class SpeechTranscribeConfig(BaseSettings):
    enabled: bool = True
    default_api: str = "SentenceRecognition"
    fallback_api: str = "CreateRecTask"
    engine_model_type: str = "16k_zh"
    source_type: int = 1
    voice_format: str = "wav"
    input_sample_rate: int = 16000
    max_audio_seconds: int = 60
    max_audio_size_mb: int = 3
    word_info: int = 0
    filter_dirty: int = 0
    filter_modal: int = 0
    filter_punc: int = 0
    convert_num_mode: int = 1
    hotword_list: str = ""
    hotword_id: str = ""
    customization_id: str = ""
    replace_text_id: str = ""


class SpeechSynthesizeConfig(BaseSettings):
    enabled: bool = True
    default_api: str = "TextToVoice"
    fallback_api: str = "CreateTtsTask"
    primary_language: int = 1
    voice_type: int = 1001
    sample_rate: int = 16000
    codec: str = "mp3"
    speed: float = 0.0
    volume: float = 0.0
    enable_subtitle: bool = False
    segment_rate: int = 0
    emotion_category: str = ""
    emotion_intensity: int = 100
    basic_text_limit_chars: int = 150
    long_text_trigger_chars: int = 120
    long_text_max_chars: int = 100000
    long_text_codec: str = "mp3"
    callback_url: str = ""


class SpeechConfig(BaseSettings):
    enabled: bool = True
    provider: str = "tencent_cloud"
    integration_mode: str = "backend_direct"
    backend_service: str = "shudao-chat-py"
    request_timeout_seconds: int = 60
    procurement: SpeechProcurementConfig = SpeechProcurementConfig()
    tencent: SpeechTencentConfig = SpeechTencentConfig()
    transcribe: SpeechTranscribeConfig = SpeechTranscribeConfig()
    synthesize: SpeechSynthesizeConfig = SpeechSynthesizeConfig()


class ThinkingSummaryConfig(BaseSettings):
    """思考过程二次总结（方案三）配置"""
    enabled: bool = True
    max_points: int = 5
    max_input_chars: int = 1500
    max_output_chars: int = 600
    temperature: float = 0.2


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.aichat = AIChatConfig(**config_data.get('aichat', {}))
        self.speech = SpeechConfig(**config_data.get('speech', {}))
        self.thinking_summary = ThinkingSummaryConfig(
            **config_data.get('thinking_summary', {}))
        self.base_url = config_data.get(
            'base_url', 'https://aqai.shudaodsj.com:22001')


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 ""
    from .crypto import encrypt_url
    encrypted = encrypt_url(original_url)
    return f"{settings.base_url}/apiv1/oss/parse?url={encrypted}"