chore(deps): 更新至最新依赖并修复测试 (#286)

Author: he0119

nonebot_plugin_wordcloud/config.py

@@ -1,6 +1,6 @@
 from datetime import datetime, time
 from pathlib import Path
-from typing import Any, Dict, List, Optional, Set, Union
+from typing import Any, Optional, Union
 from zoneinfo import ZoneInfo
 
 from nonebot import get_driver, get_plugin_config
@@ -16,7 +16,7 @@ class Config(BaseModel):
     wordcloud_width: int = 1920
     wordcloud_height: int = 1200
     wordcloud_background_color: str = "black"
-    wordcloud_colormap: Union[str, List[str]] = "viridis"
+    wordcloud_colormap: Union[str, list[str]] = "viridis"
     wordcloud_font_path: str
     wordcloud_stopwords_path: Optional[Path] = None
     wordcloud_userdict_path: Optional[Path] = None
@@ -26,8 +26,8 @@ class Config(BaseModel):
 
     如果群内不单独设置则使用这个值，默认为晚上 10 点，时区为设定的时区
     """
-    wordcloud_options: Dict[str, Any] = {}
-    wordcloud_exclude_user_ids: Set[str] = set()
+    wordcloud_options: dict[str, Any] = {}
+    wordcloud_exclude_user_ids: set[str] = set()
     """排除的用户 ID 列表（全局，不区分平台）"""
 
     @model_validator(mode="before")

nonebot_plugin_wordcloud/data_source.py

@@ -5,7 +5,7 @@
 from functools import partial
 from io import BytesIO
 from random import choice
-from typing import Dict, List, Optional
+from typing import Optional
 
 import jieba
 import jieba.analyse
@@ -37,7 +37,7 @@ def pre_precess(msg: str) -> str:
     return msg
 
 
-def analyse_message(msg: str) -> Dict[str, float]:
+def analyse_message(msg: str) -> dict[str, float]:
     """分析消息
 
     分词，并统计词频
@@ -65,7 +65,7 @@ def get_mask(key: str):
         return np.array(Image.open(default_mask_path))
 
 
-def _get_wordcloud(messages: List[str], mask_key: str) -> Optional[bytes]:
+def _get_wordcloud(messages: list[str], mask_key: str) -> Optional[bytes]:
     # 过滤掉命令
     command_start = tuple(i for i in global_config.command_start if i)
     message = " ".join(m for m in messages if not m.startswith(command_start))
@@ -98,7 +98,7 @@ def _get_wordcloud(messages: List[str], mask_key: str) -> Optional[bytes]:
         return image_bytes.getvalue()
 
 
-async def get_wordcloud(messages: List[str], mask_key: str) -> Optional[bytes]:
+async def get_wordcloud(messages: list[str], mask_key: str) -> Optional[bytes]:
     loop = asyncio.get_running_loop()
     pfunc = partial(_get_wordcloud, messages, mask_key)
     # 虽然不知道具体是哪里泄漏了，但是通过每次关闭线程池可以避免这个问题

nonebot_plugin_wordcloud/schedule.py

@@ -1,5 +1,5 @@
 from datetime import time
-from typing import Dict, Optional, Tuple
+from typing import Optional
 
 import nonebot_plugin_saa as saa
 from apscheduler.job import Job
@@ -29,7 +29,7 @@ class Scheduler:
     def __init__(self):
         # 默认定时任务的 key 为 default
         # 其他则为 ISO 8601 格式的时间字符串
-        self.schedules: Dict[str, Job] = {}
+        self.schedules: dict[str, Job] = {}
 
         # 转换到 APScheduler 的时区
         scheduler_time = get_time_with_scheduler_timezone(
@@ -160,7 +160,7 @@ async def remove_schedule(self, target: saa.PlatformTarget):
     @staticmethod
     def select_target_statement(
         target: saa.PlatformTarget, session: AsyncSession
-    ) -> Select[Tuple[Schedule]]:
+    ) -> Select[tuple[Schedule]]:
         """获取查询目标的语句
 
         MySQL 需要手动将 JSON 类型的字段转换为 JSON 类型

poetry.lock

@@ -31,40 +31,40 @@ nonebug-saa = "^0.4.1"
 nonebot-adapter-onebot = "^2.2.4"
 nonebot-plugin-orm = { extras = ["default"], version = ">=0.7.0" }
 nonebot-plugin-datastore = "^1.2.0"
+asyncpg = "*"
+aiomysql = "*"
 pytest-cov = ">=4,<6"
 pytest-xdist = "^3.1.0"
 pytest-mock = "^3.7.0"
 pytest-asyncio = ">=0.21,<0.24"
-asyncpg = "*"
-aiomysql = "*"
 respx = ">=0.20.2,<0.22.0"
 httpx = ">=0.24.1,<0.28.0"
 gevent = ">=23.7,<25.0"
 
+[build-system]
+requires = ["poetry-core>=1.0.0"]
+build-backend = "poetry.core.masonry.api"
+
 [tool.nonebot]
 plugins = ["nonebot_plugin_wordcloud"]
 adapters = [
   { name = "OneBot V12", module_name = "nonebot.adapters.onebot.v12", project_link = "nonebot-adapter-onebot", desc = "OneBot V12 协议" },
   { name = "OneBot V11", module_name = "nonebot.adapters.onebot.v11", project_link = "nonebot-adapter-onebot", desc = "OneBot V11 协议" },
 ]
 
-[tool.black]
-line-length = 88
-
-[tool.isort]
-profile = "black"
-line_length = 88
-skip_gitignore = true
-
 [tool.pytest.ini_options]
 asyncio_mode = "auto"
 
 [tool.pyright]
 pythonVersion = "3.9"
 pythonPlatform = "All"
-typeCheckingMode = "basic"
+typeCheckingMode = "standard"
 defineConstant = { PYDANTIC_V2 = true }
 
+[tool.ruff]
+line-length = 88
+target-version = "py39"
+
 [tool.ruff.lint]
 select = [
   "W",   # pycodestyle warnings
@@ -89,7 +89,3 @@ concurrency = ["thread", "gevent"]
 
 [tool.coverage.report]
 omit = ["*/compat.py", "*/migrations/*"]
-
-[build-system]
-requires = ["poetry-core>=1.0.0"]
-build-backend = "poetry.core.masonry.api"

pyproject.toml

@@ -1,4 +1,3 @@
-from typing import Dict
 from zoneinfo import ZoneInfo
 
 import pytest
@@ -17,7 +16,7 @@
         ),
     ],
 )
-async def test_default_schedule_time(app: App, default_config: Dict[str, str]):
+async def test_default_schedule_time(app: App, default_config: dict[str, str]):
     """测试设置默认定时发送时间"""
     from nonebot_plugin_wordcloud.config import Config
 
@@ -47,7 +46,7 @@ async def test_default_schedule_time(app: App, default_config: Dict[str, str]):
     ],
 )
 async def test_default_schedule_time_with_timezone(
-    app: App, default_config: Dict[str, str]
+    app: App, default_config: dict[str, str]
 ):
     """测试设置默认定时发送时间，同时设置时区"""
     from nonebot_plugin_wordcloud.config import Config

tests/test_config.py

@@ -67,7 +67,8 @@ async def test_masked_by_command(app: App, mocker: MockerFixture):
         test_image = f.read()
 
     async with app.test_matcher(wordcloud_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(message=Message("/今日词云"))
 
         ctx.receive_event(bot, event)
@@ -77,7 +78,7 @@ async def test_masked_by_command(app: App, mocker: MockerFixture):
             bot,
             event=event,
         )
-        ctx.should_finished()
+        ctx.should_finished(wordcloud_cmd)
 
     mocked_random.assert_called()
     mocked_get_messages_plain_text.assert_called_once()
@@ -203,7 +204,8 @@ async def test_set_mask_get_args(app: App, respx_mock: respx.MockRouter):
     )
 
     async with app.test_matcher(set_mask_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         message = Message("/设置词云形状")
         event = fake_group_message_event_v11(message=message, sender={"role": "owner"})
 
@@ -291,7 +293,8 @@ async def test_remove_mask(app: App):
     assert mask_group_path.exists()
 
     async with app.test_matcher(remove_mask_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         message = Message("/删除词云形状")
         event = fake_group_message_event_v11(message=message, sender={"role": "owner"})
 
@@ -312,7 +315,8 @@ async def test_set_mask_private(app: App, mocker: MockerFixture):
     mocker.patch.object(config, "superusers", {"10"})
 
     async with app.test_matcher(set_mask_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_private_message_event_v11(message=Message("/设置词云形状"))
 
         ctx.receive_event(bot, event)
@@ -333,7 +337,8 @@ async def test_remove_mask_private(app: App, mocker: MockerFixture):
     mocker.patch.object(config, "superusers", {"10"})
 
     async with app.test_matcher(remove_mask_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_private_message_event_v11(message=Message("/删除词云形状"))
 
         ctx.receive_event(bot, event)

tests/test_masked.py

@@ -1,8 +1,9 @@
 from datetime import time
 from io import BytesIO
 
-from nonebot import get_driver
-from nonebot.adapters.onebot.v11 import Bot, Message
+from nonebot import get_adapter, get_driver
+from nonebot.adapters.onebot.v11 import Adapter, Bot, Message
+from nonebot.adapters.onebot.v12 import Adapter as AdapterV12
 from nonebot.adapters.onebot.v12 import Bot as BotV12
 from nonebot.adapters.onebot.v12 import Message as MessageV12
 from nonebug import App
@@ -22,7 +23,8 @@ async def test_enable_schedule(app: App):
     from nonebot_plugin_wordcloud import schedule_cmd, schedule_service
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/开启词云每日定时发送"), sender={"role": "admin"}
         )
@@ -36,7 +38,8 @@ async def test_enable_schedule(app: App):
     assert len(schedule_service.schedules) == 1
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/开启词云每日定时发送 10:00"), sender={"role": "admin"}
         )
@@ -51,7 +54,8 @@ async def test_enable_schedule(app: App):
     assert len(schedule_service.schedules) == 2
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/开启词云每日定时发送 10:"), sender={"role": "admin"}
         )
@@ -63,7 +67,10 @@ async def test_enable_schedule(app: App):
 
     # OneBot V12
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=BotV12, platform="qq", impl="test")
+        adapter = get_adapter(AdapterV12)
+        bot = ctx.create_bot(
+            base=BotV12, adapter=adapter, auto_connect=False, platform="qq", impl="test"
+        )
         event = fake_group_message_event_v12(
             message=MessageV12("/开启词云每日定时发送")
         )
@@ -78,7 +85,10 @@ async def test_enable_schedule(app: App):
     assert len(schedule_service.schedules) == 2
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=BotV12, platform="qq", impl="test")
+        adapter = get_adapter(AdapterV12)
+        bot = ctx.create_bot(
+            base=BotV12, adapter=adapter, auto_connect=False, platform="qq", impl="test"
+        )
         event = fake_channel_message_event_v12(
             message=MessageV12("/开启词云每日定时发送 09:00")
         )
@@ -102,7 +112,8 @@ async def test_enable_schedule_private(app: App, mocker: MockerFixture):
     mocker.patch.object(config, "superusers", {"10"})
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_private_message_event_v11(message=Message("/开启词云每日定时发送"))
         ctx.receive_event(bot, event)
         ctx.should_pass_permission(schedule_cmd)
@@ -115,7 +126,8 @@ async def test_enable_schedule_without_permission(app: App, mocker: MockerFixtur
     from nonebot_plugin_wordcloud import schedule_cmd
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(message=Message("/开启词云每日定时发送"))
         ctx.receive_event(bot, event)
         ctx.should_not_pass_permission(schedule_cmd)
@@ -142,7 +154,8 @@ async def test_disable_schedule(app: App):
     assert len(schedule_service.schedules) == 2
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/关闭词云每日定时发送"), sender={"role": "admin"}
         )
@@ -165,7 +178,8 @@ async def test_schedule_status(app: App):
     from nonebot_plugin_wordcloud import schedule_cmd, schedule_service
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/词云每日定时发送状态"), sender={"role": "admin"}
         )
@@ -177,7 +191,8 @@ async def test_schedule_status(app: App):
     await schedule_service.add_schedule(TargetQQGroup(group_id=10000))
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot)
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(base=Bot, adapter=adapter, auto_connect=False)
         event = fake_group_message_event_v11(
             message=Message("/词云每日定时发送状态"), sender={"role": "admin"}
         )
@@ -191,7 +206,10 @@ async def test_schedule_status(app: App):
     await schedule_service.add_schedule(TargetQQGroup(group_id=10000), time=time(23, 0))
 
     async with app.test_matcher(schedule_cmd) as ctx:
-        bot = ctx.create_bot(base=Bot, self_id="test2")
+        adapter = get_adapter(Adapter)
+        bot = ctx.create_bot(
+            base=Bot, adapter=adapter, auto_connect=False, self_id="test2"
+        )
         event = fake_group_message_event_v11(
             message=Message("/词云每日定时发送状态"), sender={"role": "admin"}
         )
@@ -239,7 +257,7 @@ async def test_run_task_group(app: App, mocker: MockerFixture):
     )
 
     async with app.test_api() as ctx:
-        bot = ctx.create_bot(base=BotV12, impl="test", platform="qq")
+        bot = ctx.create_bot(base=BotV12, platform="qq", impl="test")
         should_send_saa(ctx, MessageFactory(Image(image)), bot, target=target)
         await schedule_service.run_task()