🎨 代码格式化

zhenxun/utils/browser.py

@@ -1,102 +1,14 @@
 from collections.abc import AsyncGenerator
 from contextlib import asynccontextmanager
-import os
 from pathlib import Path
-import sys
 from typing import Any, Literal
 
-from nonebot import get_driver
 from nonebot_plugin_alconna import UniMessage
 from nonebot_plugin_htmlrender import get_browser
-from playwright.__main__ import main
-from playwright.async_api import Browser, Page, Playwright, async_playwright
+from playwright.async_api import Page
 
-from zhenxun.configs.config import BotConfig
-from zhenxun.services.log import logger
 from zhenxun.utils.message import MessageUtils
 
-driver = get_driver()
-
-_playwright: Playwright | None = None
-_browser: Browser | None = None
-
-
-# @driver.on_startup
-# async def start_browser():
-#     global _playwright
-#     global _browser
-#     install()
-#     await check_playwright_env()
-#     _playwright = await async_playwright().start()
-#     _browser = await _playwright.chromium.launch()
-
-
-# @driver.on_shutdown
-# async def shutdown_browser():
-#     if _browser:
-#         await _browser.close()
-#     if _playwright:
-#         await _playwright.stop()  # type: ignore
-
-
-# def get_browser() -> Browser:
-#     if not _browser:
-#         raise RuntimeError("playwright is not initalized")
-#     return _browser
-
-
-def install():
-    """自动安装、更新 Chromium"""
-
-    def set_env_variables():
-        os.environ["PLAYWRIGHT_DOWNLOAD_HOST"] = (
-            "https://npmmirror.com/mirrors/playwright/"
-        )
-        if BotConfig.system_proxy:
-            os.environ["HTTPS_PROXY"] = BotConfig.system_proxy
-
-    def restore_env_variables():
-        os.environ.pop("PLAYWRIGHT_DOWNLOAD_HOST", None)
-        if BotConfig.system_proxy:
-            os.environ.pop("HTTPS_PROXY", None)
-        if original_proxy is not None:
-            os.environ["HTTPS_PROXY"] = original_proxy
-
-    def try_install_chromium():
-        try:
-            sys.argv = ["", "install", "chromium"]
-            main()
-        except SystemExit as e:
-            return e.code == 0
-        return False
-
-    logger.info("检查 Chromium 更新")
-
-    original_proxy = os.environ.get("HTTPS_PROXY")
-    set_env_variables()
-
-    success = try_install_chromium()
-
-    if not success:
-        logger.info("Chromium 更新失败，尝试从原始仓库下载，速度较慢")
-        os.environ["PLAYWRIGHT_DOWNLOAD_HOST"] = ""
-        success = try_install_chromium()
-
-    restore_env_variables()
-
-    if not success:
-        raise RuntimeError("未知错误，Chromium 下载失败")
-
-
-async def check_playwright_env():
-    """检查 Playwright 依赖"""
-    logger.info("检查 Playwright 依赖")
-    try:
-        async with async_playwright() as p:
-            await p.chromium.launch()
-    except Exception as e:
-        raise ImportError("加载失败，Playwright 依赖不全，") from e
-
 
 class BrowserIsNone(Exception):
     pass

zhenxun/utils/http_utils.py

@@ -1,7 +1,6 @@
 import asyncio
 from collections.abc import AsyncGenerator, Awaitable, Callable, Sequence
 from contextlib import asynccontextmanager
-import json
 from pathlib import Path
 import time
 from typing import Any, ClassVar, cast
@@ -17,11 +16,13 @@ from rich.progress import (
     TextColumn,
     TransferSpeedColumn,
 )
+import ujson as json
 
 from zhenxun.configs.config import BotConfig
 from zhenxun.services.log import logger
 from zhenxun.utils.decorator.retry import Retry
 from zhenxun.utils.exception import AllURIsFailedError
+from zhenxun.utils.manager.priority_manager import PriorityLifecycle
 from zhenxun.utils.user_agent import get_user_agent
 
 from .browser import AsyncPlaywright, BrowserIsNone  # noqa: F401
@@ -32,16 +33,14 @@ driver = nonebot.get_driver()
 _client: AsyncClient | None = None
 
 
-@driver.on_startup
-async def _init_client():
+@PriorityLifecycle.on_startup(priority=5)
+async def _():
     """
     在Bot启动时初始化全局httpx客户端。
     """
     global _client
-    proxy_url = BotConfig.system_proxy if BotConfig.system_proxy else None
-
     client_kwargs = {}
-    if proxy_url:
+    if proxy_url := BotConfig.system_proxy or None:
         try:
             version_parts = httpx.__version__.split(".")
             major = int("".join(c for c in version_parts[0] if c.isdigit()))
@@ -71,7 +70,7 @@ async def _init_client():
 
 
 @driver.on_shutdown
-async def _close_client():
+async def _():
     """
     在Bot关闭时关闭全局httpx客户端。
     """
@@ -172,11 +171,7 @@ class AsyncHttpx:
         use_proxy = final_config.pop("use_proxy", True)
 
         if "proxies" not in final_config and "proxy" not in final_config:
-            if use_proxy:
-                final_config["proxies"] = cls.default_proxy
-            else:
-                final_config["proxies"] = None
-
+            final_config["proxies"] = cls.default_proxy if use_proxy else None
         return final_config
 
     @classmethod
@@ -212,8 +207,7 @@ class AsyncHttpx:
         """
         [内部] 执行单次HTTP请求的私有核心方法，被重试装饰器包裹。
         """
-        response = await client.request(method, url, **kwargs)
-        return response
+        return await client.request(method, url, **kwargs)
 
     @classmethod
     async def _single_request(
@@ -245,7 +239,7 @@ class AsyncHttpx:
         """
         通用执行器，按顺序尝试多个URL，直到成功。
 
-        Args:
+        参数:
             urls: 单个URL或URL列表。
             worker: 一个接受单个URL和其他kwargs并执行请求的协程函数。
             client: 可选的HTTP客户端。