feat: switch to sqlite database

README.md

@@ -64,7 +64,7 @@ _✨ 每月🦌管签到 ✨_
 
 ## ⚙️ 配置
 
-该插件无需配置
+该插件使用了 `nonebot-plugin-localstore` 来决定用户数据的存放位置，你可能需要对其进行配置
 
 ## 🎉 使用
 ### 指令表

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "nonebot-plugin-deer-pipe"
-version = "0.1.4"
+version = "0.2.0"
 description = "A deer-pipe attendance nonebot2 plugin"
 authors = [
     {name = "SNRainiar", email = "[email protected]"},
@@ -10,6 +10,8 @@ dependencies = [
     "nonebot-plugin-alconna>=0.51.1",
     "nonebot-plugin-localstore>=0.7.1",
     "nonebot-plugin-userinfo>=0.2.5",
+    "sqlmodel>=0.0.21",
+    "aiosqlite>=0.20.0",
 ]
 requires-python = ">=3.10"
 readme = "README.md"
@@ -33,7 +35,7 @@ build-backend = "pdm.backend"
 distribution = true
 
 [tool.pdm.dev-dependencies]
-test = [
+dev = [
     "nonebot2[fastapi]>=2.3.2",
     "nonebot-adapter-console>=0.6.0",
 ]

src/nonebot_plugin_deer_pipe/__init__.py

@@ -1,18 +1,16 @@
-import json
-import os
-
-from .contants import USERDATA_PATH
+from .database import attend
 from .image import generate_image
+
 from datetime import datetime
 from nonebot.plugin import PluginMetadata, inherit_supported_adapters, require
-from typing import Any, Sequence
 
 require("nonebot_plugin_alconna")
 from nonebot_plugin_alconna import UniMessage, on_alconna
 
 require("nonebot_plugin_userinfo")
 from nonebot_plugin_userinfo import EventUserInfo, UserInfo
 
+
 # Plugin meta
 __plugin_meta__: PluginMetadata = PluginMetadata(
   name="🦌管签到",
@@ -25,34 +23,6 @@
   )
 )
 
-# Attendance
-def attendance(now: datetime, user_id: str) -> tuple[bool, Sequence[int]]:
-  if not os.path.exists(USERDATA_PATH):
-    raw_data: str = "{}"
-  else:
-    with open(USERDATA_PATH) as f:
-      raw_data: str = f.read()
-
-  data: dict[str, Any] = json.loads(raw_data)
-  userdata: dict[str, Any] | None = data.get(user_id)
-  if userdata == None:
-    userdata = { "year": now.year, "month": now.month, "deer": [] }
-    data[user_id] = userdata
-
-  if userdata["year"] != now.year or userdata["month"] != now.month:
-    userdata = { "year": now.year, "month": now.month, "deer": [] }
-    data[user_id] = userdata
-
-  if now.day in userdata["deer"]:
-    with open(USERDATA_PATH, "w") as f:
-      json.dump(data, f)
-    return (False, userdata["deer"])
-
-  userdata["deer"].append(now.day)
-  with open(USERDATA_PATH, "w") as f:
-    json.dump(data, f)
-  return (True, userdata["deer"])
-
 # Matchers
 deer_matcher = on_alconna("🦌")
 
@@ -66,7 +36,7 @@ async def handle(user_info: UserInfo = EventUserInfo()) -> None:
   )
 
   now: datetime = datetime.now()
-  ok, deer = attendance(now, user_info.user_id)
+  ok, deer = await attend(now, user_info.user_id)
   img: bytes = generate_image(now, name, deer)
 
   await UniMessage.text(

src/nonebot_plugin_deer_pipe/contants.py

@@ -10,15 +10,20 @@
 import nonebot_plugin_localstore as store
 
 
+# Plugin paths
 PLUGIN_PATH: Path = Path(os.path.dirname(os.path.realpath(__file__)))
 ASSETS_PATH: Path = PLUGIN_PATH / "assets"
 
+# Images
 CHECK_IMG: ImageFile = Image.open(ASSETS_PATH / "[email protected]")
 DEERPIPE_IMG: ImageFile = Image.open(ASSETS_PATH / "[email protected]")
 
+# Fonts
 MISANS_FONT: FreeTypeFont = ImageFont.truetype(
   ASSETS_PATH / "MiSans-Regular.ttf",
   25
 )
 
-USERDATA_PATH: Path = store.get_plugin_data_file("userdata.json")
+# Database
+DATABASE_PATH: Path = store.get_plugin_data_file("userdata.db")
+DATABASE_URL: str = f"sqlite+aiosqlite:///{DATABASE_PATH}"

src/nonebot_plugin_deer_pipe/database.py

@@ -0,0 +1,56 @@
+from .contants import DATABASE_URL
+
+from datetime import datetime
+from sqlalchemy.ext.asyncio import AsyncEngine, create_async_engine
+from sqlmodel import Field, SQLModel, select
+from sqlmodel.ext.asyncio.session import AsyncSession
+from typing import Sequence
+
+
+# Model
+class User(SQLModel, table=True):
+  user_id: str = Field(primary_key=True)
+  year: int
+  month: int
+  mask: str    = "0"
+
+# Async engine
+engin: AsyncEngine = create_async_engine(DATABASE_URL)
+initialized: bool = False
+
+# Attendance
+def get_seq(mask: str) -> list[int]:
+  return list(
+    map(
+      lambda x: x[0] + 1,
+      filter(
+        lambda x: x[1] == '1',
+        enumerate(mask)
+      )
+    )
+  )
+
+async def attend(now: datetime, user_id: str) -> tuple[bool, Sequence[int]]:
+  if not initialized:
+    async with engin.begin() as conn:
+      await conn.run_sync(SQLModel.metadata.create_all)
+
+  async with AsyncSession(engin) as session:
+    user: User | None = (
+      await session.exec(select(User).where(User.user_id == user_id))
+    ).one_or_none()
+
+    if user == None or user.year != now.year or user.month != now.month:
+      user = User(user_id=user_id, year=now.year, month=now.month)
+
+    mask: int = int(user.mask)
+    if (mask >> (now.day - 1)) & 1 == 1:
+      return (False, get_seq(user.mask))
+    else:
+      mask |= (1 << (now.day - 1))
+      user.mask = str(mask)
+
+      session.add(user)
+      await session.commit()
+
+      return (True, get_seq(str(mask)))

src/nonebot_plugin_deer_pipe/image.py

@@ -3,6 +3,7 @@
 import secrets
 
 from .contants import CHECK_IMG, DEERPIPE_IMG, MISANS_FONT, PLUGIN_PATH
+
 from PIL import Image, ImageDraw
 from datetime import datetime
 from pathlib import Path