1 Commits

Author SHA1 Message Date
Renovate Bot
92572bf5a1 chore(deps): update archlinux:base-devel docker digest to 87a967f
All checks were successful
Code check / Check code (pull_request) Successful in 1m13s
2025-10-19 00:01:20 +00:00
4 changed files with 60 additions and 148 deletions

View File

@@ -16,7 +16,7 @@ repos:
- id: uv-lock - id: uv-lock
- repo: https://github.com/astral-sh/ruff-pre-commit - repo: https://github.com/astral-sh/ruff-pre-commit
rev: v0.14.1 rev: v0.14.0
hooks: hooks:
- id: ruff-check - id: ruff-check

View File

@@ -1440,7 +1440,6 @@ class InputManager(QObject):
self.udev_context = Context() self.udev_context = Context()
self.Devices = Devices self.Devices = Devices
self.monitor_ready = False self.monitor_ready = False
self.monitor_event = threading.Event()
# Подключаем сигнал hotplug к обработчику в главном потоке # Подключаем сигнал hotplug к обработчику в главном потоке
self.gamepad_hotplug.connect(self._on_gamepad_hotplug) self.gamepad_hotplug.connect(self._on_gamepad_hotplug)
@@ -1492,7 +1491,6 @@ class InputManager(QObject):
break break
self.monitor_ready = True self.monitor_ready = True
self.monitor_event.set()
logger.info(f"Drained {drained_count} initial events, now monitoring hotplug...") logger.info(f"Drained {drained_count} initial events, now monitoring hotplug...")
# Основной цикл # Основной цикл
@@ -1594,6 +1592,7 @@ class InputManager(QObject):
except Exception as e: except Exception as e:
logger.error(f"Error in hotplug handler: {e}", exc_info=True) logger.error(f"Error in hotplug handler: {e}", exc_info=True)
def check_gamepad(self) -> None: def check_gamepad(self) -> None:
""" """
Проверка и подключение геймпада. Проверка и подключение геймпада.
@@ -1602,23 +1601,18 @@ class InputManager(QObject):
try: try:
new_gamepad = self.find_gamepad() new_gamepad = self.find_gamepad()
# Проверяем, действительно ли это новый геймпад
if new_gamepad: if new_gamepad:
if not self.gamepad or new_gamepad.path != self.gamepad.path: if not self.gamepad or new_gamepad.path != self.gamepad.path:
logger.info(f"Gamepad connected: {new_gamepad.name} at {new_gamepad.path}") logger.info(f"Gamepad connected: {new_gamepad.name} at {new_gamepad.path}")
self.stop_rumble() self.stop_rumble()
self.gamepad = new_gamepad self.gamepad = new_gamepad
if self.gamepad_thread and self.gamepad_thread.is_alive(): if self.gamepad_thread:
self.gamepad_thread.join(timeout=2.0) self.gamepad_thread.join(timeout=2.0)
def start_monitoring():
# Ожидание готовности udev monitor без busy-wait
if not self.monitor_event.wait(timeout=2.0):
logger.warning("Timeout waiting for udev monitor readiness")
self.monitor_gamepad()
self.gamepad_thread = threading.Thread( self.gamepad_thread = threading.Thread(
target=start_monitoring, target=self.monitor_gamepad,
daemon=True daemon=True
) )
self.gamepad_thread.start() self.gamepad_thread.start()
@@ -1628,11 +1622,12 @@ class InputManager(QObject):
self.toggle_fullscreen.emit(True) self.toggle_fullscreen.emit(True)
elif self.gamepad and not any(self.gamepad.path == path for path in list_devices()): elif self.gamepad and not any(self.gamepad.path == path for path in list_devices()):
# Геймпад был подключён, но теперь его нет в системе
logger.info("Gamepad no longer detected") logger.info("Gamepad no longer detected")
self.stop_rumble() self.stop_rumble()
self.gamepad = None self.gamepad = None
if self.gamepad_thread and self.gamepad_thread.is_alive(): if self.gamepad_thread:
self.gamepad_thread.join(timeout=2.0) self.gamepad_thread.join(timeout=2.0)
if read_auto_fullscreen_gamepad() and not read_fullscreen_config(): if read_auto_fullscreen_gamepad() and not read_fullscreen_config():
@@ -1641,6 +1636,7 @@ class InputManager(QObject):
except Exception as e: except Exception as e:
logger.error(f"Error checking gamepad: {e}", exc_info=True) logger.error(f"Error checking gamepad: {e}", exc_info=True)
def find_gamepad(self) -> InputDevice | None: def find_gamepad(self) -> InputDevice | None:
""" """
Находит первый доступный геймпад. Находит первый доступный геймпад.

View File

@@ -1253,15 +1253,7 @@ class MainWindow(QMainWindow):
# Показываем прогресс # Показываем прогресс
self.autoInstallProgress.setVisible(True) self.autoInstallProgress.setVisible(True)
self.autoInstallProgress.setRange(0, 0) self.autoInstallProgress.setRange(0, 0)
self.portproton_api.get_autoinstall_games_async(on_autoinstall_games_loaded)
# Store the thread to prevent premature destruction
self.autoInstallLoadThread = self.portproton_api.start_autoinstall_games_load(on_autoinstall_games_loaded)
# Optional: Clean up thread when finished (prevents leak)
if self.autoInstallLoadThread:
def on_thread_finished():
self.autoInstallLoadThread = None # Release reference
self.autoInstallLoadThread.finished.connect(on_thread_finished)
self.stackedWidget.addWidget(autoInstallPage) self.stackedWidget.addWidget(autoInstallPage)

View File

@@ -6,16 +6,13 @@ import urllib.parse
import time import time
import glob import glob
import re import re
import hashlib
from collections.abc import Callable from collections.abc import Callable
from PySide6.QtCore import QThread, Signal
from portprotonqt.downloader import Downloader from portprotonqt.downloader import Downloader
from portprotonqt.logger import get_logger from portprotonqt.logger import get_logger
from portprotonqt.config_utils import get_portproton_location from portprotonqt.config_utils import get_portproton_location
logger = get_logger(__name__) logger = get_logger(__name__)
CACHE_DURATION = 30 * 24 * 60 * 60 # 30 days in seconds CACHE_DURATION = 30 * 24 * 60 * 60 # 30 days in seconds
AUTOINSTALL_CACHE_DURATION = 3600 # 1 hour for autoinstall cache
def normalize_name(s): def normalize_name(s):
""" """
@@ -62,7 +59,6 @@ class PortProtonAPI:
self.repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) self.repo_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
self.builtin_custom_folder = os.path.join(self.repo_root, "custom_data") self.builtin_custom_folder = os.path.join(self.repo_root, "custom_data")
self._topics_data = None self._topics_data = None
self._autoinstall_cache = None # New: In-memory cache
def _get_game_dir(self, exe_name: str) -> str: def _get_game_dir(self, exe_name: str) -> str:
game_dir = os.path.join(self.custom_data_dir, exe_name) game_dir = os.path.join(self.custom_data_dir, exe_name)
@@ -235,139 +231,67 @@ class PortProtonAPI:
logger.error(f"Failed to parse {file_path}: {e}") logger.error(f"Failed to parse {file_path}: {e}")
return None, None return None, None
def _compute_scripts_signature(self, auto_dir: str) -> str: def get_autoinstall_games_async(self, callback: Callable[[list[tuple]], None]) -> None:
"""Compute a hash-based signature of the autoinstall scripts to detect changes.""" """Load auto-install games with user/builtin covers (no async download here)."""
games = []
auto_dir = os.path.join(self.portproton_location or "", "data", "scripts", "pw_autoinstall") if self.portproton_location else ""
if not os.path.exists(auto_dir): if not os.path.exists(auto_dir):
return "" callback(games)
return
scripts = sorted(glob.glob(os.path.join(auto_dir, "*"))) scripts = sorted(glob.glob(os.path.join(auto_dir, "*")))
# Simple hash: concatenate sorted filenames and hash if not scripts:
filenames_str = "".join(sorted([os.path.basename(s) for s in scripts])) callback(games)
return hashlib.md5(filenames_str.encode()).hexdigest() return
def _load_autoinstall_cache(self): xdg_data_home = os.getenv("XDG_DATA_HOME",
"""Load cached autoinstall games if fresh and scripts unchanged.""" os.path.join(os.path.expanduser("~"), ".local", "share"))
if self._autoinstall_cache is not None: base_autoinstall_dir = os.path.join(xdg_data_home, "PortProtonQt", "custom_data", "autoinstall")
return self._autoinstall_cache os.makedirs(base_autoinstall_dir, exist_ok=True)
cache_dir = get_cache_dir()
cache_file = os.path.join(cache_dir, "autoinstall_games_cache.json")
if os.path.exists(cache_file):
try:
mod_time = os.path.getmtime(cache_file)
if time.time() - mod_time < AUTOINSTALL_CACHE_DURATION:
with open(cache_file, "rb") as f:
data = orjson.loads(f.read())
# Check signature
cached_signature = data.get("scripts_signature", "")
current_signature = self._compute_scripts_signature(
os.path.join(self.portproton_location or "", "data", "scripts", "pw_autoinstall")
)
if cached_signature != current_signature:
logger.info("Scripts signature mismatch; invalidating cache")
return None
self._autoinstall_cache = data["games"]
logger.info(f"Loaded {len(self._autoinstall_cache)} cached autoinstall games")
return self._autoinstall_cache
except Exception as e:
logger.error(f"Failed to load autoinstall cache: {e}")
return None
def _save_autoinstall_cache(self, games): for script_path in scripts:
"""Save parsed autoinstall games to cache with scripts signature.""" display_name, exe_name = self.parse_autoinstall_script(script_path)
try: script_name = os.path.splitext(os.path.basename(script_path))[0]
cache_dir = get_cache_dir()
cache_file = os.path.join(cache_dir, "autoinstall_games_cache.json")
auto_dir = os.path.join(self.portproton_location or "", "data", "scripts", "pw_autoinstall")
scripts_signature = self._compute_scripts_signature(auto_dir)
data = {"games": games, "scripts_signature": scripts_signature, "timestamp": time.time()}
with open(cache_file, "wb") as f:
f.write(orjson.dumps(data))
logger.debug(f"Saved {len(games)} autoinstall games to cache with signature {scripts_signature}")
except Exception as e:
logger.error(f"Failed to save autoinstall cache: {e}")
def start_autoinstall_games_load(self, callback: Callable[[list[tuple]], None]) -> QThread | None: if not (display_name and exe_name):
"""Start loading auto-install games in a background thread. Returns the thread for management.""" continue
# Check cache first (sync, fast)
cached_games = self._load_autoinstall_cache()
if cached_games is not None:
# Emit via callback immediately if cached
QThread.msleep(0) # Yield to Qt event loop
callback(cached_games)
return None # No thread needed
# No cache: Start background thread exe_name = os.path.splitext(exe_name)[0] # Без .exe
class AutoinstallWorker(QThread): user_game_folder = os.path.join(base_autoinstall_dir, exe_name)
finished = Signal(list) os.makedirs(user_game_folder, exist_ok=True)
api: "PortProtonAPI"
portproton_location: str | None
def run(self): # Поиск обложки
games = [] cover_path = ""
auto_dir = os.path.join( user_files = set(os.listdir(user_game_folder)) if os.path.exists(user_game_folder) else set()
self.portproton_location or "", "data", "scripts", "pw_autoinstall" for ext in [".jpg", ".png", ".jpeg", ".bmp"]:
) if self.portproton_location else "" candidate = f"cover{ext}"
if not os.path.exists(auto_dir): if candidate in user_files:
self.finished.emit(games) cover_path = os.path.join(user_game_folder, candidate)
return break
scripts = sorted(glob.glob(os.path.join(auto_dir, "*"))) if not cover_path:
if not scripts: logger.debug(f"No local cover found for autoinstall {exe_name}")
self.finished.emit(games)
return
xdg_data_home = os.getenv( # Формируем кортеж игры (добавлен exe_name в конец)
"XDG_DATA_HOME", game_tuple = (
os.path.join(os.path.expanduser("~"), ".local", "share"), display_name, # name
) "", # description
base_autoinstall_dir = os.path.join( cover_path, # cover
xdg_data_home, "PortProtonQt", "custom_data", "autoinstall" "", # appid
) f"autoinstall:{script_name}", # exec_line
os.makedirs(base_autoinstall_dir, exist_ok=True) "", # controller_support
"Never", # last_launch
"0h 0m", # formatted_playtime
"", # protondb_tier
"", # anticheat_status
0, # last_played
0, # playtime_seconds
"autoinstall", # game_source
exe_name # exe_name
)
games.append(game_tuple)
for script_path in scripts: callback(games)
display_name, exe_name = self.api.parse_autoinstall_script(script_path)
script_name = os.path.splitext(os.path.basename(script_path))[0]
if not (display_name and exe_name):
continue
exe_name = os.path.splitext(exe_name)[0]
user_game_folder = os.path.join(base_autoinstall_dir, exe_name)
os.makedirs(user_game_folder, exist_ok=True)
# Find cover
cover_path = ""
user_files = (
set(os.listdir(user_game_folder))
if os.path.exists(user_game_folder)
else set()
)
for ext in [".jpg", ".png", ".jpeg", ".bmp"]:
candidate = f"cover{ext}"
if candidate in user_files:
cover_path = os.path.join(user_game_folder, candidate)
break
if not cover_path:
logger.debug(f"No local cover found for autoinstall {exe_name}")
game_tuple = (
display_name, "", cover_path, "", f"autoinstall:{script_name}",
"", "Never", "0h 0m", "", "", 0, 0, "autoinstall", exe_name
)
games.append(game_tuple)
self.api._save_autoinstall_cache(games)
self.api._autoinstall_cache = games
self.finished.emit(games)
worker = AutoinstallWorker()
worker.api = self
worker.portproton_location = self.portproton_location
worker.finished.connect(lambda games: callback(games))
worker.start()
logger.info("Started background load of autoinstall games")
return worker
def _load_topics_data(self): def _load_topics_data(self):
"""Load and cache linux_gaming_topics_min.json from the archive.""" """Load and cache linux_gaming_topics_min.json from the archive."""