263 lines
7.9 KiB
Python
263 lines
7.9 KiB
Python
import ctypes
|
|
import json
|
|
import os
|
|
import sys
|
|
from dataclasses import field, dataclass
|
|
from typing import Optional, Dict, Any, List, Union, TypeVar, Generic, Callable
|
|
|
|
|
|
def get_lib_filename(base_name):
|
|
if sys.platform == "win32":
|
|
return f"{base_name}.dll"
|
|
elif sys.platform == "darwin":
|
|
return f"lib{base_name}.dylib"
|
|
else:
|
|
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":
|
|
return cls(
|
|
text=data.get("text", ""),
|
|
# 兼容 Rust 中 PathBuf 序列化出的路径字符串列表
|
|
path=[str(p) for p in data.get("path", [])],
|
|
)
|
|
|
|
|
|
@dataclass
|
|
class FetchResultEvent:
|
|
event_id: int = 0
|
|
cards: List[FetchResultCard] = field(default_factory=list)
|
|
|
|
@classmethod
|
|
def from_dict(cls, data: Dict[str, Any]) -> "FetchResultEvent":
|
|
return cls(
|
|
event_id=data.get("event_id", 0),
|
|
cards=[
|
|
FetchResultCard.from_dict(card)
|
|
for card in data.get("cards", [])
|
|
],
|
|
)
|
|
|
|
|
|
@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", [])
|
|
],
|
|
)
|
|
|
|
|
|
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")))
|
|
self._setup_signatures()
|
|
|
|
def _setup_signatures(self):
|
|
# proxy(x: *const c_char)
|
|
self.lib.proxy.argtypes = [ctypes.c_char_p]
|
|
self.lib.proxy.restype = None
|
|
|
|
# pjsk_proxy(x: *const c_char)
|
|
self.lib.pjsk_proxy.argtypes = [ctypes.c_char_p]
|
|
self.lib.pjsk_proxy.restype = None
|
|
|
|
# ghp(x: *const c_char)
|
|
self.lib.ghp.argtypes = [ctypes.c_char_p]
|
|
self.lib.ghp.restype = None
|
|
|
|
# ghp(x: *const c_char)
|
|
self.lib.log_level.argtypes = [ctypes.c_char_p]
|
|
self.lib.log_level.restype = None
|
|
|
|
# boot(cwd: *const c_char, name: *const c_char, as_path: *const c_char) -> c_int
|
|
self.lib.boot.argtypes = [ctypes.c_char_p, ctypes.c_char_p, ctypes.c_char_p]
|
|
self.lib.boot.restype = ctypes.c_int
|
|
|
|
# stop() -> c_int
|
|
self.lib.stop.argtypes = []
|
|
self.lib.stop.restype = ctypes.c_int
|
|
|
|
# fetch_data() -> *mut c_char
|
|
# ⚠️ 这里必须用 c_void_p，不能用 c_char_p，否则会丢失指针地址导致内存泄漏
|
|
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
|
|
|
|
def _str_to_bytes(self, s: Optional[str]) -> bytes:
|
|
"""辅助函数：将 Python 字符串安全转换为 C 字符串字节"""
|
|
if s is None:
|
|
# 根据 Rust 端解析逻辑，传入空字符串将被视为 None
|
|
return b""
|
|
return s.encode('utf-8')
|
|
|
|
def set_proxy(self, proxy_url: Optional[str]):
|
|
"""设置代理，传 None 或空字符串可清除"""
|
|
self.lib.proxy(self._str_to_bytes(proxy_url))
|
|
|
|
def set_pjsk_proxy(self, proxy_url: Optional[str]):
|
|
"""设置 pjsk_proxy，传 None 或空字符串可清除"""
|
|
self.lib.pjsk_proxy(self._str_to_bytes(proxy_url))
|
|
|
|
def set_ghp(self, token: Optional[str]):
|
|
"""设置 GitHub Token，传 None 或空字符串可清除"""
|
|
self.lib.ghp(self._str_to_bytes(token))
|
|
|
|
def set_log_level(self, log_evel: Optional[str]):
|
|
"""设置 Log level，传 None 或空字符串可清除"""
|
|
self.lib.log_level(self._str_to_bytes(log_evel))
|
|
|
|
def boot(self, cwd: str, name: str, as_path: str) -> int:
|
|
"""
|
|
启动主循环与初始化状态
|
|
"""
|
|
return self.lib.boot(
|
|
self._str_to_bytes(cwd),
|
|
self._str_to_bytes(name),
|
|
self._str_to_bytes(as_path)
|
|
)
|
|
|
|
def stop(self) -> int:
|
|
"""
|
|
停止主循环与清理状态
|
|
"""
|
|
return self.lib.stop()
|
|
|
|
def fetch_data(self) -> None | list[FetchResultEvent | FetchResultUnlinkedCards]:
|
|
# 1. 获取裸指针
|
|
ptr = self.lib.fetch_data()
|
|
|
|
# 指针为空 (std::ptr::null_mut)
|
|
if not ptr:
|
|
return None
|
|
|
|
try:
|
|
# 2. 将 void_p 转换回 c_char_p 并读取里面的 bytes
|
|
json_bytes = ctypes.cast(ptr, ctypes.c_char_p).value
|
|
if not json_bytes:
|
|
return None
|
|
|
|
# 3. 解码并反序列化 JSON
|
|
json_str = json_bytes.decode('utf-8')
|
|
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 内存！
|
|
self.lib.free_string(ptr)
|
|
|
|
|
|
rlib = SekaiSyncLib()
|