QAQ/astrbot_plugin_dajiao
1
0
Fork 0
forked from saipo/astrbot_plugin_dajiao
Code Pull Requests Activity
astrbot_plugin_dajiao/main.py

285 lines
13 KiB
Python
Raw Normal View History

增加背包功能
2025-04-27 16:36:34 +08:00
import json
创建项目
2025-04-27 03:06:27 +08:00
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
解决导入bug，为什么导入类会报错(•́へ•́╬)
2025-04-27 17:11:06 +08:00
from .Tool import get_tool_name
创建项目
2025-04-27 03:06:27 +08:00
@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}!")
增加打打你的
2025-05-08 19:30:21 +08:00
@filter.command("打打你的")
增加打打你的
2025-05-08 19:30:47 +08:00
async def dajiao_other(self, event: AstrMessageEvent, message: str):
增加打打你的
2025-05-08 19:30:21 +08:00
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)
conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认127.0.0.1
,user = 'saipo' # 用户名
,passwd='Grasste0403' # 密码
,port= 3306 # 端口，默认为3306
,db='saipo' # 数据库名称
,charset='utf8' # 字符编码
)
sql = "SELECT openid FROM dajiaouser WHERE nameid = %s"
cur = conn.cursor()
cur.execute(sql, (message,))
user_name_dajiao = cur.fetchall()[0][0]
if user_name_dajiao is None:
cur.close()
conn.close()
yield event.plain_result(f"没有找到这个人哦！")
sql = "INSERT INTO dajiao (openid, timelong, volume) VALUES (%s, %s, %s)"
values = (user_name_dajiao,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())},你帮助{get_user_name(user_name_dajiao)}进行了一次打胶, TA坚持了{time}s哦，{a}.射出{V}ml,{b}!")
创建项目
2025-04-27 03:06:27 +08:00
@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}")
增加背包功能
2025-04-27 16:36:34 +08:00
@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 = ""
解决背包bug
2025-04-27 17:29:05 +08:00
if len(baglist["baglist"]) == 0:
增加背包功能
2025-04-27 16:36:34 +08:00
baglist_str = "背包空空如也"
else:
解决背包bug
2025-04-27 17:29:05 +08:00
for item in baglist["baglist"]:
解决导入bug，为什么导入类会报错(•́へ•́╬)
2025-04-27 17:11:06 +08:00
if get_tool_name(item["id"]) is None:
增加背包功能
2025-04-27 16:36:34 +08:00
cur.close()
conn.close()
yield event.plain_result(f"背包存在异常！")
else:
解决导入bug，为什么导入类会报错(•́へ•́╬)
2025-04-27 17:11:06 +08:00
baglist_str += f"{get_tool_name(item['id'])} x {item['num']}\n"
增加背包功能
2025-04-27 16:36:34 +08:00
cur.close()
conn.close()
yield event.plain_result(f"背包列表：\n{baglist_str}")
创建项目
2025-04-27 03:06:27 +08:00
测试变量作用域
2025-04-28 20:31:04 +08:00
增加产量排行榜
2025-05-08 11:48:26 +08:00
@filter.command("产量成就排行")
async def volumerank(self, event: AstrMessageEvent):
conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认
,user = 'saipo'
,passwd='Grasste0403' # 密码
,port= 3306 # 端口，默认为3306
,db='saipo' # 数据库名称
,charset='utf8' # 字符编码
)
cur = conn.cursor()
sql = "SELECT du.uid, du.nameid, t.max_volume, MAX(dj.eventTime) AS eventTime FROM dajiaouser du LEFT JOIN (SELECT openid, MAX(volume) AS max_volume FROM dajiao GROUP BY openid) t ON du.openid = t.openid LEFT JOIN dajiao dj ON du.openid = dj.openid AND t.max_volume = dj.volume GROUP BY du.uid, du.nameid, t.max_volume;"
cur.execute(sql)
date = cur.fetchall()
ranklist = []
for i in range(len(date)):
ranklist.append((date[i][0], date[i][1], date[i][2], date[i][3]))
ranklist.sort(key=lambda x: x[2], reverse=True)
ranklist_str = "产量最高排名\n"
for i in range(len(ranklist)):
修改bug
2025-05-12 16:34:15 +08:00
ranklist_str += f"{i+1} uid:{ranklist[i][0]} nameid:{ranklist[i][1]} 最高产量：{ranklist[i][2]}ml 打胶时间：{ranklist[i][3]}\n"
增加产量排行榜
2025-05-08 11:48:26 +08:00
ranklist_str += "\n"
修改bug
2025-05-12 16:34:15 +08:00
sql2 = "SELECT du.uid, du.nameid, t.min_volume, MAX(dj.eventTime) AS eventTime FROM dajiaouser du LEFT JOIN (SELECT openid, MIN(volume) AS min_volume FROM dajiao GROUP BY openid) t ON du.openid = t.openid LEFT JOIN dajiao dj ON du.openid = dj.openid AND t.min_volume = dj.volume GROUP BY du.uid, du.nameid, t.min_volume;"
增加产量排行榜
2025-05-08 11:48:26 +08:00
cur.execute(sql2)
date2 = cur.fetchall()
ranklist2 = []
for i in range(len(date2)):
ranklist2.append((date2[i][0], date2[i][1], date2[i][2], date2[i][3]))
ranklist2.sort(key=lambda x: x[2])
ranklist_str += "产量最低排名\n"
for i in range(len(ranklist2)):
修改bug
2025-05-12 16:34:15 +08:00
ranklist_str += f"{i+1} uid:{ranklist2[i][0]} nameid:{ranklist2[i][1]} 最低产量：{ranklist2[i][2]}ml 打胶时间：{ranklist2[i][3]}\n"
增加产量排行榜
2025-05-08 11:48:26 +08:00
yield event.plain_result(ranklist_str)
cur.close()
conn.close()
增加时长排行功能
2025-05-12 16:28:25 +08:00
@filter.command("时长成就排行")
async def timerank(self, event: AstrMessageEvent):
conn=pymysql.connect(host = '192.168.31.9' # 连接名称，默认
,user = 'saipo'
,passwd='Grasste0403' # 密码
,port= 3306 # 端口，默认为3306
,db='saipo' # 数据库名称
,charset='utf8' # 字符编码
)
cur = conn.cursor()
sql = "SELECT du.uid, du.nameid, t.max_timelong, MAX(dj.eventTime) AS eventTime FROM dajiaouser du LEFT JOIN (SELECT openid, MAX(timelong) AS max_timelong FROM dajiao GROUP BY openid) t ON du.openid = t.openid LEFT JOIN dajiao dj ON du.openid = dj.openid AND t.max_timelong = dj.timelong GROUP BY du.uid, du.nameid, t.max_timelong;"
cur.execute(sql)
date = cur.fetchall()
ranklist = []
for i in range(len(date)):
ranklist.append((date[i][0], date[i][1], date[i][2], date[i][3]))
ranklist.sort(key=lambda x: x[2], reverse=True)
ranklist_str = "时长最长排名\n"
for i in range(len(ranklist)):
修改bug
2025-05-12 16:34:15 +08:00
ranklist_str += f"{i+1} uid:{ranklist[i][0]} {ranklist[i][1]} 最长时长：{ranklist[i][2]}s 打胶时间：{ranklist[i][3]}\n"
增加时长排行功能
2025-05-12 16:28:25 +08:00
ranklist_str += "\n"
sql2 = "SELECT du.uid, du.nameid, t.min_timelong, MAX(dj.eventTime) AS eventTime FROM dajiaouser du LEFT JOIN (SELECT openid, MIN(timelong) AS min_timelong FROM dajiao GROUP BY openid) t ON du.openid = t.openid LEFT JOIN dajiao dj ON du.openid = dj.openid AND t.min_timelong = dj.timelong GROUP BY du.uid, du.nameid, t.min_timelong;"
cur.execute(sql2)
date2 = cur.fetchall()
ranklist2 = []
for i in range(len(date2)):
ranklist2.append((date2[i][0], date2[i][1], date2[i][2], date2[i][3]))
ranklist2.sort(key=lambda x: x[2])
ranklist_str += "时长最短排名\n"
for i in range(len(ranklist2)):
修改bug
2025-05-12 16:34:15 +08:00
ranklist_str += f"{i+1} uid:{ranklist2[i][0]} {ranklist2[i][1]} 最短时长：{ranklist2[i][2]}s 打胶时间：{ranklist2[i][3]}\n"
增加时长排行功能
2025-05-12 16:28:25 +08:00
yield event.plain_result(ranklist_str)
cur.close()
conn.close()
增加产量排行榜
2025-05-08 11:48:26 +08:00
testNumber = 0
提交测试内容
2025-04-27 11:29:59 +08:00
@filter.command("测试")
测试变量作用域
2025-04-28 20:31:04 +08:00
async def test(self, event: AstrMessageEvent):
测试变量作用域
2025-04-28 20:46:30 +08:00
global testNumber
测试变量作用域
2025-04-28 20:31:04 +08:00
testNumber = testNumber + 1
yield event.plain_result(f"累积结果: {testNumber}")
创建项目
2025-04-27 03:06:27 +08:00
增加产量排行榜
2025-05-08 11:48:26 +08:00
创建项目
2025-04-27 03:06:27 +08:00
Copy Permalink