280 lines
9.5 KiB
Python
280 lines
9.5 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'
|
||
TIMELINE_PAGESIZE = 'timeline_pagesize'
|
||
DEFAULT_RESET_PASSWORD = 'default_reset_password'
|
||
MAX_AUDIO_SIZE = 'max_audio_size'
|
||
|
||
# 声纹配置
|
||
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'
|
||
|
||
@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 = "我正在进行声纹采集，这段语音将用于身份识别和验证。\n\n声纹技术能够准确识别每个人独特的声音特征。") -> str:
|
||
"""获取声纹采集模版"""
|
||
return cls.get_config(cls.VOICEPRINT_TEMPLATE_TEXT, default)
|
||
|
||
@classmethod
|
||
def get_voiceprint_max_size(cls, default: int = 5242880) -> int:
|
||
"""获取声纹文件大小限制 (bytes), 默认5MB"""
|
||
value = cls.get_config(cls.VOICEPRINT_MAX_SIZE, default)
|
||
try:
|
||
return int(value)
|
||
except (ValueError, TypeError):
|
||
return default
|
||
|
||
@classmethod
|
||
def get_voiceprint_duration(cls, default: int = 12) -> int:
|
||
"""获取声纹采集最短时长 (秒)"""
|
||
value = cls.get_config(cls.VOICEPRINT_DURATION, 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(cls.VOICEPRINT_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(cls.VOICEPRINT_CHANNELS, default)
|
||
try:
|
||
return int(value)
|
||
except (ValueError, TypeError):
|
||
return 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
|