chore: clear typehinting errors

This commit is contained in:
aclist 2026-05-13 20:13:06 +09:00
parent fe0a6711f0
commit 866c88fbb0
14 changed files with 65 additions and 43 deletions

View File

@ -33,8 +33,8 @@ def map_id_to_record(key: str, uid: int) -> Optional["Record"]:
try:
record = get_attributes(key, uid)
ip = record["ip"]
port = record["port"]
qport = record["portQuery"]
port = int(record["port"])
qport = int(record["portQuery"])
return Record(ip, port, qport)
except Exception:
return None

View File

@ -49,7 +49,7 @@ def get_local_mods(workshop_path: Path) -> list[Path]:
return mods
def parse_meta(file: Path) -> ModMeta:
def parse_meta(file: Path) -> ModMeta | None:
mod = file / "meta.cpp"
if mod.exists() is False:
return None
@ -135,6 +135,8 @@ def remove_stale_signatures(config: Path, versions: Path) -> None:
def find_stale_mods(config: Path) -> list[int]:
def push_record(rec: str) -> list:
record = fqip_to_record(rec)
if record is None:
return []
try:
mods = get_rules(record)
except Exception:

View File

@ -33,35 +33,35 @@ class VersionMatch(Enum):
SAME_VERSION = 3
class u8:
class u8(int):
fmt = "B"
class u16:
class u16(int):
fmt = "H"
class u32:
class u32(int):
fmt = "L"
class u64:
class u64(int):
fmt = "Q"
class i8:
class i8(int):
fmt = "b"
class i16:
class i16(int):
fmt = "h"
class i32:
class i32(int):
fmt = "l"
class i64:
class i64(int):
fmt = "q"
@ -330,6 +330,7 @@ def get_version(file: Path) -> FileVersion:
magic = hex(struct.unpack("<H", (blob[0:2]))[0])
f.seek(pos)
OBJW: OPTIONAL_HDR_WIN_X86 | OPTIONAL_HDR_WIN_X64
if magic == PE32_x86:
OPTIONAL_HDR_X86.unpack(f)
OBJW = OPTIONAL_HDR_WIN_X86.unpack(f)
@ -363,7 +364,7 @@ def get_version(file: Path) -> FileVersion:
table = RESOURCE_DIRECTORY_TABLE.unpack(f)
total = table.number_of_name_entries + table.number_of_id_entries
for entry in range(total):
for _iter in range(total):
entry = RESOURCE_DIRECTORY_ENTRY.unpack(f)
if entry.name_or_id == VERSION_RESOURCE:
while entry.data_or_subdir & (1 << 31):
@ -371,7 +372,7 @@ def get_version(file: Path) -> FileVersion:
seek_to_hex(hex(offset + shift), f)
table = RESOURCE_DIRECTORY_TABLE.unpack(f)
total = table.number_of_name_entries + table.number_of_id_entries
for entry in range(total):
for _iter in range(total):
entry = RESOURCE_DIRECTORY_ENTRY.unpack(f)
break
if entry.name_or_id > VERSION_RESOURCE:
@ -383,7 +384,7 @@ def get_version(file: Path) -> FileVersion:
offset = data.data_rva - hdr.virtual_address + hdr.pointer_to_raw_data
seek_to_hex(hex(offset), f)
hdr = VS_VERSION_INFO_HDR.unpack(f)
VS_VERSION_INFO_HDR.unpack(f)
# https://learn.microsoft.com/en-us/windows/win32/menurc/vs-versioninfo
byte_len = len(VS_VERSION_INFO_ID.encode("utf-16le"))
label = f.read(byte_len).decode("utf-16le")

View File

@ -81,9 +81,9 @@ class A2SInfo:
def get_info(self) -> a2s.SourceInfo | None:
return self.info
def as_row(self) -> dict[str, Any] | None:
def as_row(self) -> dict[str, Any]:
if self.info is None:
return None
raise AttributeError("No record to convert")
ip = self.record.ip
qport = self.record.qport
return source_info_to_dict(ip, qport, self.info)
@ -239,7 +239,7 @@ def parse_json(json: list) -> list:
return rows
def source_info_to_dict(ip: str, qport: int, info: "SourceInfo") -> dict[str, Any]:
def source_info_to_dict(ip: str, qport: int, info: "SourceInfo") -> dict[str, Any] | None:
try:
name = info.server_name
mapname = info.map_name

View File

@ -28,9 +28,9 @@ def get_local_signatures(version_file: Path) -> dict[str, int]:
hashes: dict[str, int] = {}
lines = version_file.read_text().splitlines()
for line in lines:
line = line.split(",")
_id = line[0]
_hash = int(line[1])
data = line.split(",")
_id = data[0]
_hash = int(data[1])
hashes[_id] = _hash
return hashes
@ -49,7 +49,7 @@ def get_needs_update(
version_file: Path, remote_hashes: list[tuple[str, str, int, int]]
) -> list[tuple[str, str, int, int]]:
local_hashes = get_local_signatures(version_file)
needs_update: list[tuple[str, str]] = []
needs_update: list[tuple[str, str, int, int]] = []
for title, _id, _hash, size in remote_hashes:
if _id not in local_hashes:
needs_update.append((title, _id, _hash, size))
@ -64,7 +64,7 @@ def get_remote_signatures(mods: list[str]) -> list[tuple[str, str, int, int]]:
"""
Attempts to continue connecting even if signatures are empty
"""
payload: dict[str, str] = {}
payload: dict[str, str | int] = {}
payload["itemcount"] = len(mods)
for i, mod in enumerate(mods):
payload[f"publishedfileids[{i}]"] = mod
@ -76,13 +76,13 @@ def get_remote_signatures(mods: list[str]) -> list[tuple[str, str, int, int]]:
if r.status_code != 200:
return []
hashes: list[tuple[str, int, int]] = []
hashes: list[tuple[str, str, int, int]] = []
j = r.json()
rows = j["response"]["publishedfiledetails"]
for row in rows:
title = row["title"]
_id = row["publishedfileid"]
time = row["time_updated"]
title = str(row["title"])
_id = str(row["publishedfileid"])
time = int(row["time_updated"])
size = int(row["file_size"])
hashes.append((title, _id, time, size))
return hashes

View File

@ -38,8 +38,15 @@ def rc2json(file: Path) -> str:
keys: dict[str, Any] = {}
ips: list[str] = []
toggles = ["auto_install", "fullscreen"]
deprecated = ["staging_dir", "src_path", "steam_path", "debug", "branch"]
toggles = ["fullscreen"]
deprecated = [
"staging_dir",
"src_path",
"steam_path",
"debug",
"branch",
"auto_install",
]
while True:
tok = lex.get_token()

View File

@ -24,12 +24,12 @@ def lookup(path: Path, enum: Preferences) -> Any:
def get_config(path: Path) -> dict:
# TODO: is this being called twice?
# TODO: is this being called multiple times?
try:
json = read_json(path)
return json
except Exception as e:
raise e
return json
def get_favorites(path: Path) -> list[str]:

View File

@ -41,7 +41,7 @@ def is_flatpak_steam_running() -> bool:
if has_cmd(FLATPAK_CMD) is False:
return False
proc = subprocess.check_output([FLATPAK_CMD, "ps"], text=True)
lines = proc.stdout.splitlines()
lines = proc.splitlines()
if FLATPAK_APPID in lines:
return True
return False

View File

@ -28,7 +28,9 @@ class ProxyModelManager:
"""
def __init__(self, filter_man: "FilterManager") -> None:
self.filter_cache = {}
self.filter_cache: dict[tuple[str], tuple["FastInsertListStore", list[Any]]] = (
{}
)
self.proxy_model: "FastInsertListStore" = None
self.filter_man = filter_man
@ -55,16 +57,18 @@ class ProxyModelManager:
self.proxy_model[treeiter][4] = playercount.players
self.proxy_model[treeiter][6] = playercount.queue
# FIXME: typehint -> tuple
def append_row_to_history(
self, history: list[str, str, str, str, int, int, int, str, int, int, str, bool]
) -> None:
addr = history[7]
qport = history[8]
if self.control_model is None:
raise AttributeError("Trying to append row to empty model")
found = False
for i, row in enumerate(self.control_model):
if addr == row[7] and qport == row[8]:
print("record exists in history")
item = self.control_model.pop(i)
self.control_model.append(item)
found = True

View File

@ -268,8 +268,10 @@ class ServerModelManager:
def _parse_single_record(self, response: "A2SInfo", delete: bool = False) -> None:
self.preserve_on_fail = True
row = response.as_row()
if row is None:
try:
row = response.as_row()
except Exception as e:
logger.warning(e)
self.thread_man.set_cleanup_func(StoredFunc(self._cleanup_on_failure))
return
@ -301,7 +303,7 @@ class ServerModelManager:
return
config_man.add_saved_server(fqip)
# 2026-05-04
# TODO: this is valid if saved servers tab is already open,
# FIXME: this is valid if saved servers tab is already open,
# but not if app was just booted
if proxy_man.has_control_model() is False:
self._get_proxy_man().push(records)
@ -393,7 +395,7 @@ class ServerModelManager:
# TODO: distinguish signals, e.g. "servers_failed_to_load", "servers_loaded_empty"
# customize statusbar and dialog accordingly
self.emitter.emit("servers_loaded", self.enum)
# TODO: destroy wait dialog first
# FIXME: destroy wait dialog first
# see threadman.set_cleanup_func(_, destroy_first=True)
if show_dialog:
dialog = ExceptionDialog(self.controller, api_warn_msg)

View File

@ -13,9 +13,10 @@ gi.require_version("Gtk", "3.0")
from gi.repository import Gtk # noqa E402
if TYPE_CHECKING:
from queue import Queue
from dzgui.const.enum import ServerTab
from dzgui.controllers.mc import Controller
from queue import Queue
from dzgui.util.ip import Coords
logger = logging.getLogger(APP_NAME)
@ -74,13 +75,13 @@ class CalcDist:
dist = self.compare(self.ip)
self.result_queue.put([self.addr, dist, self.enum])
def compare(self, remote: str) -> int | None:
def compare(self, remote_ip: str) -> Haversine | None:
prefs = self.controller.get_prefs()
local = prefs.coords
if local is None:
return None
try:
remote = get_coords(prefs.paths.ips, remote)
remote = get_coords(prefs.paths.ips, remote_ip)
except Exception:
return None

View File

@ -130,7 +130,7 @@ def is_valid_port(port: str) -> bool:
@deprecated("use ips.csv")
def get_local_coords(ip: str):
def get_local_coords(ip: str) -> str:
url = COORDS_API + "/" + ip
# local res=$(curl -Ls "$url" | jq -r '"\(.lat)\n\(.lon)"')
return url

View File

@ -12,11 +12,15 @@ from gi.repository import Gtk, Gdk # noqa E402
logger = logging.getLogger(APP_NAME)
if TYPE_CHECKING:
from dzgui.const.enum import NotebookPage
from dzgui.controller.mc import Controller
from dzgui.controllers.emitter import Emitter
class ContextualButton(Gtk.Button):
def __init__(self, label, opens, tooltip, context):
def __init__(
self, label: str, opens: "NotebookPage", tooltip: str, context: ButtonType
) -> None:
super().__init__(
label=label,
tooltip_text=tooltip,
@ -28,7 +32,7 @@ class ContextualButton(Gtk.Button):
class ButtonBox(Gtk.Box):
def __init__(self, controller) -> None:
def __init__(self, controller: "Controller") -> None:
super().__init__(
spacing=6,
margin_top=0,

View File

@ -62,6 +62,7 @@ class ServerModTreeView(ContextMixin, TreeView): # type: ignore
path = self.get_focused_row_path()
model = self.get_model()
tree_iter = model.get_iter(path)
# FIXME: https://docs.gtk.org/gtk3/method.TreeModel.get.html
mod = model.get(tree_iter, 1)[0]
return str(mod)