756 lines
29 KiB
Python
756 lines
29 KiB
Python
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
|