怪话不可为空！！！

Reviewed-on: #27

bot.py

@@ -20,9 +20,13 @@ env_enable_minecraft = os.environ.get("ENABLE_MINECRAFT", "none")
 
 
 def main():
+    if env.upper() == 'DEBUG' or env.upper() == 'DEV':
+        console_log_level = 'DEBUG'
+    else:
+        console_log_level = 'INFO'
     init_logger(LOG_PATH, [
         BotExceptionMessage,
-    ])
+    ], console_log_level=console_log_level)
 
     nonebot.init()
 

konabot/common/log.py

@@ -18,7 +18,7 @@ def file_exception_filter(
     否则，返回 True（允许记录）。
     """
     exception_info = record.get("exception")
-    
+
     if exception_info:
         exception_type = exception_info[0]
 
@@ -29,8 +29,9 @@ def file_exception_filter(
 
 
 def init_logger(
-    log_dir: Path, 
-    ignored_exceptions: List[Type[Exception]]
+    log_dir: Path,
+    ignored_exceptions: List[Type[Exception]],
+    console_log_level: str = "INFO",
 ) -> None:
     """
     配置全局 Loguru Logger。
@@ -47,7 +48,7 @@ def init_logger(
 
     logger.add(
         sys.stderr,
-        level="INFO",
+        level=console_log_level,
         colorize=True,
         format="<green>{time:HH:mm:ss}</green> | <level>{level: <8}</level> | <cyan>{name}</cyan>:<cyan>{function}</cyan>:<cyan>{line}</cyan> - <level>{message}</level>",
     )
@@ -76,4 +77,4 @@ def init_logger(
     )
 
     logger.info("Loguru Logger 初始化完成！")
-    logger.info(f"控制台日志级别: INFO")
+    logger.info(f"控制台日志级别: {console_log_level}")

konabot/common/longtask.py

@@ -25,6 +25,7 @@ from pydantic import BaseModel, ValidationError
 from .path import DATA_PATH
 
 LONGTASK_DATA_DIR = DATA_PATH / "longtasks.json"
+QQ_PRIVATE_CHAT_CHANNEL_PREFIX = "_CHANNEL_QQ_PRIVATE_"
 
 
 class LongTaskTarget(BaseModel):
@@ -65,7 +66,7 @@ class LongTaskTarget(BaseModel):
                     } BOT_CLASS={bot.__class__.__name__}"
                 )
                 return False
-            if self.channel_id == "":
+            if self.channel_id.startswith(QQ_PRIVATE_CHAT_CHANNEL_PREFIX):
                 # 私聊模式
                 await bot.send_private_msg(
                     user_id=int(self.target_id),
@@ -196,7 +197,7 @@ async def get_long_task_target(event: BaseEvent, bot: BaseBot) -> LongTaskTarget
         return LongTaskTarget(
             platform="qq",
             self_id=str(event.self_id),
-            channel_id="",
+            channel_id=f"{QQ_PRIVATE_CHAT_CHANNEL_PREFIX}{event.self_id}",
             target_id=str(event.user_id),
         )
     if isinstance(event, ConsoleMessageEvent):

konabot/docs/sys/怪话过滤.txt

@@ -0,0 +1,9 @@
+指令介绍
+    怪话过滤 - 去除含有关键词的怪话
+
+使用方法
+    `怪话过滤 说的道理`
+        去除所有含有“说的道理”的怪话
+
+另见
+    怪话(1)

konabot/docs/user/怪话.txt

@@ -0,0 +1,12 @@
+指令介绍
+    说点怪话/说些怪话 - 让 BOT 学群友胡言乱语
+
+适用范围
+    为保证安全，只有少数授权的群聊可以使用该指令
+
+使用方法
+    `说点怪话 今天吃什么`
+        期待 Bot 会回答你什么吧
+
+    `说些怪话 明天不想上体育课`
+        Bot 会回复你三句怪话

konabot/plugins/fortune/__init__.py

@@ -0,0 +1,68 @@
+import asyncio
+from nonebot import get_plugin_config, on_command, on_message
+from nonebot.adapters import Bot
+from nonebot.adapters.onebot.v11.event import GroupMessageEvent
+from nonebot_plugin_alconna import Alconna, Args, UniMessage, UniMsg, on_alconna
+from pydantic import BaseModel
+
+from konabot.common.nb.is_admin import is_admin
+from konabot.common.path import DATA_PATH
+
+from .random_text_record import RandomTextManager
+
+
+class FortuneConfig(BaseModel):
+    plugin_fortune_collect_groups: list[int] = []
+
+
+fortune_wtf = RandomTextManager(DATA_PATH / "fortune_wtf.txt")
+fortune_insert_lock = asyncio.Lock()
+fortune_config = get_plugin_config(FortuneConfig)
+
+
+async def is_collect_target(evt: GroupMessageEvent) -> bool:
+    if evt.group_id not in fortune_config.plugin_fortune_collect_groups:
+        return False
+    return True
+
+
+evt_collector = on_message(rule=is_collect_target)
+
+@evt_collector.handle()
+async def _(msg: UniMsg):
+    txt = msg.extract_plain_text()
+    if len(txt) > 50 or not txt.strip():
+        return
+    if txt.startswith("说点怪话") or txt.startswith("说些怪话") or txt.startswith("怪话过滤"):
+        return
+    async with fortune_insert_lock:
+        fortune_wtf.insert(txt)
+
+
+cmd_guaihua = on_command("说点怪话", rule=is_collect_target)
+
+@cmd_guaihua.handle()
+async def _(bot: Bot):
+    await cmd_guaihua.send(await UniMessage().text(fortune_wtf.choice()).export(bot))
+
+
+cmd_guaihuas = on_command("说些怪话", rule=is_collect_target)
+
+@cmd_guaihuas.handle()
+async def _(bot: Bot):
+    for _ in range(3):
+        await cmd_guaihuas.send(await UniMessage().text(fortune_wtf.choice()).export(bot))
+        await asyncio.sleep(1)
+
+
+cmd_filter_guaihua = on_alconna(Alconna(
+    "怪话过滤",
+    Args["keyword", str],
+), rule=is_admin)
+
+@cmd_filter_guaihua.handle()
+async def _(keyword: str, bot: Bot):
+    async with fortune_insert_lock:
+        c = fortune_wtf.filter_out(keyword)
+    await cmd_filter_guaihua.send(await UniMessage().text(f"删除了 {c} 条怪话").export(bot))
+

konabot/plugins/fortune/random_text_record.py

@@ -0,0 +1,70 @@
+import base64
+import random
+import time
+
+from pathlib import Path
+
+
+class RandomTextManager:
+    _cache: list[tuple[float, str]]
+
+    def __init__(self, fp: Path) -> None:
+        self.fp = fp
+        self._cache = []
+        if not self.fp.exists():
+            self.fp.touch()
+        else:
+            self.load()
+
+    def load(self):
+        self._cache = []
+        with self.fp.open("r") as f:
+            for line in f.readlines():
+                if not line.strip():
+                    continue
+                if "|" not in line:
+                    continue
+                ts, cn = line.split("|")
+                try:
+                    ts = float(ts)
+                except Exception:
+                    continue
+                self._cache.append((ts, base64.b64decode(cn).decode("utf-8")))
+
+    def save(self):
+        lines = [
+            str(ts) + "|" + base64.b64encode(cn.encode("utf-8")).decode()
+            for ts, cn in self._cache
+        ]
+        with self.fp.open("w") as f:
+            f.writelines(lines)
+
+    def insert(self, text: str, timestamp: float | None = None):
+        if timestamp is None:
+            timestamp = time.time()
+        with self.fp.open("a") as f:
+            f.write(str(timestamp) + "|" + base64.b64encode(text.encode("utf-8")).decode() + "\n")
+        self._cache.append((timestamp, text))
+
+    def choice(self, now: float | None = None):
+        contents: list[str] = []
+        weights: list[float] = []
+
+        if now is None:
+            now = time.time()
+
+        for ts, cn in self._cache:
+            contents.append(cn)
+            weights.append((abs(now - ts) + 0.01) ** (-1))
+
+        return random.choices(contents, weights)[0]
+
+    def filter_out(self, keyword: str):
+        len1 = len(self._cache)
+        self._cache = [
+            (ts, cn) for ts, cn in self._cache
+            if keyword not in cn
+        ]
+        self.save()
+        return len1 - len(self._cache)
+

konabot/plugins/idiomgame/__init__.py

@@ -68,17 +68,19 @@ class TryStopState(Enum):
 
 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 = {}  # 成语首字字典
+    IDIOM_FIRST_CHAR = {}  # 所有成语包括词语的首字字典
+    AVALIABLE_IDIOM_FIRST_CHAR = {} # 真正有效的成语首字字典
 
     __inited = False
 
@@ -181,7 +183,7 @@ class IdiomGame:
         """
         判断是否有成语可以接
         """
-        return last_char in IdiomGame.IDIOM_FIRST_CHAR
+        return last_char in IdiomGame.AVALIABLE_IDIOM_FIRST_CHAR
 
     def _verify_idiom(self, idiom: str, user_id: str) -> TryVerifyState:
         # 新成语的首字应与上一条成语的尾字相同
@@ -193,6 +195,8 @@ class IdiomGame:
         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:
             self.now_playing = False
@@ -201,6 +205,8 @@ class IdiomGame:
             # 没有成语可以接了，自动跳过
             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) -> float:
@@ -223,6 +229,13 @@ class IdiomGame:
 
     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):
@@ -288,6 +301,12 @@ class IdiomGame:
                 cls.IDIOM_FIRST_CHAR[idiom[0]] = []
             cls.IDIOM_FIRST_CHAR[idiom[0]].append(idiom)
 
+        # 根据真正的成语大表，划分出有效成语首字字典
+        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(
@@ -421,7 +440,8 @@ async def _(target: DepLongTaskTarget):
     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()
@@ -458,16 +478,24 @@ async def _(event: BaseEvent, msg: UniMsg, target: DepLongTaskTarget):
         await evt.send(
             await UniMessage()
             .at(user_id)
-            .text("接不上！这个不一样！你被扣了 0.1 分！")
+            .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:
+        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)

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/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/no_luowen.py

@@ -0,0 +1,44 @@
+import nonebot
+
+from nonebot.adapters.onebot.v11.bot import Bot
+from nonebot.adapters.onebot.v11.event import GroupMessageEvent
+from nonebot_plugin_alconna import UniMsg, UniMessage
+from pydantic import BaseModel
+
+
+class NoLuowenConfig(BaseModel):
+    plugin_noluowen_qqid: int               = -1
+    plugin_noluowen_enable_group: list[int] = []
+
+config = nonebot.get_plugin_config(NoLuowenConfig)
+
+
+async def is_luowen_mentioned(evt: GroupMessageEvent, msg: UniMsg) -> bool:
+    if config.plugin_noluowen_qqid <= 0:
+        return False
+    if evt.user_id == config.plugin_noluowen_qqid:
+        return False
+    if evt.group_id not in config.plugin_noluowen_enable_group:
+        return False
+    txt = msg.extract_plain_text()
+    if "洛温" not in txt:
+        return False
+    if "罗文" in txt:
+        return False
+    if "阿特金森" in txt:
+        return False
+    return True
+
+evt_luowen_mentioned = nonebot.on_message(rule=is_luowen_mentioned)
+
+
+@evt_luowen_mentioned.handle()
+async def _(evt: GroupMessageEvent, bot: Bot):
+    msg = (
+        UniMessage()
+        .reply(str(evt.message_id))
+        .at(str(config.plugin_noluowen_qqid))
+        .text(" 好像有人念错了你的 ID")
+    )
+    await evt_luowen_mentioned.send(await msg.export(bot=bot))
+

konabot/plugins/simple_notify/__init__.py

@@ -1,10 +1,8 @@
 import asyncio as asynkio
 import datetime
-import functools
 from pathlib import Path
 from typing import Any, Literal, cast
 
-import signal
 import nonebot
 import ptimeparse
 from loguru import logger
@@ -123,7 +121,9 @@ def create_notify_task(notify: Notify, fail2remove: bool = True):
             try:
                 await asynkio.sleep((notify.notify_time - begin_time).total_seconds())
             except asynkio.CancelledError:
-                logger.debug("代办提醒被信号中止，任务退出")
+                logger.debug(
+                    f"代办提醒被信号中止，任务退出 NOTIFY={notify.notify_msg} TIME={notify.notify_time}"
+                )
                 return
         else:
             logger.warning(
@@ -254,19 +254,3 @@ async def _():
     save_notify_config(cfg)
     DATA_FILE_LOCK.release()
 
-    loop = asynkio.get_running_loop()
-
-    # 解决 asynk task 没有被 cancel 的问题
-    async def shutdown(sig: signal.Signals):
-        logger.info(f"收到 {sig.name} 指令，正在关闭所有的东西")
-        for task in ASYNK_TASKS:
-            task.cancel()
-        await asynkio.gather(*ASYNK_TASKS, return_exceptions=True)
-        logger.info("所有的代办提醒 Task 都已经退出了")
-
-    for sig in (signal.SIGINT, signal.SIGTERM):
-        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.2.0"
+version = "0.2.1"
 description = "一个用于解析中文的时间表达的库"
 optional = false
 python-versions = ">=3.9"
 groups = ["main"]
 files = [
-    {file = "ptimeparse-0.2.0-py3-none-any.whl", hash = "sha256:57055f8fd99fb69e19deac3b8a5c7ac91af86c7ac09781632e9abf318df0d6d2"},
-    {file = "ptimeparse-0.2.0.tar.gz", hash = "sha256:867c265f2e157fe4d793d20fe9c449b8ede5c855f336d7e6b2eb78551e622766"},
+    {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 = "02530953efe65da1a788845cd43f8856be62db5bfb59de691cad813f57bab25e"
+content-hash = "78a299c64ba07999fae807300b10a1c622d45b8b387aded5a34d17cf5550e777"

pyproject.toml

@@ -20,10 +20,10 @@ dependencies = [
   "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,<1.0.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]
@@ -41,4 +41,3 @@ url = "https://pypi.tuna.tsinghua.edu.cn/simple/"
 priority = "primary"
 
 [tool.poetry.dependencies]
-ptimeparse = { source = "pt-gitea-pypi" }