移动 poster 模块到 common

konabot/common/subscribe/__init__.py

@@ -0,0 +1,11 @@
+"""
+Subscribe 模块，用于向一些订阅的频道广播消息
+"""
+
+from .service import broadcast as broadcast
+from .service import dep_poster_service as dep_poster_service
+from .service import DepPosterService as DepPosterService
+from .service import PosterService as PosterService
+from .subscribe_info import PosterInfo as PosterInfo
+from .subscribe_info import POSTER_INFO_DATA as POSTER_INFO_DATA
+from .subscribe_info import register_poster_info as register_poster_info

konabot/common/subscribe/repo_local_data.py

@@ -6,7 +6,8 @@ from pydantic import BaseModel, ValidationError
 from konabot.common.longtask import LongTaskTarget
 from konabot.common.pager import PagerQuery, PagerResult
 from konabot.common.path import DATA_PATH
-from konabot.plugins.poster.repository import IPosterRepo
+
+from .repository import IPosterRepo
 
 
 class ChannelData(BaseModel):
@@ -18,9 +19,9 @@ class PosterData(BaseModel):
 
 
 def is_the_same_target(target1: LongTaskTarget, target2: LongTaskTarget) -> bool:
-    if (target1.is_private_chat and not target2.is_private_chat):
+    if target1.is_private_chat and not target2.is_private_chat:
         return False
-    if (target2.is_private_chat and not target1.is_private_chat):
+    if target2.is_private_chat and not target1.is_private_chat:
         return False
     if target1.platform != target2.platform:
         return False
@@ -58,7 +59,9 @@ class LocalPosterRepo(IPosterRepo):
         len1 = len(self.data.channels[channel].targets)
         return len0 != len1
 
-    async def get_subscribed_channels(self, target: LongTaskTarget, pager: PagerQuery) -> PagerResult[str]:
+    async def get_subscribed_channels(
+        self, target: LongTaskTarget, pager: PagerQuery
+    ) -> PagerResult[str]:
         channels: list[str] = []
         for channel_id, channel in self.data.channels.items():
             for t in channel.targets:
@@ -95,7 +98,9 @@ async def local_poster_data():
             data = PosterData()
         else:
             try:
-                data = PosterData.model_validate_json(LOCAL_POSTER_DATA_PATH.read_text())
+                data = PosterData.model_validate_json(
+                    LOCAL_POSTER_DATA_PATH.read_text()
+                )
             except ValidationError:
                 data = PosterData()
         yield data
@@ -109,4 +114,3 @@ async def local_poster():
 
 
 DepLocalPosterRepo = Annotated[LocalPosterRepo, Depends(local_poster)]
-

konabot/common/subscribe/service.py

@@ -4,9 +4,10 @@ from nonebot.params import Depends
 from nonebot_plugin_alconna import UniMessage
 from konabot.common.longtask import LongTaskTarget
 from konabot.common.pager import PagerQuery, PagerResult
-from konabot.plugins.poster.poster_info import POSTER_INFO_DATA
-from konabot.plugins.poster.repo_local_data import local_poster
-from konabot.plugins.poster.repository import IPosterRepo
+
+from .subscribe_info import POSTER_INFO_DATA
+from .repo_local_data import local_poster
+from .repository import IPosterRepo
 
 
 class PosterService:
@@ -27,7 +28,9 @@ class PosterService:
         channel = self.parse_channel_id(channel)
         return await self.repo.remove_channel_target(channel, target)
 
-    async def broadcast(self, channel: str, message: UniMessage[Any] | str) -> list[LongTaskTarget]:
+    async def broadcast(
+        self, channel: str, message: UniMessage[Any] | str
+    ) -> list[LongTaskTarget]:
         channel = self.parse_channel_id(channel)
         targets = await self.repo.get_channel_targets(channel)
         for target in targets:
@@ -35,7 +38,9 @@ class PosterService:
             await target.send_message(message, at=False)
         return targets
 
-    async def get_channels(self, target: LongTaskTarget, pager: PagerQuery) -> PagerResult[str]:
+    async def get_channels(
+        self, target: LongTaskTarget, pager: PagerQuery
+    ) -> PagerResult[str]:
         return await self.repo.get_subscribed_channels(target, pager)
 
     async def fix_data(self):
@@ -56,4 +61,3 @@ async def broadcast(channel: str, message: UniMessage[Any] | str):
 
 
 DepPosterService = Annotated[PosterService, Depends(dep_poster_service)]
-

konabot/common/subscribe/subscribe_info.py

@@ -4,7 +4,7 @@ from dataclasses import dataclass, field
 @dataclass
 class PosterInfo:
     aliases: set[str] = field(default_factory=set)
-    description: str = field(default='')
+    description: str = field(default="")
 
 
 POSTER_INFO_DATA: dict[str, PosterInfo] = {}
@@ -12,4 +12,3 @@ POSTER_INFO_DATA: dict[str, PosterInfo] = {}
 
 def register_poster_info(channel: str, info: PosterInfo):
     POSTER_INFO_DATA[channel] = info
-

konabot/plugins/kona_ph/__init__.py

@@ -6,29 +6,34 @@ from loguru import logger
 from nonebot import on_message
 import nonebot
 from nonebot.rule import to_me
-from nonebot_plugin_alconna import (Alconna, Args, UniMessage, UniMsg,
-                                    on_alconna)
+from nonebot_plugin_alconna import Alconna, Args, UniMessage, UniMsg, on_alconna
 from nonebot_plugin_apscheduler import scheduler
 
 from konabot.common import username
 from konabot.common.longtask import DepLongTaskTarget
 from konabot.common.pager import PagerQuery
-from konabot.plugins.kona_ph.core.message import (get_daily_report,
-                                                  get_daily_report_v2,
-                                                  get_puzzle_description,
-                                                  get_submission_message)
+from konabot.plugins.kona_ph.core.message import (
+    get_daily_report,
+    get_daily_report_v2,
+    get_puzzle_description,
+    get_submission_message,
+)
 from konabot.plugins.kona_ph.core.storage import get_today_date
-from konabot.plugins.kona_ph.manager import (PUZZLE_PAGE_SIZE,
-                                             create_admin_commands,
-                                             puzzle_manager)
-from konabot.plugins.poster.poster_info import PosterInfo, register_poster_info
-from konabot.plugins.poster.service import broadcast
+from konabot.plugins.kona_ph.manager import (
+    PUZZLE_PAGE_SIZE,
+    create_admin_commands,
+    puzzle_manager,
+)
+from konabot.common.subscribe import PosterInfo, register_poster_info, broadcast
 
 create_admin_commands()
-register_poster_info("每日谜题", info=PosterInfo(
-    aliases={"konaph", "kona_ph", "KonaPH", "此方谜题", "KONAPH"},
-    description="此方 BOT 每日谜题推送",
-))
+register_poster_info(
+    "每日谜题",
+    info=PosterInfo(
+        aliases={"konaph", "kona_ph", "KonaPH", "此方谜题", "KONAPH"},
+        description="此方 BOT 每日谜题推送",
+    ),
+)
 
 
 cmd_submit = on_message(rule=to_me())
@@ -44,16 +49,22 @@ async def _(msg: UniMsg, target: DepLongTaskTarget):
             if isinstance(result, str):
                 await target.send_message(result)
             else:
-                await target.send_message(get_submission_message(
-                    daily_puzzle_info=result.info,
-                    submission=result.submission,
-                    puzzle=result.puzzle,
-                ))
+                await target.send_message(
+                    get_submission_message(
+                        daily_puzzle_info=result.info,
+                        submission=result.submission,
+                        puzzle=result.puzzle,
+                    )
+                )
 
 
-cmd_query = on_alconna(Alconna(
-    r"re:(?:((?:(?:所以|话)说?)?今天的题目是什么[啊呀哇呢]?(?:\?？)?)|今日谜?题目?)"
-), rule=to_me())
+cmd_query = on_alconna(
+    Alconna(
+        r"re:(?:((?:(?:所以|话)说?)?今天的题目是什么[啊呀哇呢]?(?:\?？)?)|今日谜?题目?)"
+    ),
+    rule=to_me(),
+)
+
 
 @cmd_query.handle()
 async def _(target: DepLongTaskTarget):
@@ -64,9 +75,8 @@ async def _(target: DepLongTaskTarget):
         await target.send_message(get_puzzle_description(p))
 
 
-cmd_query_submission = on_alconna(Alconna(
-    "今日答题情况"
-), rule=to_me())
+cmd_query_submission = on_alconna(Alconna("今日答题情况"), rule=to_me())
+
 
 @cmd_query_submission.handle()
 async def _(target: DepLongTaskTarget):
@@ -77,11 +87,15 @@ async def _(target: DepLongTaskTarget):
         await target.send_message(get_daily_report_v2(manager, gid))
 
 
-cmd_history = on_alconna(Alconna(
-    "re:历史(题目|谜题)",
-    Args["page?", int],
-    Args["index_id?", str],
-), rule=to_me())
+cmd_history = on_alconna(
+    Alconna(
+        "re:历史(题目|谜题)",
+        Args["page?", int],
+        Args["index_id?", str],
+    ),
+    rule=to_me(),
+)
+
 
 @cmd_history.handle()
 async def _(target: DepLongTaskTarget, index_id: str = "", page: int = 1):
@@ -105,10 +119,10 @@ async def _(target: DepLongTaskTarget, index_id: str = "", page: int = 1):
         puzzles = sorted(puzzles, key=lambda u: u[1], reverse=True)
         count_pages = ceil(len(puzzles) / PUZZLE_PAGE_SIZE)
         if page <= 0 or page > count_pages:
-            return await target.send_message(UniMessage.text(
-                f"页数只有 1 ~ {count_pages} 啦！"
-            ))
-        puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE: page * PUZZLE_PAGE_SIZE]
+            return await target.send_message(
+                UniMessage.text(f"页数只有 1 ~ {count_pages} 啦！")
+            )
+        puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE : page * PUZZLE_PAGE_SIZE]
         for p, d in puzzles:
             info = manager.daily_puzzle[manager.daily_puzzle_of_date[d]]
             msg = msg.text(
@@ -120,22 +134,26 @@ async def _(target: DepLongTaskTarget, index_id: str = "", page: int = 1):
         await target.send_message(msg)
 
 
-cmd_leadboard = on_alconna(Alconna(
-    "re:此方(解谜|谜题)排行榜",
-    Args["page?", int],
-))
+cmd_leadboard = on_alconna(
+    Alconna(
+        "re:此方(解谜|谜题)排行榜",
+        Args["page?", int],
+    )
+)
+
 
 @cmd_leadboard.handle()
 async def _(target: DepLongTaskTarget, page: int = 1):
     async with puzzle_manager() as manager:
         result = manager.get_leadboard(PagerQuery(page, 10))
-    await target.send_message(result.to_unimessage(
-        title="此方解谜排行榜",
-        formatter=lambda data: (
-            f"✨ {data[1]} 已完成 | "
-            f"{username.get_username(data[0])}"
+    await target.send_message(
+        result.to_unimessage(
+            title="此方解谜排行榜",
+            formatter=lambda data: (
+                f"✨ {data[1]} 已完成 | {username.get_username(data[0])}"
+            ),
         )
-    ))
+    )
 
 
 @scheduler.scheduled_job("cron", hour="8")
@@ -155,4 +173,3 @@ async def _():
 
 
 driver = nonebot.get_driver()
-

konabot/plugins/kona_ph/manager.py

@@ -1,7 +1,6 @@
 import datetime
 from math import ceil
 
-from nonebot import get_plugin_config
 from nonebot.adapters import Event
 from nonebot_plugin_alconna import (
     Alconna,
@@ -14,7 +13,6 @@ from nonebot_plugin_alconna import (
     UniMessage,
     on_alconna,
 )
-from pydantic import BaseModel
 
 from konabot.common.longtask import DepLongTaskTarget
 from konabot.common.nb.exc import BotExceptionMessage
@@ -35,20 +33,11 @@ from konabot.plugins.kona_ph.core.storage import (
     get_today_date,
     puzzle_manager,
 )
-from konabot.plugins.poster.service import broadcast
+from konabot.common.subscribe import broadcast
 
 PUZZLE_PAGE_SIZE = 10
 
 
-class PuzzleConfig(BaseModel):
-    plugin_puzzle_manager: list[str] = []
-    plugin_puzzle_admin: list[str] = []
-    plugin_puzzle_playgroup: list[str] = []
-
-
-config = get_plugin_config(PuzzleConfig)
-
-
 async def check_puzzle(
     manager: PuzzleManager,
     perm: DepPermManager,

konabot/plugins/nya_echo/__init__.py

@@ -6,6 +6,7 @@ from konabot.common.nb.match_keyword import match_keyword
 
 evt_nya = on_message(rule=match_keyword("喵"))
 
+
 @evt_nya.handle()
 async def _():
     await evt_nya.send(await UniMessage().text("喵").export())
@@ -25,8 +26,9 @@ NYA_SYMBOL_MAPPING = {
     "~": "~",
     "～": "～",
     " ": " ",
-    "\n": "\n",
 }
+NYA_SYMBOL_KEEP = "—¹₁²₂³₃⁴₄⁵₅⁶₆⁷₇⁸₈⁹₉⁰₀\n"
+NYA_SYMBOL_MAPPING.update((k, k) for k in NYA_SYMBOL_KEEP)
 
 
 async def has_nya(msg: UniMsg) -> bool:
@@ -49,10 +51,10 @@ async def has_nya(msg: UniMsg) -> bool:
 
 evt_nya_v2 = on_message(rule=has_nya)
 
+
 @evt_nya_v2.handle()
 async def _(msg: UniMsg, evt: Event):
     text = msg.extract_plain_text()
-    await UniMessage.text(''.join(
-        (NYA_SYMBOL_MAPPING.get(c, '') for c in text)
-    )).send(evt)
-
+    await UniMessage.text("".join((NYA_SYMBOL_MAPPING.get(c, "") for c in text))).send(
+        evt
+    )

konabot/plugins/poster/__init__.py

@@ -3,14 +3,15 @@ from nonebot_plugin_alconna import Alconna, Args, on_alconna
 
 from konabot.common.longtask import DepLongTaskTarget
 from konabot.common.pager import PagerQuery
-from konabot.plugins.poster.poster_info import POSTER_INFO_DATA
-from konabot.plugins.poster.service import dep_poster_service
+from konabot.common.subscribe import POSTER_INFO_DATA, dep_poster_service
 
 
-cmd_subscribe = on_alconna(Alconna(
-    "订阅",
-    Args["channel", str],
-))
+cmd_subscribe = on_alconna(
+    Alconna(
+        "订阅",
+        Args["channel", str],
+    )
+)
 
 
 @cmd_subscribe.handle()
@@ -23,10 +24,12 @@ async def _(target: DepLongTaskTarget, channel: str):
         await target.send_message(f"已经订阅过「{channel}」了")
 
 
-cmd_list = on_alconna(Alconna(
-    "re:(?:查询|我的|获取)订阅(列表)?",
-    Args["page?", int],
-))
+cmd_list = on_alconna(
+    Alconna(
+        "re:(?:查询|我的|获取)订阅(列表)?",
+        Args["page?", int],
+    )
+)
 
 
 def better_channel_message(channel_id: str) -> str:
@@ -39,17 +42,24 @@ def better_channel_message(channel_id: str) -> str:
 @cmd_list.handle()
 async def _(target: DepLongTaskTarget, page: int = 1):
     async with dep_poster_service() as service:
-        result = await service.get_channels(target, PagerQuery(
-            page_index=page,
-            page_size=10,
-        ))
-        await target.send_message(result.to_unimessage(title="订阅列表", formatter=better_channel_message))
+        result = await service.get_channels(
+            target,
+            PagerQuery(
+                page_index=page,
+                page_size=10,
+            ),
+        )
+        await target.send_message(
+            result.to_unimessage(title="订阅列表", formatter=better_channel_message)
+        )
 
 
-cmd_list_available = on_alconna(Alconna(
-    "re:(查询)?可用订阅(列表)?",
-    Args["page?", int],
-))
+cmd_list_available = on_alconna(
+    Alconna(
+        "re:(查询)?可用订阅(列表)?",
+        Args["page?", int],
+    )
+)
 
 
 @cmd_list_available.handle()
@@ -58,13 +68,17 @@ async def _(target: DepLongTaskTarget, page: int = 1):
         page_index=page,
         page_size=10,
     ).apply(sorted(POSTER_INFO_DATA.keys()))
-    await target.send_message(result.to_unimessage(title="可用订阅列表", formatter=better_channel_message))
+    await target.send_message(
+        result.to_unimessage(title="可用订阅列表", formatter=better_channel_message)
+    )
 
 
-cmd_unsubscribe = on_alconna(Alconna(
-    "取消订阅",
-    Args["channel", str],
-))
+cmd_unsubscribe = on_alconna(
+    Alconna(
+        "取消订阅",
+        Args["channel", str],
+    )
+)
 
 
 @cmd_unsubscribe.handle()
@@ -79,6 +93,7 @@ async def _(target: DepLongTaskTarget, channel: str):
 
 driver = nonebot.get_driver()
 
+
 @driver.on_startup
 async def _():
     async with dep_poster_service() as service:

konabot/plugins/solar_terms/__init__.py

@@ -4,8 +4,7 @@ from nonebot.internal.adapter.event import Event
 from nonebot_plugin_alconna import UniMessage
 from nonebot_plugin_apscheduler import scheduler
 
-from konabot.plugins.poster.poster_info import PosterInfo, register_poster_info
-from konabot.plugins.poster.service import broadcast
+from konabot.common.subscribe import PosterInfo, register_poster_info, broadcast
 
 register_poster_info(
     "二十四节气",
@@ -98,4 +97,3 @@ async def _(event: Event):
 
     msg = UniMessage.text(f"现在的节气是{date.term}")
     await msg.send(event)
-

konabot/plugins/startup_notify.py

@@ -1,20 +1,23 @@
 import asyncio
 from nonebot import get_driver
 from nonebot_plugin_alconna import UniMessage
-from konabot.plugins.poster.poster_info import register_poster_info, PosterInfo
-from konabot.plugins.poster.service import broadcast
+from konabot.common.subscribe import register_poster_info, PosterInfo, broadcast
 
 
 CHANNEL_STARTUP = "启动通知"
 
 
-register_poster_info(CHANNEL_STARTUP, PosterInfo(
-    aliases=set(),
-    description="当 Bot 重启时告知",
-))
+register_poster_info(
+    CHANNEL_STARTUP,
+    PosterInfo(
+        aliases=set(),
+        description="当 Bot 重启时告知",
+    ),
+)
 
 driver = get_driver()
 
+
 @driver.on_startup
 async def _():
     # 要尽量保证接受讯息的服务存在
@@ -30,4 +33,3 @@ async def _():
         await broadcast(CHANNEL_STARTUP, UniMessage.text("此方 BOT 重启好了"))
 
     asyncio.create_task(task())
-