非event卡推送，以及手动查看

__init__.py

@@ -10,6 +10,8 @@ from .config import pconfig
 from .lib import rlib
 from .sync import sync
 
+from . import command
+
 driver = nonebot.get_driver()
 
 

command.py

@@ -0,0 +1,47 @@
+from pathlib import Path
+
+from nonebot import Bot, logger
+from nonebot.adapters.onebot.v11 import Message, MessageSegment
+from nonebot.internal.matcher import Matcher
+from nonebot.plugin.on import on_command
+
+from . import rlib, pconfig
+
+handler = on_command("烤leak最近活动")
+
+
+@handler.handle()
+async def _(matcher: Matcher, bot: Bot):
+    try:
+        data = rlib.refetch_event().unwrap()
+        message = Message()
+        inner = [MessageSegment.text(f"活动 {data.event_id} 的卡牌数据")]
+        if len(data.cards) == 0:
+            inner.append(MessageSegment.text("没有新的卡牌。"))
+        # else
+        for card in data.cards:
+            inner.append(MessageSegment.text(card.text))
+            for item in card.path:
+                p = Path(item)
+                n = MessageSegment.image(file=item)
+                n.data["summary"] = f"[{p.name}]"
+                inner.append(n)
+        for node in inner:
+            message.append(MessageSegment.node_custom(
+                user_id=int(bot.self_id),
+                nickname=pconfig.nickname,
+                content=Message([node])
+            ))
+
+        await matcher.send(message)
+
+        for card in data.cards:
+            for path in card.path:
+                try:
+                    Path(path).unlink()
+                except Exception as e:
+                    logger.error(f"删除文件 {path} 失败: {e}")
+    except Exception as e:
+        logger.error(f"获取活动数据失败: {e}")
+        await matcher.send(f"获取活动数据失败: {e}")
+        return

lib.py

@@ -3,7 +3,7 @@ import json
 import os
 import sys
 from dataclasses import field, dataclass
-from typing import Optional, Dict, Any, List
+from typing import Optional, Dict, Any, List, Union, TypeVar, Generic, Callable
 
 
 def get_lib_filename(base_name):
@@ -15,26 +15,30 @@ def get_lib_filename(base_name):
         return f"lib{base_name}.so"
 
 
+T = TypeVar("T")
+
+
 @dataclass
 class FetchResultCard:
     text: str = ""
     path: List[str] = field(default_factory=list)
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> 'FetchResultCard':
+    def from_dict(cls, data: Dict[str, Any]) -> "FetchResultCard":
         return cls(
             text=data.get("text", ""),
-            path=data.get("path", []),
+            # 兼容 Rust 中 PathBuf 序列化出的路径字符串列表
+            path=[str(p) for p in data.get("path", [])],
         )
 
 
 @dataclass
-class FetchResult:
+class FetchResultEvent:
     event_id: int = 0
     cards: List[FetchResultCard] = field(default_factory=list)
 
     @classmethod
-    def from_dict(cls, data: Dict[str, Any]) -> 'FetchResult':
+    def from_dict(cls, data: Dict[str, Any]) -> "FetchResultEvent":
         return cls(
             event_id=data.get("event_id", 0),
             cards=[
@@ -44,6 +48,95 @@ class FetchResult:
         )
 
 
+@dataclass
+class FetchResultUnlinkedCards:
+    cards: List[FetchResultCard] = field(default_factory=list)
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> "FetchResultUnlinkedCards":
+        return cls(
+            cards=[
+                FetchResultCard.from_dict(card)
+                for card in data.get("cards", [])
+            ],
+        )
+
+
+# 对应 Rust 中的 FetchResult 枚举
+FetchResultType = Union[FetchResultEvent, FetchResultUnlinkedCards]
+
+
+class FetchResult:
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any]) -> FetchResultType:
+        tag = data.get("type")
+        content = data.get("data", {})
+
+        if tag == "Event":
+            return FetchResultEvent.from_dict(content)
+        elif tag == "UnlinkedCard":
+            return FetchResultUnlinkedCards.from_dict(content)
+        else:
+            raise ValueError(f"Unknown FetchResult type: {tag}")
+
+
+@dataclass
+class ApiResultSuccess(Generic[T]):
+    data: T
+
+    @staticmethod
+    def is_ok() -> bool:
+        return True
+
+    @staticmethod
+    def is_err() -> bool:
+        return False
+
+    def unwrap(self) -> T:
+        return self.data
+
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any], parser: Callable[[Any], T] | None = None) -> "ApiResultSuccess[T]":
+        data = parser(data) if parser else data
+        return cls(data=data)
+
+
+@dataclass
+class ApiResultError(Generic[T]):
+    error: str
+
+    @staticmethod
+    def is_ok() -> bool:
+        return False
+
+    @staticmethod
+    def is_err() -> bool:
+        return True
+
+    def unwrap(self) -> T:
+        raise Exception(f"ApiResultError: {self.error}")
+
+    @classmethod
+    def from_dict(cls, data: str) -> "ApiResultError[T]":
+        return cls(error=data)
+
+
+class ApiResult(Generic[T]):
+    @classmethod
+    def from_dict(cls, data: Dict[str, Any], parser: Callable[[Any], T] | None = None) -> ApiResultSuccess[T] | \
+                                                                                          ApiResultError[T]:
+        tag = data.get("type")
+        content = data.get("data", {})
+
+        if tag == "Ok":
+            return ApiResultSuccess.from_dict(content, parser=parser)
+        elif tag == "Err":
+            return ApiResultError.from_dict(content)
+        else:
+            raise ValueError(f"Unknown ApiResult type: {tag}")
+
+
 class SekaiSyncLib:
     def __init__(self):
         self.lib = ctypes.CDLL(os.path.join(os.path.dirname(__file__), get_lib_filename("sekai_sync_lib")))
@@ -79,6 +172,9 @@ class SekaiSyncLib:
         self.lib.fetch_data.argtypes = []
         self.lib.fetch_data.restype = ctypes.c_void_p
 
+        self.lib.refetch_event.argtypes = []
+        self.lib.refetch_event.restype = ctypes.c_void_p
+
         # free_string(s: *mut c_char)
         self.lib.free_string.argtypes = [ctypes.c_void_p]
         self.lib.free_string.restype = None
@@ -122,7 +218,7 @@ class SekaiSyncLib:
         """
         return self.lib.stop()
 
-    def fetch_data(self) -> Optional[FetchResult]:
+    def fetch_data(self) -> None | list[FetchResultEvent | FetchResultUnlinkedCards]:
         # 1. 获取裸指针
         ptr = self.lib.fetch_data()
 
@@ -138,7 +234,26 @@ class SekaiSyncLib:
 
             # 3. 解码并反序列化 JSON
             json_str = json_bytes.decode('utf-8')
-            return FetchResult.from_dict(json.loads(json_str))
+            lst = json.loads(json_str)
+            return [FetchResult.from_dict(item) for item in lst]
+
+        finally:
+            # 4. 无论是否发生 JSON 解析异常，绝对保证释放 Rust 内存！
+            self.lib.free_string(ptr)
+
+    def refetch_event(self) -> ApiResultSuccess[FetchResultEvent] | ApiResultError[FetchResultEvent]:
+        # 1. 获取裸指针
+        ptr = self.lib.refetch_event()
+
+        try:
+            # 2. 将 void_p 转换回 c_char_p 并读取里面的 bytes
+            json_bytes = ctypes.cast(ptr, ctypes.c_char_p).value
+            if not json_bytes:
+                return ApiResultError(error="调用失败")
+
+            # 3. 解码并反序列化 JSON
+            json_str = json_bytes.decode('utf-8')
+            return ApiResult.from_dict(json.loads(json_str), parser=FetchResultEvent.from_dict)
 
         finally:
             # 4. 无论是否发生 JSON 解析异常，绝对保证释放 Rust 内存！

sync.py

@@ -8,7 +8,7 @@ from nonebot.permission import SUPERUSER
 from nonebot.plugin.on import on_command
 
 from .config import pconfig
-from .lib import rlib
+from .lib import rlib, FetchResultEvent
 
 group_file = Path("config") / "nonebot_sekai_update_notify" / "group.json"
 group_file.parent.mkdir(parents=True, exist_ok=True)
@@ -34,9 +34,16 @@ async def sync():
     data = rlib.fetch_data()
     if data is None:
         return
+    for component in data:
         message = Message()
-    inner = [MessageSegment.text(f"活动 {data.event_id} 的卡牌更新了！")]
+        if component is FetchResultEvent:
-    for card in data.cards:
+            inner = [MessageSegment.text(f"活动 {component.event_id} 更新了！")]
+        else:
+            inner = [MessageSegment.text("发现了新的非活动卡牌！")]
+        if len(component.cards) == 0:
+            inner.append(MessageSegment.text("但没有新的卡牌。"))
+        # else
+        for card in component.cards:
             inner.append(MessageSegment.text(card.text))
             for item in card.path:
                 p = Path(item)
@@ -56,7 +63,7 @@ async def sync():
             except Exception as e:
                 logger.error(f"发送消息到群组 {group} 失败: {e}")
 
-    for card in data.cards:
+        for card in component.cards:
             for path in card.path:
                 try:
                     Path(path).unlink()