""" 配置管理模块 """ 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()