22import os
33import subprocess
44import multiprocessing
5- import warnings
65import json
76import shutil
87import time
98import datetime
109import zipfile
1110import re
12- import shutil
1311import logging
1412import tempfile
1513import contextlib
2321 logging .basicConfig (level = logging .DEBUG , format = '%(message)s' , stream = sys .stdout )
2422
2523from pathlib import Path
26- from typing import List , Optional
24+ from collections .abc import Iterable
25+ from typing import Any , List , Optional , TypeVar , override
2726
2827from importlib import invalidate_caches
2928from BaseUtils import tuplize_version , Version , local_path , write_path , mwgg_venv_site_packages , use_worlds_venv , is_frozen
@@ -112,24 +111,29 @@ def _skip_all_installs() -> bool:
112111update_ran = _skip_update
113112need_update : List [str ] = []
114113
115- class RequirementsSet (set ):
114+ _T = TypeVar ("_T" )
115+
116+
117+ class RequirementsSet (set [_T ]):
116118 """Custom set that tracks whether updates have been run."""
117-
118- def add (self , e ):
119+
120+ @override
121+ def add (self , e : _T ) -> None :
119122 global update_ran
120123 update_ran &= _skip_update
121124 super ().add (e )
122125
123- def update (self , * s ):
126+ @override
127+ def update (self , * s : Iterable [_T ]) -> None :
124128 global update_ran
125129 update_ran &= _skip_update
126130 super ().update (* s )
127131
128132
129133# Initialize file sets
130134
131- requirements_files = RequirementsSet ({Path (local_path ("requirements.txt" ))})
132- worlds_files = {"wheels" : RequirementsSet (), "apworlds" : RequirementsSet ()}
135+ requirements_files : RequirementsSet [ Path ] = RequirementsSet ({Path (local_path ("requirements.txt" ))})
136+ worlds_files : dict [ str , RequirementsSet [ str ]] = {"wheels" : RequirementsSet (), "apworlds" : RequirementsSet ()}
133137
134138# Frozen builds: custom_worlds lives under write_path() — i.e.
135139# ~/.local/share/MultiworldGG/custom_worlds on Linux, %LOCALAPPDATA% on Windows,
@@ -206,18 +210,17 @@ def _uv_pip(*args: str) -> list[str]:
206210 return ["pip" , * args , "--python" , str (python_cmd )]
207211
208212
209- def _uv_run (args : list [str ], timeout : float = 120 , check : bool = False ) -> subprocess .CompletedProcess :
213+ def _uv_run (args : list [str ], timeout : float = 120 , check : bool = False ) -> subprocess .CompletedProcess [ str ] :
210214 """Run `uv <args>` against the first reachable uv binary."""
211215 global _uv_resolved_path , _uv_unavailable
212216
213- kwargs = {
214- "capture_output" : True ,
215- "text" : True ,
216- "stdin" : subprocess .DEVNULL ,
217- "timeout" : timeout ,
218- }
219- if is_windows ():
220- kwargs ["creationflags" ] = subprocess .CREATE_NEW_PROCESS_GROUP | subprocess .CREATE_NO_WINDOW
217+ # Windows-only: detach into a new process group with no console window so a uv
218+ # subprocess can't flash a window or steal the parent's Ctrl-C. 0 is the
219+ # cross-platform no-op default elsewhere.
220+ creationflags = (
221+ subprocess .CREATE_NEW_PROCESS_GROUP | subprocess .CREATE_NO_WINDOW
222+ if is_windows () else 0
223+ )
221224
222225 if _uv_unavailable :
223226 return subprocess .CompletedProcess (args , 127 , "" , "uv not found at any known path" )
@@ -227,7 +230,15 @@ def _uv_run(args: list[str], timeout: float = 120, check: bool = False) -> subpr
227230 for cand in candidates :
228231 cmd = [cand ] + args
229232 try :
230- result = subprocess .run (cmd , check = check , ** kwargs )
233+ result = subprocess .run (
234+ cmd ,
235+ check = check ,
236+ capture_output = True ,
237+ text = True ,
238+ stdin = subprocess .DEVNULL ,
239+ timeout = timeout ,
240+ creationflags = creationflags ,
241+ )
231242 except OSError as e :
232243 # The candidate is unusable — try next.
233244 logger .debug (f"uv not usable at { cand } ({ e !r} ); trying next candidate" )
@@ -343,7 +354,7 @@ def parse_requirements_file(file_path: Path) -> List[str]:
343354 Parse a requirements.txt file and return a list of requirement strings.
344355 Handles line continuations, comments, and various requirement formats.
345356 """
346- requirements = []
357+ requirements : list [ str ] = []
347358
348359 with open (file_path , 'r' ) as f :
349360 lines = f .readlines ()
@@ -439,9 +450,12 @@ def _resolve_variant() -> str:
439450 variant = _EXPLICIT_VARIANT
440451 else :
441452 variant = _detect_installed_variant () or DEFAULT_MWGG_IGDB_VARIANT
442- MWGG_IGDB_VARIANT = variant
443- MWGG_IGDB_BRANCH = f"game_index_{ variant } "
444- MWGG_IGDB_GIT_URL = f"git+https://github.com/{ MWGG_INDEX_REPO } @{ MWGG_IGDB_BRANCH } "
453+ # These UPPERCASE names are public, documented module globals that are
454+ # intentionally reassigned here (callers and tests read ModuleUpdate.MWGG_IGDB_*);
455+ # the "constant" rule doesn't apply and renaming would break the public API.
456+ MWGG_IGDB_VARIANT = variant # pyright: ignore[reportConstantRedefinition]
457+ MWGG_IGDB_BRANCH = f"game_index_{ variant } " # pyright: ignore[reportConstantRedefinition]
458+ MWGG_IGDB_GIT_URL = f"git+https://github.com/{ MWGG_INDEX_REPO } @{ MWGG_IGDB_BRANCH } " # pyright: ignore[reportConstantRedefinition]
445459 return variant
446460
447461
@@ -467,7 +481,9 @@ def _igdb_upgraded_recently() -> bool:
467481 return install_date is not None and install_date == datetime .date .today ()
468482
469483
470- def _venv_has_worlds () -> bool :
484+ # Consumed by the upgrader tools (tools/mwgg_upgrade.py, tools/mcp_mwgg_upgrader.py),
485+ # so it is unused *within* this module — hence the targeted ignore.
486+ def _venv_has_worlds () -> bool : # pyright: ignore[reportUnusedFunction]
471487 try :
472488 worlds_dir = _venv_worlds_dir ()
473489 return worlds_dir .exists () and any (worlds_dir .iterdir ())
@@ -578,7 +594,7 @@ def set_variant(variant: str) -> None:
578594 installed variant on subsequent _resolve_variant() calls.
579595 """
580596 global _EXPLICIT_VARIANT
581- _EXPLICIT_VARIANT = variant
597+ _EXPLICIT_VARIANT = variant # pyright: ignore[reportConstantRedefinition] # intentional reassignment of the override sentinel
582598 _resolve_variant ()
583599
584600
@@ -816,7 +832,7 @@ def install_worlds(worlds: List[str], update: bool = False, with_deps: bool = Fa
816832 install_mwgg_igdb (upgrade = True , force = True )
817833
818834 index = _get_game_index ()
819- games = index .get_all_games () if index is not None else {}
835+ games : dict [ str , dict [ str , Any ]] = index .get_all_games () if index is not None else {}
820836 # Snapshot BEFORE uninstall_worlds: a world properly installed at the current
821837 # mwgg_igdb tag stays in the set so update=True reinstalls can skip deps.
822838 installed_world_slugs = {
@@ -912,7 +928,7 @@ def update_world_from_package() -> None:
912928 import threading
913929 import queue
914930
915- result_queue = queue .Queue ()
931+ result_queue : queue . Queue [ tuple [ int , str , str ]] = queue .Queue ()
916932
917933 def _pip_install_thread ():
918934 try :
@@ -952,12 +968,14 @@ def _pip_install_thread():
952968 new_version : Optional [Version ] = None
953969 manifest : dict [str , object ] = {}
954970 try :
955- apworld_container = APWorldContainer (world )
971+ apworld_container = APWorldContainer (world_path )
956972 # Set manifest path to expected location
957973 with zipfile .ZipFile (world , 'r' ) as apworld_zip :
958974 manifest = apworld_container .read_contents (apworld_zip )
959975 if "world_version" in manifest :
960- new_version = tuplize_version (manifest ["world_version" ])
976+ # manifest is untrusted external data (dict[str, object]); coerce the
977+ # value to str so a non-string world_version can't crash tuplize_version.
978+ new_version = tuplize_version (str (manifest ["world_version" ]))
961979 logger .info (f"APworld { world } has version { new_version } " )
962980 else :
963981 logger .info (f"APworld { world } has no world_version specified" )
@@ -992,8 +1010,8 @@ def _pip_install_thread():
9921010 # According to spec: "An APWorld without a world_version is always treated as older than one with a version"
9931011 if new_version is None and installed_version is not None :
9941012 logger .info (f"There is a custom apworld file with no world version specified, please remove it from your custom_worlds directory." )
995- elif installed_version is None or new_version > installed_version :
996- if installed_version is not None :
1013+ elif installed_version is None or ( new_version is not None and new_version > installed_version ) :
1014+ if installed_version is not None and new_version is not None :
9971015 uninstall_worlds ([package_name ])
9981016 logger .info (f"New version { new_version .as_simple_string ()} > installed { installed_version .as_simple_string ()} , uninstalling old version so new version will be picked up." )
9991017 else :
@@ -1125,12 +1143,15 @@ def _install_lock():
11251143 lock_file .seek (0 )
11261144 msvcrt .locking (lock_file .fileno (), msvcrt .LK_UNLCK , 1 )
11271145 else :
1146+ # fcntl is Unix-only and this branch only runs off-Windows, but the gate
1147+ # type-checks with pythonPlatform=Windows (where typeshed hides flock/LOCK_*),
1148+ # so these are platform false positives, not real attribute errors.
11281149 import fcntl
1129- fcntl .flock (lock_file .fileno (), fcntl .LOCK_EX )
1150+ fcntl .flock (lock_file .fileno (), fcntl .LOCK_EX ) # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
11301151 try :
11311152 yield
11321153 finally :
1133- fcntl .flock (lock_file .fileno (), fcntl .LOCK_UN )
1154+ fcntl .flock (lock_file .fileno (), fcntl .LOCK_UN ) # pyright: ignore[reportAttributeAccessIssue, reportUnknownMemberType]
11341155
11351156
11361157def update (yes : bool = True , force : bool = False , worlds : Optional [List [str ]] = None ) -> None :
0 commit comments