feature(question): 新增问答模块，用户可在数据库中添加预设好的问题和回答

TODO: 计划新增该模块的web控制面板

backend.py

@@ -0,0 +1,22 @@
+from flask import Flask
+from src.clover_sqlite.models.questions import Question
+app = Flask(__name__)
+
+@app.route("/")
+def hello():
+    return "Hello World!"
+
+@app.route("/list")
+async def list_data():
+    return await Question.fetch()
+
+@app.route("/init")
+async def init_data():
+    if await Question.insert_one("你好", "你好哦"):
+        return "success"
+
+    return "failed"
+
+def start_flask():
+    print("Flask启动中...")
+    app.run(host='0.0.0.0', port=5000, debug=False, use_reloader=False)

bot.py

@@ -1,6 +1,8 @@
 import os
 import glob
+import threading
 import logging
+
 import nonebot
 from pathlib import Path
 from nonebot import logger
@@ -10,6 +12,10 @@ from apscheduler.schedulers.background import BackgroundScheduler
 from src.configs.path_config import log_path,temp_path,video_path,yuc_wiki_path
 
 nonebot.init()
+
+from backend import start_flask
+
+
 driver = nonebot.get_driver()
 driver.register_adapter(QQAdapter)  # 注册QQ适配器
 nonebot.load_from_toml("pyproject.toml")
@@ -35,5 +41,15 @@ scheduler = BackgroundScheduler()
 scheduler.add_job(clean_temp_cache, 'cron', hour=0, minute=0)
 
 if __name__ == "__main__":
+
+    # nonebot_thread = threading.Thread(target=nonebot.run(), daemon=True)
+    # nonebot_thread.start()
+
+
+
+    flask_thread = threading.Thread(target=start_flask, daemon=True)
+    flask_thread.start()
     scheduler.start()
     nonebot.run()
+
+

requirements.txt

@@ -45,3 +45,4 @@ pydantic
 matplotlib
 httpx
 crypto
+flask[async]

src/clover_sqlite/models/questions.py

@@ -0,0 +1,176 @@
+from lazy_object_proxy.utils import await_
+from tortoise import fields
+from src.clover_sqlite.data_init.db_connect import Model
+
+
+class Question(Model):
+    id = fields.IntField(primary_key=True, generated=True)
+    question = fields.CharField(max_length=155, description="问题")
+    answer = fields.CharField(max_length=400,description="回答")
+
+    class Meta:
+        table = "question"
+        table_description = "问答表"
+
+    @classmethod
+    async def _fetch_data(cls) -> list:
+        return await cls.all().order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _get_data_by_id(cls, get_id: int | None) -> list | None:
+        """
+        通过id获取问答内容
+        """
+        if not get_id:
+            return None
+        else:
+            return await cls.filter(id=get_id).order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _get_data_by_ques_keyword_exact(cls, keyword: str | None) -> list | None:
+        """
+        通过关键字精确地获取问题内容
+        """
+        if not keyword:
+            return None
+        else:
+            return await cls.filter(question__icontains=keyword).order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _get_data_by_ques_keyword_fuzzy(cls, keyword: str | None) -> list | None:
+        """
+        通过关键字模糊匹配问题内容
+        """
+        if not keyword:
+            return None
+        else:
+            key_list = list(keyword)
+
+            result = cls.filter(question__icontains=key_list[0])
+            for key in key_list:
+                result = result.filter(question__icontains=key)
+            return await result.order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _get_data_by_ans_keyword_exact(cls, keyword: str | None) -> list | None:
+        """
+        通过关键字精确地获取回答内容
+        """
+        if not keyword:
+            return None
+        else:
+            return await cls.filter(answer__icontains=keyword).order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _get_data_by_ans_keyword_fuzzy(cls, keyword: str | None) -> list | None:
+        """
+        通过关键字模糊匹配回答内容
+        """
+        if not keyword:
+            return None
+        else:
+            key_list = list(keyword)
+
+            result = cls.filter(answer__icontains=key_list[0])
+            for key in key_list:
+                result = result.filter(answer__icontains=key)
+            return await result.order_by("id").values_list("id", "question", "answer")
+
+    @classmethod
+    async def _insert_data(cls,
+                           question: str | None,
+                           answer: str | None) -> bool:
+
+        data = {
+            "question": question,
+            "answer": answer
+        }
+
+        if await cls.create(**data):
+            return True
+        else:
+            return False
+
+    @classmethod
+    async def _delete_data(cls, del_id: int | None) -> bool:
+        if del_id is None:
+            return False
+
+        if await cls.filter(id=del_id).delete():
+            return True
+        else:
+            return False
+
+    @classmethod
+    async def _alter_data(cls,
+                          alter_id: int | None,
+                          question: str | None,
+                          answer: str | None) -> bool:
+
+        updated_count = cls.filter(id=alter_id).update(
+            question=question,
+            answer=answer
+        )
+
+        print(f"更新了{updated_count}条数据")
+        return True
+
+    @classmethod
+    async def fetch(cls) -> list:
+        return await cls._fetch_data()
+
+    @classmethod
+    async def search(cls,
+                     keyword,
+                     via_id: bool = False,
+                     via_question: bool = False,
+                     via_ans: bool = False,
+                     fuzzy: bool = False) -> list | None:
+        if (via_id and via_ans and via_question is False) or (fuzzy and via_id is True):
+            print("不合法传参")
+            return None
+
+        if via_id:
+            return await cls._get_data_by_id(keyword)
+        elif via_question:
+            if fuzzy:
+                return await cls._get_data_by_ques_keyword_fuzzy(keyword)
+            else:
+                return await cls._get_data_by_ques_keyword_exact(keyword)
+        elif via_ans:
+            if fuzzy:
+                return await cls._get_data_by_ans_keyword_fuzzy(keyword)
+            else:
+                return await cls._get_data_by_ans_keyword_exact(keyword)
+        return None
+
+    @classmethod
+    async def update(cls, update_id, question, answer) -> bool:
+        return await cls._alter_data(update_id, question, answer)
+
+    @classmethod
+    async def delete_one(cls, delete_id) -> bool:
+        return await cls._delete_data(delete_id)
+
+    @classmethod
+    async def delete_many(cls, delete_id_list: list | None) -> bool:
+        for delete_id in delete_id_list:
+            await cls._delete_data(delete_id)
+
+        return True
+
+    @classmethod
+    async def insert_one(cls,
+                         question: str | None,
+                         answer: str | None) -> bool:
+        return await cls._insert_data(question, answer)
+
+    @classmethod
+    async def insert_many(cls, data_list: list | None) -> bool:
+
+        for question, answer in data_list:
+            flag = await cls._insert_data(question, answer)
+            if flag is False:
+                return flag
+
+        return True

src/plugins/check.py

@@ -10,7 +10,7 @@ from src.clover_sqlite.models.user import UserList
 
 menu = ["/重启","/今日运势","/今日塔罗","/图","/随机图","/搜番","/日报","/点歌","/摸摸头","/群老婆","/今日老婆", "/开启ai","/关闭ai","/角色列表","/添加人设", "/更新人设", "/删除人设", "/切换人设", "/管理员注册",
         "/待办", "/test","/天气","我喜欢你", "❤", "/待办查询", "/新建待办", "/删除待办" ,"/cf","/B站搜索", "/BV搜索", "/喜报", "/悲报", "/luxun","/鲁迅说",
-        "/奶龙", "/repo", "/info", "/menu", "/轻小说","/本季新番","/下季新番","/新番观察","/绝对色感" ,"/jm", "/cfrt"]
+        "/奶龙", "/repo", "/info", "/menu", "/轻小说","/本季新番","/下季新番","/新番观察","/绝对色感" ,"/jm", "/cfrt", "/问答"]
 
 send_menu = ["/menu","/今日运势","/今日塔罗","/图","/随机图","搜番","/日报","/点歌","/摸摸头","/群老婆","/待办","/天气",
              "/待办查询", "/新建待办", "/删除待办" , "/cf", "/cfrt", "/B站搜索", "/BV搜索", "/喜报", "/悲报","/鲁迅说",

src/plugins/question.py

@@ -0,0 +1,14 @@
+from nonebot.adapters.qq import MessageEvent
+from nonebot.plugin import on_command
+from nonebot.rule import to_me
+from src.clover_sqlite.models.questions import Question
+
+get_question = on_command("问答", rule=to_me(), priority=10)
+@get_question.handle()
+async def answer_question(message: MessageEvent):
+    content = message.get_plaintext().replace("/问答", "").strip(" ")
+    reply = await Question.search(content, via_question=True, fuzzy=False)
+    if len(reply) > 0:
+        await get_question.finish(reply[0][2])
+    else:
+        await get_question.finish("这个问题我没听说过哦")