saipo/astrbot_plugin_dajiao
2
0
Fork 2
Code Issues Pull Requests Actions Packages Projects Releases 2 Wiki Activity

Compare commits

base: saipo:master
Branches Tags
saipo:master
saipo:ruler229-patch-10
saipo:ruler229-patch-9
saipo:ruler229-patch-8
saipo:ruler229-patch-7
saipo:QAQ
saipo:ruler229-patch-6
saipo:ruler229-patch-5
saipo:ruler229-patch-4
saipo:ruler229-patch-3
saipo:saipo_test
saipo:ruler229-patch-2
saipo:ruler229-patch-1
saipo:develop
saipo:20250523
saipo:20250522
saipo:test
..
compare: saipo:develop
Branches Tags
saipo:master
saipo:ruler229-patch-10
saipo:ruler229-patch-9
saipo:ruler229-patch-8
saipo:ruler229-patch-7
saipo:QAQ
saipo:ruler229-patch-6
saipo:ruler229-patch-5
saipo:ruler229-patch-4
saipo:ruler229-patch-3
saipo:saipo_test
saipo:ruler229-patch-2
saipo:ruler229-patch-1
saipo:develop
saipo:20250523
saipo:20250522
saipo:test

No commits in common. "master" and "develop" have entirely different histories.
master ... develop

3 changed files with 3 additions and 410 deletions
Show Stats Expand all files Collapse all files

72
TicTacToeGame.py
View File

@ -1,72 +0,0 @@
import random
from typing import Optional
class TicTacToeGame:
def __init__(self):
self.board = [" "] * 9
self.player_symbol = "X"
self.computer_symbol = "O"
self.current_player = "X" # X先走
def print_board(self) -> str:
"""返回棋盘字符串"""
board_str = (
f"\n {self.board[0]} | {self.board[1]} | {self.board[2]} \n"
"-----------\n"
f" {self.board[3]} | {self.board[4]} | {self.board[5]} \n"
"-----------\n"
f" {self.board[6]} | {self.board[7]} | {self.board[8]} \n"
)
return board_str
def check_winner(self) -> Optional[str]:
"""检查是否有玩家获胜"""
win_combinations = [
[0, 1, 2], [3, 4, 5], [6, 7, 8], # 横向
[0, 3, 6], [1, 4, 7], [2, 5, 8], # 纵向
[0, 4, 8], [2, 4, 6] # 对角线
]
for combo in win_combinations:
if self.board[combo[0]] == self.board[combo[1]] == self.board[combo[2]] != " ":
return self.board[combo[0]]
return None
def is_board_full(self) -> bool:
"""检查棋盘是否已满"""
return " " not in self.board
def computer_move(self) -> int:
"""电脑AI移动"""
for i in range(9):
if self.board[i] == " ":
self.board[i] = self.computer_symbol
if self.check_winner() == self.computer_symbol:
return i
self.board[i] = " "
for i in range(9):
if self.board[i] == " ":
self.board[i] = self.player_symbol
if self.check_winner() == self.player_symbol:
self.board[i] = self.computer_symbol
return i
self.board[i] = " "
move_order = [4, 0, 2, 6, 8, 1, 3, 5, 7]
for move in move_order:
if self.board[move] == " ":
return move
return -1
def make_move(self, position: int) -> bool:
"""玩家移动"""
if 0 <= position <= 8 and self.board[position] == " ":
self.board[position] = self.current_player
return True
return False
def switch_player(self):
"""切换当前玩家"""
self.current_player = "O" if self.current_player == "X" else "X"

4
back.py
View File

@ -59,8 +59,8 @@ def insertUser(openid):
,charset='utf8'
)
cur = conn.cursor()
sql = "INSERT INTO dajiaouser (openid,nameid,baglist) VALUES (%s,%s,%s)"
cur.execute(sql, (openid,'小jb','{"baglist":[{"id":1,"num":1}]}'))
sql = "INSERT INTO dajiaouser (openid,nameid) VALUES (%s,%s)"
cur.execute(sql, (openid,"小jb"))
conn.commit()
cur.close()
conn.close()

337
main.py
View File

@ -5,17 +5,7 @@ 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
import matplotlib
import matplotlib.pyplot as plt
import os
from .Tool import get_tool_name
import astrbot.api.message_components as Comp
from astrbot.core.utils.session_waiter import (
session_waiter,
SessionController,
)
import aiohttp # 需要异步HTTP客户端
from .TicTacToeGame import TicTacToeGame
@register("helloworld", "YourName", "一个简单的 Hello World 插件", "1.0.0")
class MyPlugin(Star):
@ -47,67 +37,6 @@ class MyPlugin(Star):
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 = [];a = [];b = []
for i in range(0,10):
time.append(round(random.uniform(1, 600), 2))
V.append(round(random.uniform(0.01,100), 2)) if i==0 else V.append(round(random.uniform(1, V[i-1]+1), 2))
a.append(time_long(time[i]))
b.append(volume(V[i]))
sql = "INSERT INTO dajiao (openid, timelong, volume) VALUES (%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s),(%s, %s, %s)"
values = (user_name,time[0],V[0],user_name,time[1],V[1],user_name,time[2],V[2],user_name,time[3],V[3],user_name,time[4],V[4],user_name,time[5],V[5],user_name,time[6],V[6],user_name,time[7],V[7],user_name,time[8],V[8],user_name,time[9],V[9])
cur.execute(sql,values)
conn.commit()
cur.close()
conn.close()
yield event.plain_result(f"Hello,{get_user_name(event.get_sender_id())},准备开始挑战十连咯!\n第一次你坚持了{time[0]}s哦,{a[0]}.射出{V[0]}ml,{b[0]}!\n第二次你坚持了{time[1]}s哦,{a[1]}.射出{V[1]}ml,{b[1]}!\n第三次你坚持了{time[2]}s哦,{a[2]}.射出{V[2]}ml,{b[2]}!\n第四次你坚持了{time[3]}s哦,{a[3]}.射出{V[3]}ml,{b[3]}!\n第五次你坚持了{time[4]}s哦,{a[4]}.射出{V[4]}ml,{b[4]}!\n第六次你坚持了{time[5]}s哦,{a[5]}.射出{V[5]}ml,{b[5]}!\n第七次你坚持了{time[6]}s哦,{a[6]}.射出{V[6]}ml,{b[6]}!\n第八次你坚持了{time[7]}s哦,{a[7]}.射出{V[7]}ml,{b[7]}!\n第九次你坚持了{time[8]}s哦,{a[8]}.射出{V[8]}ml,{b[8]}!\n第十次你坚持了{time[9]}s哦,{a[9]}.射出{V[9]}ml,{b[9]}!\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("我的时长成就")
@ -233,277 +162,13 @@ class MyPlugin(Star):
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()
@filter.command("我的日产量")
async def mydayvolume(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 DAY(eventTime),round(sum(volume),2) FROM dajiao WHERE openid = %s and DATE_SUB(CURDATE(), INTERVAL 7 DAY) <= date(eventTime) GROUP BY DAY(eventTime);"
openid = event.get_sender_id()
cur.execute(sql,(openid,))
date = cur.fetchall()
if len(date) == 0:
cur.close()
conn.close()
yield event.plain_result(f"没有找到数据哦！")
day = []
volume = []
for i in range(len(date)):
day.append(date[i][0])
volume.append(date[i][1])
plt.plot(day, volume)
plt.title('dayvolume')
plt.xlabel('day')
plt.ylabel('volume')
plt.xticks(day)
#设置显示数据
for x, y in zip(day, volume):
plt.text(x, y, f'{y}', ha='center', va='bottom', fontsize=10)
plt.grid()
plt.savefig('dayvolume.png')
plt.close()
with open('dayvolume.png', 'rb') as f:
image_data = f.read()
image = Comp.Image.fromBytes(image_data)
chain = [
Comp.At(qq=event.get_sender_id()),
Comp.Plain("日产量折线图："),
image,
]
yield event.chain_result(chain)
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("Dora")
async def Dora(self, event: AstrMessageEvent):
async with aiohttp.ClientSession() as session:
async with session.get('https://www.doro.asia/api/random-sticker') as resp:
if resp.status == 200:
data = await resp.json()
if data.get('success', False):
image_url = data['sticker']['url']
chain = [
Comp.At(qq=event.get_sender_id()), # At 消息发送者
Comp.Plain("来看这个图："),
Comp.Image.fromURL(image_url), # 使用API返回的图片URL
]
yield event.chain_result(chain)
else:
yield "获取表情包失败"
else:
yield "API请求失败"
@filter.command("来句骚话")
async def lovelive(self, event: AstrMessageEvent):
async with aiohttp.ClientSession() as session:
async with session.get('https://api.lovelive.tools/api/SweetNothings') as resp:
if resp.status == 200:
text = await resp.text() # 获取返回的文本内容
chain = [
Comp.At(qq=event.get_sender_id()), # At 消息发送者
Comp.Plain(text) # 使用API返回的文本
]
yield event.chain_result(chain)
else:
yield "获取骚话失败"
@filter.command("部署")
async def bushutest(self, event: AstrMessageEvent):
yield event.plain_result("自动部署成功！！")
@filter.command("井字棋")
async def tic_tac_toe_handler(self, event: AstrMessageEvent):
"""井字棋游戏处理器"""
try:
# 初始化游戏
game = TicTacToeGame()
# 发送初始说明和空棋盘
instructions = (
"欢迎来到井字棋游戏!\n"
"输入1-9的数字来选择位置:\n"
" 1 | 2 | 3 \n"
"-----------\n"
" 4 | 5 | 6 \n"
"-----------\n"
" 7 | 8 | 9 \n"
"\n当前棋盘:" + game.print_board()
)
yield event.plain_result(instructions)
@session_waiter(timeout=60, record_history_chains=False)
async def game_waiter(controller: SessionController, event: AstrMessageEvent):
nonlocal game
idiom = event.message_str # 用户发来的成语，假设是 "一马当先"
# 玩家回合
if game.current_player == game.player_symbol:
move = int(idiom) - 1
if not game.make_move(move):
await event.send(event.make_result(chain=[Comp.Plain("无效移动，请重试。")]))
controller.keep(timeout=60, reset_timeout=True)
return
else:
# 电脑回合
move = game.computer_move()
game.board[move] = game.current_player
# 检查游戏状态
winner = game.check_winner()
if winner:
result = event.make_result()
result.chain = [
Comp.Plain(f"{game.print_board()}\n"),
Comp.Plain("恭喜你赢了!" if winner == game.player_symbol else "电脑赢了!")
]
await event.send(result)
controller.stop()
return
if game.is_board_full():
result = event.make_result()
result.chain = [
Comp.Plain(f"{game.print_board()}\n"),
Comp.Plain("平局!")
]
await event.send(result)
controller.stop()
return
# 切换玩家并继续游戏
game.switch_player()
# 发送更新后的棋盘
result = event.make_result()
if game.current_player == game.player_symbol:
result.chain = [
Comp.Plain(f"{game.print_board()}\n"),
Comp.Plain("轮到你走了，请输入1-9:")
]
else:
result.chain = [
Comp.Plain(f"{game.print_board()}\n"),
Comp.Plain("电脑思考中...")
]
await event.send(result)
# 重置超时时间
controller.keep(timeout=60, reset_timeout=True)
try:
await game_waiter(event)
except TimeoutError:
yield event.plain_result("井字棋游戏超时结束!")
except Exception as e:
yield event.plain_result(f"游戏出错: {str(e)}")
finally:
event.stop_event()
except Exception as e:
logger.error(f"井字棋游戏错误: {str(e)}")
yield event.plain_result("游戏发生错误，请稍后再试。")
@filter.command("帮我清理")
async def clean(self, event: AstrMessageEvent):
if isUserExist(event.get_sender_id()) != True:
insertUser(event.get_sender_id())
V = round(random.uniform(0.01,100), 2)
b = volume(V)
if V>=10:
yield event.plain_result(f"让我帮你好好清理一下哦~哧溜哧溜,嗯嗯,又射了{V}ml,{b}!")
else:
yield event.plain_result(f"让我帮你好好清理一下哦~哧溜哧溜,哎呀,全都清理干净啦,杂鱼杂鱼!")