imetting/backend/app/services/system_config_service.py

import json
import time
from threading import RLock
from typing import Optional, Dict, Any
from app.core.database import get_db_connection

class SystemConfigService:
    """系统配置服务。"""

    PUBLIC_CATEGORY = 'public'
    DEFAULT_LLM_ENDPOINT_URL = 'https://dashscope.aliyuncs.com/compatible-mode/v1'
    CACHE_TTL_SECONDS = 60

    # 配置键常量
    ASR_VOCABULARY_ID = 'asr_vocabulary_id'
    PAGE_SIZE = 'page_size'
    DEFAULT_RESET_PASSWORD = 'default_reset_password'
    MAX_AUDIO_SIZE = 'max_audio_size'
    MAX_IMAGE_SIZE = 'max_image_size'
    TOKEN_EXPIRE_DAYS = 'token_expire_days'

    # 品牌配置
    APP_NAME = 'app_name'
    DEPRECATED_BRANDING_PARAMETER_KEYS = frozenset({
        'console_subtitle',
        'preview_title',
        'login_welcome',
        'footer_text',
    })

    # 声纹配置
    VOICEPRINT_TEMPLATE_TEXT = 'voiceprint_template_text'
    VOICEPRINT_MAX_SIZE = 'voiceprint_max_size'
    VOICEPRINT_DURATION = 'voiceprint_duration'
    VOICEPRINT_SAMPLE_RATE = 'voiceprint_sample_rate'
    VOICEPRINT_CHANNELS = 'voiceprint_channels'

    # LLM模型配置
    LLM_MODEL_NAME = 'llm_model_name'
    LLM_TIMEOUT = 'llm_timeout'
    LLM_TEMPERATURE = 'llm_temperature'
    LLM_TOP_P = 'llm_top_p'
    _cache_lock = RLock()
    _config_cache: Dict[str, tuple[float, Any]] = {}
    _category_cache: Dict[str, tuple[float, Dict[str, Any]]] = {}
    _all_configs_cache: tuple[float, Dict[str, Any]] | None = None
    BUILTIN_PARAMETERS = [
        {
            "param_key": TOKEN_EXPIRE_DAYS,
            "param_name": "Token过期时间",
            "param_value": "7",
            "value_type": "number",
            "category": "system",
            "description": "控制登录 token 的过期时间，单位：天。",
            "is_active": 1,
        },
        {
            "param_key": DEFAULT_RESET_PASSWORD,
            "param_name": "默认重置密码",
            "param_value": "123456",
            "value_type": "string",
            "category": "system",
            "description": "管理员重置用户密码时使用的默认密码。",
            "is_active": 1,
        },
        {
            "param_key": PAGE_SIZE,
            "param_name": "系统分页大小",
            "param_value": "10",
            "value_type": "number",
            "category": PUBLIC_CATEGORY,
            "description": "系统通用分页数量。",
            "is_active": 1,
        },
        {
            "param_key": MAX_AUDIO_SIZE,
            "param_name": "音频上传大小限制",
            "param_value": "100",
            "value_type": "number",
            "category": PUBLIC_CATEGORY,
            "description": "音频上传大小限制，单位：MB。",
            "is_active": 1,
        },
        {
            "param_key": MAX_IMAGE_SIZE,
            "param_name": "图片上传大小限制",
            "param_value": "10",
            "value_type": "number",
            "category": PUBLIC_CATEGORY,
            "description": "图片上传大小限制，单位：MB。",
            "is_active": 1,
        },
        {
            "param_key": APP_NAME,
            "param_name": "系统名称",
            "param_value": "iMeeting",
            "value_type": "string",
            "category": PUBLIC_CATEGORY,
            "description": "前端应用标题。",
            "is_active": 1,
        },
    ]

    @classmethod
    def _is_cache_valid(cls, cached_at: float) -> bool:
        return (time.time() - cached_at) < cls.CACHE_TTL_SECONDS

    @classmethod
    def _get_cached_config(cls, cache_key: str) -> Any:
        with cls._cache_lock:
            cached = cls._config_cache.get(cache_key)
            if not cached:
                return None
            cached_at, value = cached
            if not cls._is_cache_valid(cached_at):
                cls._config_cache.pop(cache_key, None)
                return None
            return value

    @classmethod
    def _set_cached_config(cls, cache_key: str, value: Any) -> None:
        with cls._cache_lock:
            cls._config_cache[cache_key] = (time.time(), value)

    @classmethod
    def invalidate_cache(cls) -> None:
        with cls._cache_lock:
            cls._config_cache.clear()
            cls._category_cache.clear()
            cls._all_configs_cache = None

    @staticmethod
    def _parse_json_object(value: Any) -> Dict[str, Any]:
        if value is None:
            return {}
        if isinstance(value, dict):
            return dict(value)
        if isinstance(value, str):
            value = value.strip()
            if not value:
                return {}
            try:
                parsed = json.loads(value)
                return parsed if isinstance(parsed, dict) else {}
            except json.JSONDecodeError:
                return {}
        return {}

    @staticmethod
    def _normalize_string_list(value: Any) -> Optional[list[str]]:
        if value is None:
            return None
        if isinstance(value, list):
            items = [str(item).strip() for item in value if str(item).strip()]
            return items or None
        if isinstance(value, str):
            items = [item.strip() for item in value.split(",") if item.strip()]
            return items or None
        return None

    @classmethod
    def _build_audio_runtime_config(cls, audio_row: Dict[str, Any]) -> Dict[str, Any]:
        cfg: Dict[str, Any] = {}
        if not audio_row:
            return cfg

        extra_config = cls._parse_json_object(audio_row.get("extra_config"))

        if audio_row.get("endpoint_url"):
            cfg["endpoint_url"] = audio_row["endpoint_url"]
        if audio_row.get("api_key"):
            cfg["api_key"] = audio_row["api_key"]
        if audio_row.get("provider"):
            cfg["provider"] = audio_row["provider"]
        if audio_row.get("model_code"):
            cfg["model_code"] = audio_row["model_code"]
        if audio_row.get("audio_scene"):
            cfg["audio_scene"] = audio_row["audio_scene"]
        if audio_row.get("hot_word_group_id") is not None:
            cfg["hot_word_group_id"] = audio_row["hot_word_group_id"]
        if audio_row.get("request_timeout_seconds") is not None:
            cfg["request_timeout_seconds"] = int(audio_row["request_timeout_seconds"])

        language_hints = cls._normalize_string_list(extra_config.get("language_hints"))
        if language_hints is not None:
            extra_config["language_hints"] = language_hints

        cfg.update(extra_config)
        return cfg

    @classmethod
    def get_active_audio_model_config(cls, scene: str = "asr") -> Dict[str, Any]:
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                cursor.execute(
                    """
                    SELECT model_code, model_name, audio_scene, provider, endpoint_url, api_key,
                           request_timeout_seconds, hot_word_group_id, extra_config
                    FROM audio_model_config
                    WHERE audio_scene = %s AND is_active = 1
                    ORDER BY is_default DESC, updated_at DESC, config_id ASC
                    LIMIT 1
                    """,
                    (scene,),
                )
                row = cursor.fetchone()
                cursor.close()
                return cls._build_audio_runtime_config(row) if row else {}
        except Exception:
            return {}

    @classmethod
    def _get_parameter_value(cls, param_key: str):
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                cursor.execute(
                    """
                    SELECT param_value
                    FROM sys_system_parameters
                    WHERE param_key = %s AND is_active = 1
                    LIMIT 1
                    """,
                    (param_key,),
                )
                result = cursor.fetchone()
                cursor.close()
                return result["param_value"] if result else None
        except Exception:
            return None

    @classmethod
    def _get_model_config_json(cls, model_code: str):
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                # 1) llm 专表
                cursor.execute(
                    """
                    SELECT model_code, endpoint_url, api_key, llm_model_name, llm_timeout,
                           llm_temperature, llm_top_p, llm_max_tokens, llm_system_prompt
                    FROM llm_model_config
                    WHERE model_code = %s AND is_active = 1
                    ORDER BY is_default DESC, config_id ASC
                    LIMIT 1
                    """,
                    (model_code,),
                )
                llm_row = cursor.fetchone()
                if not llm_row and model_code == "llm_model":
                    cursor.execute(
                        """
                        SELECT model_code, endpoint_url, api_key, llm_model_name, llm_timeout,
                               llm_temperature, llm_top_p, llm_max_tokens, llm_system_prompt
                        FROM llm_model_config
                        WHERE is_active = 1
                        ORDER BY is_default DESC, updated_at DESC, config_id ASC
                        LIMIT 1
                        """
                    )
                    llm_row = cursor.fetchone()
                if llm_row:
                    cursor.close()
                    cfg = {}
                    if llm_row.get("endpoint_url"):
                        cfg["endpoint_url"] = llm_row["endpoint_url"]
                    if llm_row.get("api_key"):
                        cfg["api_key"] = llm_row["api_key"]
                    if llm_row.get("llm_model_name") is not None:
                        cfg["model_name"] = llm_row["llm_model_name"]
                    if llm_row.get("llm_timeout") is not None:
                        cfg["time_out"] = llm_row["llm_timeout"]
                    if llm_row.get("llm_temperature") is not None:
                        cfg["temperature"] = float(llm_row["llm_temperature"])
                    if llm_row.get("llm_top_p") is not None:
                        cfg["top_p"] = float(llm_row["llm_top_p"])
                    if llm_row.get("llm_max_tokens") is not None:
                        cfg["max_tokens"] = llm_row["llm_max_tokens"]
                    if llm_row.get("llm_system_prompt") is not None:
                        cfg["system_prompt"] = llm_row["llm_system_prompt"]
                    return cfg

                # 2) audio 专表
                if model_code in ("audio_model", "voiceprint_model"):
                    target_scene = "voiceprint" if model_code == "voiceprint_model" else "asr"
                    cursor.close()
                    audio_cfg = cls.get_active_audio_model_config(target_scene)
                    return audio_cfg or None

                cursor.execute(
                    """
                    SELECT model_code, model_name, audio_scene, provider, endpoint_url, api_key,
                           request_timeout_seconds, hot_word_group_id, extra_config
                    FROM audio_model_config
                    WHERE model_code = %s AND is_active = 1
                    ORDER BY is_default DESC, config_id ASC
                    LIMIT 1
                    """,
                    (model_code,),
                )
                audio_row = cursor.fetchone()
                cursor.close()
                if audio_row:
                    return cls._build_audio_runtime_config(audio_row)

                return None
        except Exception:
            return None

    @classmethod
    def get_config(cls, dict_code: str, default_value: Any = None) -> Any:
        """
        获取指定配置项的值

        Args:
            dict_code: 配置项编码
            default_value: 默认值，如果配置不存在则返回此值

        Returns:
            配置项的值
        """
        cached_value = cls._get_cached_config(dict_code)
        if cached_value is not None:
            return cached_value

        # 1) 新参数表
        value = cls._get_parameter_value(dict_code)
        if value is not None:
            cls._set_cached_config(dict_code, value)
            return value

        cls._set_cached_config(dict_code, default_value)
        return default_value

    @classmethod
    def get_config_attribute(cls, dict_code: str, attr_name: str, default_value: Any = None) -> Any:
        """
        从指定配置记录的扩展属性中读取指定属性值

        此方法用于处理扩展属性为复杂JSON的配置记录（如llm_model、voiceprint等）

        Args:
            dict_code: 配置项编码 (如 'llm_model', 'voiceprint')
            attr_name: 扩展属性中的属性名 (如 'time_out', 'channels', 'template_text')
            default_value: 默认值，如果配置或属性不存在则返回此值

        Returns:
            属性值
        """
        # 1) 新模型配置表
        model_json = cls._get_model_config_json(dict_code)
        if model_json is not None:
            return model_json.get(attr_name, default_value)

        return default_value

    @classmethod
    def get_model_runtime_config(cls, model_code: str) -> Optional[Dict[str, Any]]:
        """获取模型运行时配置，优先从新模型配置表读取。"""
        return cls._get_model_config_json(model_code)

    @classmethod
    def set_config(cls, dict_code: str, value: Any, label_cn: str = None) -> bool:
        """
        设置指定配置项的值

        Args:
            dict_code: 配置项编码
            value: 配置值
            label_cn: 配置项中文名称（仅在配置不存在时需要）

        Returns:
            是否设置成功
        """
        # 1) 优先写入新参数表
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)

                cursor.execute(
                    """
                    INSERT INTO sys_system_parameters
                    (param_key, param_name, param_value, value_type, category, description, is_active)
                    VALUES (%s, %s, %s, %s, %s, %s, 1)
                    ON DUPLICATE KEY UPDATE
                        param_name = VALUES(param_name),
                        param_value = VALUES(param_value),
                        value_type = VALUES(value_type),
                        category = VALUES(category),
                        description = VALUES(description),
                        is_active = 1
                    """,
                    (
                        dict_code,
                        label_cn or dict_code,
                        str(value) if value is not None else "",
                        "string",
                        "system",
                        "Migrated from legacy system_config",
                    ),
                )
                if dict_code == cls.ASR_VOCABULARY_ID:
                    cursor.execute(
                        """
                        INSERT INTO audio_model_config
                        (model_code, model_name, audio_scene, provider, request_timeout_seconds, extra_config, description, is_active, is_default)
                        VALUES (
                            'audio_model',
                            '音频识别模型',
                            'asr',
                            'dashscope',
                            300,
                            JSON_OBJECT(
                                'model', 'paraformer-v2',
                                'vocabulary_id', %s,
                                'speaker_count', 10,
                                'language_hints', JSON_ARRAY('zh', 'en'),
                                'disfluency_removal_enabled', TRUE,
                                'diarization_enabled', TRUE
                            ),
                            '语音识别模型配置',
                            1,
                            1
                        )
                        ON DUPLICATE KEY UPDATE
                            extra_config = JSON_SET(COALESCE(extra_config, JSON_OBJECT()), '$.vocabulary_id', %s),
                            is_active = 1
                        """,
                        (str(value), str(value)),
                    )
                conn.commit()
                cursor.close()
                cls.invalidate_cache()
                return True
        except Exception as e:
            print(f"Error setting config in sys_system_parameters {dict_code}: {e}")

        return False

    @classmethod
    def get_all_configs(cls) -> Dict[str, Any]:
        """
        获取所有系统配置

        Returns:
            配置字典 {dict_code: value}
        """
        with cls._cache_lock:
            if cls._all_configs_cache and cls._is_cache_valid(cls._all_configs_cache[0]):
                return dict(cls._all_configs_cache[1])
            if cls._all_configs_cache and not cls._is_cache_valid(cls._all_configs_cache[0]):
                cls._all_configs_cache = None

        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                cursor.execute(
                    """
                    SELECT param_key, param_value
                    FROM sys_system_parameters
                    WHERE is_active = 1
                    ORDER BY category, param_key
                    """
                )
                rows = cursor.fetchall()
                cursor.close()
                if rows:
                    configs = {row["param_key"]: row["param_value"] for row in rows}
                    with cls._cache_lock:
                        cls._all_configs_cache = (time.time(), configs)
                    return dict(configs)
        except Exception as e:
            print(f"Error getting all configs from sys_system_parameters: {e}")
        return {}

    @classmethod
    def get_configs_by_category(cls, category: str) -> Dict[str, Any]:
        """按分类获取启用中的参数配置。"""
        with cls._cache_lock:
            cached = cls._category_cache.get(category)
            if cached and cls._is_cache_valid(cached[0]):
                return dict(cached[1])
            if cached and not cls._is_cache_valid(cached[0]):
                cls._category_cache.pop(category, None)

        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                cursor.execute(
                    """
                    SELECT param_key, param_value
                    FROM sys_system_parameters
                    WHERE is_active = 1 AND category = %s
                    ORDER BY param_key
                    """,
                    (category,),
                )
                rows = cursor.fetchall()
                cursor.close()
                configs = {row["param_key"]: row["param_value"] for row in rows} if rows else {}
                with cls._cache_lock:
                    cls._category_cache[category] = (time.time(), configs)
                return dict(configs)
        except Exception as e:
            print(f"Error getting configs by category {category}: {e}")
            return {}

    @classmethod
    def batch_set_configs(cls, configs: Dict[str, Any]) -> bool:
        """
        批量设置配置项

        Args:
            configs: 配置字典 {dict_code: value}

        Returns:
            是否全部设置成功
        """
        success = True
        for dict_code, value in configs.items():
            if not cls.set_config(dict_code, value):
                success = False
        return success

    @classmethod
    def ensure_builtin_parameters(cls) -> None:
        """确保内建系统参数存在，避免后台参数页缺少关键配置项。"""
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor()
                for item in cls.BUILTIN_PARAMETERS:
                    cursor.execute(
                        """
                        INSERT INTO sys_system_parameters
                        (param_key, param_name, param_value, value_type, category, description, is_active)
                        VALUES (%s, %s, %s, %s, %s, %s, %s)
                        ON DUPLICATE KEY UPDATE
                            param_name = param_name
                        """,
                        (
                            item["param_key"],
                            item["param_name"],
                            item["param_value"],
                            item["value_type"],
                            item["category"],
                            item["description"],
                            item["is_active"],
                        ),
                    )
                conn.commit()
                cursor.close()
        except Exception as e:
            print(f"Error ensuring builtin parameters: {e}")

    # 便捷方法：获取特定配置
    @classmethod
    def get_asr_vocabulary_id(cls) -> Optional[str]:
        """获取ASR热词字典ID — 优先从 audio_model_config.hot_word_group_id → hot_word_group.vocabulary_id"""
        audio_cfg = cls.get_active_audio_model_config("asr")
        if audio_cfg.get("vocabulary_id"):
            return audio_cfg["vocabulary_id"]
        return cls.get_config_attribute('audio_model', 'vocabulary_id')

    # 声纹配置获取方法（直接使用通用方法）
    @classmethod
    def get_voiceprint_template(cls, default: str = "我正在进行声纹采集，这段语音将用于身份识别和验证。\n\n声纹技术能够准确识别每个人独特的声音特征。") -> str:
        """获取声纹采集模版"""
        return cls.get_config_attribute('voiceprint_model', 'template_text', default)

    @classmethod
    def get_voiceprint_max_size(cls, default: int = 5242880) -> int:
        """获取声纹文件大小限制 (bytes), 默认5MB"""
        value = cls.get_config_attribute('voiceprint_model', 'max_size_bytes', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_voiceprint_duration(cls, default: int = 12) -> int:
        """获取声纹采集最短时长 (秒)"""
        value = cls.get_config_attribute('voiceprint_model', 'duration_seconds', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_voiceprint_sample_rate(cls, default: int = 16000) -> int:
        """获取声纹采样率"""
        value = cls.get_config_attribute('voiceprint_model', 'sample_rate', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_voiceprint_channels(cls, default: int = 1) -> int:
        """获取声纹通道数"""
        value = cls.get_config_attribute('voiceprint_model', 'channels', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_page_size(cls) -> int:
        """获取系统通用分页数量。"""
        value = cls.get_config(cls.PAGE_SIZE)
        if value is None:
            raise RuntimeError("系统参数 page_size 缺失")
        try:
            return int(value)
        except (ValueError, TypeError):
            raise RuntimeError(f"系统参数 page_size 非法: {value!r}") from None

    @classmethod
    def get_default_reset_password(cls) -> str:
        """获取默认重置密码"""
        value = cls.get_config(cls.DEFAULT_RESET_PASSWORD)
        if value is None:
            raise RuntimeError("系统参数 default_reset_password 缺失")
        normalized = str(value).strip()
        if not normalized:
            raise RuntimeError("系统参数 default_reset_password 不能为空")
        return normalized

    @classmethod
    def get_max_audio_size(cls) -> int:
        """获取上传音频文件大小限制（MB）"""
        value = cls.get_config(cls.MAX_AUDIO_SIZE)
        if value is None:
            raise RuntimeError("系统参数 max_audio_size 缺失")
        try:
            return int(value)
        except (ValueError, TypeError):
            raise RuntimeError(f"系统参数 max_audio_size 非法: {value!r}") from None

    @classmethod
    def get_max_image_size(cls) -> int:
        """获取上传图片大小限制（MB）"""
        value = cls.get_config(cls.MAX_IMAGE_SIZE)
        if value is None:
            raise RuntimeError("系统参数 max_image_size 缺失")
        try:
            return int(value)
        except (ValueError, TypeError):
            raise RuntimeError(f"系统参数 max_image_size 非法: {value!r}") from None

    @classmethod
    def get_token_expire_days(cls) -> int:
        """获取访问 token 过期时间（天）。"""
        value = cls.get_config(cls.TOKEN_EXPIRE_DAYS)
        if value is None:
            raise RuntimeError("系统参数 token_expire_days 缺失")
        try:
            normalized = int(value)
        except (ValueError, TypeError):
            raise RuntimeError(f"系统参数 token_expire_days 非法: {value!r}") from None
        if normalized <= 0:
            raise RuntimeError(f"系统参数 token_expire_days 非法: {value!r}")
        return normalized

    @classmethod
    def get_public_configs(cls) -> Dict[str, Any]:
        """获取提供给前端初始化使用的公开参数。"""
        cls.ensure_builtin_parameters()
        public_configs = cls.get_configs_by_category(cls.PUBLIC_CATEGORY)
        required_keys = [
            cls.APP_NAME,
            cls.PAGE_SIZE,
            cls.MAX_AUDIO_SIZE,
            cls.MAX_IMAGE_SIZE,
        ]
        missing_keys = [key for key in required_keys if str(public_configs.get(key) or "").strip() == ""]
        if missing_keys:
            raise RuntimeError(f"公开系统参数缺失: {', '.join(missing_keys)}")

        page_size = cls.get_page_size()
        max_audio_size_mb = cls.get_max_audio_size()
        max_image_size_mb = cls.get_max_image_size()

        return {
            "app_name": str(public_configs[cls.APP_NAME]).strip(),
            "page_size": str(page_size),
            "PAGE_SIZE": page_size,
            "max_audio_size": str(max_audio_size_mb),
            "MAX_FILE_SIZE": max_audio_size_mb * 1024 * 1024,
            "max_image_size": str(max_image_size_mb),
            "MAX_IMAGE_SIZE": max_image_size_mb * 1024 * 1024,
        }

    # LLM模型配置获取方法（直接使用通用方法）
    @classmethod
    def get_llm_model_name(cls, default: str = "qwen-plus") -> str:
        """获取LLM模型名称"""
        return cls.get_config_attribute('llm_model', 'model_name', default)

    @classmethod
    def get_llm_timeout(cls, default: int = 300) -> int:
        """获取LLM超时时间（秒）"""
        value = cls.get_config_attribute('llm_model', 'time_out', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_llm_temperature(cls, default: float = 0.7) -> float:
        """获取LLM temperature参数"""
        value = cls.get_config_attribute('llm_model', 'temperature', default)
        try:
            return float(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_llm_top_p(cls, default: float = 0.9) -> float:
        """获取LLM top_p参数"""
        value = cls.get_config_attribute('llm_model', 'top_p', default)
        try:
            return float(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_llm_max_tokens(cls, default: int = 2048) -> int:
        """获取LLM最大输出token"""
        value = cls.get_config_attribute('llm_model', 'max_tokens', default)
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_llm_system_prompt(cls, default: Optional[str] = None) -> Optional[str]:
        """获取LLM系统提示词"""
        value = cls.get_config_attribute('llm_model', 'system_prompt', default)
        return value if isinstance(value, str) and value.strip() else default

    @classmethod
    def get_llm_endpoint_url(cls, default: str = DEFAULT_LLM_ENDPOINT_URL) -> str:
        """获取LLM服务Base API"""
        value = cls.get_config_attribute('llm_model', 'endpoint_url', default)
        return value if isinstance(value, str) and value.strip() else default

    @classmethod
    def get_llm_api_key(cls, default: Optional[str] = None) -> Optional[str]:
        """获取LLM服务API Key"""
        value = cls.get_config_attribute('llm_model', 'api_key', default)
        if value is None:
            return default
        value_str = str(value).strip()
        return value_str or default