perf: optimization from lint

aiotieba_reviewer/__version__.py

@@ -1 +1 @@
-__version__ = "0.6.2a1"
+__version__ = "0.6.2a2"

aiotieba_reviewer/client.py

@@ -1,4 +1,4 @@
-from typing import AsyncGenerator
+from collections.abc import AsyncGenerator
 
 import aiotieba as tb
 

aiotieba_reviewer/database/mysql.py

@@ -1,7 +1,8 @@
 import datetime
 import logging
 import ssl
-from typing import Any, Callable, Final, List, Optional, Tuple
+from collections.abc import Callable
+from typing import Any, Final
 
 import asyncmy
 from aiotieba import get_logger
@@ -27,7 +28,7 @@ def _handle_exception(
 
     def wrapper(func):
         async def awrapper(self: "MySQLDB", *args, **kwargs):
-            def _log(log_level: int, err: Optional[Exception] = None) -> None:
+            def _log(log_level: int, err: Exception | None = None) -> None:
                 logger = get_logger()
                 if logger.isEnabledFor(err_log_level):
                     if err is None:
@@ -70,7 +71,7 @@ def _log(log_level: int, err: Optional[Exception] = None) -> None:
     return wrapper
 
 
-class MySQLDB(object):
+class MySQLDB:
     """
     MySQL交互
 
@@ -223,11 +224,11 @@ async def del_forum_score(self, fid: int) -> bool:
         return True
 
     @staticmethod
-    def _default_forum_score() -> Tuple[int, int]:
+    def _default_forum_score() -> tuple[int, int]:
         return (0, 0)
 
     @_handle_exception(create_table_forum_score, _default_forum_score)
-    async def get_forum_score(self, fid: int) -> Tuple[int, int]:
+    async def get_forum_score(self, fid: int) -> tuple[int, int]:
         """
         获取表forum_score中fid的评分
 
@@ -328,11 +329,11 @@ async def get_user_id(self, user_id: int) -> int:
         return 0
 
     @staticmethod
-    def _default_user_id_full() -> Tuple[int, str, datetime.datetime]:
+    def _default_user_id_full() -> tuple[int, str, datetime.datetime]:
         return (0, '', datetime.datetime(1970, 1, 1))
 
     @_handle_exception(create_table_user_id, _default_user_id_full)
-    async def get_user_id_full(self, user_id: int) -> Tuple[int, str, datetime.datetime]:
+    async def get_user_id_full(self, user_id: int) -> tuple[int, str, datetime.datetime]:
         """
         获取表user_id_{fname}中user_id的完整信息
 
@@ -356,7 +357,7 @@ async def get_user_id_full(self, user_id: int) -> Tuple[int, str, datetime.datet
     @_handle_exception(create_table_user_id, list)
     async def get_user_id_list(
         self, lower_permission: int = 0, upper_permission: int = 50, *, limit: int = 1, offset: int = 0
-    ) -> List[int]:
+    ) -> list[int]:
         """
         获取表user_id_{fname}中user_id的列表
 
@@ -471,11 +472,11 @@ async def get_imghash(self, img_hash: int, *, hamming_dist: int = 0) -> int:
         return 0
 
     @staticmethod
-    def _default_imghash_full() -> Tuple[int, str]:
+    def _default_imghash_full() -> tuple[int, str]:
         return (0, '')
 
     @_handle_exception(create_table_imghash, _default_imghash_full)
-    async def get_imghash_full(self, img_hash: int, *, hamming_dist: int = 0) -> Tuple[int, str]:
+    async def get_imghash_full(self, img_hash: int, *, hamming_dist: int = 0) -> tuple[int, str]:
         """
         获取表imghash_{fname}中img_hash的完整信息
 

aiotieba_reviewer/database/sqlite.py

@@ -1,7 +1,8 @@
 import logging
 import sqlite3
+from collections.abc import Callable
 from pathlib import Path
-from typing import Any, Callable, Optional
+from typing import Any
 
 from aiotieba import get_logger
 
@@ -22,7 +23,7 @@ def handle_exception(
 
     def wrapper(func):
         def inner(self, *args, **kwargs):
-            def _log(log_level: int, err: Optional[Exception] = None) -> None:
+            def _log(log_level: int, err: Exception | None = None) -> None:
                 logger = get_logger()
                 if logger.isEnabledFor(err_log_level):
                     if err is None:
@@ -53,7 +54,7 @@ def _log(log_level: int, err: Optional[Exception] = None) -> None:
     return wrapper
 
 
-class SQLiteDB(object):
+class SQLiteDB:
     """
     SQLite交互
 
@@ -130,7 +131,7 @@ def del_id(self, _id: int) -> bool:
         return True
 
     @handle_exception(lambda: None)
-    def get_id(self, id_: int) -> Optional[int]:
+    def get_id(self, id_: int) -> int | None:
         """
         获取表id_{fname}中id对应的tag值
 

aiotieba_reviewer/executor.py

@@ -1,5 +1,5 @@
 import asyncio
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 import aiotieba as tb
 from aiotieba import get_logger as LOG
@@ -8,10 +8,10 @@
 from .enums import Ops
 from .punish import Punish
 
-TypePunishExecutor = Callable[[Punish], Awaitable[Optional[Punish]]]
+TypePunishExecutor = Callable[[Punish], Awaitable[Punish | None]]
 
 
-async def default_punish_executor(punish: Punish) -> Optional[Punish]:
+async def default_punish_executor(punish: Punish) -> Punish | None:
     if day := punish.day:
         client = await get_client()
         ret = await client.block(get_fname(), punish.obj.user.portrait, day=day, reason=punish.note)
@@ -52,7 +52,7 @@ async def default_punish_executor(punish: Punish) -> Optional[Punish]:
         return punish
 
 
-async def default_punish_executor_test(punish: Punish) -> Optional[Punish]:
+async def default_punish_executor_test(punish: Punish) -> Punish | None:
     if day := punish.day:
         LOG().info(f"Block. user={punish.obj.user!r} day={day} note={punish.note}")
 

aiotieba_reviewer/imgproc.py

@@ -1,4 +1,3 @@
-from typing import Tuple
 
 import cv2 as cv
 import numpy as np
@@ -105,7 +104,7 @@ async def get_imghash(image: "np.ndarray", *, hamming_dist: int = 0) -> int:
     return 0
 
 
-async def get_imghash_full(image: "np.ndarray", *, hamming_dist: int = 0) -> Tuple[int, str]:
+async def get_imghash_full(image: "np.ndarray", *, hamming_dist: int = 0) -> tuple[int, str]:
     """
     获取图像的完整信息
 

aiotieba_reviewer/perf_stat.py

@@ -2,7 +2,7 @@
 from collections import deque
 
 
-class aperf_stat(object):
+class aperf_stat:
     """
     异步函数性能统计工具
 

aiotieba_reviewer/punish.py

@@ -4,7 +4,7 @@
 from .typing import TypeObj
 
 
-class Punish(object):
+class Punish:
     """
     处罚操作
 

aiotieba_reviewer/reviewer/comment/checker.py

@@ -1,19 +1,19 @@
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 from ... import client
 from ...punish import Punish
 from ...typing import Comment
 from ..user_checker import _user_checker
 
-TypeCommentChecker = Callable[[Comment], Awaitable[Optional[Punish]]]
+TypeCommentChecker = Callable[[Comment], Awaitable[Punish | None]]
 
 
 def __id_checker(func):
     """
     装饰器: 使用历史状态缓存避免重复检查
     """
 
-    async def _(comment: Comment) -> Optional[Punish]:
+    async def _(comment: Comment) -> Punish | None:
         if client._db_sqlite.get_id(comment.pid) is not None:
             return
 

aiotieba_reviewer/reviewer/comment/runner.py

@@ -1,18 +1,18 @@
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 from ... import executor
 from ...punish import Punish
 from ...typing import Comment
 from . import checker
 
-TypeCommentRunner = Callable[[Comment], Awaitable[Optional[Punish]]]
+TypeCommentRunner = Callable[[Comment], Awaitable[Punish | None]]
 
 
 async def __null_runner(_):
     pass
 
 
-async def __default_runner(comment: Comment) -> Optional[Punish]:
+async def __default_runner(comment: Comment) -> Punish | None:
     punish = await checker.checker(comment)
     if punish is not None:
         punish = await executor.punish_executor(punish)

aiotieba_reviewer/reviewer/comments/filter.py

@@ -1,11 +1,11 @@
-from typing import Awaitable, Callable, List, Optional
+from collections.abc import Awaitable, Callable
 
 from ...punish import Punish
 from ...typing import Comment
 
-TypeCommentsFilter = Callable[[List[Comment]], Awaitable[Optional[List[Punish]]]]
+TypeCommentsFilter = Callable[[list[Comment]], Awaitable[list[Punish] | None]]
 
-_filters: List[TypeCommentsFilter] = []
+_filters: list[TypeCommentsFilter] = []
 
 _append_filter_hook = None
 

aiotieba_reviewer/reviewer/comments/producer.py

@@ -1,12 +1,12 @@
-from typing import Awaitable, Callable, List
+from collections.abc import Awaitable, Callable
 
 from ...client import get_client
 from ...typing import Comment, Post
 
-TypeCommentsProducer = Callable[[Post], Awaitable[List[Comment]]]
+TypeCommentsProducer = Callable[[Post], Awaitable[list[Comment]]]
 
 
-async def __default_producer(post: Post) -> List[Comment]:
+async def __default_producer(post: Post) -> list[Comment]:
     client = await get_client()
 
     reply_num = post.reply_num

aiotieba_reviewer/reviewer/comments/runner.py

@@ -1,20 +1,20 @@
 import asyncio
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 from ... import executor
 from ...punish import Punish
 from ...typing import Post
 from ..comment import runner as c_runner
 from . import filter, producer
 
-TypeCommentsRunner = Callable[[Post], Awaitable[Optional[Punish]]]
+TypeCommentsRunner = Callable[[Post], Awaitable[Punish | None]]
 
 
 async def __null_runner(_):
     pass
 
 
-async def __default_runner(post: Post) -> Optional[Punish]:
+async def __default_runner(post: Post) -> Punish | None:
     comments = await producer.producer(post)
     for comment in comments:
         comment.parent = post

aiotieba_reviewer/reviewer/entry.py

@@ -1,7 +1,8 @@
 import asyncio
 import contextlib
 import random
-from typing import Generator, List, NoReturn, Optional
+from collections.abc import Generator
+from typing import NoReturn
 
 from aiotieba import get_logger as LOG
 from aiotieba.enums import PostSortType
@@ -82,7 +83,7 @@ async def run_multi_pn_with_time_threshold(
     _client = await get_client()
 
     @posts.set_producer
-    async def _(thread: Thread) -> List[Post]:
+    async def _(thread: Thread) -> list[Post]:
         post_list = []
 
         last_posts = await _client.get_posts(
@@ -112,7 +113,7 @@ async def _(thread: Thread) -> List[Post]:
         await threads.runner.runner(client._fname, pn)
 
 
-async def test(tid: int, pid: int = 0, is_comment: bool = False) -> Optional[Punish]:
+async def test(tid: int, pid: int = 0, is_comment: bool = False) -> Punish | None:
     client = await get_client()
     if not pid:
         posts = await client.get_posts(tid, rn=0)

aiotieba_reviewer/reviewer/post/checker.py

@@ -1,19 +1,19 @@
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 from ... import client
 from ...punish import Punish
 from ...typing import Post
 from ..user_checker import _user_checker
 
-TypePostChecker = Callable[[Post], Awaitable[Optional[Punish]]]
+TypePostChecker = Callable[[Post], Awaitable[Punish | None]]
 
 
 def __id_checker(func):
     """
     装饰器: 使用历史状态缓存避免重复检查
     """
 
-    async def _(post: Post) -> Optional[Punish]:
+    async def _(post: Post) -> Punish | None:
         prev_reply_num = client._db_sqlite.get_id(post.pid)
         if prev_reply_num is not None:
             if post.reply_num == prev_reply_num:

aiotieba_reviewer/reviewer/post/runner.py

@@ -1,19 +1,19 @@
-from typing import Awaitable, Callable, Optional
+from collections.abc import Awaitable, Callable
 
 from ... import executor
 from ...punish import Punish
 from ...typing import Post
 from .. import comments
 from . import checker
 
-TypePostRunner = Callable[[Post], Awaitable[Optional[Punish]]]
+TypePostRunner = Callable[[Post], Awaitable[Punish | None]]
 
 
 async def __null_runner(_):
     pass
 
 
-async def __default_runner(post: Post) -> Optional[Punish]:
+async def __default_runner(post: Post) -> Punish | None:
     punish = await checker.checker(post)
     if punish is not None:
         punish = await executor.punish_executor(punish)

aiotieba_reviewer/reviewer/posts/filter.py

@@ -1,11 +1,11 @@
-from typing import Awaitable, Callable, List, Optional
+from collections.abc import Awaitable, Callable
 
 from ...punish import Punish
 from ...typing import Post
 
-TypePostsFilter = Callable[[List[Post]], Awaitable[Optional[List[Punish]]]]
+TypePostsFilter = Callable[[list[Post]], Awaitable[list[Punish] | None]]
 
-_filters: List[TypePostsFilter] = []
+_filters: list[TypePostsFilter] = []
 
 _append_filter_hook = None
 

aiotieba_reviewer/reviewer/posts/producer.py

@@ -1,14 +1,14 @@
-from typing import Awaitable, Callable, List
+from collections.abc import Awaitable, Callable
 
 from aiotieba.enums import PostSortType
 
 from ...client import get_client
 from ...typing import Post, Thread
 
-TypePostsProducer = Callable[[Thread], Awaitable[List[Post]]]
+TypePostsProducer = Callable[[Thread], Awaitable[list[Post]]]
 
 
-async def __default_producer(thread: Thread) -> List[Post]:
+async def __default_producer(thread: Thread) -> list[Post]:
     client = await get_client()
 
     last_posts = await client.get_posts(