mirror of
https://github.com/kuhyx/steam-backlog-enforcer.git
synced 2026-07-04 11:43:12 +02:00
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.
This commit is contained in:
parent
6e3040ed84
commit
e618cabf0e
234
steam_backlog_enforcer/_enforce_loop.py
Normal file
234
steam_backlog_enforcer/_enforce_loop.py
Normal file
@ -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.")
|
||||
@ -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.")
|
||||
|
||||
Loading…
Reference in New Issue
Block a user