From 235f4bca3762275f4df9fdaadf0f74c5efb4132c Mon Sep 17 00:00:00 2001 From: aclist <92275929+aclist@users.noreply.github.com> Date: Sat, 26 Jul 2025 02:45:23 +0900 Subject: [PATCH] fix: add version file check --- dzgui.sh | 1 + helpers/servers.py | 41 +++++++- helpers/servers_alt.py | 210 ----------------------------------------- 3 files changed, 37 insertions(+), 215 deletions(-) delete mode 100644 helpers/servers_alt.py diff --git a/dzgui.sh b/dzgui.sh index 7d743b7..7b8d1ed 100755 --- a/dzgui.sh +++ b/dzgui.sh @@ -1005,6 +1005,7 @@ legacy_cols(){ mv $cols_file.new $cols_file } stale_mod_signatures(){ + [[ ! -f "$versions_file" ]] && return local workshop_dir="$steam_path/steamapps/workshop/content/$aid" if [[ -d $workshop_dir ]]; then readarray -t old_mod_ids < <(awk -F, '{print $1}' $versions_file) diff --git a/helpers/servers.py b/helpers/servers.py index 5793639..a408342 100644 --- a/helpers/servers.py +++ b/helpers/servers.py @@ -1,11 +1,14 @@ +import json import os import re -import requests import socket import subprocess import sys import typing # noqa from typing import Union +from urllib import request, parse +from urllib.error import HTTPError +from dataclasses import dataclass sys.path.append("a2s") import a2s # noqa @@ -53,7 +56,7 @@ def sanitize(name: str) -> str: return name -def parse(json: list) -> list: +def parse_json(json: list) -> list: """ Server metadata is underspecified and server operators tend to insert random garbage in the headers. In case @@ -166,7 +169,14 @@ def query_direct(ip: str, qport: int) -> dict | None: return None -def query_api(key: str, param: str) -> requests.models.Response: +@dataclass +class Res: + status: int + parsed: bool + json: str + + +def query_api(key: str, param: str) -> "HTTPResponse": LIMIT = 10000 url = "https://api.steampowered.com/IGameServersService/GetServerList/v1/?" @@ -175,5 +185,26 @@ def query_api(key: str, param: str) -> requests.models.Response: "limit": LIMIT, "key": key, } - r = requests.get(url, params=payload) - return r + par = parse.urlencode(payload) + url = f"{url}{par}" + + status = 200 + parsed = True + data = None + + try: + with request.urlopen(url) as response: + if response.status != 200: + status = response.status + try: + parsed = True + data = json.load(response) + except json.decoder.JSONDecodeError: + parsed = False + data = None + except HTTPError: + status = 403 + parsed = False + data = None + finally: + return Res(status, parsed, data) diff --git a/helpers/servers_alt.py b/helpers/servers_alt.py deleted file mode 100644 index a408342..0000000 --- a/helpers/servers_alt.py +++ /dev/null @@ -1,210 +0,0 @@ -import json -import os -import re -import socket -import subprocess -import sys -import typing # noqa -from typing import Union -from urllib import request, parse -from urllib.error import HTTPError -from dataclasses import dataclass - -sys.path.append("a2s") -import a2s # noqa - -params = [ - r"\nor\1\map\chernarusplus\nor\1\map\sakhal\nor\1\map\enoch\empty\1\nor\1\map\namalsk", - r"\map\namalsk\empty\1", - r"\map\namalsk\noplayers\1", - r"\map\chernarusplus\empty\1", - r"\map\chernarusplus\noplayers\1", - r"\map\sakhal\empty\1", - r"\map\sakhal\noplayers\1", - r"\map\enoch\empty\1", - r"\map\enoch\noplayers\1", -] - - -def get_netmask() -> str: - hostname = os.uname()[1] - i = socket.gethostbyname(hostname) - netmask = i.rsplit(".", 1)[0] - return netmask - - -def test_ip(suffix: int, port: int) -> dict | None: - netmask = get_netmask() - hostname = f"{netmask}.{str(suffix)}" - ping = ["ping", "-c1", "-i", "0.1", "-w", "1"] - output = subprocess.run(ping + [hostname], capture_output=True) - if output.returncode == 0: - return query_direct(hostname, port) - return None - - -def sanitize(name: str) -> str: - name = re.sub(r"\r", r"", name) - name = re.sub(r"\n", r"", name) - name = re.sub(r"\x01", r"", name) - name = re.sub(r"\ufeff", r"", name) - name = re.sub(r"(^!\s*)", r"-", name) - name = re.sub(r"(^-\s*)", r"", name) - name = re.sub(r"(^-)", r"", name) - name = re.sub(r"(^\s*)", r"", name) - name = re.sub(r"\t", "", name) - return name - - -def parse_json(json: list) -> list: - """ - Server metadata is underspecified and server operators - tend to insert random garbage in the headers. In case - sanitization failed, discard malformed rows rather than - aborting outright. - """ - rows = [] - for row in json: - try: - name = sanitize(row["name"]) - if name == "": - continue - except KeyError: - continue - - for key in [ - "map", - "gametype", - "players", - "max_players", - "addr", - "gameport", - ]: - try: - row[key] - except KeyError: - continue - - r = row["gametype"].split(",") - if "no3rd" in r: - view = "1PP" - else: - view = "3PP" - - if "external" in r: - provider = "Unoffic." - else: - provider = "Official" - - if "mod" in r: - modded = True - else: - modded = False - - try: - r = row["gametype"].split("lqs") - queue = r[1].split(",")[0] - except IndexError: - queue = 0 - - try: - test_time = re.search(r"[0-9]{2}:[0-9]{2}", row["gametype"]) - if test_time: - time = test_time.group(0) - else: - time = "Unknown" - except AttributeError: - time = "Unknown" - - try: - ip = row["addr"].split(":")[0] + ":" + str(row["gameport"]) - qport = row["addr"].split(":")[1] - except IndexError: - continue - - mapname = row["map"].lower() - players = row["players"] - max_players = row["max_players"] - raw = [ - name, - mapname, - view, - time, - int(players), - int(max_players), - int(queue), - ip, - int(qport), - provider, - modded, - ] - rows.append(raw) - return rows - - -def query_direct(ip: str, qport: int) -> dict | None: - try: - info = a2s.info((ip, qport)) - - name = info.server_name - mapname = info.map_name - address = ip + ":" + str(qport) - gameport = str(info.port) - players = info.player_count - max_players = info.max_players - keywords = info.keywords - - res = {} - res["name"] = name - res["map"] = mapname - res["gametype"] = keywords - res["players"] = players - res["max_players"] = max_players - res["addr"] = address - res["gameport"] = gameport - return res - except TimeoutError: - return None - except KeyError: - return None - - -@dataclass -class Res: - status: int - parsed: bool - json: str - - -def query_api(key: str, param: str) -> "HTTPResponse": - LIMIT = 10000 - url = "https://api.steampowered.com/IGameServersService/GetServerList/v1/?" - - payload: dict[str, Union[int, str]] = { - "filter": r"\appid\221100" + param, - "limit": LIMIT, - "key": key, - } - par = parse.urlencode(payload) - url = f"{url}{par}" - - status = 200 - parsed = True - data = None - - try: - with request.urlopen(url) as response: - if response.status != 200: - status = response.status - try: - parsed = True - data = json.load(response) - except json.decoder.JSONDecodeError: - parsed = False - data = None - except HTTPError: - status = 403 - parsed = False - data = None - finally: - return Res(status, parsed, data)