Merge branch 'master' of ssh://gitea.service.jazzwhom.top:2221/mttu-developers/konabot

Reviewed-on: #27

konabot/common/data_man.py

@@ -0,0 +1,36 @@
+import asyncio
+from contextlib import asynccontextmanager
+from pathlib import Path
+from typing import Generic, TypeVar
+
+from pydantic import BaseModel, ValidationError
+
+T = TypeVar("T", bound=BaseModel)
+
+
+class DataManager(Generic[T]):
+    def __init__(self, cls: type[T], fp: Path) -> None:
+        self.cls = cls
+        self.fp = fp
+        self._aio_lock = asyncio.Lock()
+        self._data: T | None = None
+
+    def load(self) -> T:
+        if not self.fp.exists():
+            return self.cls()
+        try:
+            return self.cls.model_validate_json(self.fp.read_text())
+        except ValidationError:
+            return self.cls()
+
+    def save(self, data: T):
+        self.fp.write_text(data.model_dump_json())
+
+    @asynccontextmanager
+    async def get_data(self):
+        await self._aio_lock.acquire()
+        self._data = self.load()
+        yield self._data
+        self.save(self._data)
+        self._data = None
+        self._aio_lock.release()

konabot/common/longtask.py

@@ -0,0 +1,300 @@
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+import datetime
+import json
+from typing import Annotated, Any, Callable, Coroutine, cast
+import asyncio as asynkio
+import uuid
+
+from loguru import logger
+import nonebot
+from nonebot.params import Depends
+from nonebot.adapters import Event as BaseEvent
+from nonebot.adapters import Bot as BaseBot
+from nonebot.adapters.onebot.v11 import Bot as OBBot
+from nonebot.adapters.onebot.v11 import GroupMessageEvent as OBGroupMessageEvent
+from nonebot.adapters.onebot.v11 import PrivateMessageEvent as OBPrivateMessageEvent
+from nonebot.adapters.console import Bot as ConsoleBot
+from nonebot.adapters.console import MessageEvent as ConsoleMessageEvent
+from nonebot.adapters.discord import MessageEvent as DCMessageEvent
+from nonebot.adapters.discord import Bot as DCBot
+from nonebot_plugin_alconna import UniMessage
+from pydantic import BaseModel, ValidationError
+
+from .path import DATA_PATH
+
+LONGTASK_DATA_DIR = DATA_PATH / "longtasks.json"
+
+
+class LongTaskTarget(BaseModel):
+    """
+    用于定义长期任务的目标沟通对象，一般通过 DepLongTaskTarget 依赖注入获取：
+
+    ```python
+    @cmd.handle()
+    async def _(target: DepLongTaskTarget):
+        ...
+    ```
+    """
+
+    platform: str
+    "沟通对象所在的平台"
+
+    self_id: str
+    "进行沟通的对象自己的 ID"
+
+    channel_id: str
+    "沟通对象所在的群或者 Discord Channel。若为空则代表是私聊"
+
+    target_id: str
+    "沟通对象的 ID"
+
+    async def send_message(self, msg: UniMessage, at: bool = True) -> bool:
+        try:
+            bot = nonebot.get_bot(self.self_id)
+        except KeyError:
+            logger.warning(f"试图访问了不存在的 Bot。ID={self.self_id}")
+            return False
+
+        if self.platform == "qq":
+            if not isinstance(bot, OBBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            if self.channel_id == "":
+                # 私聊模式
+                await bot.send_private_msg(
+                    user_id=int(self.target_id),
+                    message=cast(Any, await msg.export(bot)),
+                    auto_escape=False,
+                )
+                return True
+            else:
+                if at:
+                    msg = UniMessage().at(self.target_id).text(" ") + msg
+                await bot.send_group_msg(
+                    group_id=int(self.channel_id),
+                    message=cast(Any, await msg.export(bot)),
+                    auto_escape=False,
+                )
+                return True
+        if self.platform == "console":
+            if not isinstance(bot, ConsoleBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            await bot.send_message(self.channel_id, cast(Any, await msg.export()))
+            return True
+        if self.platform == "discord":
+            if not isinstance(bot, DCBot):
+                logger.warning(
+                    f"编号对应的平台并非期望的平台 ID={self.self_id} PLATFORM={
+                        self.platform
+                    } BOT_CLASS={bot.__class__.__name__}"
+                )
+                return False
+            await bot.send_to(
+                channel_id=int(self.channel_id),
+                message=cast(
+                    Any, await (UniMessage().at(self.target_id) + msg).export()
+                ),
+                tts=False,
+            )
+            return True
+        logger.warning(f"没有一个平台是期望的平台 PLATFORM={self.platform}")
+        return False
+
+
+class LongTask(BaseModel):
+    uuid: str
+    data_json: str
+    target: LongTaskTarget
+    callback: str
+    deadline: datetime.datetime
+    canceled: bool = False
+
+    _aio_task: asynkio.Task | None = None
+
+    async def run(self):
+        now = datetime.datetime.now()
+        if self.deadline < now and not self.canceled:
+            await self._run_task()
+            return
+        await asynkio.sleep((self.deadline - now).total_seconds())
+        if self.canceled:
+            return
+        await self._run_task()
+
+    async def _run_task(self):
+        hdl = registered_long_task_handler.get(self.callback, None)
+        if hdl is None:
+            logger.warning(
+                f"Callback {self.callback} 未曾被注册，但是被期待调用，已忽略"
+            )
+            async with longtask_data() as datafile:
+                datafile.to_handle[self.callback] = [
+                    t
+                    for t in datafile.to_handle.get(self.callback, [])
+                    if t.uuid != self.uuid
+                ]
+                datafile.unhandled.setdefault(self.callback, []).append(self)
+
+            return
+        success = False
+        try:
+            await hdl(self)
+            success = True
+        except Exception as e:
+            logger.exception(e)
+        async with longtask_data() as datafile:
+            datafile.to_handle[self.callback] = [
+                t for t in datafile.to_handle[self.callback] if t.uuid != self.uuid
+            ]
+            if not success:
+                datafile.unhandled.setdefault(self.callback, []).append(self)
+                logger.info(
+                    f"LongTask 执行失败 UUID={self.uuid} callback={self.callback}"
+                )
+            else:
+                logger.info(
+                    f"LongTask 工作完成 UUID={self.uuid} callback={self.callback}"
+                )
+
+    def clean(self):
+        self._aio_task = None
+
+    @property
+    def data(self):
+        return json.loads(self.data_json)
+
+    async def start(self):
+        self._aio_task = asynkio.Task(self.run())
+        self._aio_task.add_done_callback(lambda _: self.clean())
+
+
+class LongTaskModuleData(BaseModel):
+    to_handle: dict[str, list[LongTask]]
+    unhandled: dict[str, list[LongTask]]
+
+
+async def get_long_task_target(event: BaseEvent, bot: BaseBot) -> LongTaskTarget | None:
+    if isinstance(event, OBGroupMessageEvent):
+        return LongTaskTarget(
+            platform="qq",
+            self_id=str(event.self_id),
+            channel_id=str(event.group_id),
+            target_id=str(event.user_id),
+        )
+    if isinstance(event, OBPrivateMessageEvent):
+        return LongTaskTarget(
+            platform="qq",
+            self_id=str(event.self_id),
+            channel_id="",
+            target_id=str(event.user_id),
+        )
+    if isinstance(event, ConsoleMessageEvent):
+        return LongTaskTarget(
+            platform="console",
+            self_id=str(event.self_id),
+            channel_id=str(event.channel.id),
+            target_id=str(event.user.id),
+        )
+    if isinstance(event, DCMessageEvent):
+        self_id = ""
+        if isinstance(bot, DCBot):
+            self_id = str(bot.self_id)
+        return LongTaskTarget(
+            platform="discord",
+            self_id=self_id,
+            channel_id=str(event.channel_id),
+            target_id=str(event.user_id),
+        )
+
+
+_TaskHandler = Callable[[LongTask], Coroutine[Any, Any, Any]]
+
+
+registered_long_task_handler: dict[str, _TaskHandler] = {}
+longtask_lock = asynkio.Lock()
+
+
+def handle_long_task(callback_id: str):
+    def _decorator(func: _TaskHandler):
+        assert callback_id not in registered_long_task_handler, (
+            "有长任务的 ID 出现冲突，请换个名字！"
+        )
+        registered_long_task_handler[callback_id] = func
+        return func
+
+    return _decorator
+
+
+def _load_longtask_data() -> LongTaskModuleData:
+    try:
+        txt = LONGTASK_DATA_DIR.read_text()
+        return LongTaskModuleData.model_validate_json(txt)
+    except (FileNotFoundError, ValidationError) as e:
+        logger.info(f"取得 LongTask 数据时出现问题：{e}")
+        return LongTaskModuleData(
+            to_handle={},
+            unhandled={},
+        )
+
+
+def _save_longtask_data(data: LongTaskModuleData):
+    LONGTASK_DATA_DIR.write_text(data.model_dump_json())
+
+
+@asynccontextmanager
+async def longtask_data():
+    async with longtask_lock:
+        data = _load_longtask_data()
+        yield data
+        _save_longtask_data(data)
+
+
+async def create_longtask(
+    handler: str,
+    data: dict[str, Any],
+    target: LongTaskTarget,
+    deadline: datetime.datetime,
+):
+    task = LongTask(
+        uuid=str(uuid.uuid4()),
+        data_json=json.dumps(data),
+        target=target,
+        callback=handler,
+        deadline=deadline,
+    )
+
+    logger.info(f"创建了新的 LongTask UUID={task.uuid} CALLBACK={task.callback}")
+    await task.start()
+
+    async with longtask_data() as d:
+        d.to_handle.setdefault(handler, []).append(task)
+
+    return task
+
+
+async def init_longtask():
+    counter = 0
+    req: set[str] = set()
+
+    async with longtask_data() as data:
+        for v in data.to_handle.values():
+            for t in v:
+                await t.start()
+                counter += 1
+                req.add(t.callback)
+
+        logger.info(f"LongTask 启动了任务 数量={counter} 期望的门类=[{','.join(req)}]")
+
+
+DepLongTaskTarget = Annotated[LongTaskTarget, Depends(get_long_task_target)]

konabot/plugins/idiomgame/__init__.py

@@ -1,23 +1,30 @@
 import asyncio as asynkio
-import base64
-from pathlib import Path
-import secrets
-import json
 import datetime
-from typing import Literal, Optional
+import json
+import secrets
 from enum import Enum
+from pathlib import Path
+from typing import Optional
 
 from loguru import logger
 from nonebot import on_message
 from nonebot.adapters import Event as BaseEvent
-from nonebot.adapters.console.event import MessageEvent as ConsoleMessageEvent
 from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
-from nonebot_plugin_alconna import (Alconna, Args, Field, Subcommand,
-                                    UniMessage, UniMsg, on_alconna)
+from nonebot_plugin_alconna import (
+    Alconna,
+    Args,
+    UniMessage,
+    UniMsg,
+    on_alconna,
+)
 
+from konabot.common.longtask import DepLongTaskTarget
 from konabot.common.path import ASSETS_PATH
 
-DATA_FILE_PATH = Path(__file__).parent.parent.parent.parent / "data" / "idiom_banned.json"
+DATA_FILE_PATH = (
+    Path(__file__).parent.parent.parent.parent / "data" / "idiom_banned.json"
+)
+
 
 def load_banned_ids() -> list[str]:
     if not DATA_FILE_PATH.exists():
@@ -27,44 +34,53 @@ def load_banned_ids() -> list[str]:
     except Exception as e:
         logger.warning(f"在解析成语接龙封禁文件时遇到问题：{e}")
         return []
-    
+
+
 def is_idiom_game_banned(group_id: str) -> bool:
     banned_ids = load_banned_ids()
     return group_id in banned_ids
 
+
 def add_banned_id(group_id: str):
     banned_ids = load_banned_ids()
     if group_id not in banned_ids:
         banned_ids.append(group_id)
         DATA_FILE_PATH.write_text(json.dumps(banned_ids, ensure_ascii=False, indent=4))
 
+
 def remove_banned_id(group_id: str):
     banned_ids = load_banned_ids()
     if group_id in banned_ids:
         banned_ids.remove(group_id)
         DATA_FILE_PATH.write_text(json.dumps(banned_ids, ensure_ascii=False, indent=4))
 
+
 class TryStartState(Enum):
     STARTED = 0
     ALREADY_PLAYING = 1
     NO_REMAINING_TIMES = 2
 
+
 class TryStopState(Enum):
     STOPPED = 0
     NOT_PLAYING = 1
 
+
 class TryVerifyState(Enum):
     VERIFIED = 0
-    NOT_IDIOM = 1
-    WRONG_FIRST_CHAR = 2
-    VERIFIED_BUT_NO_NEXT = 3
-    VERIFIED_GAME_END = 4
+    VERIFIED_AND_REAL = 1
+    NOT_IDIOM = 2
+    WRONG_FIRST_CHAR = 3
+    VERIFIED_BUT_NO_NEXT = 4
+    VERIFIED_GAME_END = 5
+
 
 class IdiomGame:
-    ALL_WORDS = [] # 所有四字词语
-    ALL_IDIOMS = [] # 所有成语
-    INSTANCE_LIST : dict[str, "IdiomGame"] = {} # 群号对应的游戏实例
-    IDIOM_FIRST_CHAR = {} # 成语首字字典
+    ALL_WORDS = []  # 所有四字词语
+    ALL_IDIOMS = []  # 所有成语
+    INSTANCE_LIST: dict[str, "IdiomGame"] = {}  # 群号对应的游戏实例
+    IDIOM_FIRST_CHAR = {}  # 所有成语包括词语的首字字典
+    AVALIABLE_IDIOM_FIRST_CHAR = {} # 真正有效的成语首字字典
 
     __inited = False
 
@@ -79,22 +95,22 @@ class IdiomGame:
         self.last_play_date = ""
         self.all_buff_score = 0
         self.lock = asynkio.Lock()
-        self.remain_rounds = 0 # 剩余回合数
+        self.remain_rounds = 0  # 剩余回合数
         IdiomGame.INSTANCE_LIST[group_id] = self
 
     def be_able_to_play(self) -> bool:
-        if(self.last_play_date != datetime.date.today()):
+        if self.last_play_date != datetime.date.today():
             self.last_play_date = datetime.date.today()
             self.remain_playing_times = 1
-        if(self.remain_playing_times > 0):
+        if self.remain_playing_times > 0:
             self.remain_playing_times -= 1
             return True
         return False
 
     def choose_start_idiom(self) -> str:
-        '''
+        """
         随机选择一个成语作为起始成语
-        '''
+        """
         self.last_idiom = secrets.choice(IdiomGame.ALL_IDIOMS)
         self.last_char = self.last_idiom[-1]
         if not self.is_nextable(self.last_char):
@@ -113,7 +129,7 @@ class IdiomGame:
             return TryStartState.NO_REMAINING_TIMES
         instance.now_playing = True
         return TryStartState.STARTED
-    
+
     def start_game(self, rounds: int = 100):
         self.now_playing = True
         self.remain_rounds = rounds
@@ -128,93 +144,102 @@ class IdiomGame:
             return TryStopState.NOT_PLAYING
         instance.now_playing = False
         return TryStopState.STOPPED
-    
+
     def clear_score_board(self):
         self.score_board = {}
         self.last_char = ""
-    
+
     def get_score_board(self) -> dict:
         return self.score_board
-    
+
     def get_all_buff_score(self) -> int:
         return self.all_buff_score
 
     async def skip_idiom(self, buff_score: int = -100) -> str:
-        '''
+        """
         跳过当前成语，选择下一个成语
-        '''
-        await self.lock.acquire()
-        self._skip_idiom_async(buff_score)
-        self.lock.release()
+        """
+        async with self.lock:
+            self._skip_idiom_async()
+            self.add_buff_score(buff_score)
         return self.last_idiom
-    
-    def _skip_idiom_async(self, buff_score: int = -100) -> str:
+
+    def _skip_idiom_async(self) -> str:
         self.last_idiom = secrets.choice(IdiomGame.ALL_IDIOMS)
         self.last_char = self.last_idiom[-1]
-        self.add_buff_score(buff_score)
-        return self.last_idiom 
-    
-    async def try_verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
-        '''
-        用户发送成语
-        '''
-        await self.lock.acquire()
-        state = self._verify_idiom(idiom, user_id)
-        self.lock.release()
-        return state
-    
-    def is_nextable(self, last_char: str) -> bool:
-        '''
-        判断是否有成语可以接
-        '''
-        return last_char in IdiomGame.IDIOM_FIRST_CHAR
+        if not self.is_nextable(self.last_char):
+            self._skip_idiom_async()
+        return self.last_idiom
 
+    async def try_verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
+        """
+        用户发送成语
+        """
+        async with self.lock:
+            state = self._verify_idiom(idiom, user_id)
+        return state
+
+    def is_nextable(self, last_char: str) -> bool:
+        """
+        判断是否有成语可以接
+        """
+        return last_char in IdiomGame.AVALIABLE_IDIOM_FIRST_CHAR
 
     def _verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
         # 新成语的首字应与上一条成语的尾字相同
         if idiom[0] != self.last_char:
             return TryVerifyState.WRONG_FIRST_CHAR
-        if(idiom not in IdiomGame.ALL_IDIOMS and idiom not in IdiomGame.ALL_WORDS):
+        if idiom not in IdiomGame.ALL_IDIOMS and idiom not in IdiomGame.ALL_WORDS:
             self.add_score(user_id, -0.1)
             return TryVerifyState.NOT_IDIOM
         self.last_idiom = idiom
         self.last_char = idiom[-1]
         self.add_score(user_id, 1)
+        if idiom in IdiomGame.ALL_IDIOMS:
+            self.add_score(user_id, 4) # 再加 4 分
         self.remain_rounds -= 1
-        if(self.remain_rounds <= 0):
+        if self.remain_rounds <= 0:
             self.now_playing = False
             return TryVerifyState.VERIFIED_GAME_END
-        if(not self.is_nextable(self.last_char)):
+        if not self.is_nextable(self.last_char):
             # 没有成语可以接了，自动跳过
             self._skip_idiom_async()
             return TryVerifyState.VERIFIED_BUT_NO_NEXT
+        if idiom in IdiomGame.ALL_IDIOMS:
+            return TryVerifyState.VERIFIED_AND_REAL # 真实成语
         return TryVerifyState.VERIFIED
-    
-    def get_user_score(self, user_id: str) -> int:
+
+    def get_user_score(self, user_id: str) -> float:
         if user_id not in self.score_board:
             return 0
-        return self.score_board[user_id]["score"]
-    
+        # 避免浮点数精度问题导致过长
+        handled_score = round(self.score_board[user_id]["score"], 1)
+        return handled_score
+
     def add_score(self, user_id: str, score: int):
         if user_id not in self.score_board:
-            self.score_board[user_id] = {
-                "name": user_id,
-                "score": 0
-            }
+            self.score_board[user_id] = {"name": user_id, "score": 0}
         self.score_board[user_id]["score"] += score
-    
+
     def add_buff_score(self, score: int):
         self.all_buff_score += score
 
     def get_playing_state(self) -> bool:
         return self.now_playing
-    
+
     def get_last_char(self) -> str:
         return self.last_char
+    
+    @classmethod
+    def random_idiom_starting_with(cls, first_char: str) -> Optional[str]:
+        cls.init_lexicon()
+        if first_char not in cls.IDIOM_FIRST_CHAR:
+            return None
+        return secrets.choice(cls.IDIOM_FIRST_CHAR[first_char])
 
     @classmethod
     def init_lexicon(cls):
-        if(cls.__inited):
+        if cls.__inited:
             return
         cls.__inited = True
 
@@ -235,28 +260,40 @@ class IdiomGame:
                     COMMON_WORDS.append(word)
 
         # 读取 THUOCL 成语库
-        with open(ASSETS_PATH / "lexicon" / "THUOCL" / "data" / "THUOCL_chengyu.txt", "r", encoding="utf-8") as f:
+        with open(
+            ASSETS_PATH / "lexicon" / "THUOCL" / "data" / "THUOCL_chengyu.txt",
+            "r",
+            encoding="utf-8",
+        ) as f:
             THUOCL_IDIOMS = [line.split(" ")[0].strip() for line in f]
 
         # 读取 THUOCL 剩下的所有 txt 文件，只保留四字词
         THUOCL_WORDS = []
         import os
+
         for filename in os.listdir(ASSETS_PATH / "lexicon" / "THUOCL" / "data"):
             if filename.endswith(".txt") and filename != "THUOCL_chengyu.txt":
-                with open(ASSETS_PATH / "lexicon" / "THUOCL" / "data" / filename, "r", encoding="utf-8") as f:
+                with open(
+                    ASSETS_PATH / "lexicon" / "THUOCL" / "data" / filename,
+                    "r",
+                    encoding="utf-8",
+                ) as f:
                     for line in f:
                         word = line.lstrip().split(" ")[0].strip()
                         if len(word) == 4:
                             THUOCL_WORDS.append(word)
 
-
         # 只有成语的大表
         cls.ALL_IDIOMS = [idiom["word"] for idiom in ALL_IDIOMS_INFOS] + THUOCL_IDIOMS
-        cls.ALL_IDIOMS = list(set(cls.ALL_IDIOMS)) # 去重
+        cls.ALL_IDIOMS = list(set(cls.ALL_IDIOMS))  # 去重
 
         # 其他四字词语表，仅表示可以有这个词
-        cls.ALL_WORDS = [word for word in cls.ALL_WORDS if len(word) == 4] + THUOCL_WORDS + COMMON_WORDS
-        cls.ALL_WORDS = list(set(cls.ALL_WORDS)) # 去重
+        cls.ALL_WORDS = (
+            [word for word in cls.ALL_WORDS if len(word) == 4]
+            + THUOCL_WORDS
+            + COMMON_WORDS
+        )
+        cls.ALL_WORDS = list(set(cls.ALL_WORDS))  # 去重
 
         # 根据成语大表，划分出成语首字字典
         for idiom in cls.ALL_IDIOMS + cls.ALL_WORDS:
@@ -264,41 +301,87 @@ class IdiomGame:
                 cls.IDIOM_FIRST_CHAR[idiom[0]] = []
             cls.IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
 
-evt = on_alconna(Alconna(
-    "我要玩成语接龙",
-    Args['rounds?', int],
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+        # 根据真正的成语大表，划分出有效成语首字字典
+        for idiom in cls.ALL_IDIOMS:
+            if idiom[0] not in cls.AVALIABLE_IDIOM_FIRST_CHAR:
+                cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]] = []
+            cls.AVALIABLE_IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
+
+
+evt = on_alconna(
+    Alconna(
+        "我要玩成语接龙",
+        Args["rounds?", int],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
 
 @evt.handle()
-async def play_game(event: BaseEvent, force = False, rounds: Optional[int] = 100):
-    group_id = str(event.get_session_id())
+async def play_game(
+    event: BaseEvent,
+    target: DepLongTaskTarget,
+    force=False,
+    rounds: Optional[int] = 100,
+):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     if is_idiom_game_banned(group_id):
-        await evt.send(await UniMessage().text("本群已被禁止使用成语接龙功能！").export())
+        await evt.send(
+            await UniMessage().text("本群已被禁止使用成语接龙功能！").export()
+        )
         return
+    rounds = rounds or 0
     if rounds <= 0:
         await evt.send(await UniMessage().text("干什么！你想玩负数局吗？").export())
         return
     state = IdiomGame.try_start_game(group_id, force)
     if state == TryStartState.ALREADY_PLAYING:
-        await evt.send(await UniMessage().text("当前已有成语接龙游戏在进行中，请稍后再试！").export())
+        await evt.send(
+            await UniMessage()
+            .text("当前已有成语接龙游戏在进行中，请稍后再试！")
+            .export()
+        )
         return
     if state == TryStartState.NO_REMAINING_TIMES:
         await evt.send(await UniMessage().text("玩玩玩，就知道玩，快去睡觉！").export())
         return
-    await evt.send(await UniMessage().text("你小子，还真有意思！\n好，成语接龙游戏开始！我说一个成语，请大家接下去！").export())
+    await evt.send(
+        await UniMessage()
+        .text(
+            "你小子，还真有意思！\n好，成语接龙游戏开始！我说一个成语，请大家接下去！"
+        )
+        .export()
+    )
     instance = IdiomGame.INSTANCE_LIST[group_id]
     instance.start_game(rounds)
     # 发布成语
-    await evt.send(await UniMessage().text(f"第一个成语：「{instance.last_idiom}」，请接！").export())
+    await evt.send(
+        await UniMessage()
+        .text(f"第一个成语：「{instance.last_idiom}」，请接！")
+        .export()
+    )
+
+
+evt = on_alconna(
+    Alconna(
+        "老子就是要玩成语接龙！！！",
+        Args["rounds?", int],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
 
-evt = on_alconna(Alconna(
-    "老子就是要玩成语接龙！！！",
-    Args['rounds?', int],
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
 
 @evt.handle()
-async def force_play_game(event: BaseEvent, rounds: Optional[int] = 100):
-    await play_game(event, force=True, rounds=rounds)
+async def force_play_game(
+    event: BaseEvent, target: DepLongTaskTarget, rounds: Optional[int] = 100
+):
+    await play_game(event, target, force=True, rounds=rounds)
+
 
 async def end_game(event: BaseEvent, group_id: str):
     instance = IdiomGame.INSTANCE_LIST[group_id]
@@ -308,19 +391,28 @@ async def end_game(event: BaseEvent, group_id: str):
         result_text += "无人得分！"
     else:
         # 按分数排序，名字用 at 的方式
-        sorted_score = sorted(score_board.items(), key=lambda x: x[1]["score"], reverse=True)
+        sorted_score = sorted(
+            score_board.items(), key=lambda x: x[1]["score"], reverse=True
+        )
         for i, (user_id, info) in enumerate(sorted_score):
-            result_text += f"{i+1}. " + UniMessage().at(user_id) + f": {info['score'] + instance.get_all_buff_score()} 分\n"
+            result_text += (
+                f"{i + 1}. "
+                + UniMessage().at(user_id)
+                + f": {round(info['score'] + instance.get_all_buff_score(), 1)} 分\n"
+            )
     await evt.send(await result_text.export())
     instance.clear_score_board()
 
-evt = on_alconna(Alconna(
-    "不玩了"
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+evt = on_alconna(
+    Alconna("不玩了"), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True
+)
+
 
 @evt.handle()
-async def _(event: BaseEvent):
-    group_id = str(event.get_session_id())
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     state = IdiomGame.try_stop_game(group_id)
     if state == TryStopState.STOPPED:
         # 发送好吧狗图片
@@ -330,22 +422,31 @@ async def _(event: BaseEvent):
         await evt.send(await UniMessage().image(raw=img_data).export())
         await end_game(event, group_id)
     else:
-        await evt.send(await UniMessage().text("当前没有成语接龙游戏在进行中！").export())
+        await evt.send(
+            await UniMessage().text("当前没有成语接龙游戏在进行中！").export()
+        )
+
 
 # 跳过
-evt = on_alconna(Alconna(
-    "跳过成语"
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+evt = on_alconna(
+    Alconna("跳过成语"), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True
+)
+
 
 @evt.handle()
-async def _(event: BaseEvent):
-    group_id = str(event.get_session_id())
+async def _(target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     instance = IdiomGame.INSTANCE_LIST.get(group_id)
     if not instance or not instance.get_playing_state():
         return
-    await evt.send(await UniMessage().text("你们太菜了！全部扣100分！").export())
+    avaliable_idiom = IdiomGame.random_idiom_starting_with(instance.get_last_char())
+    await evt.send(await UniMessage().text(f"你们太菜了，全部扣100分！明明还可以接「{avaliable_idiom}」的！").export())
     idiom = await instance.skip_idiom(-100)
-    await evt.send(await UniMessage().text(f"重新开始，下一个成语是「{idiom}」").export())
+    await evt.send(
+        await UniMessage().text(f"重新开始，下一个成语是「{idiom}」").export()
+    )
+
 
 def get_user_info(event: BaseEvent):
     if isinstance(event, DiscordMessageEvent):
@@ -356,49 +457,92 @@ def get_user_info(event: BaseEvent):
         user_name = str(event.get_user_id())
     return user_id, user_name
 
+
 # 直接读取消息
 evt = on_message()
 
+
 @evt.handle()
-async def _(event: BaseEvent, msg: UniMsg):
-    group_id = str(event.get_session_id())
+async def _(event: BaseEvent, msg: UniMsg, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     instance = IdiomGame.INSTANCE_LIST.get(group_id)
     if not instance or not instance.get_playing_state():
         return
     user_idiom = msg.extract_plain_text().strip()
-    user_id , user_name = get_user_info(event)
+    user_id, user_name = get_user_info(event)
     state = await instance.try_verify_idiom(user_idiom, user_id)
-    if(state == TryVerifyState.WRONG_FIRST_CHAR):
+    if state == TryVerifyState.WRONG_FIRST_CHAR:
         return
-    if(state == TryVerifyState.NOT_IDIOM):
-        await evt.send(await UniMessage().at(user_id).text("接不上！这个不一样！你被扣了 0.1 分！").export())
+    if state == TryVerifyState.NOT_IDIOM:
+        await evt.send(
+            await UniMessage()
+            .at(user_id)
+            .text(" 接不上！这个不一样！你被扣了 0.1 分！")
+            .export()
+        )
         return
-    await evt.send(await UniMessage().at(user_id).text(f"接对了！你有 {instance.get_user_score(user_id)} 分！").export())
-    if(state == TryVerifyState.VERIFIED_GAME_END):
+    if state == TryVerifyState.VERIFIED:
+        await evt.send(
+            await UniMessage()
+            .at(user_id)
+            .text(f" 接上了，喜提 1 分！你有 {instance.get_user_score(user_id)} 分！")
+            .export()
+        )
+    elif state == TryVerifyState.VERIFIED_AND_REAL:
+        await evt.send(
+            await UniMessage()
+            .at(user_id)
+            .text(f" 接上了，这是个真实成语，喜提 5 分！你有 {instance.get_user_score(user_id)} 分！")
+            .export()
+        )
+    if state == TryVerifyState.VERIFIED_GAME_END:
         await evt.send(await UniMessage().text("全部回合结束！").export())
         await end_game(event, group_id)
         return
-    if(state == TryVerifyState.VERIFIED_BUT_NO_NEXT):
-        await evt.send(await UniMessage().text("但是，这是条死路！你们全部都要扣 100 分！").export())
-        await evt.send(await UniMessage().text(f"重新抽取成语「{instance.last_idiom}」").export())
-    await evt.send(await UniMessage().text(f"下一个成语请以「{instance.get_last_char()}」开头！").export())
+    if state == TryVerifyState.VERIFIED_BUT_NO_NEXT:
+        await evt.send(
+            await UniMessage()
+            .text("但是，这是条死路！你们全部都要扣 100 分！")
+            .export()
+        )
+        await evt.send(
+            await UniMessage().text(f"重新抽取成语「{instance.last_idiom}」").export()
+        )
+    await evt.send(
+        await UniMessage()
+        .text(f"下一个成语请以「{instance.get_last_char()}」开头！")
+        .export()
+    )
+
+
+evt = on_alconna(
+    Alconna("禁止成语接龙"),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
 
-evt = on_alconna(Alconna(
-    "禁止成语接龙"
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
 
 @evt.handle()
-async def _(event: BaseEvent):
-    group_id = str(event.get_session_id())
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     add_banned_id(group_id)
     await evt.send(await UniMessage().text("本群已被禁止使用成语接龙功能！").export())
 
-evt = on_alconna(Alconna(
-    "开启成语接龙"
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=True)
+
+evt = on_alconna(
+    Alconna("开启成语接龙"),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=True,
+)
+
 
 @evt.handle()
-async def _(event: BaseEvent):
-    group_id = str(event.get_session_id())
+async def _(event: BaseEvent, target: DepLongTaskTarget):
+    # group_id = str(event.get_session_id())
+    group_id = target.channel_id
     remove_banned_id(group_id)
-    await evt.send(await UniMessage().text("本群已开启成语接龙功能！").export())
+    await evt.send(await UniMessage().text("本群已开启成语接龙功能！").export())

konabot/plugins/image_process/__init__.py

@@ -1,10 +1,10 @@
 import re
 from io import BytesIO
 
+import PIL.Image
 from nonebot import on_message
 from nonebot.adapters import Bot
-from nonebot_plugin_alconna import (Alconna, Args, Image, Option, UniMessage,
-                                    on_alconna)
+from nonebot_plugin_alconna import Alconna, Args, Image, Option, UniMessage, on_alconna
 
 from konabot.common.nb.exc import BotExceptionMessage
 from konabot.common.nb.extract_image import PIL_Image
@@ -29,15 +29,17 @@ def parse_timestamp(tx: str) -> float | None:
     return res
 
 
-cmd_giftool = on_alconna(Alconna(
-    "giftool",
-    Args["img", Image | None],
-    Option("--ss", Args["start_point", str]),
-    Option("--frames:v", Args["frame_count", int]),
-    Option("-t", Args["length", str]),
-    Option("-to", Args["end_point", str]),
-    Option("--speed", Args["speed_factor", float], default=1.0, alias=["-s"]),
-))
+cmd_giftool = on_alconna(
+    Alconna(
+        "giftool",
+        Args["img", Image | None],
+        Option("--ss", Args["start_point", str]),
+        Option("--frames:v", Args["frame_count", int]),
+        Option("-t", Args["length", str]),
+        Option("-to", Args["end_point", str]),
+        Option("--speed", Args["speed_factor", float], default=1.0, alias=["-s"]),
+    )
+)
 
 
 @cmd_giftool.handle()
@@ -80,81 +82,66 @@ async def _(
     if not getattr(image, "is_animated", False):
         raise BotExceptionMessage("错误：输入的不是动图（GIF）")
 
-    frames = []
-    durations = []
-    total_duration = 0.0
-
+    ##
+    # 从这里开始，采样整个 GIF 图
+    frames: list[PIL.Image.Image] = []
+    durations: list[float] = []
     try:
         for i in range(getattr(image, "n_frames")):
             image.seek(i)
             frames.append(image.copy())
-            duration = image.info.get("duration", 100)  # 单位：毫秒
+            duration = image.info.get("duration", 100) / 1000
             durations.append(duration)
-            total_duration += duration / 1000.0  # 转为秒
     except EOFError:
         pass
-
     if not frames:
         raise BotExceptionMessage("错误：读取 GIF 帧失败")
+    # 采样结束
 
-    def time_to_frame_index(target_time: float) -> int:
-        if target_time <= 0:
-            return 0
-        cum = 0.0
-        for idx, dur in enumerate(durations):
-            cum += dur / 1000.0
-            if cum >= target_time:
-                return min(idx, len(frames) - 1)
-        return len(frames) - 1
-    start_frame = 0
-    end_frame = len(frames) - 1
-    if ss is not None:
-        start_frame = time_to_frame_index(ss)
-    if to is not None:
-        end_frame = time_to_frame_index(to)
-        if end_frame < start_frame:
-            end_frame = start_frame
-    elif t is not None:
-        end_time = (ss or 0.0) + t
-        end_frame = time_to_frame_index(end_time)
-        if end_frame < start_frame:
-            end_frame = start_frame
+    ##
+    # 根据开始、结束时间或者帧数量来裁取 GIF 图
 
-    start_frame = max(0, start_frame)
-    end_frame = min(len(frames) - 1, end_frame)
-    selected_frames = frames[start_frame : end_frame + 1]
-    selected_durations = durations[start_frame : end_frame + 1]
+    begin_time = ss or 0
+    end_time = sum(durations)
+    end_time = min(begin_time + (t or end_time), to or end_time, end_time)
 
-    if frame_count is not None and frame_count > 0:
-        if frame_count >= len(selected_frames):
-            pass
-        else:
-            step = len(selected_frames) / frame_count
-            sampled_frames = []
-            sampled_durations = []
-            for i in range(frame_count):
-                idx = int(i * step)
-                sampled_frames.append(selected_frames[idx])
-                sampled_durations.append(
-                    sum(selected_durations) // len(selected_durations)
-                )
-            selected_frames = sampled_frames
-            selected_durations = sampled_durations
+    accumulated = 0.0
+    status = 0
 
-    output_img = BytesIO()
+    sel_frames: list[PIL.Image.Image] = []
+    sel_durations: list[float] = []
 
-    adjusted_durations = [
-        dur / speed_factor for dur in selected_durations
-    ]
+    for i in range(len(frames)):
+        frame = frames[i]
+        duration = durations[i]
+
+        if status == 0:
+            if accumulated + duration > begin_time:
+                status = 1
+                sel_frames.append(frame)
+                sel_durations.append(accumulated + duration - begin_time)
+        elif status == 1:
+            if accumulated + duration > end_time:
+                sel_frames.append(frame)
+                sel_durations.append(end_time - accumulated)
+                break
+            sel_frames.append(frame)
+            sel_durations.append(duration)
+
+        accumulated += duration
+
+    ##
+    # 加速！
+    sel_durations = [dur / speed_factor * 1000 for dur in durations]
 
     rframes = []
     rdur = []
 
     acc_mod_20 = 0
 
-    for i in range(len(selected_frames)):
-        fr = selected_frames[i]
-        du: float = adjusted_durations[i]
+    for i in range(len(sel_frames)):
+        fr = sel_frames[i]
+        du = round(sel_durations[i])
 
         if du >= 20:
             rframes.append(fr)
@@ -170,10 +157,12 @@ async def _(
                 if acc_mod_20 >= 20:
                     acc_mod_20 = 0
 
-    if len(rframes) == 1 and len(selected_frames) > 1:
-        rframes.append(selected_frames[max(2, len(selected_frames) // 2)])
+    if len(rframes) == 1 and len(sel_frames) > 1:
+        rframes.append(sel_frames[max(2, len(sel_frames) // 2)])
         rdur.append(20)
 
+    ##
+    # 收尾：看看透明度这块
     transparency_flag = False
     for f in rframes:
         if f.mode == "RGBA":
@@ -186,12 +175,13 @@ async def _(
 
     tf = {}
     if transparency_flag:
-        tf['transparency'] = 0
+        tf["transparency"] = 0
 
     if is_rev:
         rframes = rframes[::-1]
         rdur = rdur[::-1]
 
+    output_img = BytesIO()
     if rframes:
         rframes[0].save(
             output_img,

konabot/plugins/longtask_core.py

@@ -0,0 +1,47 @@
+import asyncio
+
+# import datetime
+from loguru import logger
+import nonebot
+
+# from nonebot_plugin_alconna import UniMessage
+from konabot.common.longtask import (
+    # DepLongTaskTarget,
+    # LongTask,
+    # create_longtask,
+    # handle_long_task,
+    init_longtask,
+)
+
+
+driver = nonebot.get_driver()
+INIT_FLAG = {"flag": False}
+
+
+@driver.on_bot_connect
+async def _():
+    if INIT_FLAG["flag"]:
+        return
+    INIT_FLAG["flag"] = True
+    logger.info("有 Bot 连接，等待 5 秒后初始化 LongTask 模块")
+    await asyncio.sleep(5)
+    await init_longtask()
+    logger.info("LongTask 初始化完成")
+
+
+# cmd1 = nonebot.on_command("test114")
+#
+#
+# @handle_long_task("test_callback_001")
+# async def _(lt: LongTask):
+#     await lt.target.send_message(UniMessage().text("Hello, world!"), at=True)
+#
+#
+# @cmd1.handle()
+# async def _(target: DepLongTaskTarget):
+#     await create_longtask(
+#         handler="test_callback_001",
+#         data={},
+#         target=target,
+#         deadline=datetime.datetime.now() + datetime.timedelta(seconds=20),
+#     )

konabot/plugins/memepack/__init__.py

@@ -2,25 +2,52 @@ from io import BytesIO
 from typing import Iterable, cast
 
 from nonebot import on_message
-from nonebot_plugin_alconna import (Alconna, Args, Field, Image, MultiVar, Option, Text,
-                                    UniMessage, UniMsg, on_alconna)
+from nonebot_plugin_alconna import (
+    Alconna,
+    Args,
+    Field,
+    Image,
+    MultiVar,
+    Option,
+    Text,
+    UniMessage,
+    UniMsg,
+    on_alconna,
+)
 
 from konabot.common.nb.extract_image import PIL_Image, extract_image_from_message
-from konabot.plugins.memepack.drawing.display import draw_cao_display, draw_snaur_display
-from konabot.plugins.memepack.drawing.saying import (draw_cute_ten,
-                                                     draw_geimao, draw_mnk,
-                                                     draw_pt, draw_suan)
+from konabot.plugins.memepack.drawing.display import (
+    draw_cao_display,
+    draw_snaur_display,
+    draw_anan_display,
+)
+from konabot.plugins.memepack.drawing.saying import (
+    draw_cute_ten,
+    draw_geimao,
+    draw_mnk,
+    draw_pt,
+    draw_suan,
+)
 
 from nonebot.adapters import Bot, Event
 
 from returns.result import Success, Failure
 
-geimao = on_alconna(Alconna(
-    "给猫说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写给猫说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"给猫哈"})
+geimao = on_alconna(
+    Alconna(
+        "给猫说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写给猫说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases={"给猫哈"},
+)
+
 
 @geimao.handle()
 async def _(saying: list[str]):
@@ -31,12 +58,21 @@ async def _(saying: list[str]):
     await geimao.send(await UniMessage().image(raw=img_bytes).export())
 
 
-pt = on_alconna(Alconna(
-    "pt说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写小帕说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"小帕说"})
+pt = on_alconna(
+    Alconna(
+        "pt说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写小帕说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases={"小帕说"},
+)
+
 
 @pt.handle()
 async def _(saying: list[str]):
@@ -47,12 +83,21 @@ async def _(saying: list[str]):
     await pt.send(await UniMessage().image(raw=img_bytes).export())
 
 
-mnk = on_alconna(Alconna(
-    "re:小?黑白子?说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写黑白子说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"mnk说"})
+mnk = on_alconna(
+    Alconna(
+        "re:小?黑白子?说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写黑白子说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases={"mnk说"},
+)
+
 
 @mnk.handle()
 async def _(saying: list[str]):
@@ -63,12 +108,21 @@ async def _(saying: list[str]):
     await mnk.send(await UniMessage().image(raw=img_bytes).export())
 
 
-suan = on_alconna(Alconna(
-    "小蒜说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写小蒜说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases=set())
+suan = on_alconna(
+    Alconna(
+        "小蒜说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写小蒜说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases=set(),
+)
+
 
 @suan.handle()
 async def _(saying: list[str]):
@@ -79,12 +133,21 @@ async def _(saying: list[str]):
     await suan.send(await UniMessage().image(raw=img_bytes).export())
 
 
-dsuan = on_alconna(Alconna(
-    "大蒜说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写大蒜说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases=set())
+dsuan = on_alconna(
+    Alconna(
+        "大蒜说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写大蒜说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases=set(),
+)
+
 
 @dsuan.handle()
 async def _(saying: list[str]):
@@ -95,12 +158,21 @@ async def _(saying: list[str]):
     await dsuan.send(await UniMessage().image(raw=img_bytes).export())
 
 
-cutecat = on_alconna(Alconna(
-    "乖猫说",
-    Args["saying", MultiVar(str, '+'), Field(
-        missing_tips=lambda: "你没有写十猫说了什么"
-    )]
-), use_cmd_start=True, use_cmd_sep=False, skip_for_unmatch=False, aliases={"十猫说"})
+cutecat = on_alconna(
+    Alconna(
+        "乖猫说",
+        Args[
+            "saying",
+            MultiVar(str, "+"),
+            Field(missing_tips=lambda: "你没有写十猫说了什么"),
+        ],
+    ),
+    use_cmd_start=True,
+    use_cmd_sep=False,
+    skip_for_unmatch=False,
+    aliases={"十猫说"},
+)
+
 
 @cutecat.handle()
 async def _(saying: list[str]):
@@ -113,13 +185,14 @@ async def _(saying: list[str]):
 
 cao_display_cmd = on_message()
 
+
 @cao_display_cmd.handle()
 async def _(msg: UniMsg, evt: Event, bot: Bot):
     flag = False
     for text in cast(Iterable[Text], msg.get(Text)):
         if text.text.strip() == "小槽展示":
             flag = True
-        elif text.text.strip() == '':
+        elif text.text.strip() == "":
             continue
         else:
             return
@@ -134,29 +207,71 @@ async def _(msg: UniMsg, evt: Event, bot: Bot):
         case Failure(err):
             await cao_display_cmd.send(
                 await UniMessage()
-                    .at(user_id=evt.get_user_id())
-                    .text(' ')
-                    .text(err)
-                    .export()
+                .at(user_id=evt.get_user_id())
+                .text(" ")
+                .text(err)
+                .export()
             )
 
 
-snaur_display_cmd = on_alconna(Alconna(
-    "卵总展示",
-    Option("--whiteness",   Args["whiteness", float],   alias=["-w"]),
-    Option("--black-level", Args["black_level", float], alias=["-b"]),
-    Option("--opacity",     Args["opacity", float],     alias=["-o"]),
-    Option("--saturation",  Args["saturation", float],  alias=["-s"]),
-    Args["image", Image | None],
-))
+snaur_display_cmd = on_alconna(
+    Alconna(
+        "卵总展示",
+        Option("--whiteness", Args["whiteness", float], alias=["-w"]),
+        Option("--black-level", Args["black_level", float], alias=["-b"]),
+        Option("--opacity", Args["opacity", float], alias=["-o"]),
+        Option("--saturation", Args["saturation", float], alias=["-s"]),
+        Args["image", Image | None],
+    )
+)
+
 
 @snaur_display_cmd.handle()
-async def _(img: PIL_Image, whiteness: float = 0.0, black_level: float = 0.2,
-            opacity: float = 0.8, saturation: float = 0.85):
+async def _(
+    img: PIL_Image,
+    whiteness: float = 0.0,
+    black_level: float = 0.2,
+    opacity: float = 0.8,
+    saturation: float = 0.85,
+):
     img_processed = await draw_snaur_display(
-        img, whiteness, black_level, opacity, saturation, 
+        img,
+        whiteness,
+        black_level,
+        opacity,
+        saturation,
     )
     img_data = BytesIO()
     img_processed.save(img_data, "PNG")
     await snaur_display_cmd.send(await UniMessage().image(raw=img_data).export())
 
+anan_display_cmd = on_message()
+@anan_display_cmd.handle()
+async def _(msg: UniMsg, evt: Event, bot: Bot):
+    flag = False
+    for text in cast(Iterable[Text], msg.get(Text)):
+        stripped = text.text.strip()
+        if stripped == "安安展示":
+            flag = True
+        elif stripped == "":
+            continue
+        else:
+            return
+    if not flag:
+        return
+
+    match await extract_image_from_message(evt.get_message(), evt, bot):
+        case Success(img):
+            img_handled = await draw_anan_display(img)
+            img_bytes = BytesIO()
+            img_handled.save(img_bytes, format="PNG")
+            await anan_display_cmd.send(await UniMessage().image(raw=img_bytes).export())
+        case Failure(err):
+            await anan_display_cmd.send(
+                await UniMessage()
+                .at(user_id=evt.get_user_id())
+                .text(" ")
+                .text(err)
+                .export()
+            )
+

konabot/plugins/memepack/drawing/display.py

@@ -27,6 +27,15 @@ SNAUR_QUAD_POINTS = np.float32(cast(Any, [
     [106, 1280],
 ]))
 
+anan_image_base = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "anan_base.png")
+anan_image_top  = PIL.Image.open(ASSETS_PATH / "img" / "meme" / "anan_top.png")
+ANAN_QUAD_POINTS = np.float32([
+    [157, 585],
+    [793, 599],
+    [781, 908],
+    [160, 908]
+])
+
 def _draw_cao_display(image: PIL.Image.Image):
     src = np.array(image.convert("RGB"))
     h, w = src.shape[:2]
@@ -139,3 +148,52 @@ async def draw_snaur_display(
         opacity, saturation,
     )
 
+
+def _draw_anan_display(image: PIL.Image.Image) -> PIL.Image.Image:
+    src = np.array(image.convert("RGBA"))
+    h, w = src.shape[:2]
+
+    src_points = np.float32([
+        [0, 0],
+        [w, 0],
+        [w, h],
+        [0, h]
+    ])
+    dst_points = ANAN_QUAD_POINTS
+
+    M = cv2.getPerspectiveTransform(src_points, dst_points)
+    output_w, output_h = anan_image_top.size
+
+    src_rgb = cv2.cvtColor(src, cv2.COLOR_RGBA2RGB) if src.shape[2] == 4 else src
+    warped_rgb = cv2.warpPerspective(
+        src_rgb,
+        M,
+        (output_w, output_h),
+        flags=cv2.INTER_LINEAR,
+        borderMode=cv2.BORDER_CONSTANT,
+        borderValue=(0, 0, 0)
+    )
+
+    mask = np.zeros((h, w), dtype=np.uint8)
+    mask[:, :] = 255
+    warped_mask = cv2.warpPerspective(
+        mask,
+        M,
+        (output_w, output_h),
+        flags=cv2.INTER_LINEAR,
+        borderMode=cv2.BORDER_CONSTANT,
+        borderValue=0
+    )
+
+    warped_rgba = cv2.cvtColor(warped_rgb, cv2.COLOR_RGB2RGBA)
+    warped_rgba[:, :, 3] = warped_mask
+
+    warped_pil = PIL.Image.fromarray(warped_rgba, 'RGBA')
+
+    result = PIL.Image.alpha_composite(anan_image_base, warped_pil)
+    result = PIL.Image.alpha_composite(result, anan_image_top)
+    return result
+
+
+async def draw_anan_display(image: PIL.Image.Image) -> PIL.Image.Image:
+    return await asyncio.to_thread(_draw_anan_display, image)

konabot/plugins/simple_notify/__init__.py

@@ -15,10 +15,10 @@ from nonebot.adapters.console.event import MessageEvent as ConsoleMessageEvent
 from nonebot.adapters.discord import Bot as DiscordBot
 from nonebot.adapters.discord.event import MessageEvent as DiscordMessageEvent
 from nonebot.adapters.onebot.v11 import Bot as OnebotV11Bot
-from nonebot.adapters.onebot.v11.event import \
-    GroupMessageEvent as OnebotV11GroupMessageEvent
-from nonebot.adapters.onebot.v11.event import \
-    MessageEvent as OnebotV11MessageEvent
+from nonebot.adapters.onebot.v11.event import (
+    GroupMessageEvent as OnebotV11GroupMessageEvent,
+)
+from nonebot.adapters.onebot.v11.event import MessageEvent as OnebotV11MessageEvent
 from nonebot_plugin_alconna import UniMessage, UniMsg
 from pydantic import BaseModel
 
@@ -68,14 +68,14 @@ def save_notify_config(config: NotifyConfigFile):
 
 
 async def notify_now(notify: Notify):
-    if notify.platform == 'console':
+    if notify.platform == "console":
         bot = [b for b in nonebot.get_bots().values() if isinstance(b, ConsoleBot)]
         if len(bot) != 1:
             logger.warning(f"提醒未成功发送出去：{nonebot.get_bots()} {notify}")
             return False
         bot = bot[0]
         await bot.send_private_message(notify.target, f"代办通知：{notify.notify_msg}")
-    elif notify.platform == 'discord':
+    elif notify.platform == "discord":
         bot = [b for b in nonebot.get_bots().values() if isinstance(b, DiscordBot)]
         if len(bot) != 1:
             logger.warning(f"提醒未成功发送出去：{nonebot.get_bots()} {notify}")
@@ -83,7 +83,7 @@ async def notify_now(notify: Notify):
         bot = bot[0]
         channel = await bot.create_DM(recipient_id=int(notify.target))
         await bot.send_to(channel.id, f"代办通知：{notify.notify_msg}")
-    elif notify.platform == 'qq':
+    elif notify.platform == "qq":
         bot = [b for b in nonebot.get_bots().values() if isinstance(b, OnebotV11Bot)]
         if len(bot) != 1:
             logger.warning(f"提醒未成功发送出去：{nonebot.get_bots()} {notify}")
@@ -92,17 +92,22 @@ async def notify_now(notify: Notify):
         if notify.target_env is None:
             await bot.send_private_msg(
                 user_id=int(notify.target),
-                message=cast(Any, await UniMessage.text(f"代办通知：{notify.notify_msg}").export(
-                    bot=bot,
-                )),
+                message=cast(
+                    Any,
+                    await UniMessage.text(f"代办通知：{notify.notify_msg}").export(
+                        bot=bot,
+                    ),
+                ),
             )
         else:
             await bot.send_group_msg(
                 group_id=int(notify.target_env),
-                message=cast(Any,
-                    await UniMessage().at(
-                        notify.target
-                    ).text(f" 代办通知：{notify.notify_msg}").export(bot=bot)
+                message=cast(
+                    Any,
+                    await UniMessage()
+                    .at(notify.target)
+                    .text(f" 代办通知：{notify.notify_msg}")
+                    .export(bot=bot),
                 ),
             )
     else:
@@ -127,15 +132,17 @@ def create_notify_task(notify: Notify, fail2remove: bool = True):
             )
         res = await notify_now(notify)
         if fail2remove or res:
-            await DATA_FILE_LOCK.acquire()
-            cfg = load_notify_config()
-            cfg.notifies = [n for n in cfg.notifies if n.get_str() != notify.get_str()]
-            if not res:
-                cfg.unsent.append(notify)
-            save_notify_config(cfg)
-            DATA_FILE_LOCK.release()
+            async with DATA_FILE_LOCK:
+                cfg = load_notify_config()
+                cfg.notifies = [
+                    n for n in cfg.notifies if n.get_str() != notify.get_str()
+                ]
+                if not res:
+                    cfg.unsent.append(notify)
+                save_notify_config(cfg)
         else:
             pass
+
     return asynkio.create_task(mission())
 
 
@@ -155,7 +162,8 @@ async def _(msg: UniMsg, mEvt: Event):
     notify_time, notify_text = segments
     # target_time = get_target_time(notify_time)
     try:
-        target_time = ptimeparse.parse(notify_time)
+        # target_time = ptimeparse.parse(notify_time)
+        target_time = ptimeparse.Parser().parse(notify_time)
         logger.info(f"从 {notify_time} 解析出了时间：{target_time}")
     except Exception:
         logger.info(f"无法从 {notify_time} 中解析出时间")
@@ -201,8 +209,12 @@ async def _(msg: UniMsg, mEvt: Event):
     save_notify_config(cfg)
     DATA_FILE_LOCK.release()
 
-    await evt.send(await UniMessage().at(mEvt.get_user_id()).text(
-        f" 了解啦！将会在 {notify.notify_time} 提醒你哦~").export())
+    await evt.send(
+        await UniMessage()
+        .at(mEvt.get_user_id())
+        .text(f" 了解啦！将会在 {notify.notify_time} 提醒你哦~")
+        .export()
+    )
     logger.info(f"创建了一条于 {notify.notify_time} 的代办提醒")
 
 
@@ -253,8 +265,8 @@ async def _():
         logger.info("所有的代办提醒 Task 都已经退出了")
 
     for sig in (signal.SIGINT, signal.SIGTERM):
-        loop.add_signal_handler(sig, functools.partial(
-            asynkio.create_task, shutdown(sig)
-        ))
+        loop.add_signal_handler(
+            sig, functools.partial(asynkio.create_task, shutdown(sig))
+        )
 
     await asynkio.gather(*ASYNK_TASKS)

poetry.lock

@@ -2402,14 +2402,14 @@ reference = "mirrors"
 
 [[package]]
 name = "ptimeparse"
-version = "0.1.2"
+version = "0.2.1"
 description = "一个用于解析中文的时间表达的库"
 optional = false
 python-versions = ">=3.9"
 groups = ["main"]
 files = [
-    {file = "ptimeparse-0.1.2-py3-none-any.whl", hash = "sha256:0eea791396e53b63330fadb40d9f0a2e6272bd5467246f10d1d6971bc606edff"},
-    {file = "ptimeparse-0.1.2.tar.gz", hash = "sha256:658be90a3cc2994c09c4ea2f276d257e7eb84bc330be79950baefe32b19779a2"},
+    {file = "ptimeparse-0.2.1-py3-none-any.whl", hash = "sha256:cf1115784d5d983da2d5b7af327108bf04c218c795d63291e71f76d7c6ffd2d4"},
+    {file = "ptimeparse-0.2.1.tar.gz", hash = "sha256:9b640e0a315d19b1e3821a290d236a051d8320348970ce3a835ed675bd2d832f"},
 ]
 
 [package.source]
@@ -3807,4 +3807,4 @@ reference = "mirrors"
 [metadata]
 lock-version = "2.1"
 python-versions = ">=3.12,<4.0"
-content-hash = "d6a325b769fb3ed207c1a8891e65ea20bae20166fa281d6fa620faf54ad15bd8"
+content-hash = "78a299c64ba07999fae807300b10a1c622d45b8b387aded5a34d17cf5550e777"

pyproject.toml

@@ -2,30 +2,28 @@
 name = "konabot"
 version = "0.1.0"
 description = "在 MTTU 内部使用的 bot"
-authors = [
-    {name = "passthem",email = "Passthem183@gmail.com"}
-]
+authors = [{ name = "passthem", email = "Passthem183@gmail.com" }]
 readme = "README.md"
 requires-python = ">=3.12,<4.0"
 dependencies = [
-    "nonebot2[all] (>=2.4.3,<3.0.0)",
-    "nonebot-adapter-onebot (>=2.4.6,<3.0.0)",
-    "nonebot-adapter-console (>=0.9.0,<0.10.0)",
-    "nonebot-adapter-discord (>=0.1.8,<0.2.0)",
-    "nonebot-adapter-minecraft (>=1.5.2,<2.0.0)",
-    "nonebot-plugin-alconna (>=0.59.4,<0.60.0)",
-    "nonebot-plugin-apscheduler (>=0.5.0,<0.6.0)",
-    "requests (>=2.32.5,<3.0.0)",
-    "beautifulsoup4 (>=4.13.5,<5.0.0)",
-    "lxml (>=6.0.2,<7.0.0)",
-    "pillow (>=11.3.0,<12.0.0)",
-    "imagetext-py (>=2.2.0,<3.0.0)",
-    "opencv-python-headless (>=4.12.0.88,<5.0.0.0)",
-    "returns (>=0.26.0,<0.27.0)",
-    "ptimeparse (>=0.1.1,<0.2.0)",
-    "skia-python (>=138.0,<139.0)",
-    "nonebot-plugin-analysis-bilibili (>=2.8.1,<3.0.0)",
-    "qrcode (>=8.2,<9.0)",
+  "nonebot2[all] (>=2.4.3,<3.0.0)",
+  "nonebot-adapter-onebot (>=2.4.6,<3.0.0)",
+  "nonebot-adapter-console (>=0.9.0,<0.10.0)",
+  "nonebot-adapter-discord (>=0.1.8,<0.2.0)",
+  "nonebot-adapter-minecraft (>=1.5.2,<2.0.0)",
+  "nonebot-plugin-alconna (>=0.59.4,<0.60.0)",
+  "nonebot-plugin-apscheduler (>=0.5.0,<0.6.0)",
+  "requests (>=2.32.5,<3.0.0)",
+  "beautifulsoup4 (>=4.13.5,<5.0.0)",
+  "lxml (>=6.0.2,<7.0.0)",
+  "pillow (>=11.3.0,<12.0.0)",
+  "imagetext-py (>=2.2.0,<3.0.0)",
+  "opencv-python-headless (>=4.12.0.88,<5.0.0.0)",
+  "returns (>=0.26.0,<0.27.0)",
+  "skia-python (>=138.0,<139.0)",
+  "nonebot-plugin-analysis-bilibili (>=2.8.1,<3.0.0)",
+  "qrcode (>=8.2,<9.0)",
+  "ptimeparse (>=0.2.1,<0.3.0)",
 ]
 
 [build-system]
@@ -43,4 +41,3 @@ url = "https://pypi.tuna.tsinghua.edu.cn/simple/"
 priority = "primary"
 
 [tool.poetry.dependencies]
-ptimeparse = {source = "pt-gitea-pypi"}