更少的量，更解耦的数据，更健壮的系统

konabot/common/longtask.py

@@ -51,6 +51,10 @@ class LongTaskTarget(BaseModel):
     target_id: str
     "沟通对象的 ID"
 
+    @property
+    def is_private_chat(self):
+        return self.channel_id.startswith(QQ_PRIVATE_CHAT_CHANNEL_PREFIX)
+
     async def send_message(self, msg: UniMessage | str, at: bool = True) -> bool:
         try:
             bot = nonebot.get_bot(self.self_id)

konabot/docs/sys/konaph.txt

@@ -0,0 +1,4 @@
+指令介绍
+    konaph - KonaBot 的 PuzzleHunt 管理工具
+
+详细介绍请直接输入 konaph 获取使用指引（该指令权限仅对部分人开放。如果你有权限的话才有响应。建议在此方 BOT 私聊使用该指令。）

konabot/plugins/kona_ph/__init__.py

@@ -13,15 +13,15 @@ create_admin_commands()
 
 
 async def is_play_group(target: DepLongTaskTarget):
-    if target.channel_id in config.plugin_puzzle_playgroup:
+    if target.is_private_chat:
         return True
-    if target.target_id in target.channel_id:
+    if target.channel_id in config.plugin_puzzle_playgroup:
         return True
     return False
 
 
 cmd_submit = on_alconna(Alconna(
-    "提交答案",
+    "re:提交(?:答案|题解|[fF]lag)",
     Args["flag", str],
 ), rule=is_play_group)
 
@@ -33,7 +33,7 @@ async def _(flag: str, target: DepLongTaskTarget):
 
 
 cmd_query = on_alconna(Alconna(
-    r"re:(?:(?:所以|话)说?)?今天的题目是什么[啊呀哇呢]?(?:\?？)?"
+    r"re:(?:((?:(?:所以|话)说?)?今天的题目是什么[啊呀哇呢]?(?:\?？)?)|今日谜?题目?)"
 ), rule=is_play_group)
 
 @cmd_query.handle()
@@ -92,6 +92,10 @@ async def _(target: DepLongTaskTarget, index_id: str = "", page: int = 1):
 @scheduler.scheduled_job("cron", hour="8")
 async def _():
     async with puzzle_manager() as manager:
+        msg2 = manager.get_report_yesterday()
+        if msg2 is not None:
+            await qq_broadcast(config.plugin_puzzle_playgroup, msg2)
+
         puzzle = manager.get_today_puzzle()
         if puzzle is not None:
             logger.info(f"找到了题目 {puzzle.raw_id}，发送")
@@ -99,3 +103,4 @@ async def _():
         else:
             logger.info("自动任务：没有找到题目，跳过")
 
+

konabot/plugins/kona_ph/core/storage.py

@@ -37,9 +37,6 @@ class Puzzle(BaseModel):
     flag: str
 
     ready: bool = False
-    published: bool = False
-    pinned: bool = False
-
     created_at: datetime.datetime = Field(default_factory=datetime.datetime.now)
 
     def get_image_path(self) -> Path:
@@ -53,6 +50,7 @@ class Puzzle(BaseModel):
             result = result.text("\n\n").image(raw=self.get_image_path().read_bytes())
 
         result = result.text("\n\n出题者：").at(self.author_id)
+        result = result.text("\n\n输入「提交答案 答案」来提交你的解答")
 
         return result
 
@@ -110,68 +108,81 @@ class PuzzleManager(BaseModel):
     daily_puzzle_of_date: dict[datetime.date, str] = {}
 
     puzzle_pinned: str = ""
-    unpublished_puzzles: set[str] = set()
-    unready_puzzles: set[str] = set()
-    published_puzzles: set[str] = set()
 
     index_id_counter: int = 1
     submissions: dict[str, dict[str, list[PuzzleSubmission]]] = {}
-    last_pubish_date: datetime.date = Field(
-        default_factory=lambda: get_today_date() - datetime.timedelta(days=1)
-    )
     last_checked_date: datetime.date = Field(
         default_factory=lambda: get_today_date() - datetime.timedelta(days=1)
     )
 
+    @property
+    def last_publish_date(self):
+        return max(self.daily_puzzle_of_date.keys())
+
+    @property
+    def unpublished_puzzles(self):
+        return set((
+            p.raw_id for p in self.puzzle_data.values()
+            if not self.is_puzzle_published(p.raw_id) and p.ready
+        ))
+
+    @property
+    def unready_puzzles(self):
+        return set((
+            p.raw_id for p in self.puzzle_data.values()
+            if not self.is_puzzle_published(p.raw_id) and not p.ready
+        ))
+
+    @property
+    def published_puzzles(self):
+        return set((
+            p.raw_id for p in self.puzzle_data.values()
+            if self.is_puzzle_published(p.raw_id)
+        ))
+
+    def is_puzzle_published(self, raw_id: str):
+        return raw_id in [i.raw_id for i in self.daily_puzzle.values()]
+
     def publish_puzzle(self, raw_id: str):
         assert raw_id in self.puzzle_data
 
-        self.unpublished_puzzles -= set(raw_id)
+        today = get_today_date()
-        self.unready_puzzles -= set(raw_id)
+
         p = self.puzzle_data[raw_id]
         p.index_id = str(self.index_id_counter)
         p.ready = True
-        p.published = True
-        p.pinned = False
         self.puzzle_pinned = ""
-        self.last_pubish_date = get_today_date()
+        self.last_checked_date = today
-        self.last_checked_date = self.last_pubish_date
         self.daily_puzzle[p.index_id] = DailyPuzzleInfo(
             raw_id=raw_id,
-            time=self.last_pubish_date,
+            time=today,
         )
-        self.daily_puzzle_of_date[self.last_pubish_date] = p.index_id
+        self.daily_puzzle_of_date[today] = p.index_id
-        self.published_puzzles.add(raw_id)
 
         self.index_id_counter += 1
 
-    def admin_mark_ready(self, raw_id: str, ready: bool = True):
+    def fix(self):
-        if raw_id not in self.puzzle_data:
+        # 尝试修复今日的数据
-            return
+        for p in self.puzzle_data.values():
-        if ready:
+            if self.is_puzzle_published(p.raw_id):
-            self.unready_puzzles -= set(raw_id)
-            if raw_id not in self.published_puzzles:
-                self.unpublished_puzzles.add(raw_id)
-            p = self.puzzle_data[raw_id]
                 p.ready = True
-            p.published = raw_id in self.published_puzzles
+
+        if self.puzzle_pinned not in self.unpublished_puzzles:
+            self.puzzle_pinned = ""
+
+        # 撤回重复发布的题
+        already_published: set[str] = set()
+        for date in list(self.daily_puzzle_of_date.keys()):
+            index_id = self.daily_puzzle_of_date[date]
+            info = self.daily_puzzle[index_id]
+            if info.raw_id in already_published:
+                del self.daily_puzzle[index_id]
+                del self.daily_puzzle_of_date[date]
             else:
-            self.unready_puzzles.add(raw_id)
+                already_published.add(info.raw_id)
-            self.unpublished_puzzles -= set(raw_id)
-            p = self.puzzle_data[raw_id]
-            p.ready = False
-            p.published = False
-            # if p.raw_id == self.puzzle_pinned:
-            #     self.puzzle_pinned = ""
 
     def admin_pin_puzzle(self, raw_id: str):
-        if self.puzzle_pinned:
-            p = self.puzzle_data.get(self.puzzle_pinned)
-            if p is not None:
-                p.pinned = False
         if raw_id in self.puzzle_data:
-            p = self.puzzle_data[raw_id]
-            p.pinned = True
             self.puzzle_pinned = raw_id
         else:
             self.puzzle_pinned = ""
@@ -251,9 +262,7 @@ class PuzzleManager(BaseModel):
             author_id=user,
             flag="konaph{this_is_a_flag}",
             ready=False,
-            published=False,
         )
-        self.unready_puzzles.add(p.raw_id)
         self.puzzle_data[p.raw_id] = p
         return p
 

konabot/plugins/kona_ph/manager.py

@@ -8,7 +8,7 @@ from pydantic import BaseModel
 from konabot.common.longtask import DepLongTaskTarget
 from konabot.common.nb.extract_image import download_image_bytes
 from konabot.common.nb.qq_broadcast import qq_broadcast
-from konabot.plugins.kona_ph.core.storage import Puzzle, get_today_date, puzzle_manager
+from konabot.plugins.kona_ph.core.storage import Puzzle, PuzzleManager, get_today_date, puzzle_manager
 
 PUZZLE_PAGE_SIZE = 10
 
@@ -30,12 +30,12 @@ def is_puzzle_admin(target: DepLongTaskTarget):
     return target.target_id in config.plugin_puzzle_admin
 
 
-def get_puzzle_info_message(puzzle: Puzzle) -> UniMessage[Any]:
+def get_puzzle_info_message(manager: PuzzleManager, puzzle: Puzzle) -> UniMessage[Any]:
-    status = "✅ 已准备，待发布" if puzzle.ready and not puzzle.published else \
+    status = "✅ 已准备，待发布" if puzzle.ready and not manager.is_puzzle_published(puzzle.raw_id) else \
-    (f"🟢 已发布: #{puzzle.index_id}" if puzzle.published else "⚙️ 未准备")
+    (f"🟢 已发布: #{puzzle.index_id}" if manager.is_puzzle_published(puzzle.raw_id) else "⚙️ 未准备")
 
     status_suffix = ""
-    if puzzle.pinned:
+    if puzzle.raw_id == manager.puzzle_pinned:
         status_suffix += " | 📌 已被管理员置顶"
 
     msg = UniMessage.text(
@@ -130,7 +130,7 @@ def create_admin_commands():
                 return await target.send_message(UniMessage.text(
                     "题目早就准备好啦！"
                 ))
-            manager.admin_mark_ready(raw_id, True)
+            p.ready = True
             await target.send_message(UniMessage.text(
                 f"谜题「{p.title}」已经准备就绪！"
             ))
@@ -151,12 +151,12 @@ def create_admin_commands():
                 return await target.send_message(UniMessage.text(
                     f"谜题「{p.title}」已经是未取消状态了！"
                 ))
-            if p.published:
+            if manager.is_puzzle_published(p.raw_id):
                 return await target.send_message(UniMessage.text(
                     "已发布的谜题不能取消准备状态！"
                 ))
 
-            manager.admin_mark_ready(raw_id, False)
+            p.ready = False
             await target.send_message(UniMessage.text(
                 f"谜题「{p.title}」已经取消准备！"
             ))
@@ -174,7 +174,7 @@ def create_admin_commands():
                     "这不是你的题，你没有权限查看详细信息！"
                 ))
 
-            await target.send_message(get_puzzle_info_message(p))
+            await target.send_message(get_puzzle_info_message(manager, p))
 
     @cmd_admin.assign("my")
     async def _(target: DepLongTaskTarget, page: int = 1):
@@ -193,9 +193,9 @@ def create_admin_commands():
             message = UniMessage.text("==== 我的谜题 ====\n\n")
             for p in puzzles:
                 message = message.text("- ")
-                if p.pinned:
+                if manager.puzzle_pinned == p.raw_id:
                     message = message.text("[📌]")
-                if p.published:
+                if manager.is_puzzle_published(p.raw_id):
                     message = message.text(f"[#{p.index_id}] ")
                 elif p.ready:
                     message = message.text("[✅] ")
@@ -207,7 +207,7 @@ def create_admin_commands():
             await target.send_message(message)
 
     @cmd_admin.assign("all")
-    async def _(target: DepLongTaskTarget, ready: Query[bool] = Query("ready"), page: int = 1):
+    async def _(target: DepLongTaskTarget, ready: Query[bool] = Query("all.ready"), page: int = 1):
         if not is_puzzle_admin(target):
             return await target.send_message(UniMessage.text("你没有权限查看所有的哦"))
         async with puzzle_manager() as manager:
@@ -224,9 +224,9 @@ def create_admin_commands():
             message = UniMessage.text("==== 所有谜题 ====\n\n")
             for p in puzzles:
                 message = message.text("- ")
-                if p.pinned:
+                if p.raw_id == manager.puzzle_pinned:
                     message = message.text("[📌]")
-                if p.published:
+                if manager.is_puzzle_published(p.raw_id):
                     message = message.text(f"[#{p.index_id}] ")
                 elif p.ready:
                     message = message.text("[✅] ")
@@ -276,7 +276,7 @@ def create_admin_commands():
         description: str | None = None,
         flag: str | None = None,
         image: Image | None = None,
-        remove_image: Query[bool] = Query("--remove-image"),
+        remove_image: Query[bool] = Query("modify.remove-image"),
     ):
         if raw_id == "":
             return await target.send_message(
@@ -307,7 +307,7 @@ def create_admin_commands():
             elif remove_image.available:
                 p.remove_image()
 
-            info2 = get_puzzle_info_message(p)
+            info2 = get_puzzle_info_message(manager, p)
 
             return await target.send_message("修改好啦！看看效果：\n\n" + info2)