开了一个Grasscutter服务器,玩的人比较少,然后虽然有了网页版webtools但是查看在线玩家还是不是很方便。
于是,我基于liujiaqi7998/GrasscuttersWebDashboard: Grasscutters的WEB控制面板 (github.com)插件开发了基于他的websocket协议通信的上下线通知机器人。
我是采用websocket-client这个包来进行操作的。
pip install websocket-client
因为requests那个包虽然做爬虫请求之类的比较常用,但是他都是建立在http协议上的,并不支持wss协议。
另外win7或win server2008 等一些win可能会报错:
[WinError 10042] 在 getsockopt 或 setsockopt 调用中指定的一个未知的、无效的或不受支持的选项或层次。
注释掉/site-packages/websocket/_socket.py中
# if hasattr(socket, "TCP_KEEPCNT"): # DEFAULT_SOCKET_OPTION.append((socket.SOL_TCP, socket.TCP_KEEPCNT, 3))这两行即可。
直接放代码:
机器人采用go-cqhttp
代码中带有【】的部分需要填入对应信息才能使用。
原理:先建立websocket连接,然后监听消息。如果收到类似“json_info['eventName'] == 'cmd_msg'”这种信息的时候(比如XXX已连接或者XXX已断开连接),就会发送一条websocket来查询当前在线玩家信息。之后就会与两个列表:uid_list和player_list进行比较,查看多出来的uid和player或者少了的uid和player,然后如果多出来了就发送已上线的信息,少了就发送已下线的信息,并从对应的列表变量中移除,保证列表中的变量信息和实际上服务器上的人数一致。
这种方式可以避免一些异常情况,比如虽然显示XXX已连接,但是实际上并没有连接,或者是新创建的账号,这时候实际上是没有建立连接的,跟别提断开连接了,所以靠这个参考就会有一些错误。
""" @FILE_NAME : websocket_Test -*- coding : utf-8 -*- @Author : Zhaokugua @Time : 2022/5/21 18:09 """ import websocket import ssl import json from time import sleep import requests # 发送群消息封装函数 def send_group_message(group_id, message): url = 'http://【CQhttp机器人地址】/send_group_msg' params = { 'group_id': group_id, 'message': message } try: requests.get(url, params=params) except: return False return True class Player: def __init__(self, uid, nickname): self.uid = uid self.nickname = nickname def send_message(self, message): message = f'用户{self.nickname}({self.uid}){message}!\n当前服务器共{len(uid_list)}人:' for player in player_list: message = message + f'\n{player.nickname}({player.uid})' send_group_message('【要发送提醒消息的群号】', message) uid_list = [] # 已上线的uid列表 player_list = [] # 已上线的player实例 def on_message(ws, message): # 连接wss通信之后接受到消息会执行 # print(message) player_online_message(message) def on_error(ws, error): # 出现错误时会执行 print(error) def on_close(ws, *args): # 关闭wss连接时会执行 print('连接已关闭。') print(args) def on_open(ws): # 开启wss连接时会执行 pass def player_online_message(message): json_info = json.loads(message) req = { 'type': 'Player', 'data': 0, } if json_info['eventName'] == 'cmd_msg': print(json_info) data_info = json_info['data'] if '已连接' in data_info: sleep(5) # 立即发送可能还没连接上导致获取不到,睡一会 ws.send(json.dumps(req)) # 发送消息获取上线的用户信息 elif '已断开连接' in data_info: sleep(5) # 立即发送可能还没连接上导致获取不到,睡一会 ws.send(json.dumps(req)) # 发送消息获取下线的用户信息 if json_info['eventName'] == 'PlayerList': print(json_info) players = json_info['data'] temp_uids = [] # 临时存放本次获取信息中所有的uid for player in players: temp_uids.append(player['uid']) if player['uid'] in uid_list: pass else: uid_list.append(player['uid']) ply = Player(uid=player['uid'], nickname=player['nickname']) player_list.append(ply) ply.send_message('已上线') lost_uid = set(uid_list) - set(temp_uids) # 获取缺少的玩家 for uid in lost_uid: for player in player_list: if player.uid == uid: player_list.remove(player) uid_list.remove(uid) player.send_message('已下线') # print(json_info) if __name__ == '__main__': websocket.enableTrace(False) ws = websocket.WebSocketApp("wss://【grasscutter的地址】/Dashboard/GrasscuttersWebDashboard", on_message=on_message, on_error=on_error, on_close=on_close) ws.on_open = on_open ws.run_forever(ping_timeout=30, sslopt={"cert_reqs": ssl.CERT_NONE}) # 解决SSL问题,忽略证书
本文地址:https://blog.jixiaob.cn/?post=84
版权声明:若无注明,本文皆为“赵苦瓜のBlog~”原创,转载请保留文章出处。