Nonebot-金币管理系统(频道插件)

介绍

今天利用空闲时间写了一个金币的插件，做到优先同步内存，再同步数据库，在用户1000+的高频率使用下，已经做到毫无压力了，主要思路如下

• 开机做好预处理，把全部用户数据加载出来，并缓存在内存中

• 动态维护 对内存里的数据进行加减

• 再统一提交给数据库

如何使用

本插件下载链接 👇 👇

传送门

☝ ☝ ☝

添加到你的频道插件目录下 并在 pyproject.toml里添加该文件夹

我这里把插件放在plugin文件夹里了

plugin_dirs = ["plugin"] # 添加该插件所在的文件夹

代码展示

1.核心代码

## 作者:bingyue
## QQ: 35***19417
## 一个高效率的写法，适合用于金币好感度

from nonebot import on_command, on_fullmatch
from nonebot.permission import SUPERUSER
from nonebot.params import CommandArg
from nonebot.adapters.qqguild import Message, MessageEvent
from . import sqlite_conn

a = on_command("增加金币",permission=SUPERUSER)
b = on_command("减少金币",permission=SUPERUSER)
c = on_fullmatch("查询金币")

@a.handle()
async def _(b: Message = CommandArg()):
    c = list(b)
    await add_jb(c[0].data["user_id"],int(c[1].data["text"].strip()))
    await a.finish("添加金币成功！")

@b.handle()
async def _(d: Message = CommandArg()):
    c = list(d)
    await del_jb(c[0].data["user_id"],int(c[1].data["text"]))
    await b.finish("减少金币成功")

@c.handle()
async def _(event: MessageEvent):
    j = await find_jb(event.get_user_id())
    if j is None:
        await c.finish("你目前的金币为0")
    else:
        await c.finish(f"你目前的金币为{j}")

async def add_jb(id:str, e:int) -> None:
    """
    增加好感度

    id: 你的频道号
    
    e: 增加的具体数据
    """
    j = await find_jb(id)
    if j is None:
        sqlite_conn.conn_jb.cursor().execute(f'insert into `jb` (`id`,`jb`) values("{id}",{e})')
        sqlite_conn.h[id] = e
    else:
        sqlite_conn.conn_jb.cursor().execute(f'UPDATE `jb` SET jb={j+e} WHERE id="{id}"')
        sqlite_conn.h[id] += e


async def find_jb(id:str) -> int:
    """
    查询好感度

    id: 你的频道号id
    """
    if id not in sqlite_conn.h:
        return None
    else:
        return sqlite_conn.h[id]

async def del_jb(id:str, e:int) -> None:
    """
    减少好感度

    id: 你的频道号

    e: 减少的具体数据
    """
    j = await find_jb(id)
    if j is None:
        sqlite_conn.conn_jb.cursor().execute(f'insert into `jb` (`id`,`jb`) values("{id}",{-e})')
        sqlite_conn.h[id] = -e
    else:
        sqlite_conn.conn_jb.cursor().execute(f'UPDATE `jb` SET jb={j-e} WHERE id="{id}"')
        sqlite_conn.h[id] -= e

2.连接数据库，预处理

## 作者:bingyue
## QQ: 35***19417
## 一个高效率的写法，适合用于金币好感度

from nonebot.log import logger
import os
import sqlite3

h = {}

if not os.path.isdir("data"):
    logger.error("未检测到data文件,正在创建")
    os.mkdir("data")
    logger.success("创建data文件夹成功")
    
if not os.path.isfile("data/jb.db"):
    conn_jb = sqlite3.connect("data/jb.db")
    logger.warning("正在加载数据库！！")
    conn_jb.cursor().execute(f"CREATE TABLE `jb` (`id` text, `jb` int(12))")
    conn_jb.commit()
    logger.success("创建表格完毕")

conn_jb = sqlite3.connect("data/jb.db")

q = conn_jb.cursor().execute(f"SELECT * from `jb`").fetchall()
for i in q:
    h[i[0]] = i[1]

3.定时同步给数据库

## 作者:bingyue
## QQ: 35***19417
## 一个高效率的写法，适合用于金币好感度

from nonebot_plugin_apscheduler import scheduler
from . import sqlite_conn
from nonebot.log import logger

@scheduler.scheduled_job('interval', seconds=60)
async def eveny_day() -> None:
    sqlite_conn.conn_jb.commit()
    logger.success("同步数据成功！")