🎨 按ruff的检查调整程序代码

nonebot_bison/admin_page/__init__.py

@@ -1,17 +1,16 @@
 import os
 from pathlib import Path
-from typing import Union
 
-from nonebot import get_driver, on_command
-from nonebot.adapters.onebot.v11 import Bot
-from nonebot.adapters.onebot.v11.event import PrivateMessageEvent
-from nonebot.drivers.fastapi import Driver
 from nonebot.log import logger
 from nonebot.rule import to_me
 from nonebot.typing import T_State
+from nonebot import get_driver, on_command
+from nonebot.drivers.fastapi import Driver
+from nonebot.adapters.onebot.v11 import Bot
+from nonebot.adapters.onebot.v11.event import PrivateMessageEvent
 
-from ..plugin_config import plugin_config
 from .api import router as api_router
+from ..plugin_config import plugin_config
 from .token_manager import token_manager as tm
 
 STATIC_PATH = (Path(__file__).parent / "dist").resolve()
@@ -28,11 +27,9 @@ def init_fastapi():
     class SinglePageApplication(StaticFiles):
         def __init__(self, directory: os.PathLike, index="index.html"):
             self.index = index
-            super().__init__(
-                directory=directory, packages=None, html=True, check_dir=True
-            )
+            super().__init__(directory=directory, packages=None, html=True, check_dir=True)
 
-        def lookup_path(self, path: str) -> tuple[str, Union[os.stat_result, None]]:
+        def lookup_path(self, path: str) -> tuple[str, os.stat_result | None]:
             full_path, stat_res = super().lookup_path(path)
             if stat_res is None:
                 return super().lookup_path(self.index)
@@ -45,9 +42,7 @@ def init_fastapi():
             description="nonebot-bison webui and api",
         )
         nonebot_app.include_router(api_router)
-        nonebot_app.mount(
-            "/", SinglePageApplication(directory=static_path), name="bison-frontend"
-        )
+        nonebot_app.mount("/", SinglePageApplication(directory=static_path), name="bison-frontend")
 
         app = driver.server_app
         app.mount("/bison", nonebot_app, "nonebot-bison")
@@ -63,10 +58,9 @@ def init_fastapi():
     if host in ["0.0.0.0", "127.0.0.1"]:
         host = "localhost"
     logger.opt(colors=True).info(
-        f"Nonebot Bison frontend will be running at: "
-        f"<b><u>http://{host}:{port}/bison</u></b>"
+        f"Nonebot Bison frontend will be running at: " f"<b><u>http://{host}:{port}/bison</u></b>"
     )
-    logger.opt(colors=True).info(f"该页面不能被直接访问，请私聊bot <b><u>后台管理</u></b> 以获取可访问地址")
+    logger.opt(colors=True).info("该页面不能被直接访问，请私聊bot <b><u>后台管理</u></b> 以获取可访问地址")
 
 
 def register_get_token_handler():
@@ -93,6 +87,4 @@ if (STATIC_PATH / "index.html").exists():
     else:
         logger.warning("your driver is not fastapi, webui feature will be disabled")
 else:
-    logger.warning(
-        "Frontend file not found, please compile it or use docker or pypi version"
-    )
+    logger.warning("Frontend file not found, please compile it or use docker or pypi version")

nonebot_bison/admin_page/api.py

@@ -1,35 +1,30 @@
 import nonebot
 from fastapi import status
-from fastapi.exceptions import HTTPException
-from fastapi.param_functions import Depends
 from fastapi.routing import APIRouter
-from fastapi.security.oauth2 import OAuth2PasswordBearer
+from fastapi.param_functions import Depends
+from fastapi.exceptions import HTTPException
 from nonebot_plugin_saa import TargetQQGroup
+from fastapi.security.oauth2 import OAuth2PasswordBearer
 from nonebot_plugin_saa.utils.auto_select_bot import get_bot
 
-from ..apis import check_sub_target
-from ..config import (
-    NoSuchSubscribeException,
-    NoSuchTargetException,
-    NoSuchUserException,
-    config,
-)
-from ..config.db_config import SubscribeDupException
-from ..platform import platform_manager
-from ..types import Target as T_Target
 from ..types import WeightConfig
-from ..utils.get_bot import get_groups
+from ..apis import check_sub_target
 from .jwt import load_jwt, pack_jwt
+from ..types import Target as T_Target
+from ..utils.get_bot import get_groups
+from ..platform import platform_manager
 from .token_manager import token_manager
+from ..config.db_config import SubscribeDupException
+from ..config import NoSuchUserException, NoSuchTargetException, NoSuchSubscribeException, config
 from .types import (
-    AddSubscribeReq,
+    TokenResp,
     GlobalConf,
-    PlatformConfig,
     StatusResp,
+    SubscribeResp,
+    PlatformConfig,
+    AddSubscribeReq,
     SubscribeConfig,
     SubscribeGroupDetail,
-    SubscribeResp,
-    TokenResp,
 )
 
 router = APIRouter(prefix="/api", tags=["api"])
@@ -44,9 +39,7 @@ async def get_jwt_obj(token: str = Depends(oath_scheme)):
     return obj
 
 
-async def check_group_permission(
-    groupNumber: int, token_obj: dict = Depends(get_jwt_obj)
-):
+async def check_group_permission(groupNumber: int, token_obj: dict = Depends(get_jwt_obj)):
     groups = token_obj["groups"]
     for group in groups:
         if int(groupNumber) == group["id"]:
@@ -95,15 +88,13 @@ async def auth(token: str) -> TokenResp:
             jwt_obj = {
                 "id": qq,
                 "type": "admin",
-                "groups": list(
-                    map(
-                        lambda info: {
-                            "id": info["group_id"],
-                            "name": info["group_name"],
-                        },
-                        await get_groups(),
-                    )
-                ),
+                "groups": [
+                    {
+                        "id": info["group_id"],
+                        "name": info["group_name"],
+                    }
+                    for info in await get_groups()
+                ],
             }
             ret_obj = TokenResp(
                 type="admin",
@@ -134,18 +125,16 @@ async def get_subs_info(jwt_obj: dict = Depends(get_jwt_obj)) -> SubscribeResp:
     for group in groups:
         group_id = group["id"]
         raw_subs = await config.list_subscribe(TargetQQGroup(group_id=group_id))
-        subs = list(
-            map(
-                lambda sub: SubscribeConfig(
-                    platformName=sub.target.platform_name,
-                    targetName=sub.target.target_name,
-                    cats=sub.categories,
-                    tags=sub.tags,
-                    target=sub.target.target,
-                ),
-                raw_subs,
+        subs = [
+            SubscribeConfig(
+                platformName=sub.target.platform_name,
+                targetName=sub.target.target_name,
+                cats=sub.categories,
+                tags=sub.tags,
+                target=sub.target.target,
             )
-        )
+            for sub in raw_subs
+        ]
         res[group_id] = SubscribeGroupDetail(name=group["name"], subscribes=subs)
     return res
 
@@ -174,9 +163,7 @@ async def add_group_sub(groupNumber: int, req: AddSubscribeReq) -> StatusResp:
 @router.delete("/subs", dependencies=[Depends(check_group_permission)])
 async def del_group_sub(groupNumber: int, platformName: str, target: str):
     try:
-        await config.del_subscribe(
-            TargetQQGroup(group_id=groupNumber), target, platformName
-        )
+        await config.del_subscribe(TargetQQGroup(group_id=groupNumber), target, platformName)
     except (NoSuchUserException, NoSuchSubscribeException):
         raise HTTPException(status.HTTP_400_BAD_REQUEST, "no such user or subscribe")
     return StatusResp(ok=True, msg="")
@@ -204,13 +191,9 @@ async def get_weight_config():
 
 
 @router.put("/weight", dependencies=[Depends(check_is_superuser)])
-async def update_weigth_config(
-    platformName: str, target: str, weight_config: WeightConfig
-):
+async def update_weigth_config(platformName: str, target: str, weight_config: WeightConfig):
     try:
-        await config.update_time_weight_config(
-            T_Target(target), platformName, weight_config
-        )
+        await config.update_time_weight_config(T_Target(target), platformName, weight_config)
     except NoSuchTargetException:
         raise HTTPException(status.HTTP_400_BAD_REQUEST, "no such subscribe")
     return StatusResp(ok=True, msg="")

nonebot_bison/admin_page/jwt.py

@@ -1,7 +1,6 @@
-import datetime
 import random
 import string
-from typing import Optional
+import datetime
 
 import jwt
 
@@ -16,8 +15,8 @@ def pack_jwt(obj: dict) -> str:
     )
 
 
-def load_jwt(token: str) -> Optional[dict]:
+def load_jwt(token: str) -> dict | None:
     try:
         return jwt.decode(token, _key, algorithms=["HS256"])
-    except:
+    except Exception:
         return None

nonebot_bison/admin_page/token_manager.py

@@ -1,6 +1,5 @@
 import random
 import string
-from typing import Optional
 
 from expiringdict import ExpiringDict
 
@@ -9,7 +8,7 @@ class TokenManager:
     def __init__(self):
         self.token_manager = ExpiringDict(max_len=100, max_age_seconds=60 * 10)
 
-    def get_user(self, token: str) -> Optional[tuple]:
+    def get_user(self, token: str) -> tuple | None:
         res = self.token_manager.get(token)
         assert res is None or isinstance(res, tuple)
         return res