QQ频道机器人开源-蓝图计划
作者：互联网
蓝图计划 是 [镜芯] 旗下由一铭负责开发的QQ频道图片对战机器人
项目提示
由于编写的比较仓促，很多细节都没有去处理，需要你们自己去填坑，就当提升一下技能吧
项目结构
xiangsu.py 是图像处理模块，负责处理所有图像事务
ymsql.py 是游戏逻辑以及数据库接口
xs_bot.py 负责与频道机器人建立连接，传受数据，控制对局
bk-demo2.png 一个背景模板
运行流程
xs_bot.py 收到消息后，初始化消息，处理完毕后传入到ymsql.py
ymsql.py 对传入的消息进行处理，传出处理结果
xiangsu.py 生成对局图片到缓存路径，返回图片名称，url
坑位提示
1.SQL未每次执行都写入数据库
2.SQL注入并未完美解决
3.ymsql.py写的太垃圾了
4.需要自己去修改胜利条件，对局人数，用你善于发现的眼睛找一找
5.由于数据库版本及数据库自身问题的原因，可能存在神奇的问题
6.如果运行模式改为异步
7.没有做全局配置
8.你可能没有好好看这个文档
运行环境
运行环境：
环境：无限制 Python3.7 Mysql5.6
一个web服务器
一个可用的qq频道机器人账号
申请地址：https://q.qq.com/#/
（其他应用场景需要自己对接xiangsu.py模块去开发使用）
安装方法
安装方法
1.安装web服务器，安装mysql，安装python3.7，可使用宝塔一键安装
2.修改ymsql.py内数据库相关信息
3.修改xs_bot.py内机器人botid和token
4.修改所有源码内文件缓存路径
5.安装依赖Pillow qq-bot
6.导入xiangsu.sql
运行主程序
python xs_bot.py
源码下载
部分代码展示
# 数据库操作方法
class AdminSql:
'''
playid : 对局ID
membid : 对局成员ID
'''
# 链接数据库
def __init__(self,playid='',membid=''):
self.db_con = pymysql.connect(host='localhost',port=3306,db=db,user=user,password=password)
self.cursor = self.db_con.cursor()
# 用户顺序颜色列表
self.user_colors = {1:(106,123,222),2:(106,196,222)
,3:(42,100,114),4:(87,87,87)
,5:(199,75,75),6:(76,205,97)}
# 关闭数据库连接
def Closedb(self):
self.db_con.close()
# 创建一个对局
# 需传入 用户id，用户昵称，用户头像url，频道id，频道昵称
# 根据用户参加游戏的顺序给定用户颜色，默认为第一个(106,123,222)
def Creat_Play(self,uid,name,author,chanid,chanme):
# 首先查询用户是否在对局中
seart = self.Saech_User_Play_stau(uid)
user_stau = int(seart['code'])
# 202则表示该用户不存在
# 可进行对局创建
# 同时初始化写入该用户信息
if user_stau == 202:
# 获取一个可用的gameid
gameid = self._Get_New_Gameid()
# 对局创建的时间
create = int(time.time())
# 写入一个新的对局
#new_games = '''INSERT INTO `xiangsu`.`Ex_games` (`id`, `gameid`, `users`, `status`, `modtime`, `gametype`, `mateid`, `souruse`, `create`, `templ`) VALUES (null, '%s', '[{"id":"%s","name":"%s","author":"%s","color":(106,123,222),"had":"0"}]', '1', '%s', '0', '{"id":"%s","name":"%s"}', '{"id":"%s","name":"%s"}', '%s', '0')'''%(gameid,uid,name,author,create,chanid,chanme,uid,name,create)
new_games = '''INSERT INTO `xiangsu`.`Ex_games` (`gameid`, `users`, `status`, `modtime`, `gametype`, `mateid`, `souruse`, `create`, `templ`) VALUES (%s, '[{"id":"%s","name":"%s","author":"%s","color":(106,123,222),"had":"0"}]', 1, '%s', 0, '{"id":"%s","name":"%s"}', '{"id":"%s","name":"%s"}', '%s', '0')'''%(gameid,uid,name,author,create,chanid,chanme,uid,name,create)
print(new_games)
self.cursor.execute(new_games)
# 新增一个用户
self._Insert_user_info(uid,name,chanid,chanme,gameid)
# 设置对局状态为等待中
self.Set_Game_Done(gameid,1)
msg = {"code":200,"msg":"创建对局成功","gameid":gameid}
return msg
# 203表示参数无效
if user_stau == 203:
msg = {"code":203,"msg":"接收到了无效参数"}
return msg
# 201表示用户存在，但不在对局中
# 为用户创建新对局，更新用户对局信息
if user_stau == 201:
# 获取一个最新的可用对局ID
gameid = self._Get_New_Gameid()
# 对局创建的时间
create = int(time.time())
# 写入一个新的对局
#new_games = '''INSERT INTO `xiangsu`.`Ex_games` (`id`, `gameid`, `users`, `status`, `modtime`, `gametype`, `mateid`, `souruse`, `create`, `templ`) VALUES (null, '%s', '[{"id":"%s","name":"%s","author":"%s","color":(106,123,222),"had":"0"}]', '1', '%s', '0', '{"id":"%s","name":"%s"}', '{"id":"%s","name":"%s"}', '%s', '0')'''%(gameid,uid,name,author,create,chanid,chanme,uid,name,create)
new_games = '''INSERT INTO `xiangsu`.`Ex_games` (`gameid`, `users`, `status`, `modtime`, `gametype`, `mateid`, `souruse`, `create`, `templ`) VALUES (%s, '[{"id":"%s","name":"%s","author":"%s","color":(106,123,222),"had":"0"}]', 1, '%s', 0, '{"id":"%s","name":"%s"}', '{"id":"%s","name":"%s"}', '%s', '0')'''%(gameid,uid,name,author,create,chanid,chanme,uid,name,create)
print(new_games)
self.cursor.execute(new_games)
# 更新用户的对局状态
self._Update_User_Game(uid,0,chanid,chanme,gameid)
# 设置对局状态为等待中
self.Set_Game_Done(gameid,1)
msg = {"code":200,"msg":"创建对局成功","gameid":gameid}
return msg
# 200表示用户在对局中
# 返回对局信息
if user_stau == 210:
return seart
# 加入对局 要传入gameid uid username author chanid chanme
# 传出加入对局是否成功
def Insert_Game(self,gameid,uid,username,author,chanid,chanme):
# 先查询对局成员、最后修改时间、对局所在频道id
self.cursor.execute("SELECT users,modtime,mateid,status FROM `Ex_games` WHERE `gameid`=%s"%(gameid))
sear = self.cursor.fetchall()
# 对局所在频道
# 若对局不存在
if len(sear) == 0:
msg = {"code":202,"msg":"该对局不存在"}
return msg
# 若对局已结束
if int(sear[0][3]) == 2:
return {"code":204,"msg":"该对局已结束"}
# 若对局正在进行中
if int(sear[0][3]) == 0:
return {"code":209,"msg":"该对局已结束"}
# 判断玩家是否在对局中
user_info = json.loads(json.dumps(self.Saech_User_Play_stau(uid)))
user_stau = user_info['code']
# 若玩家在对局中
# 返回对局信息，结束语句
if user_stau == 210:
msg = user_info
return msg
# 若玩家不存在，创建玩家
if user_stau == 202:
self._Insert_user_info(uid,username,chanid,chanme,gameid)
# 其他情况下，认为玩家已存在，并且不在游戏中
game_chanid = int(json.loads(sear[0][2])['id'])
# 若对局所在频道不是该频道
if game_chanid != int(chanid):
msg = {"code":201,"msg":"该对局不在当前频道,请选择频道内的对局或创建对局"}
return msg
# 若对局所在频道属于该频道
if game_chanid == int(chanid):
sear = json.loads(json.dumps(self.Search_Game_Stau(gameid)))
# 判断对局人数是否已满
game_user_num = literal_eval(sear['data'][2])
if len(game_user_num) >= 6:
msg = {"code":203,"msg":"当前对局人数已满,请选择其他对局或发起对局"}
return msg
# 人数未满则将玩家加入该对局
else:
# 新加入玩家的信息
# 新玩家的颜色
new_color = self.user_colors[len(game_user_num)+1]
new_user_info = {"id":f"{uid}","name":f"{username}","author":f"{author}","color":new_color,"had":"0"}
game_user_num.append(new_user_info)
# 对局更新的时间
update = int(time.time())
upda_users = '''UPDATE `xiangsu`.`Ex_games` SET `users` = "%s" WHERE `Ex_games`.`gameid` = %s'''%(str(game_user_num),gameid)
upda_update = '''UPDATE `xiangsu`.`Ex_games` SET `modtime` = '%s' WHERE `Ex_games`.`gameid` = %s'''%(update,gameid)
self.cursor.execute(upda_users)
self.cursor.execute(upda_update)
self.db_con.commit()
# 更新玩家对局状态
self._Update_User_Game(uid,0,chanid,chanme,gameid)
# index为6，表示人数已满，开始对局
# 否则继续等待玩家
if len(game_user_num) == 6:
# 设定对局状态为进行中
self.Set_Game_Done(gameid,0)
msg = {"code":200,"msg":f"玩家 {username} 已加入对局 {gameid}","color":f"{new_color}","index":len(game_user_num)}
return msg
# 更新用户的对局状态
def _Update_User_Game(self,uid,code,chanid,chanme,gameid):
# 更新用户的对局状态
upt_user = '''UPDATE `xiangsu`.`Ex_user_info` SET `whplay` = '{"code":%s,"chanid":"%s","channame":"%s","gameid":%s}' WHERE `Ex_user_info`.`id` = %s'''%(code,chanid,chanme,gameid,uid)
self.cursor.execute(upt_user)
self.db_con.commit()
msg = {"code":201,"msg":"更改用户对局状态","gameid":gameid}
return msg
# 新增一个用户
def _Insert_user_info(self,uid,name,chanid,chanme,gameid):
# 新增一个用户
new_user = '''INSERT INTO `xiangsu`.`Ex_user_info` (`id`, `name`, `gamenub`, `gamevic`, `whplay`) VALUES (%s, '%s', '0', '0', '{"code":0,"chanid":"%s","channame":"%s","gameid":%s}')'''%(uid,name,chanid,chanme,gameid)
print("新增用户语句：",new_user)
self.cursor.execute(new_user)
self.db_con.commit()
msg = {"code":200,"msg":f"新增用户 {name}"}
return msg
# 查询对局所有信息 要传入gameid
def Search_Game_Stau(self,gameid):
self.cursor.execute(f"SELECT * FROM `Ex_games` WHERE `gameid`={gameid}")
sear = self.cursor.fetchall()
if len(sear) == 0:
msg = {"code":201,"msg":"未查询到对局信息"}
else:
numbp = literal_eval(sear[0][2])
templ = sear[0][-1]
msg = {"code":200,"data":list(sear[0]),"num":len(numbp),"users":numbp,"templ":f"{templ}"}
return msg
# 获得一个可用的对局ID
# 创建对局需要用到
def _Get_New_Gameid(self):
self.cursor.execute(f"SELECT MAX(gameid) from Ex_games")
return self.cursor.fetchall()[0][0] + 1
# 查询用户是否在对局中
# 需传入用户id
def Saech_User_Play_stau(self,uid):
if uid == '' or uid == None:
msg = {"code":203,"msg":"无效参数"}
# 从user表中查询
self.cursor.execute(f"SELECT * FROM `Ex_user_info` WHERE `id`={uid}")
# 查询数据
seardt = self.cursor.fetchall()
# 为0则不存在
if len(seardt) == 0:
msg = {"code":202,"msg":"未查询到该用户"}
else:
# 用户对局信息
st = json.loads(seardt[0][4])
# 用户昵称
user = seardt[0][1]
# 格式化输出
if st['code'] == 0:
chanid,chanme,gameid = st['chanid'],st['channame'],st['gameid']
msg = {"code":210,"msg":f"{user} 正在频道 {chanme} 进行对战","chanid":chanid,"chanme":f"{chanme}","gameid":gameid}
if st['code'] == 1:
msg = {"code":201,"msg":f"{user} 未在对局中"}
# 关闭数据库
return msg
# 开始一个新对局 需要gameid
# 确保对应对局已拥有 6 人
# 如果头像重复，则只生成一个头像
def Start_Game(self,gameid):
# 先获取玩家列表
users = self.Search_Game_Stau(gameid)['users']
# 获取玩家头像列表 传入像素模块使用
authors = []
for i in users:
authors.append(i['author'])
# 生成对局图片
game_back = xiangsu.BackGround(aust=authors)
game_back = json.loads(json.dumps(game_back.Author_Add()))
print(game_back)
# 对局图片及url
templ_url = game_back['url']
templ = game_back['templ']
# 更新对局templ 对局信息 对局状态
upd_templ = '''UPDATE `xiangsu`.`Ex_games` SET `templ` = '%s' WHERE `Ex_games`.`gameid` = %s'''%(templ,gameid)
upd_status = '''UPDATE `xiangsu`.`Ex_games` SET `status` = '%s' WHERE `Ex_games`.`gameid` = %s'''%(0,gameid)
self.cursor.execute(upd_templ)
self.cursor.execute(upd_status)
self.db_con.commit()
return game_back
# 占领一个方块 生成一个像素格
# 传入uid 根据uid来查找对局图片name，判断玩家是否有对局
# 需要 指定bt=1 传入像素位置gratt 传入像素颜色gratcolor
def Game_Add_grat(self,uid,gratt,chanid):
# 先判断玩家对局状态，获取对局templ
user_info = self._Search_Uidto_Templ(uid)
if user_info['code'] == 202 or user_info['code'] == 201 or user_info['code'] == 203:
return user_info
if user_info['code'] == 200:
# 判断对局是否结束
if user_info['data'][3] == 2:
return {"code":205,"msg":"该对局已结束"}
# 若对局在等待中
if user_info['data'][3] == 1:
user_info = self.Saech_User_Play_stau(uid)
gameid = user_info['gameid']
self.cursor.execute(f"SELECT * FROM `Ex_games` WHERE `gameid`={gameid}")
sear = self.cursor.fetchall()
numbp = literal_eval(sear[0][2])
return {"code":206,"msg":"该对局在等待中,您可以先加入对局","num":len(numbp)}
gamesuse = self.Saech_User_Play_stau(uid)
if int(gamesuse['code']) == 210 and int(gamesuse['chanid']) == int(chanid):
return {"code":206,"msg":"该对局在等待中,您可以先加入对局"}
else:
return {"code":207 ,"msg":"请前往对局所在的频道进行对战"}
templ = user_info['templ']
# 获取玩家方块颜色
#print("ymsql输出：",uid,user_info['data'][2],templ)
grat_color = self._Search_Usergrat_Color(uid,user_info['data'][2])
# 实例化
# 新的templ
templ2 = "xiangsu"+ str(time.time()) + ".png"
game_admin = xiangsu.BackGround(bt=1,gratt=gratt,gratcolor=grat_color,name=templ,name2=templ2)
print("ymsql:",templ,templ2,gratt,grat_color)
# 生成像素格
add_grat = game_admin.Add_grat()
# 获取gameid
gameid = self.Saech_User_Play_stau(uid)['gameid']
# 更新对局templ
upd_templ = '''UPDATE `xiangsu`.`Ex_games` SET `templ` = '%s' WHERE `Ex_games`.`gameid` = %s'''%(templ2,gameid)
print("更新新的templ：",upd_templ)
self.cursor.execute(upd_templ)
self.db_con.commit()
# 给玩家增加一个占领的方块
incgrat = self._Inc_Game_grat(uid,user_info['data'][2])
# 获取对局上一次的修改时间
# modtime = self._Get_Game_Modif(uid)
# 判断玩家是否获胜 ： 占领144个方块
if int(self.Search_user_grat_Num(uid)) >= 30:
msg = {"code":211,"msg":"对局已结束","king":uid}
# 为对局玩家增加对局次数
gameid = self.Saech_User_Play_stau(uid)['gameid']
self.Insert_Game_User_king(uid,gameid)
# 更改对局中所有玩家的状态为不在对局中
# 更改对局状态为已结束
self.Set_Game_User_Done(uid,chanid)
self.Set_Game_Done(gameid,2)
return msg
else:
msg = {"code":201,"msg":"占领成功"}
return add_grat
# 查找uid对应的对局templ
# 同时判断玩家的对局状态
def _Search_Uidto_Templ(self,uid):
data = self.Saech_User_Play_stau(uid)
# 210说明玩家有对局，继续查找templ
if data['code'] == 210:
templ = self.Search_Game_Stau(data['gameid'])
return templ
# 玩家没有参与过对局
if data['code'] == 202:
return data
# 玩家当前不在对局中
if data['code'] == 201:
return data
# 获取玩家方块颜色 需要传入uid和users信息
def _Search_Usergrat_Color(self,uid,users):
users = literal_eval(users)
for i in users:
dt = json.loads(json.dumps(i))
if int(dt['id']) == int(uid):
color = dt['color']
print("ymsql:",color)
return tuple(color)
# 给玩家增加一个占领的方块
def _Inc_Game_grat(self,uid,users):
users = literal_eval(users)
busers = []
for i in users:
dt = json.loads(json.dumps(i))
if int(dt['id']) == int(uid):
dt['had'] = str(int(dt['had']) + 1)
busers.append(dt)
inc_grat = '''UPDATE `xiangsu`.`Ex_games` SET `users` = "%s" WHERE `Ex_games`.`gameid` = %s'''%(busers,self.Saech_User_Play_stau(uid)['gameid'])
self.cursor.execute(inc_grat)
self.db_con.commit()
# 获取对局上一次的修改时间
# 根据uid，用于占领方块格子时使用
def _Get_Game_Modif(self,uid):
data = self.Saech_User_Play_stau(uid)
# 210说明玩家有对局，继续查找templ
if data['code'] == 210:
modtime = self.Search_Game_Stau(data['gameid'])
return modtime['data'][4]
# 玩家没有参与过对局
if data['code'] == 202:
return data
# 玩家当前不在对局中
if data['code'] == 201:
return data
# 获取玩家方块数量
def Search_user_grat_Num(self,uid):
user_info = self._Search_Uidto_Templ(uid)
users = literal_eval(user_info['data'][2])
for i in users:
dt = json.loads(json.dumps(i))
if int(dt['id']) == int(uid):
gratnum = dt['had']
return int(gratnum)
# 为对局中的胜出玩家增加一次胜利次数和对局次数
# 所有玩家增加一次对局次数
# king为胜利玩家uid
def Insert_Game_User_king(self,king,gameid):
data = self.Search_Game_Stau(gameid)
self._Update_user_Games_king(king)
for i in literal_eval(data['data'][2]):
uid = json.loads(json.dumps(i))['id']
self._Update_user_Games(uid)
# 为玩家增加一次对局次数
def _Update_user_Games(self,uid):
updt = '''UPDATE `xiangsu`.`Ex_user_info` SET `gamenub` = gamenub+1 WHERE `Ex_user_info`.`id` = %s'''%(uid)
self.cursor.execute(updt)
self.db_con.commit()
# 为玩家增加一次胜利次数
def _Update_user_Games_king(self,uid):
updt = '''UPDATE `xiangsu`.`Ex_user_info` SET `gamevic` = gamevic+1 WHERE `Ex_user_info`.`id` = %s'''%(uid)
self.cursor.execute(updt)
self.db_con.commit()
# 设置对局状态
# 0进行中 1等待中 2已结束
def Set_Game_Done(self,gameid,sta):
updt = '''UPDATE `xiangsu`.`Ex_games` SET `status` = '%s' WHERE `Ex_games`.`gameid` = %s'''%(sta,gameid)
self.cursor.execute(updt)
self.db_con.commit()
# 设置对局中所有玩家状态为不在对局中
def Set_Game_User_Done(self,uid,chanid):
# 获取gameid
gameid = self.Saech_User_Play_stau(uid)['gameid']
dataa = self.Search_Game_Stau(gameid)
chanme = json.loads(dataa['data'][6])['name']
users = dataa['users']
# 获取所有id
#ids = []
for i in users:
#ids.append(i['id'])
self._Update_User_Game(i['id'],1,chanid,chanme,gameid)
# 背景图像操作类
class BackGround:
'''
必须在满足以下条件后调用
1 拥有六个玩家
'''
def __init__(self,aust='',name='',path='',bt=0,gratt='',gratcolor='',name2=''):
'''
path :存储路径
name :图片名称 要打开的图像，已有的对局需要此参数
bt :0需要生成背景 1需要打开已有背景
aust :玩家头像列表
gratt :要生成的像素格的位置 如(1,1)
gratcolor : 要生成的像素格的颜色
'''
# 像素格颜色
self.gratcolor = gratcolor
# 生成要存储的背景名称
# 或要打开的图片
if name == '' or name == None:
self.name = "xiangsu" + str(time.time()) + ".png"
else:
self.name = name
# 图片最新的名字
try:
self.name2 = name2
except:
pass
if path == '' or path == None:
self.path = "C:/wwwroot/xiangsu.siriusbot.cn/cache/" + self.name
else:
self.path = path
# 生成新背景或使用已有背景
if bt == 0:
self.back = Image.open("bk-demo2.png")
self.aust = aust
elif bt == 1:
self.back = Image.open(self.path)
# 头像处理后的大小 像素
self.px, self.py = 50,50
# 颜色列表
self.Authors_Colors = [(106,123,222),(106,196,222)
,(42,100,114),(87,87,87)
,(199,75,75),(76,205,97)]
# 每个玩家的头像生成位置
self.Authors_Add = [(26,16),(127,16)
,(227,16),(26,82)
,(127,82),(227,82)]
# 像素格的大小 默认15
self.grat = (15,15)
# 每个像素格的x位置
self.gratsx = {1:41,2:60,3:79,4:98,5:117,6:136,7:156,8:175,9:194,10:213,11:232,12:251}
# 每个像素格的y位置
self.gratsy = {1:157,2:176,3:195,4:213,5:232,6:251,7:270,8:289,9:308,10:327,11:346,12:365}
# 输入的要生成的像素格位置
self.gratt = gratt
# 生成一个像素格
def Add_grat(self):
'''
需要初始化传入gratt
'''
# 生成方块 传入大小，颜色
grat = Image.new('RGBA',self.grat,self.gratcolor)
# 计算要生成像素格的位置
try:
grtobk = (int(self.gratsx[int(self.gratt[0])]),int(self.gratsy[int(self.gratt[1])]))
except KeyError:
return {"code":100,"msg":"位置超出了预期"}
# 生成到背景图的位置
self.back.paste(grat,grtobk,grat)
self.back.save("C:/wwwroot/xiangsu.siriusbot.cn/cache/" + self.name2)
print("xiangsu:图片的新名字：",self.name2)
print("xiangsu:图片的旧名字：",self.name)
self.back.close()
return {"code":200,"url":f"https://xiangsu.siriusbot.cn/cache/{self.name2}","templ":f"{self.name2}"}
#grat.close()
# 下载头像
def _Get_Author(self):
'''
传入一个头像url 传出一个Image对象
author :头像URL
'''
author = request.urlopen(unquote(self.author))
authorio = io.BytesIO(author.read())
self.author = Image.open(authorio)
# 处理gif，取第一帧
if "Content-Type: image/gif" in str(author.headers):
self.author.seek(0)
self.author = self.author.convert("RGBA")
print("已处理该gif")
author.close()
self.author = self.author.resize((self.px,self.py))
# 处理头像圆角
self._Author_Circle()
# 处理头像圆角
def _Author_Circle(self):
'''
需要传入头像对象到self.author
'''
alpha_layer = Image.new('L', (self.px, self.py), 0)
draw = ImageDraw.Draw(alpha_layer)
draw.ellipse((0,0,self.px,self.py), fill=255)
self.author.putalpha(alpha_layer)
#alpha_alyer.close()
# 将头像合成到背景中指定位置
def _Author_To_Back(self):
'''
需提前传入头像对象到self.author
'''
self.back.paste(self.author,self.autobk,self.author)
# 生成玩家头像到demo背景
def Author_Add(self):
try:
if len(self.aust) > 6:
return "aust数量超出预期"
for self.author in self.aust:
if self.author != None:
self.autobk = self.Authors_Add[self.aust.index(self.author)]
self._Get_Author()
self._Author_To_Back()
#测试用
# self.back.save("adddemo.png")
# 实际使用
self.back.save(self.path)
self.back.close()
authorurl = "https://xiangsu.siriusbot.cn/cache/" + self.name
return {"code":200,"url":f"{authorurl}","templ":self.name}
except:
return {"code":100}
标签：QQ,gameid,uid,对局,self,蓝图,开源,user,msg 来源： https://www.cnblogs.com/ymer/p/16351423.html