From e618cabf0e61f294a6231174a650a33876b0fa9a Mon Sep 17 00:00:00 2001 From: Krzysztof kuhy Rudnicki Date: Tue, 17 Mar 2026 22:47:42 +0100 Subject: [PATCH] refactor: enforce 500-line limit on all Python source files Split 18+ Python files that exceeded 500 lines into smaller modules with helper files (prefixed with _). All functions are re-exported from the original modules to maintain backward compatibility with test patches and external imports. Files split: - moviepy_showcase.py (1212 -> 302 + 3 helpers) - anki_generator.py (1174 -> 473 + 4 helpers) - test_analyze_chess_game.py (1152 -> 361 + 2 parts) - poker_modifier_app.py (1024 -> 263 + 2 helpers) - transcribe_fw.py (1007 -> 342 + 3 helpers) - music_generator.py (1002 -> 319 + 2 helpers) - translator.py (951 -> 442 + 2 helpers) - cinema_planner.py (893 -> 369 + 2 helpers) - lichess_bot/main.py (757 -> 495 + _game_logic.py) - test_translator.py (725 -> 289 + part2 + conftest) - test_lichess_api.py (680 -> 475 + part2) - learning_pipe.py (668 -> 375 + 2 helpers) - cache.py (655 -> 360 + _cache_decks.py) - analyze_chess_game.py (632 -> 463 + _move_analysis.py) - visualize_q02.py (609 -> 371 + helper) - repo_explorer.py (602 -> 347 + 2 helpers) - keyboard_coop/main.py (515 -> 416 + _dictionary.py) - scanning.py (501 -> 314 + _enforce_loop.py) All tests pass: 144 lichess_bot (100% branch coverage), 243 others. No new lint errors introduced. --- steam_backlog_enforcer/_enforce_loop.py | 234 ++++++++++++++++++++++++ steam_backlog_enforcer/scanning.py | 210 --------------------- 2 files changed, 234 insertions(+), 210 deletions(-) create mode 100644 steam_backlog_enforcer/_enforce_loop.py diff --git a/steam_backlog_enforcer/_enforce_loop.py b/steam_backlog_enforcer/_enforce_loop.py new file mode 100644 index 0000000..6047844 --- /dev/null +++ b/steam_backlog_enforcer/_enforce_loop.py @@ -0,0 +1,234 @@ +"""Enforcement daemon loop and related helpers.""" + +from __future__ import annotations + +import logging +import time + +from python_pkg.steam_backlog_enforcer.config import ( + Config, + State, + load_snapshot, +) +from python_pkg.steam_backlog_enforcer.enforcer import ( + enforce_allowed_game, + send_notification, +) +from python_pkg.steam_backlog_enforcer.game_install import ( + PROTECTED_APP_IDS, + _echo, + get_installed_games, + install_game, + is_game_installed, + uninstall_game, + uninstall_other_games, +) +from python_pkg.steam_backlog_enforcer.library_hider import hide_other_games +from python_pkg.steam_backlog_enforcer.steam_api import SteamAPIClient +from python_pkg.steam_backlog_enforcer.store_blocker import block_store + +logger = logging.getLogger(__name__) + + +# ────────────────────────────────────────────────────────────── +# Helpers +# ────────────────────────────────────────────────────────────── + + +def get_all_owned_app_ids(config: Config) -> list[int]: + """Get all owned game app IDs from the snapshot or Steam API.""" + snapshot = load_snapshot() + if snapshot: + return [d["app_id"] for d in snapshot] + + # Fall back to a quick API call. + try: + client = SteamAPIClient(config.steam_api_key, config.steam_id) + owned = client.get_owned_games() + return [g["appid"] for g in owned] + except (OSError, RuntimeError, ValueError): + logger.warning("Could not fetch owned game list for hiding.") + return [] + + +# ────────────────────────────────────────────────────────────── +# Enforce mode (daemon loop) +# ────────────────────────────────────────────────────────────── + +# How often the enforce loop runs (seconds). +ENFORCE_INTERVAL = 3 + + +def _guard_installed_games(allowed_app_id: int | None) -> int: + """Remove any unauthorized game manifests + files. Runs every loop. + + Returns number of games removed this pass. + """ + installed = get_installed_games() + count = 0 + for app_id, name in installed: + if app_id == allowed_app_id: + continue + if app_id in PROTECTED_APP_IDS: + continue + + logger.warning( + "Unauthorized game detected — removing: %s (AppID=%d)", name, app_id + ) + if uninstall_game(app_id, name): + count += 1 + send_notification( + "Game Removed!", + f"Uninstalled {name} (AppID={app_id}). " + f"Only the assigned game is allowed.", + ) + return count + + +def _enforce_setup(config: Config, state: State) -> None: + """Perform initial setup for enforcement mode. + + Args: + config: Enforcer configuration. + state: Current enforcer state. + """ + # Initial store block. + if config.block_store: + if block_store(): + _echo(" Steam store: BLOCKED") + else: + _echo(" Steam store: FAILED (need sudo?)") + + # Initial cleanup. + if config.uninstall_other_games: + _echo(" Uninstalling non-assigned games...") + count = uninstall_other_games(state.current_app_id) + _echo(f" Uninstalled {count} games") + + # Auto-install the assigned game. + _enforce_auto_install(config, state) + + # Hide all other games in the Steam library. + _enforce_hide_games(config, state) + + +def _enforce_auto_install(config: Config, state: State) -> None: + """Auto-install the assigned game if not already installed. + + Args: + config: Enforcer configuration. + state: Current enforcer state. + """ + app_id = state.current_app_id + if app_id is None: + return + if not is_game_installed(app_id): + _echo(f" Auto-installing {state.current_game_name}...") + if install_game( + app_id, + state.current_game_name, + config.steam_id, + use_steam_protocol=True, + ): + send_notification( + "Game Installing", + f"{state.current_game_name} is being downloaded.", + ) + else: + _echo(" Could not auto-install. Install manually from Steam.") + else: + _echo(f" Assigned game already installed: {state.current_game_name}") + + +def _enforce_hide_games(config: Config, state: State) -> None: + """Hide non-assigned games in the Steam library. + + Args: + config: Enforcer configuration. + state: Current enforcer state. + """ + owned_ids = get_all_owned_app_ids(config) + if owned_ids: + hidden = hide_other_games(owned_ids, state.current_app_id) + if hidden > 0: + _echo(f" Library: hid {hidden} games (only assigned game visible)") + else: + _echo(" Library: games already hidden") + else: + _echo(" Library hiding: skipped (no owned game list — run 'scan' first)") + + +def _enforce_loop_iteration(config: Config, state: State) -> None: + """Perform one iteration of the enforcement loop. + + Args: + config: Enforcer configuration. + state: Current enforcer state. + """ + # A) Kill unauthorized game processes. + if config.kill_unauthorized_games: + violations = enforce_allowed_game( + state.current_app_id, + kill_unauthorized=True, + ) + for pid, app_id in violations: + _echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})") + send_notification( + "Game Blocked!", + f"Killed unauthorized game (AppID={app_id}). " + f"Focus on {state.current_game_name}!", + ) + + # B) Remove any newly-installed unauthorized games. + if config.uninstall_other_games: + removed = _guard_installed_games(state.current_app_id) + if removed > 0: + _echo(f" Guard removed {removed} unauthorized game(s)") + + # C) Re-install assigned game if it was somehow removed. + app_id = state.current_app_id + if app_id is not None and not is_game_installed(app_id): + logger.info( + "Assigned game disappeared — re-installing %s", + state.current_game_name, + ) + install_game( + app_id, + state.current_game_name, + config.steam_id, + ) + + +def do_enforce(config: Config, state: State) -> None: + """Run the enforcer: block store, uninstall other games, kill processes. + + This is a persistent loop that continuously: + 1. Keeps the Steam store blocked. + 2. Removes any newly-installed unauthorized games. + 3. Auto-installs the assigned game if missing. + 4. Kills any running unauthorized game processes. + """ + if state.current_app_id is None: + _echo("No game assigned. Run 'scan' first.") + return + + _echo(f"Enforcing: {state.current_game_name} (AppID={state.current_app_id})") + _enforce_setup(config, state) + + _echo(f" Enforce loop: ACTIVE (every {ENFORCE_INTERVAL}s)") + _echo(" Guarding: processes + installs + store") + _echo(" Press Ctrl+C to stop.\n") + try: + while True: + # Reload state from disk so CLI changes (e.g. new game + # assignment via ``done`` / ``scan``) take effect immediately + # without needing to restart the daemon. + fresh = State.load() + state.current_app_id = fresh.current_app_id + state.current_game_name = fresh.current_game_name + state.finished_app_ids = fresh.finished_app_ids + + _enforce_loop_iteration(config, state) + time.sleep(ENFORCE_INTERVAL) + except KeyboardInterrupt: + _echo("\nEnforcer stopped.") diff --git a/steam_backlog_enforcer/scanning.py b/steam_backlog_enforcer/scanning.py index 3ebb532..3bf4fd0 100644 --- a/steam_backlog_enforcer/scanning.py +++ b/steam_backlog_enforcer/scanning.py @@ -13,26 +13,20 @@ from python_pkg.steam_backlog_enforcer.config import ( save_snapshot, ) from python_pkg.steam_backlog_enforcer.enforcer import ( - enforce_allowed_game, send_notification, ) from python_pkg.steam_backlog_enforcer.game_install import ( - PROTECTED_APP_IDS, _echo, - get_installed_games, install_game, is_game_installed, - uninstall_game, uninstall_other_games, ) from python_pkg.steam_backlog_enforcer.hltb import fetch_hltb_times_cached -from python_pkg.steam_backlog_enforcer.library_hider import hide_other_games from python_pkg.steam_backlog_enforcer.protondb import ( ProtonDBRating, fetch_protondb_ratings, ) from python_pkg.steam_backlog_enforcer.steam_api import GameInfo, SteamAPIClient -from python_pkg.steam_backlog_enforcer.store_blocker import block_store logger = logging.getLogger(__name__) @@ -313,207 +307,3 @@ def detect_tampering(config: Config, state: State) -> None: "Tampering Detected!", f"Achievements unlocked on {len(suspicious)} non-assigned games!", ) - - -# ────────────────────────────────────────────────────────────── -# Helpers -# ────────────────────────────────────────────────────────────── - - -def get_all_owned_app_ids(config: Config) -> list[int]: - """Get all owned game app IDs from the snapshot or Steam API.""" - snapshot = load_snapshot() - if snapshot: - return [d["app_id"] for d in snapshot] - - # Fall back to a quick API call. - try: - client = SteamAPIClient(config.steam_api_key, config.steam_id) - owned = client.get_owned_games() - return [g["appid"] for g in owned] - except (OSError, RuntimeError, ValueError): - logger.warning("Could not fetch owned game list for hiding.") - return [] - - -# ────────────────────────────────────────────────────────────── -# Enforce mode (daemon loop) -# ────────────────────────────────────────────────────────────── - -# How often the enforce loop runs (seconds). -ENFORCE_INTERVAL = 3 - - -def _guard_installed_games(allowed_app_id: int | None) -> int: - """Remove any unauthorized game manifests + files. Runs every loop. - - Returns number of games removed this pass. - """ - installed = get_installed_games() - count = 0 - for app_id, name in installed: - if app_id == allowed_app_id: - continue - if app_id in PROTECTED_APP_IDS: - continue - - logger.warning( - "Unauthorized game detected — removing: %s (AppID=%d)", name, app_id - ) - if uninstall_game(app_id, name): - count += 1 - send_notification( - "Game Removed!", - f"Uninstalled {name} (AppID={app_id}). " - f"Only the assigned game is allowed.", - ) - return count - - -def _enforce_setup(config: Config, state: State) -> None: - """Perform initial setup for enforcement mode. - - Args: - config: Enforcer configuration. - state: Current enforcer state. - """ - # Initial store block. - if config.block_store: - if block_store(): - _echo(" Steam store: BLOCKED") - else: - _echo(" Steam store: FAILED (need sudo?)") - - # Initial cleanup. - if config.uninstall_other_games: - _echo(" Uninstalling non-assigned games...") - count = uninstall_other_games(state.current_app_id) - _echo(f" Uninstalled {count} games") - - # Auto-install the assigned game. - _enforce_auto_install(config, state) - - # Hide all other games in the Steam library. - _enforce_hide_games(config, state) - - -def _enforce_auto_install(config: Config, state: State) -> None: - """Auto-install the assigned game if not already installed. - - Args: - config: Enforcer configuration. - state: Current enforcer state. - """ - app_id = state.current_app_id - if app_id is None: - return - if not is_game_installed(app_id): - _echo(f" Auto-installing {state.current_game_name}...") - if install_game( - app_id, - state.current_game_name, - config.steam_id, - use_steam_protocol=True, - ): - send_notification( - "Game Installing", - f"{state.current_game_name} is being downloaded.", - ) - else: - _echo(" Could not auto-install. Install manually from Steam.") - else: - _echo(f" Assigned game already installed: {state.current_game_name}") - - -def _enforce_hide_games(config: Config, state: State) -> None: - """Hide non-assigned games in the Steam library. - - Args: - config: Enforcer configuration. - state: Current enforcer state. - """ - owned_ids = get_all_owned_app_ids(config) - if owned_ids: - hidden = hide_other_games(owned_ids, state.current_app_id) - if hidden > 0: - _echo(f" Library: hid {hidden} games (only assigned game visible)") - else: - _echo(" Library: games already hidden") - else: - _echo(" Library hiding: skipped (no owned game list — run 'scan' first)") - - -def _enforce_loop_iteration(config: Config, state: State) -> None: - """Perform one iteration of the enforcement loop. - - Args: - config: Enforcer configuration. - state: Current enforcer state. - """ - # A) Kill unauthorized game processes. - if config.kill_unauthorized_games: - violations = enforce_allowed_game( - state.current_app_id, - kill_unauthorized=True, - ) - for pid, app_id in violations: - _echo(f" Killed unauthorized game: AppID={app_id} (PID={pid})") - send_notification( - "Game Blocked!", - f"Killed unauthorized game (AppID={app_id}). " - f"Focus on {state.current_game_name}!", - ) - - # B) Remove any newly-installed unauthorized games. - if config.uninstall_other_games: - removed = _guard_installed_games(state.current_app_id) - if removed > 0: - _echo(f" Guard removed {removed} unauthorized game(s)") - - # C) Re-install assigned game if it was somehow removed. - app_id = state.current_app_id - if app_id is not None and not is_game_installed(app_id): - logger.info( - "Assigned game disappeared — re-installing %s", - state.current_game_name, - ) - install_game( - app_id, - state.current_game_name, - config.steam_id, - ) - - -def do_enforce(config: Config, state: State) -> None: - """Run the enforcer: block store, uninstall other games, kill processes. - - This is a persistent loop that continuously: - 1. Keeps the Steam store blocked. - 2. Removes any newly-installed unauthorized games. - 3. Auto-installs the assigned game if missing. - 4. Kills any running unauthorized game processes. - """ - if state.current_app_id is None: - _echo("No game assigned. Run 'scan' first.") - return - - _echo(f"Enforcing: {state.current_game_name} (AppID={state.current_app_id})") - _enforce_setup(config, state) - - _echo(f" Enforce loop: ACTIVE (every {ENFORCE_INTERVAL}s)") - _echo(" Guarding: processes + installs + store") - _echo(" Press Ctrl+C to stop.\n") - try: - while True: - # Reload state from disk so CLI changes (e.g. new game - # assignment via ``done`` / ``scan``) take effect immediately - # without needing to restart the daemon. - fresh = State.load() - state.current_app_id = fresh.current_app_id - state.current_game_name = fresh.current_game_name - state.finished_app_ids = fresh.finished_app_ids - - _enforce_loop_iteration(config, state) - time.sleep(ENFORCE_INTERVAL) - except KeyboardInterrupt: - _echo("\nEnforcer stopped.")