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

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/plugins/kona_ph/__init__.py

@@ -13,9 +13,9 @@ 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
 
@@ -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()

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:
@@ -111,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)
-        self.unready_puzzles -= set(raw_id)
+        today = get_today_date()
+
         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 = self.last_pubish_date
+        self.last_checked_date = today
         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.published_puzzles.add(raw_id)
+        self.daily_puzzle_of_date[today] = p.index_id
 
         self.index_id_counter += 1
 
-    def admin_mark_ready(self, raw_id: str, ready: bool = True):
-        if raw_id not in self.puzzle_data:
-            return
-        if ready:
-            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
-        else:
-            self.unready_puzzles.add(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 fix(self):
+        # 尝试修复今日的数据
+        for p in self.puzzle_data.values():
+            if self.is_puzzle_published(p.raw_id):
+                p.ready = True
+
+        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:
+                already_published.add(info.raw_id)
 
     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 = ""
@@ -252,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]:
-    status = "✅ 已准备，待发布" if puzzle.ready and not puzzle.published else \
-    (f"🟢 已发布: #{puzzle.index_id}" if puzzle.published else "⚙️ 未准备")
+def get_puzzle_info_message(manager: PuzzleManager, puzzle: Puzzle) -> UniMessage[Any]:
+    status = "✅ 已准备，待发布" if puzzle.ready and not manager.is_puzzle_published(puzzle.raw_id) 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("[✅] ")
@@ -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("[✅] ")
@@ -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)