🎨 Various code style / behaviour fixes

- Merged platform sources when possible
- Added URLExecutableSource class
- Moved replaced_by_schema_key to utils/decorators
- Better retryable exception handling in some managers
-  Split SteamHelper into SteamFileHelper and SteamAPIHelper
- Delegated SteamRateLimiter creation to SteamAPIManager init
- Using additional_data for appid in SteamAPIManager
- Added Windows support for Legendary
- Stylistic changed suggested by pylint
This commit is contained in:
GeoffreyCoulaud
2023-06-10 02:59:41 +02:00
parent 070d875ff8
commit 842f9fe522
17 changed files with 182 additions and 224 deletions

View File

@@ -6,12 +6,15 @@ import yaml
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.utils.decorators import (
replaced_by_env_path,
replaced_by_path,
replaced_by_schema_key,
)
from src.utils.decorators import replaced_by_env_path, replaced_by_path
class BottlesSourceIterator(SourceIterator):
@@ -49,22 +52,16 @@ class BottlesSourceIterator(SourceIterator):
yield (game, additional_data)
class BottlesSource(Source):
class BottlesSource(URLExecutableSource):
"""Generic Bottles source"""
name = "Bottles"
location_key = "bottles-location"
def __iter__(self) -> SourceIterator:
return BottlesSourceIterator(self)
class BottlesLinuxSource(BottlesSource, LinuxSource):
variant = "linux"
executable_format = 'xdg-open bottles:run/"{bottle_name}"/"{game_name}"'
iterator_class = BottlesSourceIterator
url_format = 'bottles:run/"{bottle_name}"/"{game_name}"'
available_on = set(("linux",))
@property
@BottlesSource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_path("~/.var/app/com.usebottles.bottles/data/bottles/")
@replaced_by_env_path("XDG_DATA_HOME", "bottles/")
@replaced_by_path("~/.local/share/bottles/")

View File

@@ -9,13 +9,15 @@ from typing import Optional, TypedDict
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
URLExecutableSource,
SourceIterationResult,
SourceIterator,
WindowsSource,
)
from src.utils.decorators import replaced_by_env_path, replaced_by_path
from src.utils.decorators import (
replaced_by_env_path,
replaced_by_path,
replaced_by_schema_key,
)
class HeroicLibraryEntry(TypedDict):
@@ -112,40 +114,24 @@ class HeroicSourceIterator(SourceIterator):
yield result
class HeroicSource(Source):
class HeroicSource(URLExecutableSource):
"""Generic heroic games launcher source"""
name = "Heroic"
location_key = "heroic-location"
iterator_class = HeroicSourceIterator
url_format = "heroic://launch/{app_name}"
available_on = set(("linux", "win32"))
@property
def game_id_format(self) -> str:
"""The string format used to construct game IDs"""
return self.name.lower() + "_{service}_{game_id}"
def __iter__(self):
return HeroicSourceIterator(source=self)
class HeroicLinuxSource(HeroicSource, LinuxSource):
variant = "linux"
executable_format = "xdg-open heroic://launch/{app_name}"
@property
@HeroicSource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_path("~/.var/app/com.heroicgameslauncher.hgl/config/heroic/")
@replaced_by_env_path("XDG_CONFIG_HOME", "heroic/")
@replaced_by_path("~/.config/heroic/")
def location(self) -> Path:
raise FileNotFoundError()
class HeroicWindowsSource(HeroicSource, WindowsSource):
variant = "windows"
executable_format = "start heroic://launch/{app_name}"
@property
@HeroicSource.replaced_by_schema_key()
@replaced_by_env_path("appdata", "heroic/")
def location(self) -> Path:
raise FileNotFoundError()

View File

@@ -5,13 +5,15 @@ from time import time
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
SourceIterationResult,
SourceIterator,
WindowsSource,
URLExecutableSource,
)
from src.utils.decorators import (
replaced_by_env_path,
replaced_by_path,
replaced_by_schema_key,
)
from src.utils.decorators import replaced_by_env_path, replaced_by_path
class ItchSourceIterator(SourceIterator):
@@ -54,33 +56,17 @@ class ItchSourceIterator(SourceIterator):
yield (game, additional_data)
class ItchSource(Source):
class ItchSource(URLExecutableSource):
name = "Itch"
location_key = "itch-location"
def __iter__(self) -> SourceIterator:
return ItchSourceIterator(self)
class ItchLinuxSource(ItchSource, LinuxSource):
variant = "linux"
executable_format = "xdg-open itch://caves/{cave_id}/launch"
iterator_class = ItchSourceIterator
url_format = "itch://caves/{cave_id}/launch"
available_on = set(("linux", "win32"))
@property
@ItchSource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_path("~/.var/app/io.itch.itch/config/itch/")
@replaced_by_env_path("XDG_DATA_HOME", "itch/")
@replaced_by_path("~/.config/itch")
def location(self) -> Path:
raise FileNotFoundError()
class ItchWindowsSource(ItchSource, WindowsSource):
variant = "windows"
executable_format = "start itch://caves/{cave_id}/launch"
@property
@ItchSource.replaced_by_schema_key()
@replaced_by_env_path("appdata", "itch/")
def location(self) -> Path:
raise FileNotFoundError()

View File

@@ -1,19 +1,18 @@
import logging
from pathlib import Path
from typing import Generator
import json
import logging
from json import JSONDecodeError
from pathlib import Path
from time import time
from typing import Generator
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
SourceIterationResult,
SourceIterator,
from src.importer.sources.source import Source, SourceIterationResult, SourceIterator
from src.utils.decorators import (
replaced_by_env_path,
replaced_by_path,
replaced_by_schema_key,
)
from src.utils.decorators import replaced_by_env_path, replaced_by_path
class LegendarySourceIterator(SourceIterator):
@@ -74,22 +73,14 @@ class LegendarySourceIterator(SourceIterator):
class LegendarySource(Source):
name = "Legendary"
location_key = "legendary-location"
def __iter__(self) -> SourceIterator:
return LegendarySourceIterator(self)
# TODO add Legendary windows variant
class LegendaryLinuxSource(LegendarySource, LinuxSource):
variant = "linux"
executable_format = "legendary launch {app_name}"
iterator_class = LegendarySourceIterator
available_on = set(("linux", "win32"))
@property
@LegendarySource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_env_path("XDG_CONFIG_HOME", "legendary/")
@replaced_by_path("~/.config/legendary/")
@replaced_by_path("~\\.config\\legendary\\")
def location(self) -> Path:
raise FileNotFoundError()

View File

@@ -4,12 +4,11 @@ from time import time
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.utils.decorators import replaced_by_path
from src.utils.decorators import replaced_by_path, replaced_by_schema_key
class LutrisSourceIterator(SourceIterator):
@@ -47,7 +46,6 @@ class LutrisSourceIterator(SourceIterator):
game_id=row[2], game_internal_id=row[0]
),
"executable": self.source.executable_format.format(game_id=row[2]),
"developer": None, # TODO get developer metadata on Lutris
}
game = Game(values, allow_side_effects=False)
@@ -59,26 +57,20 @@ class LutrisSourceIterator(SourceIterator):
yield (game, additional_data)
class LutrisSource(Source):
class LutrisSource(URLExecutableSource):
"""Generic lutris source"""
name = "Lutris"
location_key = "lutris-location"
iterator_class = LutrisSourceIterator
url_format = "lutris:rungameid/{game_id}"
available_on = set(("linux",))
@property
def game_id_format(self):
return super().game_id_format + "_{game_internal_id}"
def __iter__(self):
return LutrisSourceIterator(source=self)
class LutrisLinuxSource(LutrisSource, LinuxSource):
variant = "linux"
executable_format = "xdg-open lutris:rungameid/{game_id}"
@property
@LutrisSource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_path("~/.var/app/net.lutris.Lutris/data/lutris/")
@replaced_by_path("~/.local/share/lutris/")
def location(self):

View File

@@ -1,13 +1,11 @@
import sys
from abc import abstractmethod
from collections.abc import Iterable, Iterator
from functools import wraps
from pathlib import Path
from typing import Generator, Any, TypedDict
from typing import Generator, Any
from src import shared
from src.game import Game
from src.utils.decorators import replaced_by_path
# Type of the data returned by iterating on a Source
SourceIterationResult = None | Game | tuple[Game, tuple[Any]]
@@ -45,13 +43,9 @@ class Source(Iterable):
"""Source of games. E.g an installed app with a config file that lists game directories"""
name: str
variant: str
location_key: str
available_on: set[str]
def __init__(self) -> None:
super().__init__()
self.available_on = set()
iterator_class: type[SourceIterator]
variant: str = None
available_on: set[str] = set()
@property
def full_name(self) -> str:
@@ -83,6 +77,14 @@ class Source(Iterable):
return False
return sys.platform in self.available_on
@property
def location_key(self) -> str:
"""
The schema key pointing to the user-set location for the source.
May be overriden by inherinting classes.
"""
return f"{self.name.lower()}-location"
def update_location_schema_key(self):
"""Update the schema value for this source's location if possible"""
try:
@@ -91,19 +93,9 @@ class Source(Iterable):
return
shared.schema.set_string(self.location_key, location)
@classmethod
def replaced_by_schema_key(cls): # Decorator builder
"""Replace the returned path with schema's path if valid"""
def decorator(original_function): # Built decorator (closure)
@wraps(original_function)
def wrapper(*args, **kwargs): # func's override
override = shared.schema.get_string(cls.location_key)
return replaced_by_path(override)(original_function)(*args, **kwargs)
return wrapper
return decorator
def __iter__(self) -> SourceIterator:
"""Get an iterator for the source"""
return self.iterator_class(self)
@property
@abstractmethod
@@ -115,22 +107,21 @@ class Source(Iterable):
def executable_format(self) -> str:
"""The executable format used to construct game executables"""
@abstractmethod
def __iter__(self) -> SourceIterator:
"""Get the source's iterator, to use in for loops"""
# pylint: disable=abstract-method
class URLExecutableSource(Source):
"""Source class that use custom URLs to start games"""
class WindowsSource(Source):
"""Mixin for sources available on Windows"""
url_format: str
def __init__(self) -> None:
super().__init__()
self.available_on.add("win32")
class LinuxSource(Source):
"""Mixin for sources available on Linux"""
def __init__(self) -> None:
super().__init__()
self.available_on.add("linux")
@property
def executable_format(self) -> str:
match sys.platform:
case "win32":
return "start " + self.url_format
case "linux":
return "xdg-open " + self.url_format
case other:
raise NotImplementedError(
f"No URL handler command available for {other}"
)

View File

@@ -6,14 +6,16 @@ from typing import Iterable
from src import shared
from src.game import Game
from src.importer.sources.source import (
LinuxSource,
Source,
SourceIterationResult,
SourceIterator,
WindowsSource,
URLExecutableSource,
)
from src.utils.decorators import replaced_by_env_path, replaced_by_path
from src.utils.steam import SteamHelper, SteamInvalidManifestError
from src.utils.decorators import (
replaced_by_env_path,
replaced_by_path,
replaced_by_schema_key,
)
from src.utils.steam import SteamFileHelper, SteamInvalidManifestError
class SteamSourceIterator(SourceIterator):
@@ -22,11 +24,11 @@ class SteamSourceIterator(SourceIterator):
def get_manifest_dirs(self) -> Iterable[Path]:
"""Get dirs that contain steam app manifests"""
libraryfolders_path = self.source.location / "steamapps" / "libraryfolders.vdf"
with open(libraryfolders_path, "r") as file:
with open(libraryfolders_path, "r", encoding="utf-8") as file:
contents = file.read()
return [
Path(path) / "steamapps"
for path in re.findall('"path"\s+"(.*)"\n', contents, re.IGNORECASE)
for path in re.findall('"path"\\s+"(.*)"\n', contents, re.IGNORECASE)
]
def get_manifests(self) -> Iterable[Path]:
@@ -50,15 +52,15 @@ class SteamSourceIterator(SourceIterator):
manifests = self.get_manifests()
for manifest in manifests:
# Get metadata from manifest
steam = SteamHelper()
steam = SteamFileHelper()
try:
local_data = steam.get_manifest_data(manifest)
except (OSError, SteamInvalidManifestError):
continue
# Skip non installed games
INSTALLED_MASK: int = 4
if not int(local_data["stateflags"]) & INSTALLED_MASK:
installed_mask = 4
if not int(local_data["stateflags"]) & installed_mask:
continue
# Skip duplicate appids
@@ -85,40 +87,24 @@ class SteamSourceIterator(SourceIterator):
/ "librarycache"
/ f"{appid}_library_600x900.jpg"
)
additional_data = {"local_image_path": image_path}
additional_data = {"local_image_path": image_path, "steam_appid": appid}
# Produce game
yield (game, additional_data)
class SteamSource(Source):
class SteamSource(URLExecutableSource):
name = "Steam"
location_key = "steam-location"
def __iter__(self):
return SteamSourceIterator(source=self)
class SteamLinuxSource(SteamSource, LinuxSource):
variant = "linux"
executable_format = "xdg-open steam://rungameid/{game_id}"
iterator_class = SteamSourceIterator
url_format = "steam://rungameid/{game_id}"
available_on = set(("linux", "win32"))
@property
@SteamSource.replaced_by_schema_key()
@replaced_by_schema_key
@replaced_by_path("~/.var/app/com.valvesoftware.Steam/data/Steam/")
@replaced_by_env_path("XDG_DATA_HOME", "Steam/")
@replaced_by_path("~/.steam/")
@replaced_by_path("~/.local/share/Steam/")
def location(self):
raise FileNotFoundError()
class SteamWindowsSource(SteamSource, WindowsSource):
variant = "windows"
executable_format = "start steam://rungameid/{game_id}"
@property
@SteamSource.replaced_by_schema_key()
@replaced_by_env_path("programfiles(x86)", "Steam")
def location(self):
raise FileNotFoundError()

View File

@@ -34,12 +34,12 @@ from src import shared
from src.details_window import DetailsWindow
from src.game import Game
from src.importer.importer import Importer
from src.importer.sources.bottles_source import BottlesLinuxSource
from src.importer.sources.heroic_source import HeroicLinuxSource, HeroicWindowsSource
from src.importer.sources.itch_source import ItchLinuxSource, ItchWindowsSource
from src.importer.sources.legendary_source import LegendaryLinuxSource
from src.importer.sources.lutris_source import LutrisLinuxSource
from src.importer.sources.steam_source import SteamLinuxSource, SteamWindowsSource
from src.importer.sources.bottles_source import BottlesSource
from src.importer.sources.heroic_source import HeroicSource
from src.importer.sources.itch_source import ItchSource
from src.importer.sources.legendary_source import LegendarySource
from src.importer.sources.lutris_source import LutrisSource
from src.importer.sources.steam_source import SteamSource
from src.preferences import PreferencesWindow
from src.store.managers.display_manager import DisplayManager
from src.store.managers.file_manager import FileManager
@@ -190,21 +190,25 @@ class CartridgesApplication(Adw.Application):
def on_import_action(self, *_args):
importer = Importer()
if shared.schema.get_boolean("lutris"):
importer.add_source(LutrisLinuxSource())
importer.add_source(LutrisSource())
if shared.schema.get_boolean("steam"):
importer.add_source(SteamLinuxSource())
importer.add_source(SteamWindowsSource())
importer.add_source(SteamSource())
if shared.schema.get_boolean("heroic"):
importer.add_source(HeroicLinuxSource())
importer.add_source(HeroicWindowsSource())
importer.add_source(HeroicSource())
if shared.schema.get_boolean("bottles"):
importer.add_source(BottlesLinuxSource())
importer.add_source(BottlesSource())
if shared.schema.get_boolean("itch"):
importer.add_source(ItchLinuxSource())
importer.add_source(ItchWindowsSource())
importer.add_source(ItchSource())
if shared.schema.get_boolean("legendary"):
importer.add_source(LegendaryLinuxSource())
importer.add_source(LegendarySource())
importer.run()
def on_remove_game_action(self, *_args):

View File

@@ -34,7 +34,7 @@ class AsyncManager(Manager):
task.set_task_data((game, additional_data))
task.run_in_thread(self._task_thread_func)
def _task_thread_func(self, _task, _source_object, data, cancellable):
def _task_thread_func(self, _task, _source_object, data, _cancellable):
"""Task thread entry point"""
game, additional_data, *_rest = data
self.execute_resilient_manager_logic(game, additional_data)

View File

@@ -2,13 +2,12 @@ from pathlib import Path
import requests
from gi.repository import GdkPixbuf, Gio
from requests import HTTPError
from urllib3.exceptions import SSLError
from requests.exceptions import HTTPError, SSLError
from src import shared
from src.game import Game
from src.store.managers.manager import Manager
from src.store.managers.local_cover_manager import LocalCoverManager
from src.store.managers.manager import Manager
from src.utils.save_cover import resize_cover, save_cover

View File

@@ -64,7 +64,7 @@ class Manager:
"""Execute the manager logic and handle its errors by reporting them or retrying"""
try:
self.manager_logic(game, additional_data)
except Exception as error:
except Exception as error: # pylint: disable=broad-exception-caught
logging_args = (
type(error).__name__,
self.name,

View File

@@ -2,12 +2,11 @@ from pathlib import Path
import requests
from gi.repository import Gio
from requests import HTTPError
from urllib3.exceptions import SSLError
from requests.exceptions import HTTPError, SSLError
from src.game import Game
from src.store.managers.manager import Manager
from src.store.managers.local_cover_manager import LocalCoverManager
from src.store.managers.manager import Manager
from src.utils.save_cover import resize_cover, save_cover

View File

@@ -1,18 +1,20 @@
from urllib3.exceptions import SSLError
from json import JSONDecodeError
from requests.exceptions import HTTPError, SSLError
from src.game import Game
from src.store.managers.async_manager import AsyncManager
from src.store.managers.itch_cover_manager import ItchCoverManager
from src.store.managers.local_cover_manager import LocalCoverManager
from src.store.managers.steam_api_manager import SteamAPIManager
from src.utils.steamgriddb import HTTPError, SGDBAuthError, SGDBHelper
from src.utils.steamgriddb import SGDBAuthError, SGDBHelper
class SGDBManager(AsyncManager):
"""Manager in charge of downloading a game's cover from steamgriddb"""
run_after = set((SteamAPIManager, LocalCoverManager, ItchCoverManager))
retryable_on = set((HTTPError, SSLError))
retryable_on = set((HTTPError, SSLError, ConnectionError, JSONDecodeError))
def manager_logic(self, game: Game, _additional_data: dict) -> None:
try:

View File

@@ -1,12 +1,12 @@
from urllib3.exceptions import SSLError
from requests.exceptions import HTTPError, SSLError
from src.game import Game
from src.store.managers.async_manager import AsyncManager
from src.utils.steam import (
HTTPError,
SteamGameNotFoundError,
SteamHelper,
SteamAPIHelper,
SteamNotAGameError,
SteamRateLimiter,
)
@@ -15,15 +15,22 @@ class SteamAPIManager(AsyncManager):
retryable_on = set((HTTPError, SSLError))
def manager_logic(self, game: Game, _additional_data: dict) -> None:
steam_api_helper: SteamAPIHelper = None
steam_rate_limiter: SteamRateLimiter = None
def __init__(self) -> None:
super().__init__()
self.steam_rate_limiter = SteamRateLimiter()
self.steam_api_helper = SteamAPIHelper(self.steam_rate_limiter)
def manager_logic(self, game: Game, additional_data: dict) -> None:
# Skip non-steam games
if not game.source.startswith("steam_"):
appid = additional_data.get("steam_appid", None)
if appid is None:
return
# Get online metadata
appid = str(game.game_id).split("_")[-1]
steam = SteamHelper()
try:
online_data = steam.get_api_data(appid=appid)
online_data = self.steam_api_helper.get_api_data(appid=appid)
except (SteamNotAGameError, SteamGameNotFoundError):
game.update_values({"blacklisted": True})
else:

View File

@@ -2,6 +2,8 @@ from pathlib import Path
from os import PathLike, environ
from functools import wraps
from src import shared
def replaced_by_path(override: PathLike): # Decorator builder
"""Replace the method's returned path with the override
@@ -36,3 +38,18 @@ def replaced_by_env_path(env_var_name: str, suffix: PathLike | None = None):
return wrapper
return decorator
def replaced_by_schema_key(original_method): # Built decorator (closure)
"""
Replace the original method's value by the path pointed at in the schema
by the class' location key (if that override exists)
"""
@wraps(original_method)
def wrapper(*args, **kwargs): # func's override
source = args[0]
override = shared.schema.get_string(source.location_key)
return replaced_by_path(override)(original_method)(*args, **kwargs)
return wrapper

View File

@@ -4,7 +4,7 @@ import re
from typing import TypedDict
import requests
from requests import HTTPError
from requests.exceptions import HTTPError
from src import shared
from src.utils.rate_limiter import PickHistory, RateLimiter
@@ -27,7 +27,7 @@ class SteamInvalidManifestError(SteamError):
class SteamManifestData(TypedDict):
"""Dict returned by SteamHelper.get_manifest_data"""
"""Dict returned by SteamFileHelper.get_manifest_data"""
name: str
appid: str
@@ -35,7 +35,7 @@ class SteamManifestData(TypedDict):
class SteamAPIData(TypedDict):
"""Dict returned by SteamHelper.get_api_data"""
"""Dict returned by SteamAPIHelper.get_api_data"""
developers: str
@@ -73,34 +73,35 @@ class SteamRateLimiter(RateLimiter):
shared.state_schema.set_string("steam-limiter-tokens-history", timestamps_str)
class SteamHelper:
"""Helper around the Steam API"""
base_url = "https://store.steampowered.com/api"
rate_limiter: SteamRateLimiter = None
def __init__(self) -> None:
# Instanciate the rate limiter on the class to share across instances
# Can't be done at class creation time, schema isn't available yet
if self.__class__.rate_limiter is None:
self.__class__.rate_limiter = SteamRateLimiter()
class SteamFileHelper:
"""Helper for steam file formats"""
def get_manifest_data(self, manifest_path) -> SteamManifestData:
"""Get local data for a game from its manifest"""
with open(manifest_path) as file:
with open(manifest_path, "r", encoding="utf-8") as file:
contents = file.read()
data = {}
for key in SteamManifestData.__required_keys__:
regex = f'"{key}"\s+"(.*)"\n'
for key in SteamManifestData.__required_keys__: # pylint: disable=no-member
regex = f'"{key}"\\s+"(.*)"\n'
if (match := re.search(regex, contents, re.IGNORECASE)) is None:
raise SteamInvalidManifestError()
data[key] = match.group(1)
return SteamManifestData(**data)
class SteamAPIHelper:
"""Helper around the Steam API"""
base_url = "https://store.steampowered.com/api"
rate_limiter: RateLimiter
def __init__(self, rate_limiter: RateLimiter) -> None:
self.rate_limiter = rate_limiter
def get_api_data(self, appid) -> SteamAPIData:
"""
Get online data for a game from its appid.

View File

@@ -3,7 +3,7 @@ from pathlib import Path
import requests
from gi.repository import Gio
from requests import HTTPError
from requests.exceptions import HTTPError
from src import shared
from src.utils.create_dialog import create_dialog