优化IO问题

plugins/jd_sign_token/main.py

@@ -1,7 +1,5 @@
 from loguru import logger
-import requests
 import re
-from json import dumps as jsonDumps
 from typing import Dict, Any, List, Optional, Tuple
 
 from base.plugin_common.message_plugin_interface import MessagePluginInterface
@@ -10,133 +8,142 @@ from utils.decorator.plugin_decorators import plugin_stats_decorator
 from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
 from wechat_ipad import WechatAPIClient
 
+import aiohttp
+import asyncio
+from typing import Optional, List, Dict, Any
+
 
 class QL:
     def __init__(self, address: str, id: str, secret: str) -> None:
-        """
+        """初始化"""
-        初始化
-        """
         self.address = address
         self.id = id
         self.secret = secret
         self.valid = True
         self.auth = None
+        self._session: Optional[aiohttp.ClientSession] = None
 
-    def log(self, content: str) -> None:
+    async def login(self) -> bool:
-        """
+        """异步登录"""
-        日志
-        """
-        print(content)
-
-    def login(self) -> bool:
-        """
-        登录
-        """
         url = f"{self.address}/open/auth/token?client_id={self.id}&client_secret={self.secret}"
         try:
-            # 添加超时参数
+            async with aiohttp.ClientSession().get(url, timeout=aiohttp.ClientTimeout(total=15)) as response:
-            rjson = requests.get(url, timeout=(5, 15)).json()
+                if response.status == 200:
-            if (rjson['code'] == 200):
+                    rjson = await response.json()
-                self.auth = f"{rjson['data']['token_type']} {rjson['data']['token']}"
+                    if rjson['code'] == 200:
-                self.valid = True
+                        self.auth = f"{rjson['data']['token_type']} {rjson['data']['token']}"
-                return True
+                        self.valid = True
-            else:
+                        return True
-                self.log(f"登录失败：{rjson['message']}")
+                    else:
-                self.valid = False
+                        logger.info(f"登录失败：{rjson['message']}")
-                return False
+                        self.valid = False
+                        return False
+                else:
+                    logger.info(f"登录失败：HTTP {response.status}")
+                    self.valid = False
+                    return False
         except Exception as e:
             self.valid = False
-            self.log(f"登录失败：{str(e)}")
+            logger.info(f"登录失败：{str(e)}")
             return False
 
-    def getEnvs(self) -> list:
+    async def getEnvs(self) -> list:
-        """
+        """异步获取环境变量"""
-        获取环境变量
-        """
         # 每次操作前先登录，确保token有效
-        if not self.login():
+        if not await self.login():
             return []
 
         url = f"{self.address}/open/envs?searchValue="
         headers = {"Authorization": self.auth}
         try:
-            # 添加超时参数
+            async with aiohttp.ClientSession().get(url, headers=headers, timeout=aiohttp.ClientTimeout(total=15)) as response:
-            rjson = requests.get(url, headers=headers, timeout=(5, 15)).json()
+                if response.status == 200:
-            if (rjson['code'] == 200):
+                    rjson = await response.json()
-                return rjson['data']
+                    if rjson['code'] == 200:
-            else:
+                        return rjson['data']
-                self.log(f"获取环境变量失败：{rjson['message']}")
+                    else:
-                return []
+                        logger.info(f"获取环境变量失败：{rjson['message']}")
+                        return []
+                else:
+                    logger.info(f"获取环境变量失败：HTTP {response.status}")
+                    return []
         except Exception as e:
-            self.log(f"获取环境变量失败：{str(e)}")
+            logger.info(f"获取环境变量失败：{str(e)}")
             return []
 
-    def deleteEnvs(self, ids: list) -> bool:
+    async def deleteEnvs(self, ids: list) -> bool:
-        """
+        """异步删除环境变量"""
-        删除环境变量
-        """
         # 每次操作前先登录，确保token有效
-        if not self.login():
+        if not await self.login():
             return False
 
         url = f"{self.address}/open/envs"
         headers = {"Authorization": self.auth, "content-type": "application/json"}
         try:
-            rjson = requests.delete(url, headers=headers, data=jsonDumps(ids)).json()
+            async with aiohttp.ClientSession().delete(url, headers=headers, json=ids,
-            if (rjson['code'] == 200):
+                                      timeout=aiohttp.ClientTimeout(total=15)) as response:
-                self.log(f"删除环境变量成功：{len(ids)}")
+                if response.status == 200:
-                return True
+                    rjson = await response.json()
-            else:
+                    if rjson['code'] == 200:
-                self.log(f"删除环境变量失败：{rjson['message']}")
+                        logger.info(f"删除环境变量成功：{len(ids)}")
-                return False
+                        return True
+                    else:
+                        logger.info(f"删除环境变量失败：{rjson['message']}")
+                        return False
+                else:
+                    logger.info(f"删除环境变量失败：HTTP {response.status}")
+                    return False
         except Exception as e:
-            self.log(f"删除环境变量失败：{str(e)}")
+            logger.info(f"删除环境变量失败：{str(e)}")
             return False
 
-    def addEnvs(self, envs: list) -> bool:
+    async def addEnvs(self, envs: list) -> bool:
-        """
+        """异步添加环境变量"""
-        新建环境变量
+        if not await self.login():
-        """
-        # 每次操作前先登录，确保token有效
-        if not self.login():
             return False
 
         url = f"{self.address}/open/envs"
         headers = {"Authorization": self.auth, "content-type": "application/json"}
         try:
-            # 添加超时参数
+            async with aiohttp.ClientSession().post(url, headers=headers, json=envs,
-            rjson = requests.post(url, headers=headers, data=jsonDumps(envs), timeout=(5, 15)).json()
+                                    timeout=aiohttp.ClientTimeout(total=15)) as response:
-            if (rjson['code'] == 200):
+                if response.status == 200:
-                self.log(f"新建环境变量成功：{len(envs)}")
+                    rjson = await response.json()
-                return True
+                    if rjson['code'] == 200:
-            else:
+                        logger.info(f"添加环境变量成功：{len(envs)}")
-                self.log(f"新建环境变量失败：{rjson['message']}")
+                        return True
-                return False
+                    else:
+                        logger.info(f"添加环境变量失败：{rjson['message']}")
+                        return False
+                else:
+                    logger.info(f"添加环境变量失败：HTTP {response.status}")
+                    return False
         except Exception as e:
-            self.log(f"新建环境变量失败：{str(e)}")
+            logger.info(f"添加环境变量失败：{str(e)}")
             return False
 
-    def updateEnv(self, env: dict) -> bool:
+    async def updateEnv(self, env: dict) -> bool:
-        """
+        """异步更新环境变量"""
-        更新环境变量
+        if not await self.login():
-        """
-        # 每次操作前先登录，确保token有效
-        if not self.login():
             return False
 
         url = f"{self.address}/open/envs"
         headers = {"Authorization": self.auth, "content-type": "application/json"}
         try:
-            # 添加超时参数
+            async with aiohttp.ClientSession().put(url, headers=headers, json=env, timeout=aiohttp.ClientTimeout(total=15)) as response:
-            rjson = requests.put(url, headers=headers, data=jsonDumps(env), timeout=(5, 15)).json()
+                if response.status == 200:
-            if (rjson['code'] == 200):
+                    rjson = await response.json()
-                self.log(f"更新环境变量成功")
+                    if rjson['code'] == 200:
-                return True
+                        logger.info(f"更新环境变量成功：{env.get('id')}")
-            else:
+                        return True
-                self.log(f"更新环境变量失败：{rjson['message']}")
+                    else:
-                return False
+                        logger.info(f"更新环境变量失败：{rjson['message']}")
+                        return False
+                else:
+                    logger.info(f"更新环境变量失败：HTTP {response.status}")
+                    return False
         except Exception as e:
-            self.log(f"更新环境变量失败：{str(e)}")
+            logger.info(f"更新环境变量失败：{str(e)}")
             return False
 
 

utils/markdown_to_image.py

@@ -1,4 +1,5 @@
 import subprocess
+import time
 
 import markdown
 from playwright.async_api import async_playwright
@@ -235,7 +236,8 @@ async def convert_md_str_to_image(md_content, output_image):
     """
     将 Markdown 字符串转换为图片（异步）。
     """
-    temp_html = 'temp_output.html'
+    timestamp = int(time.time())
+    temp_html = f"temp_output_{timestamp}.html"
     md_str_to_html(md_content, temp_html)
     await html_to_image(temp_html, output_image)
     os.remove(temp_html)