238 lines
8.0 KiB
Python
238 lines
8.0 KiB
Python
|
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_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
|