imetting_backend/app/services/system_config_service.py

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


class SystemConfigService:
    """系统配置服务 - 从 dict_data 表中读取和保存 system_config 类型的配置"""

    DICT_TYPE = 'system_config'

    # 配置键常量
    ASR_VOCABULARY_ID = 'asr_vocabulary_id'
    VOICEPRINT_TEMPLATE = 'voiceprint_template'
    TIMELINE_PAGESIZE = 'timeline_pagesize'
    DEFAULT_RESET_PASSWORD = 'default_reset_password'
    MAX_AUDIO_SIZE = 'max_audio_size'

    # LLM模型配置
    LLM_MODEL_NAME = 'llm_model_name'
    LLM_TIMEOUT = 'llm_timeout'
    LLM_TEMPERATURE = 'llm_temperature'
    LLM_TOP_P = 'llm_top_p'

    @classmethod
    def get_dict_extension(cls, dict_code: str, dict_type: str = 'system_config') -> Dict[str, Any]:
        """
        获取指定码表项的扩展属性

        Args:
            dict_code: 码表编码
            dict_type: 字典类型，默认为 system_config

        Returns:
            扩展属性字典
        """
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                query = """
                    SELECT extension_attr
                    FROM dict_data
                    WHERE dict_type = %s AND dict_code = %s AND status = 1
                    LIMIT 1
                """
                cursor.execute(query, (dict_type, dict_code))
                result = cursor.fetchone()
                cursor.close()

                if result and result['extension_attr']:
                    try:
                        return json.loads(result['extension_attr']) if isinstance(result['extension_attr'], str) else result['extension_attr']
                    except (json.JSONDecodeError, AttributeError):
                        pass

                return {}

        except Exception as e:
            print(f"Error getting dict extension {dict_code}: {e}")
            return {}

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

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

        Returns:
            配置项的值
        """
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                query = """
                    SELECT extension_attr
                    FROM dict_data
                    WHERE dict_type = %s AND dict_code = %s AND status = 1
                    LIMIT 1
                """
                cursor.execute(query, (cls.DICT_TYPE, dict_code))
                result = cursor.fetchone()
                cursor.close()

                if result and result['extension_attr']:
                    try:
                        ext_attr = json.loads(result['extension_attr']) if isinstance(result['extension_attr'], str) else result['extension_attr']
                        return ext_attr.get('value', default_value)
                    except (json.JSONDecodeError, AttributeError):
                        pass

                return default_value

        except Exception as e:
            print(f"Error getting config {dict_code}: {e}")
            return default_value

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

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

        Returns:
            是否设置成功
        """
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)

                # 检查配置是否存在
                cursor.execute(
                    "SELECT id FROM dict_data WHERE dict_type = %s AND dict_code = %s",
                    (cls.DICT_TYPE, dict_code)
                )
                existing = cursor.fetchone()

                extension_attr = json.dumps({"value": value}, ensure_ascii=False)

                if existing:
                    # 更新现有配置
                    update_query = """
                        UPDATE dict_data
                        SET extension_attr = %s, update_time = NOW()
                        WHERE dict_type = %s AND dict_code = %s
                    """
                    cursor.execute(update_query, (extension_attr, cls.DICT_TYPE, dict_code))
                else:
                    # 插入新配置
                    if not label_cn:
                        label_cn = dict_code

                    insert_query = """
                        INSERT INTO dict_data (
                            dict_type, dict_code, parent_code, label_cn,
                            extension_attr, status, sort_order
                        ) VALUES (%s, %s, 'ROOT', %s, %s, 1, 0)
                    """
                    cursor.execute(insert_query, (cls.DICT_TYPE, dict_code, label_cn, extension_attr))

                conn.commit()
                cursor.close()
                return True

        except Exception as e:
            print(f"Error setting config {dict_code}: {e}")
            return False

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

        Returns:
            配置字典 {dict_code: value}
        """
        try:
            with get_db_connection() as conn:
                cursor = conn.cursor(dictionary=True)
                query = """
                    SELECT dict_code, label_cn, extension_attr
                    FROM dict_data
                    WHERE dict_type = %s AND status = 1
                    ORDER BY sort_order
                """
                cursor.execute(query, (cls.DICT_TYPE,))
                results = cursor.fetchall()
                cursor.close()

                configs = {}
                for row in results:
                    if row['extension_attr']:
                        try:
                            ext_attr = json.loads(row['extension_attr']) if isinstance(row['extension_attr'], str) else row['extension_attr']
                            configs[row['dict_code']] = ext_attr.get('value')
                        except (json.JSONDecodeError, AttributeError):
                            configs[row['dict_code']] = None
                    else:
                        configs[row['dict_code']] = None

                return configs

        except Exception as e:
            print(f"Error getting all configs: {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 get_asr_vocabulary_id(cls) -> Optional[str]:
        """获取ASR热词字典ID"""
        return cls.get_config(cls.ASR_VOCABULARY_ID)

    @classmethod
    def get_voiceprint_template(cls, default: str = "") -> str:
        """获取声纹采集模版"""
        return cls.get_config(cls.VOICEPRINT_TEMPLATE, default)

    @classmethod
    def get_timeline_pagesize(cls, default: int = 10) -> int:
        """获取会议时间轴每页数量"""
        value = cls.get_config(cls.TIMELINE_PAGESIZE, str(default))
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    @classmethod
    def get_default_reset_password(cls, default: str = "111111") -> str:
        """获取默认重置密码"""
        return cls.get_config(cls.DEFAULT_RESET_PASSWORD, default)

    @classmethod
    def get_max_audio_size(cls, default: int = 100) -> int:
        """获取上传音频文件大小限制（MB）"""
        value = cls.get_config(cls.MAX_AUDIO_SIZE, str(default))
        try:
            return int(value)
        except (ValueError, TypeError):
            return default

    # LLM模型配置获取方法
    @classmethod
    def get_llm_model_name(cls, default: str = "qwen-plus") -> str:
        """获取LLM模型名称"""
        return cls.get_config(cls.LLM_MODEL_NAME, default)

    @classmethod
    def get_llm_timeout(cls, default: int = 120) -> int:
        """获取LLM超时时间（秒）"""
        value = cls.get_config(cls.LLM_TIMEOUT, str(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(cls.LLM_TEMPERATURE, str(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(cls.LLM_TOP_P, str(default))
        try:
            return float(value)
        except (ValueError, TypeError):
            return default