Ping servers in the servers.dat

ping_servers.py

#!/usr/bin/env python3
# pip3 install nbt mcstatus


"""
Ping servers in the servers.dat

Works on Mac, should work on Windows and Linux.
"""


from nbt.nbt import NBTFile
import os
from mcstatus import MinecraftServer
from collections import namedtuple
import platform
import multiprocessing


Server = namedtuple('Server', ['name', 'ip', 'port'])
PingResult = namedtuple('PingResult', ['server', 'status', 'error'])


MINECRAFT_DIRS = {
    'Darwin': os.path.join(os.getenv('HOME'),
                           'Library/Application Support/minecraft'),
    'Linux': os.path.join(os.getenv('HOME'),
                          '.minecraft'),
    'Windows': os.path.join(os.getenv('HOME'),
                            '.minecraft'),
}


def get_minecraft_dir():
    return MINECRAFT_DIRS[platform.system()]


def get_servers():
    minecraft_dir = get_minecraft_dir()
    os.chdir(minecraft_dir)
    with open('servers.dat', 'rb') as buf:
        s = NBTFile(buffer=buf)
    for server in s['servers']:
        name = server['name'].valuestr()
        address = server['ip'].valuestr()
        ip, port = get_host_port(address)
        yield Server(name=name, ip=ip, port=port)


def get_host_port(address):
    address = address.strip(' \t')
    if address[0] == '[' or address.count(':') == 1:
        host, _, port = address.rpartition(':')
        return host.strip('[]'), int(port)
    return address, 25565


def ping_server(server):
    s = MinecraftServer(server.ip, server.port)
    try:
        res = s.status()
        error = None
    except IOError as ex:
        res = None
        error = ex
    return PingResult(server, res, error)


def ping_servers(servers):
    for server in servers:
        yield ping_server(server)


def ping_servers_async(servers):
    pool = multiprocessing.Pool(10)
    yield from pool.map(ping_server, servers)


def main():
    for s in ping_servers_async(get_servers()):
        head = '{0} ({1}:{2})'.format(
            s.server.name,
            s.server.ip,
            s.server.port,
        )
        if not s.status:
            body = 'is offline.'
        else:
            body = 'has {0}/{1} players, {2} ms lag.'.format(
                s.status.players.online,
                s.status.players.max,
                round(s.status.latency),
            )
        print(' '.join([head, body]))


if __name__ == '__main__':
    main()