astrbot_plugin_dajiao/main.py

import json
import Tool
from astrbot.api.event import filter, AstrMessageEvent, MessageEventResult
from astrbot.api.star import Context, Star, register
from astrbot.api import logger
import random
from .back import time_long, volume, isUserExist, insertUser, seconds_to_hms, ml_to_l_ml, get_user_name
import pymysql

@register("helloworld", "YourName", "一个简单的 Hello World 插件", "1.0.0")
class MyPlugin(Star):
    def __init__(self, context: Context):
        super().__init__(context)
    
    
    @filter.command("打打我的")
    async def dajiao(self, event: AstrMessageEvent):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        time = round(random.uniform(1, 600), 2)
        V = round(random.uniform(0.01,100), 2)
        a = time_long(time)
        b = volume(V)
        user_name = event.get_sender_id()
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认127.0.0.1
        ,user = 'saipo' # 用户名
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "INSERT INTO dajiao (openid, timelong, volume) VALUES (%s, %s, %s)"
        values = (user_name,time,V)
        cur.execute(sql,values)
        conn.commit()
        cur.close()
        conn.close()
        yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())}, 你坚持了{time}s哦，{a}.射出{V}ml,{b}!") 



    @filter.command("我的时长成就")
    async def timebeat(self, event: AstrMessageEvent):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认127.0.0.1
        ,user = 'saipo' # 用户名
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "SELECT eventTime, openid, timelong, volume FROM dajiao WHERE openid = %s AND timelong = (SELECT MAX(timelong) FROM dajiao WHERE openid = %s) LIMIT 1"
        openid = event.get_sender_id()
        cur.execute(sql,(openid,openid))
        date1 = cur.fetchone()
        sql2 = "SELECT eventTime, openid, timelong, volume FROM dajiao WHERE openid = %s AND timelong = (SELECT MIN(timelong) FROM dajiao WHERE openid = %s) LIMIT 1"
        cur.execute(sql2,(openid,openid))
        date2 = cur.fetchone()
        cur.close()
        conn.close()
        yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())}\n你的时长最佳记录为{date1[2]}s,发生在{date1[0]},射出{date1[3]}ml！\n你的时长最弱记录为{date2[2]}s,发生在{date2[0]},射出{date2[3]}ml！")

    @filter.command("我的产量成就")
    async def volumebeat(self, event: AstrMessageEvent):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认127.0.0.1
        ,user = 'saipo' # 用户名
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "SELECT eventTime, openid, timelong, volume FROM dajiao WHERE openid = %s AND volume = (SELECT MAX(volume) FROM dajiao WHERE openid = %s) LIMIT 1"
        openid = event.get_sender_id()
        cur.execute(sql,(openid,openid))
        date1 = cur.fetchone()
        sql2 = "SELECT eventTime, openid, timelong, volume FROM dajiao WHERE openid = %s AND volume = (SELECT MIN(volume) FROM dajiao WHERE openid = %s) LIMIT 1"
        cur.execute(sql2,(openid,openid))
        date2 = cur.fetchone()
        cur.close()
        conn.close()
        yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())}\n你的产量最佳记录为{date1[3]}ml,发生在{date1[0]},时长为{date1[2]}s！\n你的产量最弱记录为{date2[3]}ml,发生在{date2[0]},时长为{date2[2]}s！") 

    @filter.command("我的个人信息")
    async def userinfo(self, event: AstrMessageEvent):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认
        ,user = 'saipo'
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "SELECT uid, openid , nameid FROM dajiaouser WHERE openid = %s"
        openid = event.get_sender_id()
        cur.execute(sql,(openid,))
        date1 = cur.fetchone()
        sql2 = "SELECT sum(timelong),sum(volume) FROM dajiao WHERE openid = %s group by openid"
        cur.execute(sql2,(openid,))
        date2 = cur.fetchone()
        cur.close()
        conn.close()
        yield event.plain_result(f"个人信息\nuid：{date1[0]}\nopenid：{date1[1]}\nname：{date1[2]}\n总时长：{seconds_to_hms(date2[0])}\n总产量：{ml_to_l_ml(date2[1])}")


    @filter.command("修改昵称")
    async def updatename(self, event: AstrMessageEvent, name: str):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认
        ,user = 'saipo'
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "UPDATE dajiaouser SET nameid = %s WHERE openid = %s"
        openid = event.get_sender_id()
        cur.execute(sql,(name,openid))
        conn.commit()
        cur.close()
        conn.close()
        yield event.plain_result(f"昵称修改成功！\n新的昵称为：{name}")


    @filter.command("我的背包")
    async def bag(self, event: AstrMessageEvent):
        if isUserExist(event.get_sender_id()) != True:
            insertUser(event.get_sender_id())
        conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认
        ,user = 'saipo'
        ,passwd='Grasste0403' # 密码
        ,port= 3306 # 端口，默认为3306
        ,db='saipo' # 数据库名称
        ,charset='utf8' # 字符编码
        )
        cur = conn.cursor()
        sql = "SELECT baglist FROM dajiaouser WHERE openid = %s"
        openid = event.get_sender_id()
        cur.execute(sql,(openid,))
        date = cur.fetchone()
        baglist = json.loads(date[0])
        baglist_str = ""
        if len(baglist) == 0:
            baglist_str = "背包空空如也"
        else:
            for item in baglist:
                if Tool.get_tool_name(item["id"]) is None:
                    cur.close()
                    conn.close()
                    yield event.plain_result(f"背包存在异常！")
                else:
                    baglist_str += f"{Tool.get_tool_name(item['id'])} x {item['num']}\n"
        cur.close()
        conn.close()
        yield event.plain_result(f"背包列表：\n{baglist_str}")
        
    @filter.command("测试")
    async def test(self, event: AstrMessageEvent, name: str):
        yield event.plain_result(f"测试结果")