Update: 提升至最新版本 / 逻辑增加 / 配置项增加 / 文档转移

Signed-off-by: Shanshui2024 <[email protected]>

Author: Shanshui2024

.env.example

@@ -0,0 +1,22 @@
+# 连接配置
+IP=127.0.0.1    # 监听IP地址 可不填 默认为0.0.0.0
+PORT=8443       # 监听端口 可不填 默认为8443
+WEBHOOK_PATH=/webhook   # 回调路径 可不填 默认为根 / 
+
+## SSL配置 可不指定路径 不指定时默认为./data/cert.pem和./data/key.pem
+SSL_CERT=./data/cert.pem
+SSL_KEY=./data/key.pem
+
+
+# 机器人配置
+APPID=appid # QQ开放平台展示的AppID
+BOT_SECRET=appsecret # QQ开放平台展示的AppSecret
+
+
+# 通用配置
+LOG_LEVEL=INFO # 日志等级 可不填 默认为INFO 可选值:INFO DEBUG
+
+
+DATA_DIR=./data # 数据目录 可不填 默认为./data
+LOG_DIR=./logs # 日志目录 可不填 默认为./logs
+PLUGINS_DIR=./plugins # 插件目录 可不填 默认为./plugins

Core/Auth.py

@@ -8,131 +8,113 @@
 import json
 import threading
 from Core.Logger import logger
+from Core.Config import config
 from datetime import timedelta
 
-# 从 QQ 机器人控制台获取的 Bot Secret
-BOT_SECRET = "您的BOTSecret"
-
-# 从 QQ 机器人控制台获取的 App ID 和 Client Secret
-APP_ID = "您的APPID"
-# 全局变量存储 access_token 和过期时间
-access_token = None
-expires_at = 0
-
-start_time = None
-
-# 验证签名
-def verify_signature(bot_secret: str, signature_hex: str, timestamp: str, http_body: bytes) -> bool:
-    seed = bot_secret.encode("utf-8")
-    while len(seed) < nacl.bindings.crypto_sign_SEEDBYTES:
-        seed += seed
-    seed = seed[:nacl.bindings.crypto_sign_SEEDBYTES]
-    signing_key = nacl.signing.SigningKey(seed)
-    verify_key = signing_key.verify_key
-
-    msg = timestamp.encode("utf-8") + http_body
-    try:
-        signature = nacl.encoding.HexEncoder.decode(signature_hex)
-        verify_key.verify(msg, signature)
-        return True
-    except nacl.exceptions.BadSignatureError:
-        return False
-
-# 生成签名
-def generate_signature(bot_secret: str, plain_token: str, event_ts: str) -> str:
-    seed = bot_secret.encode("utf-8")
-    while len(seed) < nacl.bindings.crypto_sign_SEEDBYTES:
-        seed += seed
-    seed = seed[:nacl.bindings.crypto_sign_SEEDBYTES]
-    signing_key = nacl.signing.SigningKey(seed)
-
-    msg = event_ts.encode("utf-8") + plain_token.encode("utf-8")
-    signed_msg = signing_key.sign(msg)
-    signature = signed_msg.signature.hex()
-    logger.debug(f"框架 >>> 生成签名: {signature}")
-    return signature
-
-# 验证回调请求
-async def validate_callback(request: Request):
-    signature_hex = request.headers.get("X-Signature-Ed25519")
-    timestamp = request.headers.get("X-Signature-Timestamp")
-    http_body = await request.body()
-
-    if not verify_signature(BOT_SECRET, signature_hex, timestamp, http_body):
-        raise HTTPException(status_code=401, detail="Invalid signature")
-# 获取 access_token 的函数
-def get_access_token(app_id: str, client_secret: str) -> dict:
-    url = "https://bots.qq.com/app/getAppAccessToken"
-    headers = {"Content-Type": "application/json"}
-    data = {"appId": app_id, "clientSecret": client_secret}
-    response = requests.post(url, headers=headers, data=json.dumps(data))
-    if response.status_code == 200:
-        return response.json()
-    else:
-        logger.error(f"框架 >>> 获取 access token 失败: {response.text}")
-        raise Exception(f"Failed to get access token: {response.text}")
-
-# 守护线程函数
-def token_refresh_thread(app_id: str, client_secret: str):
-    global access_token, expires_at, start_time
-    start_time = time.time()
-    while True:
+
+class Auth:
+    def __init__(self, Bot_Secret: str, AppId: str):
+        self.BOTSECRET = Bot_Secret
+        self.APPID = AppId
+        self.access_token = None
+        self.expires_at = 0
+        self.start_time = None
+
+    def verify_signature(self, signature_hex: str, timestamp: str, http_body: bytes) -> bool:
+        seed = self.BOTSECRET.encode("utf-8")
+        while len(seed) < nacl.bindings.crypto_sign_SEEDBYTES:
+            seed += seed
+        seed = seed[:nacl.bindings.crypto_sign_SEEDBYTES]
+        signing_key = nacl.signing.SigningKey(seed)
+        verify_key = signing_key.verify_key
+
+        msg = timestamp.encode("utf-8") + http_body
         try:
-            # 获取新的 access_token
-            response = get_access_token(app_id, client_secret)
-            access_token = response["access_token"]
-            expires_in = int(response["expires_in"])
-            expires_at = time.time() + int(expires_in) - 45  # 提前45s刷新 / 详见：https://bot.q.qq.com/wiki/develop/api-v2/dev-prepare/interface-framework/api-use.html#%E8%8E%B7%E5%8F%96%E8%B0%83%E7%94%A8%E5%87%AD%E8%AF%81
-            logger.debug(f"框架 >>> Access token 已刷新: {access_token}, 将于 {expires_in} s 后过期")
-        except Exception as e:
-            logger.error(f"框架 >>> 刷新 access token 失败: {e}")
-            raise e
-        wait_time = expires_at - time.time() # 等待时间
-        if wait_time > 0:
-            time.sleep(wait_time)
+            signature = nacl.encoding.HexEncoder.decode(signature_hex)
+            verify_key.verify(msg, signature)
+            return True
+        except nacl.exceptions.BadSignatureError:
+            return False
+
+    def generate_signature(self, plain_token: str, event_ts: str) -> str:
+        seed = self.BOTSECRET.encode("utf-8")
+        while len(seed) < nacl.bindings.crypto_sign_SEEDBYTES:
+            seed +=seed
+        seed = seed[:nacl.bindings.crypto_sign_SEEDBYTES]
+        signing_key = nacl.signing.SigningKey(seed)
+
+        msg = event_ts.encode("utf-8") + plain_token.encode("utf-8")
+        signed_msg = signing_key.sign(msg)
+        signature = signed_msg.signature.hex()
+        logger.debug(f"框架 >>> 生成签名: {signature}")
+        return signature
+
+    async def validate_callback(self, request: Request):
+        signature_hex = request.headers.get("X-Signature-Ed25519")
+        timestamp = request.headers.get("X-Signature-Timestamp")
+        http_body = await request.body()
+
+        if not self.verify_signature(signature_hex, timestamp, http_body):
+            raise HTTPException(status_code=401, detail="Invalid signature")
+
+    def get_access_token(self) -> dict:
+        url = "https://bots.qq.com/app/getAppAccessToken"
+        headers = {"Content-Type": "application/json"}
+        data = {"appId": self.APPID, "clientSecret": self.BOTSECRET}
+        response = requests.post(url, headers=headers, data=json.dumps(data))
+        if response.status_code == 200:
+            return response.json()
         else:
-            logger.warning("框架 >>> Access token 已失效...立即获取新token")
-            pass
-
-def start_token_refresh(app_id: str, client_secret: str):
-    thread = threading.Thread(target=token_refresh_thread, args=(app_id, client_secret))
-    thread.daemon = True  # 设置为守护线程
-    thread.start()
-    logger.debug("框架 >>> Access Token刷新线程启动")
-
-def get_current_access_token():
-    if access_token is None or time.time() >= expires_at:
-        raise Exception("Access token is not available or has expired.")
-    return access_token
-
-
-def format_timedelta(delta: timedelta) -> str:
-    """
-    将 timedelta 对象格式化为 'XX天 XX时 XX分 XX秒' 的字符串
-    """
-    days = delta.days
-    hours, remainder = divmod(delta.seconds, 3600)
-    minutes, seconds = divmod(remainder, 60)
-    return f"{days}天 {hours:02d}时 {minutes:02d}分 {seconds:02d}秒"
-
-def get_current_run_time(string_out: bool = True) -> timedelta or str:
-    """
-    获取当前运行时间
-
-    Args:
-        string_out (bool): 是否返回格式化后的字符串。默认为 True，返回格式化后的字符串。
-                           如果设置为 False，则返回 timedelta 对象。
-
-    Returns:
-        timedelta or str: 根据 string_out 参数返回 timedelta 对象或格式化后的字符串。
-    """
-    global start_time
-    if start_time is None:
-        start_time = time.time()
-    now = time.time()
-    elapsed_time = timedelta(seconds=int(now - start_time))
-
-    if string_out:
-        return format_timedelta(elapsed_time)
-    else:
-        return elapsed_time
+            logger.error(f"框架 >>> 获取 access token 失败: {response.text}")
+            raise Exception(f"Failed to access get token: {response.text}")
+
+    def token_refresh_thread(self):
+        self.start_time = time.time()
+        while True:
+            try:
+                response = self.get_access_token()
+                logger.debug(f'框架 >>> 获得返回内容：{response}')
+                self.access_token = response["access_token"]
+                expires_in = int(response["expires_in"])
+                self.expires_at = time.time() + int(expires_in) - 45
+                logger.debug(f"框架 >>> Access token 已刷新: {self.access_token}, 将于 {expires_in} s 后过期")
+            except Exception as e:
+                logger.error(f"框架 >>> 刷新 access token 失败: {e}")
+                raise e
+            wait_time = self.expires_at - time.time()
+            if wait_time > 0:
+                time.sleep(wait_time)
+            else:
+                logger.warning("框架 >>> Access token 已失效...立即获取新token")
+
+    def start_token_refresh(self):
+        thread = threading.Thread(target=self.token_refresh_thread)
+        thread.daemon = True
+        thread.start()
+        logger.debug("框架 >>> Access Token刷新线程启动")
+
+    def get_current_access_token(self):
+        if self.access_token is None or time.time() >= self.expires_at:
+            raise Exception("Access token is not available or has expired.")
+        return self.access_token
+
+    def format_timedelta(self, delta: timedelta) -> str:
+        days = delta.days
+        hours, remainder = divmod(delta.seconds, 3600)
+        minutes, seconds = divmod(remainder, 60)
+        return f"{days}天 {hours:02d}时 {minutes:02d}分 {seconds:02d}秒"
+
+    def get_current_run_time(self, string_out: bool = True):
+        if self.start_time is None:
+            self.start_time = time.time()
+        now = time.time()
+        elapsed_time = timedelta(seconds=int(now - self.start_time))
+
+        if string_out:
+            return self.format_timedelta(elapsed_time)
+        else:
+            return elapsed_time
+
+
+# 实例化 Auth 类
+auth = Auth(config.bot_secret, str(config.appid))

Core/Config.py

@@ -0,0 +1,81 @@
+import os
+from dotenv import load_dotenv
+
+# 加载 .env 文件中的环境变量
+load_dotenv(".env")
+
+class Config:
+    """
+    配置类，用于管理应用程序的配置参数。
+
+    参数可以通过构造函数传入，或者从环境变量中读取。如果未提供参数且环境变量也未设置，
+    则使用默认值。
+
+    Attributes:
+        ip (str): Webhook 监听IP地址，默认 '127.0.0.1'。
+        port (int): Webhook 监听端口，默认 8443。
+        path (str): Webhook 监听路径，默认 '/webhook'。
+        appid (int): 机器人AppID。无默认值，用于登录机器人、发送消息。 
+        ssl_cert (str): SSL证书路径，默认 './data/cert.pem'。
+        ssl_key (str): SSL证书私钥路径，默认 './data/key.pem'。
+        bot_secret (str): 机器人AppSecret。无默认值，用于登录机器人、发送消息。
+        log_level (str): 日志级别，默认 'INFO'。
+        log_dir (str): 日志文件存储目录，默认 './logs'。
+        data_dir (str): 数据文件存储目录，默认 './data'。
+        plugins_dir (str): 插件文件存储目录，默认 './plugins'。
+    """
+
+    def __init__(self, ip=None, port=None, path=None, appid=None, bot_secret=None, ssl_cert=None, ssl_key=None, log_level=None, log_dir=None, data_dir=None, plugins_dir=None):
+        """
+        初始化配置实例。
+
+        Args:
+            ip (str, optional): IP地址。默认为环境变量中的 'IP' 或 '0.0.0.0'。
+            port (int, optional): 端口号。默认为环境变量中的 'PORT' 或 8443。
+            path (str, optional): Webhook 监听路径，默认为环境变量中的 'WEBHOOK_PATH' 或 '/'。
+            appid (int): 机器人AppID。无默认值，用于登录机器人、发送消息。
+            ssl_cert (str): SSL证书路径，默认 './data/cert.pem'。
+            ssl_key (str): SSL证书私钥路径，默认 './data/key.pem'。
+            bot_secret (str): 机器人AppSecret。无默认值，用于登录机器人、发送消息。
+            log_level (str, optional): 日志级别。默认为环境变量中的 'LOG_LEVEL' 或 'INFO'。
+            log_dir (str, optional): 日志目录。默认为环境变量中的 'LOG_DIR' 或 './logs'。
+            data_dir (str, optional): 数据目录。默认为环境变量中的 'DATA_DIR' 或 './data'。
+            plugins_dir (str, optional): 插件目录。默认为环境变量中的 'PLUGINS_DIR' 或 './plugins'。
+        """
+        self.ip = os.getenv('IP', ip or '0.0.0.0')
+        try:
+            self.port = int(os.getenv('PORT', port or 8443))
+        except ValueError:
+            raise ValueError("无效的端口号，请确保 PORT 是一个有效的整数")
+        self.path = os.getenv('WEBHOOK_PATH', path or '/')
+        self.appid = str(os.getenv('APPID', appid))
+        self.bot_secret = os.getenv('BOT_SECRET', bot_secret)
+        self.ssl_cert = os.getenv('SSL_CERT', ssl_cert or './data/cert.pem')
+        self.ssl_key = os.getenv('SSL_KEY', ssl_key or './data/key.pem')
+        self.log_level = os.getenv('LOG_LEVEL', log_level or 'INFO')
+        self.log_dir = os.getenv('LOG_DIR', log_dir or './logs')
+        self.data_dir = os.getenv('DATA_DIR', data_dir or './data')
+        self.plugins_dir = os.getenv('PLUGINS_DIR', plugins_dir or './plugins')
+
+        # 检查必填项
+        required_fields = {
+            'appid': self.appid,
+            'bot_secret': self.bot_secret,
+            'ssl_cert': self.ssl_cert,
+            'ssl_key': self.ssl_key
+        }
+        missing_fields = [k for k, v in required_fields.items() if v is None]
+        if missing_fields:
+            raise ValueError(f"配置项丢失！以下配置项必须有值: {', '.join(missing_fields)}")
+
+    def __repr__(self):
+        """
+        返回配置实例的字符串表示。
+
+        Returns:
+            str: 配置实例的字符串表示。
+        """
+        return f"Config(ip={self.ip}, port={self.port}, path={self.path}, appid={self.appid}, bot_secret={self.bot_secret}, ssl_cert={self.ssl_cert}, ssl_key={self.ssl_key}, log_level={self.log_level}, log_dir={self.log_dir}, data_dir={self.data_dir}, plugins_dir={self.plugins_dir})"
+
+# 创建 Config 实例
+config = Config()