开了一个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问题，忽略证书

搜索