➕ cave module

.gitignore

@@ -149,6 +149,7 @@ temp/
 *.temp
 
 config/
+database/
 
 # PyCharm
 #  JetBrains specific template is maintainted in a separate JetBrains.gitignore that can

library/path.py

@@ -9,3 +9,7 @@
 config: Path = Path(root, "config")
 
 config.mkdir(parents=True, exist_ok=True)
+
+database: Path = Path(root, "database")
+
+database.mkdir(parents=True, exist_ok=True)

module/__init__.py

@@ -1,7 +1,8 @@
 import os
+import pathlib
 
-__all__ = [
-    s.removesuffix(".py")
-    for s in os.listdir(os.path.dirname(__file__))
-    if s.endswith(".py") and not s.startswith("_")
-]
+__all__ = []
+
+for path in pathlib.Path(os.path.dirname(__file__)).iterdir():
+    if (path.is_dir() or path.name.endswith(".py")) and not path.name.startswith("_"):
+        __all__.append(path.stem)

module/cave/__init__.py

@@ -0,0 +1,81 @@
+from graia.ariadne.message.element import At, Image, Plain
+from graia.ariadne.model import Group, Member
+from asyncio.exceptions import TimeoutError
+from . import db
+
+from graia.ariadne.app import Ariadne
+from graia.ariadne.event.message import GroupMessage
+from graia.ariadne.message.chain import MessageChain
+from graia.ariadne.message.commander.saya import CommandSchema
+from graia.saya.channel import Channel
+from graia.broadcast import Broadcast
+from graia.broadcast.interrupt.waiter import Waiter
+from graia.broadcast.interrupt import InterruptControl
+from graia.ariadne.message.formatter import Formatter
+
+channel = Channel.current()
+
+
+@channel.use(CommandSchema("[回声洞|.cave|cave] [add|-a] {...content:raw}"))
+async def add(content: "MessageChain", broadcast: Broadcast, sender: Member, app: Ariadne, group: Group):
+    if not content:
+        inc = InterruptControl(broadcast)
+
+        await app.sendMessage(group, MessageChain([At(sender), " 请发送要存入回声洞的消息\n" "发送 “取消” 来取消操作"]))
+
+        @Waiter.create_using_function([GroupMessage])
+        async def wait_for_chain(chain: MessageChain, i_sender: Member):
+            if i_sender.id == sender.id and i_sender.group.id == sender.group.id:
+                await chain.download_binary()
+                chain = chain.include(Plain, Image)
+                if chain:
+                    if chain.asDisplay() != "取消":
+                        id: int = db.add_cave(sender.name, chain)
+                        await app.sendMessage(group, MessageChain([f"添加成功，ID: {id}"]))
+                else:
+                    await app.sendMessage(group, MessageChain(["请发送图片或文本"]))
+                return True
+
+        try:
+            await inc.wait(wait_for_chain, timeout=60.0)
+        except TimeoutError:
+            await app.sendMessage(group, MessageChain(["操作超时"]))
+    else:
+        await content.download_binary()
+        content = content.include(Plain, Image)
+        if content:
+            id: int = db.add_cave(sender.name, content)
+            await app.sendMessage(group, MessageChain([f"添加成功，ID: {id}"]))
+        else:
+            await app.sendMessage(group, MessageChain(["请发送图片或文本"]))
+
+
+send_fmt = Formatter(
+    "回声洞 [{id}]: {create_time}\n" "共被找到 {query_time} 次\n" "{chain}\n" "          ————{name}\n"
+)
+
+
+@channel.use(CommandSchema("[回声洞|.cave|cave] [del|-d] {id}"))
+async def delete(id: int, app: Ariadne, group: Group):
+    if db.del_cave(id):
+        await app.sendMessage(group, MessageChain(["删除成功"]))
+    else:
+        await app.sendMessage(group, MessageChain(["没有找到这条回声洞"]))
+
+
+@channel.use(CommandSchema("[回声洞|.cave|cave] [get|-g] {id}"))
+async def get(id: int, app: Ariadne, group: Group):
+    content = db.get_cave(id)
+    if content:
+        await app.sendMessage(group, MessageChain(send_fmt.format(**content)))
+    else:
+        await app.sendMessage(group, MessageChain([f"ID: {id} 不存在"]))
+
+
+@channel.use(CommandSchema("[回声洞|.cave|cave]"))
+async def random_cave(app: Ariadne, group: Group):
+    content = db.random_cave()
+    if content:
+        await app.sendMessage(group, MessageChain(send_fmt.format(**content)))
+    else:
+        await app.sendMessage(group, MessageChain(["没有回声洞条目, 使用 “.cave -a” 添加一条吧！"]))

module/cave/db.py

@@ -0,0 +1,78 @@
+from datetime import datetime
+import random
+from typing import List, Optional, TypedDict
+from graia.ariadne.message.chain import MessageChain
+from graia.ariadne.message.element import MultimediaElement
+from peewee import Model, CharField, BigIntegerField, SqliteDatabase, DateTimeField, TextField, fn
+from library.path import database
+
+cave_db = SqliteDatabase(database.joinpath("cave.sqlite"))
+
+
+class CaveEntry(Model):
+    name = CharField()
+    content = TextField()
+    time = DateTimeField(default=datetime.now)
+    query_time = BigIntegerField(default=0)
+
+    class Meta:
+        database = cave_db
+        db_table = "entry"
+
+
+CaveEntry.create_table()
+
+
+def add_cave(name: str, content: MessageChain) -> int:
+    for elem in content:
+        if isinstance(elem, MultimediaElement):
+            elem.id = None
+            elem.url = None
+    entry = CaveEntry(name=name, content=content.asPersistentString())
+    entry.save()
+    return entry.id
+
+
+def del_cave(id: int) -> bool:
+    if CaveEntry.get_or_none(CaveEntry.id == id):
+        CaveEntry.delete().where(CaveEntry.id == id).execute()
+        return True
+    return False
+
+
+class CaveDict(TypedDict):
+    name: str
+    chain: MessageChain
+    query_time: str
+    create_time: str
+    id: str
+
+
+def random_cave() -> Optional[CaveDict]:
+    if CaveEntry.select().count() == 0:
+        return None
+    entries: List[CaveEntry] = CaveEntry.select().order_by(fn.Random())[:5]
+    entry: CaveEntry = random.choice(entries)
+    entry_dict: CaveDict = CaveDict(
+        name=entry.name,
+        chain=MessageChain.fromPersistentString(entry.content),
+        create_time=entry.time.strftime("%Y-%m-%d %H:%M:%S"),
+        query_time=str(entry.query_time),
+        id=str(entry.id),
+    )
+    CaveEntry.update(query_time=entry.query_time + 1).where(CaveEntry.id == entry.id).execute()
+    return entry_dict
+
+
+def get_cave(id: int) -> Optional[CaveDict]:
+    entry: CaveEntry = CaveEntry.get_or_none(CaveEntry.id == id)
+    if entry is None:
+        return None
+    entry_dict: CaveDict = CaveDict(
+        name=entry.name,
+        chain=MessageChain.fromPersistentString(entry.content),
+        create_time=entry.time.strftime("%Y-%m-%d %H:%M:%S"),
+        query_time=str(entry.query_time),
+        id=str(entry.id),
+    )
+    return entry_dict

pyproject.toml

@@ -6,9 +6,9 @@ authors = [
     {name = "BlueGlassBlock", email = "[email protected]"},
 ]
 dependencies = [
-    "graia-ariadne[full]>=0.5.2.post1",
+    "graia-ariadne[full]>=0.5.3.post2",
     "toml>=0.10.2",
-]
+    "peewee>=3.14.8"]
 requires-python = ">=3.9"
 license = {text = "AGPL-3.0"}