0.2.0

README.md

@@ -145,6 +145,11 @@ Telegram：[@lgc2333](https://t.me/lgc2333)
 
 ## 📝 更新日志
 
+### 0.2.0
+
+- 自动更新图片列表
+- 缓存获取到的图片和图片列表
+
 ### 0.1.1
 
 - 发送梗图会回复指令消息

nonebot_plugin_nonememe/__init__.py

@@ -1,12 +1,13 @@
 from nonebot import require
 from nonebot.plugin import PluginMetadata
 
+require("nonebot_plugin_apscheduler")
 require("nonebot_plugin_saa")
 
 from . import __main__ as __main__  # noqa: E402
 from .config import ConfigModel  # noqa: E402
 
-__version__ = "0.1.1"
+__version__ = "0.2.0"
 __plugin_meta__ = PluginMetadata(
     name="NoneMeme",
     description="NoneBot 群大佬的日常",

nonebot_plugin_nonememe/__main__.py

@@ -10,16 +10,13 @@
 from nonebot_plugin_saa import Image, MessageFactory, Text
 
 from .config import config
-from .data_source import MemeItem, fetch_meme, meme_list, search_meme_items
+from .data_source import MemeItem, get_meme, meme_list, search_meme_items
 
 
 async def finish_with_meme(meme_item: MemeItem) -> NoReturn:
-    image_bytes = await fetch_meme(meme_item.path)
+    image_bytes = await get_meme(meme_item)
     await MessageFactory(
-        [
-            Text(f"# {meme_item.name}"),
-            Image(image_bytes),
-        ],
+        [Text(f"# {meme_item.name}"), Image(image_bytes)],
     ).finish(reply=True)
 
 
@@ -38,7 +35,7 @@ async def _(matcher: Matcher, state: T_State, arg_msg: Message = CommandArg()):
 
     searched = search_meme_items(arg, use_regex=use_regex)
     if not searched:
-        await matcher.finish("没有找到相关 NoneMeme")
+        await matcher.finish("没有找到相关图片")
     if len(searched) == 1:
         await finish_with_meme(searched[0])
 

nonebot_plugin_nonememe/config.py

@@ -1,15 +1,28 @@
-from typing import Optional
+from typing import Optional, TypedDict, Union
+from typing_extensions import NotRequired
 
 from nonebot import get_driver
 from pydantic import BaseModel
 
 
+class CronDict(TypedDict):
+    year: NotRequired[Union[str, int]]
+    month: NotRequired[Union[str, int]]
+    day: NotRequired[Union[str, int]]
+    week: NotRequired[Union[str, int]]
+    day_of_week: NotRequired[Union[str, int]]
+    hour: NotRequired[Union[str, int]]
+    minute: NotRequired[Union[str, int]]
+    second: NotRequired[Union[str, int]]
+
+
 class ConfigModel(BaseModel):
     nonememe_proxy: Optional[str] = None
     nonememe_repo_prefix: str = (
         "https://raw.githubusercontent.com/NoneMeme/NoneMeme/main"
     )
 
+    nonememe_update_cron: CronDict = {"hour": 1}
     nonememe_search_limit: int = 5
 
 

nonebot_plugin_nonememe/data_source.py

@@ -1,19 +1,32 @@
+import json
 import re
 import urllib.parse
-from dataclasses import dataclass
 from pathlib import Path
 from typing import List, cast
 
+import anyio
 import json5
 from httpx import AsyncClient
 from nonebot import get_driver, logger
+from nonebot_plugin_apscheduler import scheduler
+from pydantic import BaseModel, parse_raw_as
 
 from .config import config
 
+DATA_DIR = Path.cwd() / "data" / "nonememe"
+LIST_CACHE_PATH = DATA_DIR / "cached_list.json"
+MEME_CACHE_DIR = DATA_DIR / "cache"
 
-@dataclass
-class MemeItem:
+if not DATA_DIR.exists():
+    DATA_DIR.mkdir(parents=True)
+# if MEME_CACHE_DIR.exists():
+#     shutil.rmtree(MEME_CACHE_DIR)
+# MEME_CACHE_DIR.mkdir(parents=True)
+
+
+class MemeItem(BaseModel):
     name: str
+    suffix: str
     path: str
 
 
@@ -43,9 +56,19 @@ async def fetch_meme(path: str) -> bytes:
         return resp.content
 
 
+async def get_meme(meme: MemeItem) -> bytes:
+    cache_path = anyio.Path(MEME_CACHE_DIR / f"{meme.name}{meme.suffix}")
+    if await cache_path.exists():
+        return await cache_path.read_bytes()
+
+    data = await fetch_meme(meme.path)
+    await cache_path.write_bytes(data)
+    return data
+
+
 def build_meme_item(meme_path: str) -> MemeItem:
-    name = Path(urllib.parse.unquote(meme_path)).stem
-    return MemeItem(name=name, path=meme_path)
+    path_obj = Path(urllib.parse.unquote(meme_path))
+    return MemeItem(name=path_obj.stem, suffix=path_obj.suffix, path=meme_path)
 
 
 async def fetch_meme_list() -> List[MemeItem]:
@@ -58,13 +81,38 @@ async def fetch_meme_list() -> List[MemeItem]:
     return [build_meme_item(item) for item in items]
 
 
-async def init_meme_list():
+async def update_meme_list():
+    logger.info("Updating meme list")
+
+    cache_json_path = anyio.Path(LIST_CACHE_PATH)
+
+    try:
+        got_meme_list = await fetch_meme_list()
+        await cache_json_path.write_text(
+            json.dumps(
+                [x.dict() for x in got_meme_list],
+                indent=2,
+                ensure_ascii=False,
+            ),
+        )
+
+    except Exception:
+        if not await cache_json_path.exists():
+            raise
+
+        logger.warning("Failed to fetch meme list, use cache instead")
+        got_meme_list = parse_raw_as(
+            List[MemeItem],
+            await cache_json_path.read_text(encoding="u8"),
+        )
+
     meme_list.clear()
-    meme_list.extend(await fetch_meme_list())
+    meme_list.extend(got_meme_list)
     logger.opt(colors=True).success(
-        f"Succeed to init meme list, Loaded <y>{len(meme_list)}</y> memes",
+        f"Succeed to update meme list, Loaded <y>{len(meme_list)}</y> memes",
     )
 
 
 driver = get_driver()
-driver.on_startup(init_meme_list)
+driver.on_startup(update_meme_list)
+scheduler.add_job(update_meme_list, "cron", **config.nonememe_update_cron)

pyproject.toml

@@ -1,14 +1,17 @@
 [project]
 name = "nonebot-plugin-nonememe"
-version = "0.1.1.post1"
+version = "0.2.0"
 description = "The daily life of the NoneBot group members"
 authors = [{ name = "student_2333", email = "[email protected]" }]
 dependencies = [
     "nonebot2>=2.0.0",
-    "pydantic>=1.10.4,<2",
     "nonebot-plugin-send-anything-anywhere>=0.2.7",
+    "nonebot-plugin-apscheduler>=0.3.0",
+    "pydantic>=1.10.4,<2",
     "httpx>=0.24.1",
     "json5>=0.9.14",
+    "anyio>=3.7.1",
+    "typing-extensions>=4.7.1",
 ]
 requires-python = ">=3.8,<4.0"
 readme = "README.md"