wolfenstein-et/etmasters.py

#!/usr/bin/env python3
"""List Wolfenstein: Enemy Territory servers known by the master servers.

Asks each master for its server list (getservers), then every server for its info
(getinfo): name, map, players, mod. Prints a table, or writes JSON with --json.

    ./etmasters.py
    ./etmasters.py --json servers.json
    ./etmasters.py --master etmaster.net:27950
"""
import argparse
import json
import re
import select
import socket
import sys
import time

OOB = b"\xff\xff\xff\xff"
# 84: ET 2.60b / ET: Legacy protocol
GETSERVERS = OOB + b"getservers 84 empty full"
GETINFO = OOB + b"getinfo etmasters"
DEFAULT_MASTERS = ["etmaster.idsoftware.com:27950", "etmaster.net:27950"]
# ^1red ^7white... color codes in server names
COLOR_CODE = re.compile(r"\^.")


def query_master(master, timeout):
    """Server addresses listed by one master, None if it doesn't answer."""
    host, port = master.rsplit(":", 1)
    try:
        ip = socket.gethostbyname(host)
    except OSError:
        return None

    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.settimeout(timeout)
    sock.sendto(GETSERVERS, (ip, int(port)))

    servers, answered = set(), False
    try:
        while True:
            data, _ = sock.recvfrom(65535)
            if b"getserversResponse" not in data:
                continue
            answered = True
            body = data.split(b"getserversResponse", 1)[1]
            # \<4 bytes IP><2 bytes port>\...\EOT
            for entry in body.split(b"\\")[1:]:
                if len(entry) >= 6 and not entry.startswith(b"EOT"):
                    addr = socket.inet_ntoa(entry[:4])
                    servers.add((addr, int.from_bytes(entry[4:6], "big")))
            # Every packet ends with EOT, the list spans several: read until timeout.
    except socket.timeout:
        pass
    finally:
        sock.close()

    return servers if answered else None


def query_infos(addresses, timeout):
    """getinfo to every server at once, {address: info dict} for those answering."""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setblocking(False)
    for addr in addresses:
        try:
            sock.sendto(GETINFO, addr)
        except OSError:
            pass

    infos, deadline = {}, time.monotonic() + timeout
    while len(infos) < len(addresses):
        left = deadline - time.monotonic()
        if left <= 0 or not select.select([sock], [], [], left)[0]:
            break
        try:
            data, addr = sock.recvfrom(65535)
        except OSError:
            continue
        if data.startswith(OOB + b"infoResponse"):
            infos[addr] = parse_info(data)

    sock.close()
    return infos


def parse_info(data):
    """infoResponse\n\\key\\value\\... -> dict."""
    body = data.split(b"\n", 1)[1].decode("latin-1").strip("\\\n\x00")
    parts = body.split("\\")
    return dict(zip(parts[::2], parts[1::2]))


def server_entry(addr, masters, info):
    entry = {"address": f"{addr[0]}:{addr[1]}", "masters": sorted(masters), "online": info is not None}
    if info:
        entry.update(
            name=COLOR_CODE.sub("", info.get("hostname", "")).strip(),
            map=info.get("mapname", ""),
            players=int(info.get("clients", 0) or 0),
            humans=int(info["humans"]) if info.get("humans", "").isdigit() else None,
            max_players=int(info.get("sv_maxclients", 0) or 0),
            # game: mod folder (nq, etpro, legacy...), gamename is always "et"
            mod=info.get("game") or info.get("gamename", ""),
            password=info.get("needpass") == "1",
        )
    return entry


def print_table(entries, masters_status):
    for master, count in masters_status.items():
        state = "no answer" if count is None else f"{count} servers"
        print(f"{master:32} {state}")
    print()

    online = [e for e in entries if e["online"]]
    online.sort(key=lambda e: (-e["players"], e["name"].lower()))

    header = f"{'ADDRESS':21} {'PLAYERS':>7}  {'MAP':16} {'MOD':10} NAME"
    print(header)
    print("-" * len(header))
    for e in online:
        players = f"{e['players']}/{e['max_players']}"
        lock = " [pw]" if e["password"] else ""
        print(f"{e['address']:21} {players:>7}  {e['map'][:16]:16} {e['mod'][:10]:10} {e['name'][:60]}{lock}")

    total = sum(e["players"] for e in online)
    print(f"\n{len(online)} servers online ({len(entries) - len(online)} not answering), {total} players")


def main():
    parser = argparse.ArgumentParser(description="List ET servers known by the master servers.")
    parser.add_argument("--master", action="append", metavar="HOST:PORT",
                        help=f"master to query, repeatable (default: {', '.join(DEFAULT_MASTERS)})")
    parser.add_argument("--json", metavar="PATH", help="write servers as JSON to PATH instead of printing a table")
    parser.add_argument("--timeout", type=float, default=3.0, help="seconds to wait for answers (default: 3)")
    args = parser.parse_args()

    masters = args.master or DEFAULT_MASTERS

    masters_status, listed_by = {}, {}
    for master in masters:
        servers = query_master(master, args.timeout)
        masters_status[master] = None if servers is None else len(servers)
        for addr in servers or ():
            listed_by.setdefault(addr, set()).add(master)

    infos = query_infos(list(listed_by), args.timeout)
    entries = [server_entry(addr, ms, infos.get(addr)) for addr, ms in listed_by.items()]

    if args.json:
        with open(args.json, "w") as f:
            json.dump({"masters": masters_status, "servers": entries}, f, indent=2, ensure_ascii=False)
        print(f"{len(entries)} servers written to {args.json}", file=sys.stderr)
    else:
        print_table(entries, masters_status)


if __name__ == "__main__":
    main()