1073775152: "server_name",

536904239: "xuid",

536904219: "unknown_int64_1",

1073775141: "server_desc",

268468743: "map_id",

1073775142: "map_name",

1073775143: "map_hash_1",

1073775144: "gametype_name",

268468744: "unknown_int32_1",

268468745: "gametype_id",

268468746: "map_id_2",

1073775145: "map_name_2",

1073775146: "map_hash_2",

1073775147: "gametype_name_2",

268468747: "unknown_int32_2",

268468748: "unknown_int32_3",

268468749: "unknown_int32_4",

268468750: "version_1",

268468751: "version_2",

268468752: "party_privacy",

268468753: "game_status",

268468754: "unknown_int32_6",

268468755: "unknown_int32_7",

1: ("00a_introduction", "The Heretic"),

101: ("01a_tutorial", "Armory"),

105: ("01b_spacestation", "Cairo Station"),

301: ("03a_oldmombasa", "Outskirts"),

305: ("03b_newmombasa", "Metropolis"),

401: ("04a_gasgiant", "The Arbiter"),

405: ("04b_floodlab", "Oracle"),

501: ("05a_deltaapproach", "Delta Halo"),

505: ("05b_deltatowers", "Regret"),

601: ("06a_sentinelwalls", "Sacred Icon"),

605: ("06b_floodzone", "Quarantine Zone"),

701: ("07a_highcharity", "Gravemind"),

801: ("07b_forerunnership", "High Charity"),

705: ("08a_deltacliffs", "Uprising"),

805: ("08b_deltacontrol", "The Great Journey"),

80: ("ascension", "Ascension"),

1201: ("backwash", "Backwash"),

100: ("beavercreek", "Beaver Creek"),

60: ("burial_mounds", "Burial Mounds"),

110: ("coagulation", "Coagulation"),

70: ("colossus", "Colossus"),

1300: ("containment", "Containment"),

10: ("cyclotron", "Ivory Tower"),

1302: ("deltatap", "Sanctuary"),

1400: ("derelict", "Desolation"),

3001: ("derelict", "Desolation"),

1200: ("dune", "Relic"),

1001: ("elongation", "Elongation"),

120: ("foundation", "Foundation"),

1002: ("gemini", "Gemini"),

800: ("headlong", "Headlong"),

1402: ("highplains", "Tombstone"),

3000: ("highplains", "Tombstone"),

50: ("lockout", "Lockout"),

20: ("midship", "Midship"),

444678: ("needle", "Uplift"),

91101: ("street_sweeper", "District"),

1101: ("triplicate", "Terminal"),

1000: ("turf", "Turf"),

1109: ("warlock", "Warlock"),

40: ("waterworks", "Waterworks"),

30: ("zanzibar", "Zanzibar"),

0: "None",

1: "CTF",

2: "Slayer",

3: "Oddball",

4: "KOTH",

5: "Race",

6: "Headhunter",

7: "Juggernaut",

8: "Territories",

9: "Assault",

10: "Stub",

HALO = "QW88cv"

HALOD = "yG3d9w"

HALOMACD = "e4Rd9J"

HALOMAC = "e4Rd9J"

HALOM = "e4Rd9J"

address: str

port: int

game: Optional[str]

if cnt == 0: return 0, n1, n2

mask = 1

if cnt > 1:

while mask < cnt: mask = (mask << 1) + 1

i = 0

while True:

n1 = encxkey[n1 & 0xff] + id_bytes[n2]

n2 += 1

if n2 >= idlen:

n2 = 0

n1 += idlen

tmp = n1 & mask

i += 1

if i > 11: tmp %= cnt

if tmp <= cnt: break

if idlen < 1: return

for i in range(256): encxkey[i] = i

n1 = n2 = 0

for i in range(255, -1, -1):

t1, n1, n2 = _enctypex_func5(encxkey, i, id_bytes, idlen, n1, n2)

t2 = encxkey[i]

encxkey[i] = encxkey[t1]

encxkey[t1] = t2

encxkey[256] = encxkey[1]

encxkey[257] = encxkey[3]

encxkey[258] = encxkey[5]

encxkey[259] = encxkey[7]

a = encxkey[256]

b = encxkey[257]

c = encxkey[a]

encxkey[256] = (a + 1) & 0xff

encxkey[257] = (b + c) & 0xff

a = encxkey[260]

b = encxkey[257]

b = encxkey[b]

c = encxkey[a]

encxkey[a] = b

a = encxkey[259]

b = encxkey[257]

a = encxkey[a]

encxkey[b] = a

a = encxkey[256]

b = encxkey[259]

a = encxkey[a]

encxkey[b] = a

a = encxkey[256]

encxkey[a] = c

b = encxkey[258]

a = encxkey[c]

c = encxkey[259]

b = (b + a) & 0xff

encxkey[258] = b

a = b

c = encxkey[c]

b = encxkey[257]

b = encxkey[b]

a = encxkey[a]

c = (c + b) & 0xff

b = encxkey[260]

b = encxkey[b]

c = (c + b) & 0xff

b = encxkey[c]

c = encxkey[256]

c = encxkey[c]

a = (a + c) & 0xff

c = encxkey[b]

b = encxkey[a]

encxkey[260] = d & 0xff

c = c ^ b ^ (d & 0xff)

encxkey[259] = c & 0xff

keylen = len(key)

for i in range(datalen):

idx1 = (key[i % keylen] * i) & 7

idx2 = i & 7

encxvalidate[idx1] = (encxvalidate[idx1] ^ encxvalidate[idx2] ^ data[i]) & 0xff

datalen = len(data)

if datalen < 1: return None, 0

a = (data[0] ^ 0xec) + 2

if datalen < a: return None, 0

b = data[a - 1] ^ 0xea

if datalen < (a + b): return None, 0

encxvalidate = bytearray(validate[:8])

_enctypex_funcx(encxkey, key, encxvalidate, bytes(data[a:a+b]), b)

a += b

encxkey = bytearray(261)

data_array = bytearray(data)

result, offset = _enctypex_init(encxkey, key, validate, data_array)

if result is None: return None

result_array = bytearray(result)

_enctypex_func6(encxkey, result_array, len(result_array))

def __init__(self):

self.socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

self.socket.settimeout(0.1)

def close(self):

try: self.socket.close()

except: pass

def send_query(self, address: str, port: int, data: str):

try:

self.socket.sendto(data.encode('utf-8'), (address, port))

return True

except: return False

async def write(self, address: str, port: int, data: str):

return await asyncio.get_event_loop().run_in_executor(None, self.send_query, address, port, data)

def receive_responses(self, end_delay: float = 0.5) -> Optional[List[Dict]]:

responses = []

last_receive_time = time.time()

while True:

try:

data, addr = self.socket.recvfrom(4096)

responses.append({'address': addr[0], 'port': addr[1], 'data': data})

last_receive_time = time.time()

except socket.timeout:

if time.time() - last_receive_time > end_delay: break

except: break

return responses if responses else None

async def read_all(self, end_delay: float = 0.5) -> Optional[List[Dict]]:

def __init__(self, host: str, port: int, timeout: float = 5.0):

self.host, self.port, self.timeout = host, port, timeout

self.socket = None

async def connect(self):

def _conn():

self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

self.socket.settimeout(self.timeout)

self.socket.connect((self.host, self.port))

await asyncio.get_event_loop().run_in_executor(None, _conn)

async def request(self, data: bytes) -> bytes:

loop = asyncio.get_event_loop()

await loop.run_in_executor(None, self.socket.sendall, data)

response = b''

while True:

try:

chunk = await asyncio.wait_for(loop.run_in_executor(None, self.socket.recv, 4096), timeout=1.0)

if not chunk: break

response += chunk

except asyncio.TimeoutError: break

return response

def close(self):

def cstring(s: str) -> List[int]: return [ord(c) for c in s] + [0]

data = [0, 0, 0] + [1, 3, 0, 0, 0, 0] + cstring(game) + cstring(game) + cstring(validation_key) + [0]*5

data[1] = len(data)

if len(data) < GAMESPY_IP_PORT_LENGTH: return None

scanner = 0

request_ip = '.'.join(str(b) for b in data[scanner:scanner+4])

common_port = (data[scanner+4] << 8) | data[scanner+5]

scanner += GAMESPY_IP_PORT_LENGTH

if common_port == 0xFFFF: return None

def extract_pascal_string(offset: int):

if offset >= len(data): return "", offset

size = data[offset]

return data[offset+1:offset+1+size].decode('ascii', errors='ignore'), offset + size + 1

_, scanner = extract_pascal_string(scanner)

_, scanner = extract_pascal_string(scanner)

servers = []

while scanner < len(data):

flag = data[scanner]

scanner += 1

extra_len = sum([3 if flag & GameSpyFlags.A else 0, 4 if flag & GameSpyFlags.B else 0, 2 if flag & GameSpyFlags.C else 0, 2 if flag & GameSpyFlags.D else 0])

if scanner + GAMESPY_IP_PORT_LENGTH > len(data): break

ip = '.'.join(str(b) for b in data[scanner:scanner+4])

port = (data[scanner+4] << 8) | data[scanner+5]

scanner += GAMESPY_IP_PORT_LENGTH + extra_len - 1

if flag == 0 and ip == "255.255.255.255": break

servers.append(GameSpyServerAddress(address=ip, port=port))

game_map = {'halom':'HALOM','halor':'HALOR','halod':'HALOD','halomac':'HALOMAC','halomacd':'HALOMACD','halo':'HALO'}

game_enum = game_map.get(game.lower())

if not game_enum: return []

client = GameSpyTCPClient(host, port, timeout)

try:

await client.connect()

vkey = _make_validation_key()

encrypted = await client.request(_encode_master_server_request(game, vkey))

decrypted = _gamespy_decryptx(GameKeys[game_enum].value, vkey, encrypted)

decoded = _decode_master_server_response(decrypted) if decrypted else None

return decoded['servers'] if decoded else []

except: return []

servers = []

game_map = {'ce':'halom','pc':'halor','trial':'halod','mac':'halomac','macdemo':'halomacd','beta':'halo'}

for arg in args:

if isinstance(arg, str):

game_key = game_map.get(arg.lower(), arg)

s_list = await _get_gamespy_master_server_list(game_key, master_host, master_port, timeout)

servers.extend([GameSpyServer(s.address, s.port, arg) for s in s_list])

else: servers.append(GameSpyServer(arg.address, arg.port))

if not servers: return None

address_game_map = {f"{s.address}:{s.port}": s.game for s in servers}

client = GameSpyUDPClient()

try:

for s in servers: await client.write(s.address, s.port, '\\')

await asyncio.sleep(0.2)

responses = await client.read_all(end_delay=timeout)

if not responses: return None

return [GameSpyServerResponse(r['address'], r['port'], address_game_map.get(f"{r['address']}:{r['port']}"), r['data'].decode('utf-8', errors='ignore')) for r in responses]

result, i = [], 0

while i < len(s):

c = s[i]

if ord(c) < 32 and c not in '\t\n\r': i += 1; continue

if c == '^' and i + 1 < len(s) and (s[i+1].isdigit() or s[i+1].islower()): i += 2; continue

if c == '\\': result.append('/'); i += 1; continue

if 32 <= ord(c) < 127 or ord(c) >= 128: result.append(c)

i += 1

parts = response_str.split('\\')

if parts and parts[0] == '': parts = parts[1:]

info = {}

for i in range(0, len(parts) - 1, 2):

key, value = parts[i], _clean_gamespy_string(parts[i+1])

if value.isdigit(): value = int(value)

elif re.match(r'^\d+\.\d+$', value): value = float(value)

info[key] = value

"""Clean and normalize string fields from Cartographer API."""

if not isinstance(s, str):

return s

s = s.strip()

# Remove wrapping quotes

if s.startswith('"'):

s = s[1:]

if s.endswith('"'):

s = s[:-1]

# Strip control chars but keep all other unicode

s = ''.join(c for c in s if ord(c) >= 0x20 and ord(c) != 0x7f)

"""Decode properties array into named fields."""

out = {}

raw_props = []

pm = PROPERTY_MAP

for prop in pp:

pid = prop.get('dwPropertyId')

ptype = prop.get('type')

val = prop.get('value')

if isinstance(val, str):

val = clean_string_field(val)

name = pm.get(pid)

out_key = name if name else f"prop_{hex(pid) if isinstance(pid,int) else pid}"

out[out_key] = { 'value': val, 'type': ptype }

raw_props.append({ 'dwPropertyId': pid, 'type': ptype, 'value': val })

out['_raw'] = raw_props

"""Summarize Cartographer server data into a clean format."""

if data is None:

return {}

summary = {}

summary['xuid'] = data.get('xuid') or data.get('XUID') or None

summary['players'] = { 'filled': data.get('dwFilledPublicSlots'), 'max': data.get('dwMaxPublicSlots') }

pp = data.get('pProperties') or []

decoded = decode_properties(pp)

server_name = decoded.get('server_name', {}).get('value') or decoded.get('prop_0x40008230', {}).get('value') or ''

server_name = server_name.strip()

map_name = decoded.get('map_name', {}).get('value') or decoded.get('map_name_2', {}).get('value') or ''

map_id = decoded.get('map_id', {}).get('value') or decoded.get('map_id_2', {}).get('value')

gt1 = decoded.get('gametype_name', {}).get('value') or ''

gt2 = decoded.get('gametype_name_2', {}).get('value') or ''

gtid_raw = decoded.get('gametype_id', {}).get('value')

gametype_display = ''

if gtid_raw is not None:

try:

gtid_int = int(gtid_raw)

gametype_display = GAMETYPE_ID_TO_NAME.get(gtid_int, str(gtid_int))

except Exception:

gametype_display = str(gtid_raw)

variant_display = gt1 or gt2 or ''

if not map_name and map_id is not None:

try:

mid = int(map_id)

info = MAP_ID_TO_INFO.get(mid)

if info and len(info) > 1:

map_name = info[1]

else:

map_name = f"<map id {mid}>"

except Exception:

map_name = f"<map id {map_id}>"

description = (decoded.get('server_desc', {}).get('value') or '').strip()

summary['server_name'] = server_name

summary['map_name'] = map_name

summary['map_id'] = map_id

summary['gametype'] = gametype_display

summary['variant'] = variant_display

summary['description'] = description

summary['decoded_properties'] = decoded

# Preserve some raw top-level Cartographer fields that are not part

# of the property list but are present in the upstream JSON.

# These are useful for debugging/network info (saddr, abenet, abonline, xnaddr).

try:

if 'saddr' in data:

summary['saddr'] = data.get('saddr')

if 'abenet' in data:

summary['abenet'] = data.get('abenet')

if 'abonline' in data:

summary['abonline'] = data.get('abonline')

if 'xnaddr' in data:

summary['xnaddr'] = data.get('xnaddr')

except Exception:

# don't fail summarization if unexpected types are present

pass

"""Fetch details for a single Cartographer server."""

url = f"{CARTOGRAPHER_SERVER_URL}/{server_id}"

try:

r = await client.get(url, timeout=15.0)

r.raise_for_status()

json_data = r.json()

summary = summarize_server(json_data)

# Preserve the original upstream payload for debugging and inspection

try:

summary['_raw_cartographer'] = json_data

except Exception:

pass

if summary:

return summary

# Remote returned an empty/null payload — return a consistent failure object

logger.warning(f"Cartographer server {server_id} returned empty data")

return {'xuid': server_id, 'server_name': '', 'map_name': '', 'gametype': '', 'variant': '', 'description': '<failed>'}

except Exception as e:

logger.warning(f"Failed to fetch Cartographer server {server_id}: {e}")

"""Fetch historical ElDewrito stats from legacy API."""

try:

async with httpx.AsyncClient() as client:

logger.info(f"Fetching legacy ElDewrito stats from {LEGACY_ELDEWRITO_STATS_URL}...")

response = await client.get(LEGACY_ELDEWRITO_STATS_URL, timeout=30.0)

response.raise_for_status()

data = response.json()

if "players" in data and "servers" in data:

if isinstance(data["players"], list) and isinstance(data["servers"], list):

logger.info(f"Successfully fetched {len(data['players'])} historical ElDewrito data points")

return data

logger.warning("Legacy stats API returned unexpected format")

return None

except Exception as e:

logger.warning(f"Failed to fetch legacy ElDewrito stats: {e}")

"""Populate ElDewrito database with legacy stats data."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

players_dict = {entry[0]: entry[1] for entry in legacy_data.get("players", [])}

servers_dict = {entry[0]: entry[1] for entry in legacy_data.get("servers", [])}

all_timestamps = set(players_dict.keys()) | set(servers_dict.keys())

records_added = 0

for timestamp_ms in sorted(all_timestamps):

timestamp_dt = datetime.fromtimestamp(timestamp_ms / 1000, tz=timezone.utc)

player_count = players_dict.get(timestamp_ms, 0)

server_count = servers_dict.get(timestamp_ms, 0)

cursor.execute("""

""", (player_count, server_count, timestamp_dt))

records_added += 1

conn.commit()

conn.close()

logger.info(f"Successfully populated ElDewrito database with {records_added} historical records")

except Exception as e:

"""Initialize the stats tables and populate ElDewrito stats with legacy data if empty."""

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

# ElDewrito stats table

cursor.execute("""

""")

cursor.execute("""

""")

# Cartographer stats table

cursor.execute("""

""")

cursor.execute("""

""")

# Halo CE stats table

cursor.execute("""

""")

cursor.execute("""

""")

# Halo PC stats table

cursor.execute("""

""")

cursor.execute("""

""")

# Only check ElDewrito stats for legacy data population

cursor.execute("SELECT COUNT(*) FROM server_stats")

eldewrito_count = cursor.fetchone()[0]

conn.commit()

conn.close()

logger.info("Database initialized")

# Only populate legacy data for ElDewrito stats table

if eldewrito_count == 0:

logger.info("ElDewrito stats table is empty, attempting to fetch legacy data...")

legacy_data = await fetch_legacy_eldewrito_stats()

if legacy_data and (legacy_data.get("players") or legacy_data.get("servers")):

populate_from_legacy_eldewrito_stats(legacy_data)

else:

"""Save current ElDewrito stats to database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

now = datetime.now(timezone.utc)

cursor.execute("""

""", (player_count, server_count, now))

conn.commit()

conn.close()

logger.info(f"Saved ElDewrito stats: {server_count} servers, {player_count} players")

except Exception as e:

"""Save current Cartographer stats to database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

now = datetime.now(timezone.utc)

cursor.execute("""

""", (player_count, server_count, now))

conn.commit()

conn.close()

logger.info(f"Saved Cartographer stats: {server_count} servers, {player_count} players")

except Exception as e:

"""Save current Halo CE stats to database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

now = datetime.now(timezone.utc)

cursor.execute("""

""", (player_count, server_count, now))

conn.commit()

conn.close()

logger.info(f"Saved Halo CE stats: {server_count} servers, {player_count} players")

except Exception as e:

"""Save current Halo PC stats to database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

now = datetime.now(timezone.utc)

cursor.execute("""

""", (player_count, server_count, now))

conn.commit()

conn.close()

logger.info(f"Saved Halo PC stats: {server_count} servers, {player_count} players")

except Exception as e:

"""Retrieve historical ElDewrito stats from database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

cursor.execute("""

""")

rows = cursor.fetchall()

conn.close()

players = [[row[0], row[1]] for row in rows]

servers = [[row[0], row[2]] for row in rows]

return {

"players": players,

"servers": servers

}

except Exception as e:

logger.error(f"Failed to retrieve ElDewrito stats: {e}")

return {

"players": [],

"servers": []

"""Retrieve historical Cartographer stats from database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

cursor.execute("""

""")

rows = cursor.fetchall()

conn.close()

players = [[row[0], row[1]] for row in rows]

servers = [[row[0], row[2]] for row in rows]

return {

"players": players,

"servers": servers

}

except Exception as e:

logger.error(f"Failed to retrieve Cartographer stats: {e}")

return {

"players": [],

"servers": []

"""Retrieve historical Halo CE stats from database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

cursor.execute("""

""")

rows = cursor.fetchall()

conn.close()

players = [[row[0], row[1]] for row in rows]

servers = [[row[0], row[2]] for row in rows]

return {

"players": players,

"servers": servers

}

except Exception as e:

logger.error(f"Failed to retrieve Halo CE stats: {e}")

return {

"players": [],

"servers": []

"""Retrieve historical Halo PC stats from database."""

try:

conn = sqlite3.connect(DB_PATH)

cursor = conn.cursor()

cursor.execute("""

""")

rows = cursor.fetchall()

conn.close()

players = [[row[0], row[1]] for row in rows]

servers = [[row[0], row[2]] for row in rows]

return {

"players": players,

"servers": servers

}

except Exception as e:

logger.error(f"Failed to retrieve Halo PC stats: {e}")

return {

"players": [],

"servers": []

"""Main logic: Pulls master lists, dedupes, queries servers, updates cache."""

global eldewrito_cache

# 1. Load Master Server URLs from local JSON

try:

async with await anyio.open_file(ELDEWRITO_MASTER_LIST, 'r') as f:

data = await f.read()

config = json.loads(data)

master_entries = config.get("masterServers", [])

# Extract only the 'list' attribute

master_urls = [m['list'] for m in master_entries if 'list' in m]

except Exception as e:

logger.error(f"Error reading {ELDEWRITO_MASTER_LIST}: {e}")

return

async with httpx.AsyncClient() as client:

# 2. Query all master servers concurrently

logger.info(f"Querying {len(master_urls)} master servers...")

master_tasks = [fetch_master_list(client, url) for url in master_urls]

results = await asyncio.gather(*master_tasks)

# 3. Deduplicate IP:Port combos

unique_servers: Set[str] = set()

for server_list in results:

for ip_port in server_list:

unique_servers.add(ip_port)

logger.info(f"Found {len(unique_servers)} unique game servers. Querying details...")

# 4. Query all game servers concurrently

# We limit concurrency slightly to avoid file descriptor limits if the list is huge,

# but for <100 servers, full concurrency is fine.

game_tasks = [fetch_game_server_info(client, srv) for srv in unique_servers]

game_results = await asyncio.gather(*game_tasks)

# 5. Build the final data structure

successful_servers = {}

total_players = 0

for res in game_results:

if res:

ip_port, data = res

successful_servers[ip_port] = data

# Safely add player count

if "numPlayers" in data:

try:

total_players += int(data["numPlayers"])

except ValueError:

pass

# 6. Format Final JSON

new_cache = {

"count": {

"players": total_players,

"servers": len(successful_servers)

},

"updatedAt": get_current_http_date(),

"servers": successful_servers

}

# Atomically update global cache

eldewrito_cache = new_cache

"""Fetch Cartographer server list and update cache with summarized data."""

global cartographer_cache, cartographer_summarized_cache