Ruler229/astrbot_plugin_dajiao
1
0
Fork 0
forked from saipo/astrbot_plugin_dajiao
Code Pull Requests Activity
astrbot_plugin_dajiao/main.py
QAQ e693a02e51 更新 main.py 
测试
2025-05-21 20:37:22 +08:00

317 lines
15 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import json
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
from .Tool import get_tool_name
@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 dajiaoX(self, event: AstrMessageEvent):
if isUserExist(event.get_sender_id()) != True:
insertUser(event.get_sender_id())
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()
time = []
V = [1]
a = [1]
b = [1]
for i in range(0,10):
time.append(round(random.uniform(1, 600), 2))
#V.append(round(random.uniform(0.01,100), 2))
#a.append(time_long(time))
#b.append(volume(V))
#sql = "INSERT INTO dajiao (openid, timelong, volume) VALUES (%s, %s, %s)"
#values = (user_name,time[i],V[i])
#cur.execute(sql,values)
#conn.commit()
#cur.close()
#conn.close()
#yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())}, 第一次你坚持了{time[0]}s哦，{a[0]}.射出{V[0]}ml,{b[0]}!\n第二次你坚持了{time[1]}s哦，{a[1]}.射出{V[1]}ml,{b[1]}!\n哎呀，一点也打不出来了!")
yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())}, 第一次你坚持了{time[0]}s哦，{a[0]}.射出{V[0]}ml,{b[0]}!\n一点也打不出来了!")
@filter.command("打打你的")
async def dajiao_other(self, event: AstrMessageEvent, message: str):
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}!")
@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["baglist"]) == 0:
baglist_str = "背包空空如也"
else:
for item in baglist["baglist"]:
if get_tool_name(item["id"]) is None:
cur.close()
conn.close()
yield event.plain_result(f"背包存在异常！")
else:
baglist_str += f"{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 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)):
ranklist_str += f"{i+1} uid:{ranklist[i][0]} nameid:{ranklist[i][1]} 最高产量：{ranklist[i][2]}ml 打胶时间：{ranklist[i][3]}\n"
ranklist_str += "\n"
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;"
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)):
ranklist_str += f"{i+1} uid:{ranklist2[i][0]} nameid:{ranklist2[i][1]} 最低产量：{ranklist2[i][2]}ml 打胶时间：{ranklist2[i][3]}\n"
yield event.plain_result(ranklist_str)
cur.close()
conn.close()
@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)):
ranklist_str += f"{i+1} uid:{ranklist[i][0]} {ranklist[i][1]} 最长时长：{ranklist[i][2]}s 打胶时间：{ranklist[i][3]}\n"
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)):
ranklist_str += f"{i+1} uid:{ranklist2[i][0]} {ranklist2[i][1]} 最短时长：{ranklist2[i][2]}s 打胶时间：{ranklist2[i][3]}\n"
yield event.plain_result(ranklist_str)
cur.close()
conn.close()
testNumber = 0
@filter.command("测试")
async def test(self, event: AstrMessageEvent):
global testNumber
testNumber = testNumber + 1
yield event.plain_result(f"累积结果: {testNumber}")
@filter.command("麦否？")
async def QAQtest(self, event: AstrMessageEvent):
yield event.plain_result("包的")
View Git Blame Copy Permalink