"""
|
配置管理模块
|
"""
|
|
import os
|
import json
|
import threading
|
from loguru import logger
|
|
|
class Settings:
|
"""应用配置"""
|
|
def __init__(self, config_file: str = "config.json"):
|
"""初始化配置，从JSON文件加载"""
|
self.config_file = config_file
|
self._lock = threading.Lock() # 用于线程安全的配置更新
|
self._load_config()
|
|
def _load_config(self):
|
"""从JSON文件加载配置"""
|
if not os.path.exists(self.config_file):
|
raise FileNotFoundError(f"配置文件 {self.config_file} 不存在")
|
|
try:
|
with open(self.config_file, 'r', encoding='utf-8') as f:
|
config_data = json.load(f)
|
self._set_config_from_dict(config_data)
|
except Exception as e:
|
raise Exception(f"加载配置文件失败: {e}")
|
|
def _set_config_from_dict(self, config_data: dict):
|
"""从字典设置配置"""
|
# 数据库配置
|
self.database_url = config_data["database"]["url"]
|
|
# Redis配置
|
self.redis_url = config_data["redis"]["url"]
|
|
# E云管家配置
|
ecloud_config = config_data["ecloud"]
|
self.ecloud_base_url = ecloud_config["base_url"]
|
self.ecloud_authorization = ecloud_config["authorization"]
|
self.ecloud_w_id = ecloud_config["w_id"]
|
|
# DifyAI配置
|
dify_config = config_data["dify"]
|
self.dify_base_url = dify_config["base_url"]
|
self.dify_api_key = dify_config["api_key"]
|
self.dify_streaming_enabled = dify_config["streaming_enabled"]
|
self.dify_streaming_timeout = dify_config["streaming_timeout"]
|
|
# 服务配置
|
server_config = config_data["server"]
|
self.server_host = server_config["host"]
|
self.server_port = server_config["port"]
|
self.debug = server_config["debug"]
|
|
# 日志配置
|
logging_config = config_data["logging"]
|
self.log_level = logging_config["level"]
|
self.log_file = logging_config["file"]
|
|
# 消息处理配置
|
msg_config = config_data["message_processing"]
|
self.max_retry_count = msg_config["max_retry_count"]
|
self.retry_delay = msg_config["retry_delay"]
|
self.queue_timeout = msg_config["queue_timeout"]
|
|
# 客服配置
|
customer_service_config = config_data["customer_service"]
|
self.customer_service_names = customer_service_config["names"]
|
self.customer_service_default_name = customer_service_config.get("default_name", "智能客服")
|
|
# 好友忽略配置
|
friend_ignore_config = config_data["friend_ignore"]
|
self.friend_ignore_enabled = friend_ignore_config["enabled"]
|
self.friend_ignore_whitelist = friend_ignore_config["whitelist"]
|
|
# 静默模式配置
|
silence_mode_config = config_data.get("silence_mode", {})
|
self.silence_mode_enabled = silence_mode_config.get("enabled", True)
|
self.silence_duration_minutes = silence_mode_config.get("duration_minutes", 10)
|
|
# 在线状态监控配置
|
online_status_config = config_data["online_status_monitor"]
|
self.online_status_enabled = online_status_config["enabled"]
|
self.online_status_check_interval = online_status_config["check_interval_minutes"]
|
self.online_status_notification_message = online_status_config["notification_message"]
|
|
# 邮件通知配置
|
email_config = config_data["email_notification"]
|
self.email_enabled = email_config["enabled"]
|
self.email_smtp_server = email_config["smtp_server"]
|
self.email_smtp_port = email_config["smtp_port"]
|
self.email_smtp_username = email_config["smtp_username"]
|
self.email_smtp_password = email_config["smtp_password"]
|
self.email_from_email = email_config["from_email"]
|
self.email_to_emails = email_config["to_emails"]
|
|
# 短信通知配置
|
sms_config = config_data["sms_notification"]
|
self.sms_enabled = sms_config["enabled"]
|
self.sms_api_url = sms_config["api_url"]
|
self.sms_username = sms_config["username"]
|
self.sms_password = sms_config["password"]
|
self.sms_phone_numbers = sms_config["phone_numbers"]
|
|
# 关键词过滤配置
|
keyword_filter_config = config_data.get("keyword_filter", {})
|
self.keyword_filter_enabled = keyword_filter_config.get("enabled", False)
|
self.keyword_filter_keywords = keyword_filter_config.get("keywords", [])
|
|
# 消息聚合配置
|
message_aggregation_config = config_data.get("message_aggregation", {})
|
self.message_aggregation_enabled = message_aggregation_config.get("enabled", True)
|
self.message_aggregation_timeout = message_aggregation_config.get("timeout_seconds", 15)
|
|
# 结束字符串配置
|
self.end_str_list = config_data.get("end_str_list", [])
|
|
# 测试群组关键字配置
|
self.test_group_keywords = config_data.get("test_group_keywords", [])
|
|
def update_ecloud_w_id(self, new_w_id: str) -> bool:
|
"""
|
动态更新E云管家的w_id配置
|
|
Args:
|
new_w_id: 新的w_id值
|
|
Returns:
|
更新成功返回True，失败返回False
|
"""
|
if not new_w_id or new_w_id == self.ecloud_w_id:
|
return True
|
|
with self._lock:
|
try:
|
# 更新内存中的配置
|
old_w_id = self.ecloud_w_id
|
self.ecloud_w_id = new_w_id
|
|
# 更新配置文件
|
if self._update_config_file_w_id(new_w_id):
|
logger.info(f"成功更新w_id: {old_w_id} -> {new_w_id}")
|
return True
|
else:
|
# 如果文件更新失败，回滚内存配置
|
self.ecloud_w_id = old_w_id
|
logger.error(f"更新配置文件失败，回滚w_id: {new_w_id} -> {old_w_id}")
|
return False
|
|
except Exception as e:
|
logger.error(f"更新w_id异常: {str(e)}")
|
return False
|
|
def _update_config_file_w_id(self, new_w_id: str) -> bool:
|
"""
|
更新配置文件中的w_id
|
|
Args:
|
new_w_id: 新的w_id值
|
|
Returns:
|
更新成功返回True，失败返回False
|
"""
|
try:
|
# 读取当前配置文件
|
with open(self.config_file, 'r', encoding='utf-8') as f:
|
config_data = json.load(f)
|
|
# 更新w_id
|
config_data["ecloud"]["w_id"] = new_w_id
|
|
# 写回配置文件
|
with open(self.config_file, 'w', encoding='utf-8') as f:
|
json.dump(config_data, f, indent=2, ensure_ascii=False)
|
|
return True
|
|
except Exception as e:
|
logger.error(f"更新配置文件w_id失败: {str(e)}")
|
return False
|
|
def get_current_w_id(self) -> str:
|
"""
|
获取当前的w_id
|
|
Returns:
|
当前的w_id值
|
"""
|
with self._lock:
|
return self.ecloud_w_id
|
|
|
# 全局配置实例
|
settings = Settings()
|
