lint

Author: lgc2333

examples/external_example/templates/example_template/__init__.py

@@ -3,12 +3,11 @@
 import json
 from collections import deque
 from pathlib import Path
-from typing import Any, Dict, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 import jinja2 as jj
 from cookit.jinja import make_register_jinja_filter_deco
 from nonebot import get_plugin_config, require
-from nonebot_plugin_picstatus.bg_provider import BgData
 from nonebot_plugin_picstatus.templates import pic_template
 from nonebot_plugin_picstatus.templates.pw_render import (
     ROUTE_URL,
@@ -31,6 +30,9 @@
 
 from nonebot_plugin_htmlrender import get_new_page
 
+if TYPE_CHECKING:
+    from nonebot_plugin_picstatus.bg_provider import BgData
+
 RES_DIR = Path(__file__).parent / "res"
 
 template_env = jj.Environment(
@@ -76,7 +78,7 @@ class TemplateConfig(BaseModel):
         "time_counter",
     },
 )
-async def example_template(collected: Dict[str, Any], bg: BgData, **_):
+async def example_template(collected: dict[str, Any], bg: "BgData", **_):
     template = template_env.get_template("index.html.jinja")
     html = await template.render_async(d=collected, config=template_config)
 

nonebot_plugin_picstatus/__main__.py

@@ -24,7 +24,7 @@ def check_empty_arg_rule(arg: BaseMessage = CommandArg()):
 def trigger_rule():
     rule = Rule(check_empty_arg_rule)
     if config.ps_need_at:
-        rule = rule & to_me()
+        rule &= to_me()
     return rule
 
 

nonebot_plugin_picstatus/bg_provider.py

@@ -1,15 +1,18 @@
 import asyncio as aio
 import mimetypes
 import random
-from pathlib import Path
-from typing import Awaitable, Callable, Dict, List, NamedTuple, Optional, TypeVar
+from collections.abc import Awaitable
+from typing import TYPE_CHECKING, Callable, NamedTuple, Optional, TypeVar
 
 import anyio
 from httpx import AsyncClient, Response
 from nonebot import logger
 
 from .config import DEFAULT_BG_PATH, config
 
+if TYPE_CHECKING:
+    from pathlib import Path
+
 
 class BgData(NamedTuple):
     data: bytes
@@ -19,10 +22,10 @@ class BgData(NamedTuple):
 BGProviderType = Callable[[], Awaitable[BgData]]
 TBP = TypeVar("TBP", bound=BGProviderType)
 
-registered_bg_providers: Dict[str, BGProviderType] = {}
+registered_bg_providers: dict[str, BGProviderType] = {}
 
 
-def get_bg_files() -> List[Path]:
+def get_bg_files() -> list["Path"]:
     if not config.ps_bg_local_path.exists():
         logger.warning("Custom background path does not exist, fallback to default")
         return [DEFAULT_BG_PATH]
@@ -133,8 +136,8 @@ def __init__(self, preload_count: int):
         if preload_count < 1:
             raise ValueError("preload_count must be greater than 0")
         self.preload_count = preload_count
-        self.backgrounds: List[BgData] = []
-        self.tasks: List[aio.Task[None]] = []
+        self.backgrounds: list[BgData] = []
+        self.tasks: list[aio.Task[None]] = []
         self.task_signal: Optional[aio.Future[None]] = None
         self.signal_wait_lock = aio.Lock()
 

nonebot_plugin_picstatus/collectors/__init__.py

@@ -3,19 +3,9 @@
 import time
 from abc import abstractmethod
 from collections import deque
+from collections.abc import Awaitable
 from pathlib import Path
-from typing import (
-    Any,
-    Awaitable,
-    Callable,
-    Deque,
-    Dict,
-    Generic,
-    Optional,
-    Type,
-    TypeVar,
-    Union,
-)
+from typing import Any, Callable, Generic, Optional, TypeVar, Union
 from typing_extensions import override
 
 from nonebot import logger
@@ -68,13 +58,13 @@ async def get(self) -> T:
         return data
 
 
-class BasePeriodicCollector(Collector[T, Deque[T]], Generic[T]):
+class BasePeriodicCollector(Collector[T, deque[T]], Generic[T]):
     def __init__(self, size: int = config.ps_default_collect_cache_size) -> None:
         super().__init__()
         self.data = deque(maxlen=size)
 
     @override
-    async def get(self) -> Deque[T]:
+    async def get(self) -> deque[T]:
         return self.data
 
     async def collect(self):
@@ -88,12 +78,12 @@ async def collect(self):
             self.data.append(data)
 
 
-registered_collectors: Dict[str, Type[Collector]] = {}
-enabled_collectors: Dict[str, Collector] = {}
+registered_collectors: dict[str, type[Collector]] = {}
+enabled_collectors: dict[str, Collector] = {}
 
 
 def collector(name: str):
-    def deco(cls: Type[TC]) -> Type[TC]:
+    def deco(cls: type[TC]) -> type[TC]:
         if name in registered_collectors:
             raise ValueError(f"Collector {name} already exists")
         registered_collectors[name] = cls
@@ -121,7 +111,7 @@ async def enable_collectors(*names: str):
     await collect_perodic_collectors()
 
 
-def functional_collector(cls: Type[Collector], name: Optional[str] = None):
+def functional_collector(cls: type[Collector], name: Optional[str] = None):
     def deco(func: TCF) -> TCF:
         collector_name = name or func.__name__
         if not collector_name:
@@ -175,7 +165,7 @@ async def _get(self) -> R:
         return await self._calc(past, self.last_obj, time_passed)
 
 
-async def collect_all() -> Dict[str, Any]:
+async def collect_all() -> dict[str, Any]:
     async def get(name: str):
         return name, await enabled_collectors[name].get()
 

nonebot_plugin_picstatus/collectors/bot.py

@@ -1,10 +1,9 @@
 import asyncio
 from dataclasses import dataclass
 from datetime import datetime
-from typing import List, Optional, Tuple
+from typing import TYPE_CHECKING, Optional
 
 from nonebot import get_bots, logger
-from nonebot.adapters import Bot as BaseBot
 from nonebot.matcher import current_bot
 
 from ..config import config
@@ -17,6 +16,9 @@
 from ..util import format_timedelta
 from . import normal_collector
 
+if TYPE_CHECKING:
+    from nonebot.adapters import Bot as BaseBot
+
 try:
     from nonebot.adapters.onebot.v11 import Bot as OBV11Bot
 except ImportError:
@@ -33,7 +35,7 @@ class BotStatus:
     msg_sent: str
 
 
-async def get_ob11_msg_num(bot: BaseBot) -> Tuple[Optional[int], Optional[int]]:
+async def get_ob11_msg_num(bot: "BaseBot") -> tuple[Optional[int], Optional[int]]:
     if not (config.ps_ob_v11_use_get_status and OBV11Bot and isinstance(bot, OBV11Bot)):
         return None, None
 
@@ -54,7 +56,7 @@ async def get_ob11_msg_num(bot: BaseBot) -> Tuple[Optional[int], Optional[int]]:
     return msg_rec, msg_sent
 
 
-async def get_bot_status(bot: BaseBot, now_time: datetime) -> BotStatus:
+async def get_bot_status(bot: "BaseBot", now_time: datetime) -> BotStatus:
     nick = (
         ((info := bot_info_cache[bot.self_id]).user_displayname or info.user_name)
         if (not config.ps_use_env_nick) and (bot.self_id in bot_info_cache)
@@ -85,7 +87,7 @@ async def get_bot_status(bot: BaseBot, now_time: datetime) -> BotStatus:
 
 
 @normal_collector()
-async def bots() -> List[BotStatus]:
+async def bots() -> list[BotStatus]:
     now_time = datetime.now().astimezone()
     return (
         [await get_bot_status(current_bot.get(), now_time)]

nonebot_plugin_picstatus/collectors/cpu.py

@@ -19,7 +19,7 @@ class CpuFreq:
 async def cpu_brand() -> str:
     try:
         brand = (
-            cast(str, get_cpu_info().get("brand_raw", ""))
+            cast("str", get_cpu_info().get("brand_raw", ""))
             .split("@", maxsplit=1)[0]
             .strip()
         )

nonebot_plugin_picstatus/collectors/disk.py

@@ -1,5 +1,5 @@
 from dataclasses import dataclass
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
 import psutil
 from nonebot import logger
@@ -35,7 +35,7 @@ class DiskIO:
 
 
 @periodic_collector()
-async def disk_usage() -> List[DiskUsageType]:
+async def disk_usage() -> list[DiskUsageType]:
     def get_one(disk: sdiskpart) -> Optional[DiskUsageType]:
         mountpoint = disk.mountpoint
 
@@ -71,13 +71,13 @@ def get_one(disk: sdiskpart) -> Optional[DiskUsageType]:
 
 
 @collector("disk_io")
-class DiskIOCollector(TimeBasedCounterCollector[Dict[str, sdiskio], List[DiskIO]]):
+class DiskIOCollector(TimeBasedCounterCollector[dict[str, sdiskio], list[DiskIO]]):
     async def _calc(
         self,
-        past: Dict[str, sdiskio],
-        now: Dict[str, sdiskio],
+        past: dict[str, sdiskio],
+        now: dict[str, sdiskio],
         time_passed: float,
-    ) -> List[DiskIO]:
+    ) -> list[DiskIO]:
         def calc_one(name: str, past_it: sdiskio, now_it: sdiskio) -> Optional[DiskIO]:
             if match_list_regexp(config.ps_ignore_disk_ios, name):
                 # logger.info(f"IO统计 磁盘 {name} 匹配 {regex.re.pattern}，忽略")
@@ -98,5 +98,5 @@ def calc_one(name: str, past_it: sdiskio, now_it: sdiskio) -> Optional[DiskIO]:
             res.sort(key=lambda x: x.read + x.write, reverse=True)
         return res
 
-    async def _get_obj(self) -> Dict[str, sdiskio]:
+    async def _get_obj(self) -> dict[str, sdiskio]:
         return psutil.disk_io_counters(perdisk=True)

nonebot_plugin_picstatus/collectors/misc.py

@@ -3,7 +3,7 @@
 import time
 from datetime import datetime
 from pathlib import Path
-from typing import Dict, Optional, Tuple, Union
+from typing import Optional, Union
 
 import nonebot
 import psutil
@@ -13,9 +13,9 @@
 from . import first_time_collector, normal_collector, periodic_collector
 
 
-def parse_env(env: str) -> Dict[str, Optional[str]]:
+def parse_env(env: str) -> dict[str, Optional[str]]:
     env_lines = env.strip().splitlines()
-    env_dict: Dict[str, Optional[str]] = {}
+    env_dict: dict[str, Optional[str]] = {}
 
     for line in env_lines:
         if "=" not in line:
@@ -28,7 +28,7 @@ def parse_env(env: str) -> Dict[str, Optional[str]]:
     return env_dict
 
 
-def parse_env_file(env_file: Union[str, Path]) -> Optional[Dict[str, Optional[str]]]:
+def parse_env_file(env_file: Union[str, Path]) -> Optional[dict[str, Optional[str]]]:
     if not isinstance(env_file, Path):
         env_file = Path(env_file)
     if not env_file.exists():
@@ -38,7 +38,7 @@ def parse_env_file(env_file: Union[str, Path]) -> Optional[Dict[str, Optional[st
 
 
 # Thanks to https://github.com/nonedesktop/nonebot-plugin-guestool/blob/main/nonebot_plugin_guestool/info.py
-def get_linux_name_version() -> Optional[Tuple[str, str]]:
+def get_linux_name_version() -> Optional[tuple[str, str]]:
     env = parse_env_file("/etc/os-release")
     if env and (name := env.get("NAME")) and (version_id := env.get("VERSION_ID")):
         return name, version_id

nonebot_plugin_picstatus/collectors/network.py

@@ -1,7 +1,7 @@
 import asyncio
 import time
 from dataclasses import dataclass
-from typing import Dict, List, Optional, Union
+from typing import Optional, Union
 
 import psutil
 from httpx import AsyncClient, ReadTimeout
@@ -37,13 +37,13 @@ class NetworkConnectionError:
 
 
 @collector("network_io")
-class NetworkIOCollector(TimeBasedCounterCollector[Dict[str, snetio], List[NetworkIO]]):
+class NetworkIOCollector(TimeBasedCounterCollector[dict[str, snetio], list[NetworkIO]]):
     async def _calc(
         self,
-        past: Dict[str, snetio],
-        now: Dict[str, snetio],
+        past: dict[str, snetio],
+        now: dict[str, snetio],
         time_passed: float,
-    ) -> List[NetworkIO]:
+    ) -> list[NetworkIO]:
         def calc_one(name: str, past_it: snetio, now_it: snetio) -> Optional[NetworkIO]:
             if match_list_regexp(config.ps_ignore_nets, name):
                 # logger.info(f"网卡IO统计 {name} 匹配 {regex.re.pattern}，忽略")
@@ -64,12 +64,12 @@ def calc_one(name: str, past_it: snetio, now_it: snetio) -> Optional[NetworkIO]:
             res.sort(key=lambda x: x.sent + x.recv, reverse=True)
         return res
 
-    async def _get_obj(self) -> Dict[str, snetio]:
+    async def _get_obj(self) -> dict[str, snetio]:
         return psutil.net_io_counters(pernic=True)
 
 
 @normal_collector()
-async def network_connection() -> List[NetworkConnectionType]:
+async def network_connection() -> list[NetworkConnectionType]:
     def format_conn_error(error: Exception) -> str:
         if isinstance(error, ReadTimeout):
             return "超时"

nonebot_plugin_picstatus/collectors/process.py

@@ -1,6 +1,6 @@
 import asyncio
 from dataclasses import dataclass
-from typing import List, Optional, Union, cast
+from typing import Optional, Union, cast
 
 import psutil
 
@@ -17,7 +17,7 @@ class ProcessStatus:
 
 
 @periodic_collector()
-async def process_status() -> List[ProcessStatus]:
+async def process_status() -> list[ProcessStatus]:
     if not config.ps_proc_len:
         return []
 
@@ -44,7 +44,7 @@ def sorter(x: ProcessStatus):
         return x.cpu
 
     proc_list = cast(
-        List[Union[Optional[ProcessStatus], Exception]],
+        "list[Union[Optional[ProcessStatus], Exception]]",
         await asyncio.gather(
             *(parse_one(proc) for proc in psutil.process_iter()),
             return_exceptions=True,