yj
2025-07-28 69945b730fd3f6b6138ce50e49fc3392fcd74d71
config.py
@@ -4,6 +4,8 @@
import os
import json
import threading
from loguru import logger
class Settings:
@@ -12,95 +14,173 @@
    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:
            if os.path.exists(self.config_file):
                with open(self.config_file, 'r', encoding='utf-8') as f:
                    config_data = json.load(f)
                self._set_config_from_dict(config_data)
            else:
                # 如果配置文件不存在,使用默认值
                self._set_default_config()
            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:
            print(f"加载配置文件失败: {e}")
            self._set_default_config()
            raise Exception(f"加载配置文件失败: {e}")
    def _set_config_from_dict(self, config_data: dict):
        """从字典设置配置"""
        # 数据库配置
        self.database_url = config_data.get("database", {}).get("url", "mysql+pymysql://root:TAI%402019%23Zjun@120.24.39.179:3306/ecloud_dify")
        self.database_url = config_data["database"]["url"]
        # Redis配置
        self.redis_url = config_data.get("redis", {}).get("url", "redis://localhost:6379/0")
        self.redis_url = config_data["redis"]["url"]
        # E云管家配置
        ecloud_config = config_data.get("ecloud", {})
        self.ecloud_base_url = ecloud_config.get("base_url", "http://125.122.152.142:9899")
        self.ecloud_authorization = ecloud_config.get("authorization", "eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiIxMzYxMTQ1MjE3NSIsInBhc3N3b3JkIjoiJDJhJDEwJEU3Ry5LOEJzekphM2JGQlh0SG8vOXVrUk1NalVweGVVemguUDRnMkJBdHN2YXpBb0JIQWJpIn0.Gd2vbeJjL5pUGFhUngWPLkDTLhD3GUaEPXOkdoTf4KRh9o2FtST1OZJxmZuGdUy7WIYlIPVueoVyIu5iHOyi8A")
        self.ecloud_w_id = ecloud_config.get("w_id", "")
        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.get("dify", {})
        self.dify_base_url = dify_config.get("base_url", "https://api.dify.ai/v1")
        self.dify_api_key = dify_config.get("api_key", "app-OMnBr7zsf5UTV83Ey8QcSErA")
        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.get("server", {})
        self.server_host = server_config.get("host", "0.0.0.0")
        self.server_port = server_config.get("port", 7979)
        self.debug = server_config.get("debug", True)
        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.get("logging", {})
        self.log_level = logging_config.get("level", "INFO")
        self.log_file = logging_config.get("file", "logs/app.log")
        logging_config = config_data["logging"]
        self.log_level = logging_config["level"]
        self.log_file = logging_config["file"]
        # 消息处理配置
        msg_config = config_data.get("message_processing", {})
        self.max_retry_count = msg_config.get("max_retry_count", 3)
        self.retry_delay = msg_config.get("retry_delay", 5)
        self.queue_timeout = msg_config.get("queue_timeout", 300)
        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.get("customer_service", {})
        self.customer_service_names = customer_service_config.get("names", ["客服1", "客服2"])
        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", "智能客服")
    def _set_default_config(self):
        """设置默认配置"""
        # 数据库配置
        self.database_url = "mysql+pymysql://root:TAI%402019%23Zjun@120.24.39.179:3306/ecloud_dify"
        # 好友忽略配置
        friend_ignore_config = config_data["friend_ignore"]
        self.friend_ignore_enabled = friend_ignore_config["enabled"]
        self.friend_ignore_whitelist = friend_ignore_config["whitelist"]
        # Redis配置
        self.redis_url = "redis://localhost:6379/0"
        # 静默模式配置
        silence_mode_config = config_data["silence_mode"]
        self.silence_mode_enabled = silence_mode_config["enabled"]
        self.silence_duration_minutes = silence_mode_config["duration_minutes"]
        # E云管家配置
        self.ecloud_base_url = "http://125.122.152.142:9899"
        self.ecloud_authorization = "eyJhbGciOiJIUzUxMiJ9.eyJzdWIiOiIxMzYxMTQ1MjE3NSIsInBhc3N3b3JkIjoiJDJhJDEwJEU3Ry5LOEJzekphM2JGQlh0SG8vOXVrUk1NalVweGVVemguUDRnMkJBdHN2YXpBb0JIQWJpIn0.Gd2vbeJjL5pUGFhUngWPLkDTLhD3GUaEPXOkdoTf4KRh9o2FtST1OZJxmZuGdUy7WIYlIPVueoVyIu5iHOyi8A"
        self.ecloud_w_id = ""
        # 在线状态监控配置
        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"]
        # DifyAI配置
        self.dify_base_url = "https://api.dify.ai/v1"
        self.dify_api_key = "app-OMnBr7zsf5UTV83Ey8QcSErA"
        # 邮件通知配置
        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"]
        # 服务配置
        self.server_host = "0.0.0.0"
        self.server_port = 7979
        self.debug = True
        # 短信通知配置
        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"]
        # 日志配置
        self.log_level = "INFO"
        self.log_file = "logs/app.log"
        # 关键词过滤配置
        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", [])
        # 消息处理配置
        self.max_retry_count = 3
        self.retry_delay = 5
        self.queue_timeout = 300
    def update_ecloud_w_id(self, new_w_id: str) -> bool:
        """
        动态更新E云管家的w_id配置
        # 客服配置
        self.customer_service_names = ["客服1", "客服2"]
        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
# 全局配置实例