✨ 适配ceobecanteen平台

Co-authored-by: phidiaLam <[email protected]>

Author: AzideCupric

nonebot_bison/admin_page/token_manager.py

@@ -1,12 +1,13 @@
 import random
 import string
+from datetime import timedelta
 
-from expiringdict import ExpiringDict
+from expiringdictx import ExpiringDict
 
 
 class TokenManager:
     def __init__(self):
-        self.token_manager = ExpiringDict(max_len=100, max_age_seconds=60 * 10)
+        self.token_manager = ExpiringDict[str, tuple](capacity=100, default_age=timedelta(minutes=10))
 
     def get_user(self, token: str) -> tuple | None:
         res = self.token_manager.get(token)

nonebot_bison/platform/arknights.py

@@ -6,6 +6,7 @@
 from pydantic import Field, BaseModel
 
 from ..post import Post
+from ..utils import text_fletten
 from ..types import Target, RawPost, Category
 from .platform import NewMessage, StatusChange
 from ..utils.scheduler_config import SchedulerConfig
@@ -93,17 +94,14 @@ async def parse(self, raw_post: BulletinListItem) -> Post:
         )
         data = ArkBulletinResponse.parse_obj(raw_data.json()).data
 
-        def title_escape(text: str) -> str:
-            return text.replace("\n", " - ")
-
         # gen title, content
         if data.header:
             # header是title的更详细版本
             # header会和content一起出现
-            title = data.header
+            title = text_fletten(data.header, replace=" - ")
         else:
             # 只有一张图片
-            title = title_escape(data.title)
+            title = text_fletten(data.title, replace=" - ")
 
         return Post(
             self,

nonebot_bison/platform/ceobecanteen/__init__.py

@@ -0,0 +1 @@
+from .platform import CeobeCanteen as CeobeCanteen

nonebot_bison/platform/ceobecanteen/cache.py

@@ -0,0 +1,99 @@
+from functools import partial
+from datetime import timedelta
+from types import MappingProxyType
+from collections.abc import Callable
+
+from expiringdictx import ExpiringDict
+from httpx import AsyncClient, AsyncHTTPTransport
+from hishel import Controller, AsyncCacheTransport, AsyncInMemoryStorage
+
+from .const import DATASOURCE_URL
+from .utils import process_response
+from .models import CeobeSource, CeobeTarget, DataSourceResponse
+
+cache_transport = AsyncCacheTransport(
+    AsyncHTTPTransport(),
+    storage=AsyncInMemoryStorage(),
+    controller=Controller(
+        always_revalidate=True,
+    ),
+)
+
+CeobeClient = partial(
+    AsyncClient,
+    transport=cache_transport,
+)
+
+
+class CeobeDataSourceCache:
+    """数据源缓存"""
+
+    def __init__(self):
+        self._cache = ExpiringDict[str, CeobeTarget](capacity=100, default_age=timedelta(days=7))
+        self.client = CeobeClient()
+        self.url = DATASOURCE_URL
+        self.init_requested = False
+
+    @property
+    def cache(self) -> MappingProxyType[str, CeobeTarget]:
+        return MappingProxyType(self._cache)
+
+    async def refresh_data_sources(self):
+        """请求数据源API刷新缓存"""
+        data_sources_resp = await self.client.get(self.url)
+        data_sources = process_response(data_sources_resp, DataSourceResponse).data
+        for ds in data_sources:
+            self._cache[ds.unique_id] = ds
+        return self.cache
+
+    async def get_all(self):
+        if not self.init_requested:
+            await self.refresh_data_sources()
+            self.init_requested = True
+        return self.cache
+
+    def select_one(self, cond_func: Callable[[CeobeTarget], bool]) -> CeobeTarget | None:
+        """根据条件获取数据源
+
+        不会刷新缓存
+        """
+        cache = self._cache.values()
+        return next(filter(cond_func, cache), None)
+
+    async def get_by_unique_id(self, unique_id: str) -> CeobeTarget | None:
+        """根据unique_id获取数据源
+
+        如果在缓存中找不到，会刷新缓存
+        """
+        if target := self._cache.get(unique_id):
+            return target
+        await self.refresh_data_sources()
+        return self._cache.get(unique_id)
+
+    async def get_by_nickname(self, nickname: str) -> CeobeTarget | None:
+        """根据nickname获取数据源
+
+        如果在缓存中找不到，会刷新缓存
+        """
+
+        def cond_func(target: CeobeTarget):
+            return target.nickname == nickname
+
+        if target := self.select_one(cond_func):
+            return target
+        await self.refresh_data_sources()
+        return self.select_one(cond_func)
+
+    async def get_by_source(self, source: CeobeSource) -> CeobeTarget | None:
+        """根据source获取数据源
+
+        如果在缓存中找不到，会刷新缓存
+        """
+
+        def cond_func(target: CeobeTarget):
+            return target.db_unique_key == source.data and target.datasource == source.type
+
+        if target := self.select_one(cond_func):
+            return target
+        await self.refresh_data_sources()
+        return self.select_one(cond_func)

nonebot_bison/platform/ceobecanteen/const.py

@@ -0,0 +1,4 @@
+DATASOURCE_URL = "https://server.ceobecanteen.top/api/v1/canteen/config/datasource/list"
+COMB_ID_URL = "https://server.ceobecanteen.top/api/v1/canteen/user/getDatasourceComb"
+COOKIE_ID_URL = "http://cdn.ceobecanteen.top/datasource-comb"
+COOKIES_URL = "https://server-cdn.ceobecanteen.top/api/v1/cdn/cookie/mainList/cookieList"

nonebot_bison/platform/ceobecanteen/exception.py

@@ -0,0 +1 @@
+class CeobeResponseError(Exception): ...

nonebot_bison/platform/ceobecanteen/models.py

@@ -0,0 +1,120 @@
+from typing import Literal, TypeVar
+
+from pydantic import BaseModel
+
+
+class CeobeTarget(BaseModel):
+    """账户结构"""
+
+    avatar: str
+    """数据源头像"""
+    datasource: str
+    """数据源类型"""
+    db_unique_key: str
+    """数据源相关唯一id"""
+    nickname: str
+    """数据源昵称"""
+    platform: str
+    """平台代码"""
+    unique_id: str
+    """数据源唯一标识（用于前后端交互标识）"""
+    jump_url: str | None = None
+    """跳转url（null就是没办法跳转）"""
+
+
+class DataSourceResponse(BaseModel):
+    code: int
+    message: str
+    data: list[CeobeTarget]
+
+
+class CeobeImage(BaseModel):
+    origin_url: str
+    """原图"""
+    compress_url: str | None = None
+    """压缩图，为null就是没有原图对应压缩图"""
+
+
+class CeobeDefaultCookie(BaseModel):
+    text: str
+    images: list[CeobeImage] | None
+
+
+class CeobeRetweeted(BaseModel):
+    author_name: str
+    author_avatar: str
+    text: str
+    images: list[CeobeImage] | None = None
+
+
+class CeobeItem(BaseModel):
+    id: str
+    """单条id"""
+    url: str
+    """跳转链接"""
+    type: str | None = None
+    """类型"""
+    is_long_text: bool | None = None
+    """是否长文"""
+    is_retweeted: bool = False
+    """是否转发"""
+    retweeted: CeobeRetweeted | None = None
+
+    class Config:
+        extra = "allow"
+
+
+class CeobeSource(BaseModel):
+    data: str
+    """数据源id"""
+    type: str
+    """数据源类型"""
+
+
+class CeobeTimestamp(BaseModel):
+    fetcher: int
+    """蹲饼时间，毫秒"""
+    platform_precision: Literal["none", "day", "hour", "minute", "second", "ms"]
+    """平台时间精度"""
+    platform: int | None = None
+    """平台时间戳，毫秒"""
+
+
+class CeobeCookie(BaseModel):
+    datasource: str
+    """数据源名字"""
+    icon: str
+    """数据源头像"""
+    timestamp: CeobeTimestamp
+    """时间戳"""
+    default_cookie: CeobeDefaultCookie
+    """原始饼"""
+    item: CeobeItem
+    """数据源信息，有平台的特殊字段"""
+    source: CeobeSource
+    """数据源"""
+
+
+class CeobeData(BaseModel):
+    cookies: list[CeobeCookie]
+    next_page_id: str | None = None
+
+
+class CookiesResponse(BaseModel):
+    code: int
+    message: str
+    data: CeobeData
+
+
+class CombIdResponse(BaseModel):
+    code: int
+    message: str
+    data: dict[Literal["datasource_comb_id"], str]
+
+
+class CookieIdResponse(BaseModel):
+    cookie_id: str
+    update_cookie_id: str
+
+
+ResponseModel = TypeVar("ResponseModel", bound=CookiesResponse | CombIdResponse | CookieIdResponse | DataSourceResponse)