Merge pull request #150 from kra-mo/heroic-fixes-and-improvements

Heroic fixes and improvements
This commit is contained in:
kramo
2023-07-25 20:23:54 +02:00
committed by GitHub
14 changed files with 331 additions and 120 deletions

View File

@@ -26,17 +26,13 @@ import yaml
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import (
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.importer.sources.source import SourceIterable, URLExecutableSource
class BottlesSourceIterator(SourceIterator):
class BottlesSourceIterable(SourceIterable):
source: "BottlesSource"
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""Generator method producing games"""
data = self.source.data_location["library.yml"].read_text("utf-8")
@@ -84,7 +80,7 @@ class BottlesSource(URLExecutableSource):
"""Generic Bottles source"""
name = _("Bottles")
iterator_class = BottlesSourceIterator
iterable_class = BottlesSourceIterable
url_format = 'bottles:run/"{bottle_name}"/"{game_name}"'
available_on = {"linux"}

View File

@@ -25,13 +25,13 @@ from gi.repository import GLib, Gtk
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import Source, SourceIterationResult, SourceIterator
from src.importer.sources.source import Source, SourceIterable
class FlatpakSourceIterator(SourceIterator):
class FlatpakSourceIterable(SourceIterable):
source: "FlatpakSource"
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""Generator method producing games"""
added_time = int(time())
@@ -115,7 +115,7 @@ class FlatpakSource(Source):
"""Generic Flatpak source"""
name = _("Flatpak")
iterator_class = FlatpakSourceIterator
iterable_class = FlatpakSourceIterable
executable_format = "flatpak run {flatpak_id}"
available_on = {"linux"}

View File

@@ -20,21 +20,47 @@
import json
import logging
from abc import abstractmethod
from hashlib import sha256
from json import JSONDecodeError
from pathlib import Path
from time import time
from typing import Optional, TypedDict
from typing import Iterable, Optional, TypedDict
from functools import cached_property
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import (
URLExecutableSource,
SourceIterable,
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
def path_json_load(path: Path):
"""
Load JSON from the file at the given path
:raises OSError: if the file can't be opened
:raises JSONDecodeError: if the file isn't valid JSON
"""
with path.open("r", encoding="utf-8") as open_file:
return json.load(open_file)
class InvalidLibraryFileError(Exception):
pass
class InvalidInstalledFileError(Exception):
pass
class InvalidStoreFileError(Exception):
pass
class HeroicLibraryEntry(TypedDict):
app_name: str
installed: Optional[bool]
@@ -44,115 +70,304 @@ class HeroicLibraryEntry(TypedDict):
art_square: str
class HeroicSubSource(TypedDict):
service: str
path: tuple[str]
class SubSourceIterable(Iterable):
"""Class representing a Heroic sub-source"""
class HeroicSourceIterator(SourceIterator):
source: "HeroicSource"
source_iterable: "HeroicSourceIterable"
name: str
service: str
image_uri_params: str = ""
relative_library_path: Path
library_json_entries_key: str = "library"
sub_sources: dict[str, HeroicSubSource] = {
"sideload": {
"service": "sideload",
"path": ("sideload_apps", "library.json"),
},
"legendary": {
"service": "epic",
"path": ("store_cache", "legendary_library.json"),
},
"gog": {
"service": "gog",
"path": ("store_cache", "gog_library.json"),
},
}
def __init__(self, source, source_iterable) -> None:
self.source = source
self.source_iterable = source_iterable
def game_from_library_entry(
@cached_property
def library_path(self) -> Path:
path = self.source.config_location.root / self.relative_library_path
logging.debug("Using Heroic %s library.json path %s", self.name, path)
return path
def process_library_entry(
self, entry: HeroicLibraryEntry, added_time: int
) -> SourceIterationResult:
"""Helper method used to build a Game from a Heroic library entry"""
"""Build a Game from a Heroic library entry"""
# Skip games that are not installed
if not entry["is_installed"]:
return None
app_name = entry["app_name"]
# Build game
app_name = entry["app_name"]
runner = entry["runner"]
service = self.sub_sources[runner]["service"]
values = {
"source": f"{self.source.id}_{service}",
"source": f"{self.source.id}_{self.service}",
"added": added_time,
"name": entry["title"],
"developer": entry.get("developer", None),
"game_id": self.source.game_id_format.format(
service=service, game_id=app_name
service=self.service, game_id=app_name
),
"executable": self.source.executable_format.format(app_name=app_name),
"hidden": self.source_iterable.is_hidden(app_name),
}
game = Game(values)
# Get the image path from the heroic cache
# Filenames are derived from the URL that heroic used to get the file
uri: str = entry["art_square"]
if service == "epic":
uri += "?h=400&resize=1&w=300"
uri: str = entry["art_square"] + self.image_uri_params
digest = sha256(uri.encode()).hexdigest()
image_path = self.source.config_location.root / "images-cache" / digest
additional_data = {"local_image_path": image_path}
return (game, additional_data)
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""
Iterate through the games with a generator
:raises InvalidLibraryFileError: on initial call if the library file is bad
"""
added_time = int(time())
try:
iterator = iter(
path_json_load(self.library_path)[self.library_json_entries_key]
)
except (OSError, JSONDecodeError, TypeError, KeyError) as error:
raise InvalidLibraryFileError(
f"Invalid {self.library_path.name}"
) from error
for entry in iterator:
try:
yield self.process_library_entry(entry, added_time)
except KeyError as error:
logging.warning(
"Skipped invalid %s game %s",
self.name,
entry.get("app_name", "UNKNOWN"),
exc_info=error,
)
continue
class StoreSubSourceIterable(SubSourceIterable):
"""
Class representing a "store" sub source.
Games can be installed or not, this class does the check accordingly.
"""
relative_installed_path: Path
installed_app_names: set[str]
@cached_property
def installed_path(self) -> Path:
path = self.source.config_location.root / self.relative_installed_path
logging.debug("Using Heroic %s installed.json path %s", self.name, path)
return path
@abstractmethod
def get_installed_app_names(self) -> set[str]:
"""
Get the sub source's installed app names as a set.
:raises InvalidInstalledFileError: if the installed file data cannot be read
Whenever possible, `__cause__` is set with the original exception
"""
def is_installed(self, app_name: str) -> bool:
return app_name in self.installed_app_names
def process_library_entry(self, entry, added_time):
# Skip games that are not installed
app_name = entry["app_name"]
if not self.is_installed(app_name):
logging.warning(
"Skipped %s game %s (%s): not installed",
self.service,
entry["title"],
app_name,
)
return None
# Process entry as normal
return super().process_library_entry(entry, added_time)
def __iter__(self):
"""
Iterate through the installed games with a generator
:raises InvalidLibraryFileError: on initial call if the library file is bad
:raises InvalidInstalledFileError: on initial call if the installed file is bad
"""
self.installed_app_names = self.get_installed_app_names()
yield from super().__iter__()
class SideloadIterable(SubSourceIterable):
name = "sideload"
service = "sideload"
relative_library_path = Path("sideload_apps") / "library.json"
library_json_entries_key = "games"
class LegendaryIterable(StoreSubSourceIterable):
name = "legendary"
service = "epic"
image_uri_params = "?h=400&resize=1&w=300"
relative_library_path = Path("store_cache") / "legendary_library.json"
# relative_installed_path = (
# Path("legendary") / "legendaryConfig" / "legendary" / "installed.json"
# )
@cached_property
def installed_path(self) -> Path:
"""
Get the right path depending on the Heroic version
TODO after heroic 2.9 has been out for a while
We should use the commented out relative_installed_path
and remove this property override.
"""
heroic_config_path = self.source.config_location.root
# Heroic >= 2.9
if (path := heroic_config_path / "legendaryConfig").is_dir():
logging.debug("Using Heroic >= 2.9 legendary file")
# Heroic <= 2.8
elif heroic_config_path.is_relative_to(shared.flatpak_dir):
# Heroic flatpak
path = shared.flatpak_dir / "com.heroicgameslauncher.hgl" / "config"
logging.debug("Using Heroic flatpak <= 2.8 legendary file")
else:
# Heroic native
logging.debug("Using Heroic native <= 2.8 legendary file")
path = Path.home() / ".config"
path = path / "legendary" / "installed.json"
logging.debug("Using Heroic %s installed.json path %s", self.name, path)
return path
def get_installed_app_names(self):
try:
return set(path_json_load(self.installed_path).keys())
except (OSError, JSONDecodeError, AttributeError) as error:
raise InvalidInstalledFileError(
f"Invalid {self.installed_path.name}"
) from error
class GogIterable(StoreSubSourceIterable):
name = "gog"
service = "gog"
library_json_entries_key = "games"
relative_library_path = Path("store_cache") / "gog_library.json"
relative_installed_path = Path("gog_store") / "installed.json"
def get_installed_app_names(self):
try:
return {
app_name
for entry in path_json_load(self.installed_path)["installed"]
if (app_name := entry.get("appName")) is not None
}
except (OSError, JSONDecodeError, KeyError, AttributeError) as error:
raise InvalidInstalledFileError(
f"Invalid {self.installed_path.name}"
) from error
class NileIterable(StoreSubSourceIterable):
name = "nile"
service = "amazon"
relative_library_path = Path("store_cache") / "nile_library.json"
relative_installed_path = Path("nile_config") / "nile" / "installed.json"
def get_installed_app_names(self):
try:
installed_json = path_json_load(self.installed_path)
return {
app_name
for entry in installed_json
if (app_name := entry.get("id")) is not None
}
except (OSError, JSONDecodeError, AttributeError) as error:
raise InvalidInstalledFileError(
f"Invalid {self.installed_path.name}"
) from error
class HeroicSourceIterable(SourceIterable):
source: "HeroicSource"
hidden_app_names: set[str] = set()
def is_hidden(self, app_name: str) -> bool:
return app_name in self.hidden_app_names
def get_hidden_app_names(self) -> set[str]:
"""Get the hidden app names from store/config.json
:raises InvalidStoreFileError: if the store is invalid for some reason
"""
try:
store = path_json_load(self.source.config_location["store_config.json"])
self.hidden_app_names = {
app_name
for game in store["games"]["hidden"]
if (app_name := game.get("appName")) is not None
}
except KeyError:
logging.warning('No ["games"]["hidden"] key in Heroic store file')
except (OSError, JSONDecodeError, TypeError) as error:
logging.error("Invalid Heroic store file", exc_info=error)
raise InvalidStoreFileError() from error
def __iter__(self):
"""Generator method producing games from all the Heroic sub-sources"""
for sub_source_name, sub_source in self.sub_sources.items():
# Skip disabled sub-sources
if not shared.schema.get_boolean("heroic-import-" + sub_source["service"]):
self.get_hidden_app_names()
# Get games from the sub sources
for sub_source_class in (
SideloadIterable,
LegendaryIterable,
GogIterable,
NileIterable,
):
sub_source = sub_source_class(self.source, self)
if not shared.schema.get_boolean("heroic-import-" + sub_source.service):
logging.debug("Skipping Heroic %s: disabled", sub_source.service)
continue
# Load games from JSON
file = self.source.config_location.root.joinpath(*sub_source["path"])
try:
contents = json.load(file.open())
key = "library" if sub_source_name == "legendary" else "games"
library = contents[key]
except (JSONDecodeError, OSError, KeyError):
# Invalid library.json file, skip it
logging.warning("Couldn't open Heroic file: %s", str(file))
sub_source_iterable = iter(sub_source)
yield from sub_source_iterable
except (InvalidLibraryFileError, InvalidInstalledFileError) as error:
logging.error(
"Skipping bad Heroic sub-source %s",
sub_source.service,
exc_info=error,
)
continue
added_time = int(time())
for entry in library:
try:
result = self.game_from_library_entry(entry, added_time)
except KeyError as error:
# Skip invalid games
logging.warning(
"Invalid Heroic game skipped in %s", str(file), exc_info=error
)
continue
yield result
class HeroicSource(URLExecutableSource):
"""Generic Heroic Games Launcher source"""
name = _("Heroic")
iterator_class = HeroicSourceIterator
iterable_class = HeroicSourceIterable
url_format = "heroic://launch/{app_name}"
available_on = {"linux", "win32"}
config_location = Location(
schema_key="heroic-location",
candidates=(
shared.flatpak_dir / "com.heroicgameslauncher.hgl" / "config" / "heroic",
shared.config_dir / "heroic",
shared.home / ".config" / "heroic",
shared.flatpak_dir / "com.heroicgameslauncher.hgl" / "config" / "heroic",
shared.appdata_dir / "heroic",
),
paths={
"config.json": (False, "config.json"),
"store_config.json": (False, Path("store") / "config.json"),
},
)

View File

@@ -25,18 +25,14 @@ from time import time
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import (
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.importer.sources.source import SourceIterable, URLExecutableSource
from src.utils.sqlite import copy_db
class ItchSourceIterator(SourceIterator):
class ItchSourceIterable(SourceIterable):
source: "ItchSource"
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""Generator method producing games"""
# Query the database
@@ -80,7 +76,7 @@ class ItchSourceIterator(SourceIterator):
class ItchSource(URLExecutableSource):
name = _("itch")
iterator_class = ItchSourceIterator
iterable_class = ItchSourceIterable
url_format = "itch://caves/{cave_id}/launch"
available_on = {"linux", "win32"}

View File

@@ -21,15 +21,14 @@ import json
import logging
from json import JSONDecodeError
from time import time
from typing import Generator
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import Source, SourceIterationResult, SourceIterator
from src.importer.sources.source import Source, SourceIterationResult, SourceIterable
class LegendarySourceIterator(SourceIterator):
class LegendarySourceIterable(SourceIterable):
source: "LegendarySource"
def game_from_library_entry(
@@ -65,7 +64,7 @@ class LegendarySourceIterator(SourceIterator):
game = Game(values)
return (game, data)
def generator_builder(self) -> Generator[SourceIterationResult, None, None]:
def __iter__(self):
# Open library
file = self.source.config_location["installed.json"]
try:
@@ -94,7 +93,7 @@ class LegendarySource(Source):
executable_format = "legendary launch {app_name}"
available_on = {"linux"}
iterator_class = LegendarySourceIterator
iterable_class = LegendarySourceIterable
config_location: Location = Location(
schema_key="legendary-location",
candidates=(

View File

@@ -24,18 +24,14 @@ from time import time
from src import shared
from src.game import Game
from src.importer.sources.location import Location
from src.importer.sources.source import (
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.importer.sources.source import SourceIterable, URLExecutableSource
from src.utils.sqlite import copy_db
class LutrisSourceIterator(SourceIterator):
class LutrisSourceIterable(SourceIterable):
source: "LutrisSource"
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""Generator method producing games"""
# Query the database
@@ -91,7 +87,7 @@ class LutrisSource(URLExecutableSource):
"""Generic Lutris source"""
name = _("Lutris")
iterator_class = LutrisSourceIterator
iterable_class = LutrisSourceIterable
url_format = "lutris:rungameid/{game_id}"
available_on = {"linux"}

View File

@@ -29,25 +29,16 @@ from src.importer.sources.location import Location
SourceIterationResult = None | Game | tuple[Game, tuple[Any]]
class SourceIterator(Iterator):
class SourceIterable(Iterable):
"""Data producer for a source of games"""
source: "Source" = None
generator: Generator = None
def __init__(self, source: "Source") -> None:
super().__init__()
self.source = source
self.generator = self.generator_builder()
def __iter__(self) -> "SourceIterator":
return self
def __next__(self) -> SourceIterationResult:
return next(self.generator)
@abstractmethod
def generator_builder(self) -> Generator[SourceIterationResult, None, None]:
def __iter__(self) -> Generator[SourceIterationResult, None, None]:
"""
Method that returns a generator that produces games
* Should be implemented as a generator method
@@ -66,7 +57,7 @@ class Source(Iterable):
data_location: Optional[Location] = None
cache_location: Optional[Location] = None
config_location: Optional[Location] = None
iterator_class: type[SourceIterator]
iterable_class: type[SourceIterable]
@property
def full_name(self) -> str:
@@ -98,7 +89,7 @@ class Source(Iterable):
def executable_format(self) -> str:
"""The executable format used to construct game executables"""
def __iter__(self) -> SourceIterator:
def __iter__(self) -> Generator[SourceIterationResult, None, None]:
"""
Get an iterator for the source
:raises UnresolvableLocationError: Not iterable if any of the locations are unresolvable
@@ -108,7 +99,7 @@ class Source(Iterable):
if location is None:
continue
location.resolve()
return self.iterator_class(self)
return iter(self.iterable_class(self))
# pylint: disable=abstract-method

View File

@@ -18,6 +18,7 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
import re
from pathlib import Path
from time import time
@@ -25,16 +26,12 @@ from typing import Iterable
from src import shared
from src.game import Game
from src.importer.sources.source import (
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.utils.steam import SteamFileHelper, SteamInvalidManifestError
from src.importer.sources.location import Location
from src.importer.sources.source import SourceIterable, URLExecutableSource
from src.utils.steam import SteamFileHelper, SteamInvalidManifestError
class SteamSourceIterator(SourceIterator):
class SteamSourceIterable(SourceIterable):
source: "SteamSource"
def get_manifest_dirs(self) -> Iterable[Path]:
@@ -62,7 +59,7 @@ class SteamSourceIterator(SourceIterator):
)
return manifests
def generator_builder(self) -> SourceIterationResult:
def __iter__(self):
"""Generator method producing games"""
appid_cache = set()
manifests = self.get_manifests()
@@ -74,17 +71,20 @@ class SteamSourceIterator(SourceIterator):
steam = SteamFileHelper()
try:
local_data = steam.get_manifest_data(manifest)
except (OSError, SteamInvalidManifestError):
except (OSError, SteamInvalidManifestError) as error:
logging.debug("Couldn't load appmanifest %s", manifest, exc_info=error)
continue
# Skip non installed games
installed_mask = 4
if not int(local_data["stateflags"]) & installed_mask:
logging.debug("Skipped %s: not installed", manifest)
continue
# Skip duplicate appids
appid = local_data["appid"]
if appid in appid_cache:
logging.debug("Skipped %s: appid already seen during import", manifest)
continue
appid_cache.add(appid)
@@ -112,7 +112,7 @@ class SteamSourceIterator(SourceIterator):
class SteamSource(URLExecutableSource):
name = _("Steam")
available_on = {"linux", "win32"}
iterator_class = SteamSourceIterator
iterable_class = SteamSourceIterable
url_format = "steam://rungameid/{game_id}"
data_location = Location(

View File

@@ -73,7 +73,7 @@ def setup_logging():
"PIL": {
"handlers": ["lib_console_handler", "file_handler"],
"propagate": False,
"level": "NOTSET",
"level": "WARNING",
},
"urllib3": {
"handlers": ["lib_console_handler", "file_handler"],

View File

@@ -68,6 +68,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
heroic_config_file_chooser_button = Gtk.Template.Child()
heroic_import_epic_switch = Gtk.Template.Child()
heroic_import_gog_switch = Gtk.Template.Child()
heroic_import_amazon_switch = Gtk.Template.Child()
heroic_import_sideload_switch = Gtk.Template.Child()
bottles_expander_row = Gtk.Template.Child()
@@ -181,6 +182,7 @@ class PreferencesWindow(Adw.PreferencesWindow):
"lutris-import-flatpak",
"heroic-import-epic",
"heroic-import-gog",
"heroic-import-amazon",
"heroic-import-sideload",
"flatpak-import-launchers",
"sgdb",

View File

@@ -18,6 +18,8 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
from gi.repository import GdkPixbuf
from src import shared
@@ -35,6 +37,7 @@ class LocalCoverManager(Manager):
def manager_logic(self, game: Game, additional_data: dict) -> None:
if image_path := additional_data.get("local_image_path"):
if not image_path.is_file():
logging.error("Local image path is not a file: %s", image_path)
return
save_cover(game.game_id, resize_cover(image_path))
elif icon_path := additional_data.get("local_icon_path"):