skullbite/cartridges
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/main'

Browse Source
This commit is contained in:
Rilic
2023-08-01 10:52:33 +01:00
parent 311ed3b09c 1aff1347e3
commit 417a02e8b6
56 changed files with 2364 additions and 1328 deletions
Download Patch File Download Diff File Expand all files Collapse all files

53
src/importer/sources/bottles_source.py
View File

@@ -20,33 +20,30 @@
from pathlib import Path
from time import time
from typing import NamedTuple
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.location import Location, LocationSubPath
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")
data = self.source.locations.data["library.yml"].read_text("utf-8")
library: dict = yaml.safe_load(data)
added_time = int(time())
for entry in library.values():
# Build game
values = {
"source": self.source.id,
"source": self.source.source_id,
"added": added_time,
"name": entry["name"],
"game_id": self.source.game_id_format.format(game_id=entry["id"]),
@@ -62,11 +59,11 @@ class BottlesSourceIterator(SourceIterator):
# as Cartridges can't access directories picked via Bottles' file picker portal
bottles_location = Path(
yaml.safe_load(
self.source.data_location["data.yml"].read_text("utf-8")
self.source.locations.data["data.yml"].read_text("utf-8")
)["custom_bottles_path"]
)
except (FileNotFoundError, KeyError):
bottles_location = self.source.data_location.root / "bottles"
bottles_location = self.source.locations.data.root / "bottles"
bottle_path = entry["bottle"]["path"]
@@ -80,23 +77,31 @@ class BottlesSourceIterator(SourceIterator):
yield (game, additional_data)
class BottlesLocations(NamedTuple):
data: Location
class BottlesSource(URLExecutableSource):
"""Generic Bottles source"""
source_id = "bottles"
name = _("Bottles")
iterator_class = BottlesSourceIterator
iterable_class = BottlesSourceIterable
url_format = 'bottles:run/"{bottle_name}"/"{game_name}"'
available_on = {"linux"}
data_location = Location(
schema_key="bottles-location",
candidates=(
shared.flatpak_dir / "com.usebottles.bottles" / "data" / "bottles",
shared.data_dir / "bottles/",
shared.home / ".local" / "share" / "bottles",
),
paths={
"library.yml": (False, "library.yml"),
"data.yml": (False, "data.yml"),
},
locations = BottlesLocations(
Location(
schema_key="bottles-location",
candidates=(
shared.flatpak_dir / "com.usebottles.bottles" / "data" / "bottles",
shared.data_dir / "bottles/",
shared.home / ".local" / "share" / "bottles",
),
paths={
"library.yml": LocationSubPath("library.yml"),
"data.yml": LocationSubPath("data.yml"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)

45
src/importer/sources/flatpak_source.py
View File

@@ -19,25 +19,26 @@
from pathlib import Path
from time import time
from typing import NamedTuple
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.location import Location, LocationSubPath
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())
icon_theme = Gtk.IconTheme.new()
icon_theme.add_search_path(str(self.source.data_location["icons"]))
icon_theme.add_search_path(str(self.source.locations.data["icons"]))
blacklist = (
{"hu.kramo.Cartridges", "hu.kramo.Cartridges.Devel"}
@@ -53,7 +54,7 @@ class FlatpakSourceIterator(SourceIterator):
}
)
for entry in (self.source.data_location["applications"]).iterdir():
for entry in (self.source.locations.data["applications"]).iterdir():
if entry.suffix != ".desktop":
continue
@@ -76,7 +77,7 @@ class FlatpakSourceIterator(SourceIterator):
continue
values = {
"source": self.source.id,
"source": self.source.source_id,
"added": added_time,
"name": name,
"game_id": self.source.game_id_format.format(game_id=flatpak_id),
@@ -111,22 +112,30 @@ class FlatpakSourceIterator(SourceIterator):
yield (game, additional_data)
class FlatpakLocations(NamedTuple):
data: Location
class FlatpakSource(Source):
"""Generic Flatpak source"""
source_id = "flatpak"
name = _("Flatpak")
iterator_class = FlatpakSourceIterator
iterable_class = FlatpakSourceIterable
executable_format = "flatpak run {flatpak_id}"
available_on = {"linux"}
data_location = Location(
schema_key="flatpak-location",
candidates=(
"/var/lib/flatpak/",
shared.data_dir / "flatpak",
),
paths={
"applications": (True, "exports/share/applications"),
"icons": (True, "exports/share/icons"),
},
locations = FlatpakLocations(
Location(
schema_key="flatpak-location",
candidates=(
"/var/lib/flatpak/",
shared.data_dir / "flatpak",
),
paths={
"applications": LocationSubPath("exports/share/applications", True),
"icons": LocationSubPath("exports/share/icons", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)

377
src/importer/sources/heroic_source.py
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, NamedTuple, 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.location import Location, LocationSubPath
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,119 +70,320 @@ 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.locations.config.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
# Build game
app_name = entry["app_name"]
runner = entry["runner"]
service = self.sub_sources[runner]["service"]
# Build game
values = {
"source": f"{self.source.id}_{service}",
"source": f"{self.source.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),
"executable": self.source.executable_format.format(runner=runner, 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
image_path = self.source.locations.config.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.locations.config.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.locations.config.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.locations.config["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 HeroicLocations(NamedTuple):
config: Location
class HeroicSource(URLExecutableSource):
"""Generic Heroic Games Launcher source"""
source_id = "heroic"
name = _("Heroic")
iterator_class = HeroicSourceIterator
url_format = "heroic://launch/{app_name}"
iterable_class = HeroicSourceIterable
url_format = "heroic://launch/{runner}/{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.appdata_dir / "heroic",
),
paths={
"config.json": (False, "config.json"),
},
locations = HeroicLocations(
Location(
schema_key="heroic-location",
candidates=(
shared.config_dir / "heroic",
shared.home / ".config" / "heroic",
shared.flatpak_dir
/ "com.heroicgameslauncher.hgl"
/ "config"
/ "heroic",
shared.appdata_dir / "heroic",
),
paths={
"config.json": LocationSubPath("config.json"),
"store_config.json": LocationSubPath("store/config.json"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)
@property
def game_id_format(self) -> str:
"""The string format used to construct game IDs"""
return self.id + "_{service}_{game_id}"
return self.source_id + "_{service}_{game_id}"

47
src/importer/sources/itch_source.py
View File

@@ -21,22 +21,19 @@
from shutil import rmtree
from sqlite3 import connect
from time import time
from typing import NamedTuple
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.location import Location, LocationSubPath
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
@@ -55,7 +52,7 @@ class ItchSourceIterator(SourceIterator):
caves.game_id = games.id
;
"""
db_path = copy_db(self.source.config_location["butler.db"])
db_path = copy_db(self.source.locations.config["butler.db"])
connection = connect(db_path)
cursor = connection.execute(db_request)
@@ -65,7 +62,7 @@ class ItchSourceIterator(SourceIterator):
for row in cursor:
values = {
"added": added_time,
"source": self.source.id,
"source": self.source.source_id,
"name": row[1],
"game_id": self.source.game_id_format.format(game_id=row[0]),
"executable": self.source.executable_format.format(cave_id=row[4]),
@@ -78,19 +75,29 @@ class ItchSourceIterator(SourceIterator):
rmtree(str(db_path.parent))
class ItchLocations(NamedTuple):
config: Location
class ItchSource(URLExecutableSource):
source_id = "itch"
name = _("itch")
iterator_class = ItchSourceIterator
iterable_class = ItchSourceIterable
url_format = "itch://caves/{cave_id}/launch"
available_on = {"linux", "win32"}
config_location = Location(
schema_key="itch-location",
candidates=(
shared.flatpak_dir / "io.itch.itch" / "config" / "itch",
shared.config_dir / "itch",
shared.home / ".config" / "itch",
shared.appdata_dir / "itch",
),
paths={"butler.db": (False, "db/butler.db")},
locations = ItchLocations(
Location(
schema_key="itch-location",
candidates=(
shared.flatpak_dir / "io.itch.itch" / "config" / "itch",
shared.config_dir / "itch",
shared.home / ".config" / "itch",
shared.appdata_dir / "itch",
),
paths={
"butler.db": LocationSubPath("db/butler.db"),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)

46
src/importer/sources/legendary_source.py
View File

@@ -21,15 +21,15 @@ import json
import logging
from json import JSONDecodeError
from time import time
from typing import Generator
from typing import NamedTuple
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.location import Location, LocationSubPath
from src.importer.sources.source import Source, SourceIterationResult, SourceIterable
class LegendarySourceIterator(SourceIterator):
class LegendarySourceIterable(SourceIterable):
source: "LegendarySource"
def game_from_library_entry(
@@ -43,7 +43,7 @@ class LegendarySourceIterator(SourceIterator):
app_name = entry["app_name"]
values = {
"added": added_time,
"source": self.source.id,
"source": self.source.source_id,
"name": entry["title"],
"game_id": self.source.game_id_format.format(game_id=app_name),
"executable": self.source.executable_format.format(app_name=app_name),
@@ -51,7 +51,7 @@ class LegendarySourceIterator(SourceIterator):
data = {}
# Get additional metadata from file (optional)
metadata_file = self.source.config_location["metadata"] / f"{app_name}.json"
metadata_file = self.source.locations.config["metadata"] / f"{app_name}.json"
try:
metadata = json.load(metadata_file.open())
values["developer"] = metadata["metadata"]["developer"]
@@ -65,9 +65,9 @@ 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"]
file = self.source.locations.config["installed.json"]
try:
library: dict = json.load(file.open())
except (JSONDecodeError, OSError):
@@ -89,20 +89,28 @@ class LegendarySourceIterator(SourceIterator):
yield result
class LegendaryLocations(NamedTuple):
config: Location
class LegendarySource(Source):
source_id = "legendary"
name = _("Legendary")
executable_format = "legendary launch {app_name}"
available_on = {"linux"}
iterable_class = LegendarySourceIterable
iterator_class = LegendarySourceIterator
config_location: Location = Location(
schema_key="legendary-location",
candidates=(
shared.config_dir / "legendary",
shared.home / ".config" / "legendary",
),
paths={
"installed.json": (False, "installed.json"),
"metadata": (True, "metadata"),
},
locations = LegendaryLocations(
Location(
schema_key="legendary-location",
candidates=(
shared.config_dir / "legendary",
shared.home / ".config" / "legendary",
),
paths={
"installed.json": LocationSubPath("installed.json"),
"metadata": LocationSubPath("metadata", True),
},
invalid_subtitle=Location.CONFIG_INVALID_SUBTITLE,
)
)

36
src/importer/sources/location.py
View File

@@ -1,13 +1,18 @@
import logging
from pathlib import Path
from typing import Callable, Mapping, Iterable
from typing import Mapping, Iterable, NamedTuple
from os import PathLike
from src import shared
PathSegment = str | PathLike | Path
PathSegments = Iterable[PathSegment]
Candidate = PathSegments | Callable[[], PathSegments]
Candidate = PathSegments
class LocationSubPath(NamedTuple):
segment: PathSegment
is_directory: bool = False
class UnresolvableLocationError(Exception):
@@ -24,31 +29,42 @@ class Location:
* When resolved, the schema is updated with the picked chosen
"""
# The variable is the name of the source
CACHE_INVALID_SUBTITLE = _("Select the {} cache directory.")
# The variable is the name of the source
CONFIG_INVALID_SUBTITLE = _("Select the {} configuration directory.")
# The variable is the name of the source
DATA_INVALID_SUBTITLE = _("Select the {} data directory.")
schema_key: str
candidates: Iterable[Candidate]
paths: Mapping[str, tuple[bool, PathSegments]]
paths: Mapping[str, LocationSubPath]
invalid_subtitle: str
root: Path = None
def __init__(
self,
schema_key: str,
candidates: Iterable[Candidate],
paths: Mapping[str, tuple[bool, PathSegments]],
paths: Mapping[str, LocationSubPath],
invalid_subtitle: str,
) -> None:
super().__init__()
self.schema_key = schema_key
self.candidates = candidates
self.paths = paths
self.invalid_subtitle = invalid_subtitle
def check_candidate(self, candidate: Path) -> bool:
"""Check if a candidate root has the necessary files and directories"""
for type_is_dir, subpath in self.paths.values():
subpath = Path(candidate) / Path(subpath)
if type_is_dir:
if not subpath.is_dir():
for segment, is_directory in self.paths.values():
path = Path(candidate) / segment
if is_directory:
if not path.is_dir():
return False
else:
if not subpath.is_file():
if not path.is_file():
return False
return True
@@ -81,4 +97,4 @@ class Location:
def __getitem__(self, key: str):
"""Get the computed path from its key for the location"""
self.resolve()
return self.root / self.paths[key][1]
return self.root / self.paths[key].segment

74
src/importer/sources/lutris_source.py
View File

@@ -20,22 +20,19 @@
from shutil import rmtree
from sqlite3 import connect
from time import time
from typing import NamedTuple
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.location import Location, LocationSubPath
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
@@ -55,7 +52,7 @@ class LutrisSourceIterator(SourceIterator):
"import_steam": shared.schema.get_boolean("lutris-import-steam"),
"import_flatpak": shared.schema.get_boolean("lutris-import-flatpak"),
}
db_path = copy_db(self.source.data_location["pga.db"])
db_path = copy_db(self.source.locations.config["pga.db"])
connection = connect(db_path)
cursor = connection.execute(request, params)
@@ -68,7 +65,7 @@ class LutrisSourceIterator(SourceIterator):
"added": added_time,
"hidden": row[4],
"name": row[1],
"source": f"{self.source.id}_{row[3]}",
"source": f"{self.source.source_id}_{row[3]}",
"game_id": self.source.game_id_format.format(
runner=row[3], game_id=row[0]
),
@@ -77,7 +74,7 @@ class LutrisSourceIterator(SourceIterator):
game = Game(values)
# Get official image path
image_path = self.source.cache_location["coverart"] / f"{row[2]}.jpg"
image_path = self.source.locations.cache["coverart"] / f"{row[2]}.jpg"
additional_data = {"local_image_path": image_path}
# Produce game
@@ -87,40 +84,49 @@ class LutrisSourceIterator(SourceIterator):
rmtree(str(db_path.parent))
class LutrisLocations(NamedTuple):
config: Location
cache: Location
class LutrisSource(URLExecutableSource):
"""Generic Lutris source"""
source_id = "lutris"
name = _("Lutris")
iterator_class = LutrisSourceIterator
iterable_class = LutrisSourceIterable
url_format = "lutris:rungameid/{game_id}"
available_on = {"linux"}
# FIXME possible bug: location picks ~/.var... and cache_lcoation picks ~/.local...
# FIXME possible bug: config picks ~/.var... and cache picks ~/.local...
data_location = Location(
schema_key="lutris-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "data" / "lutris",
shared.data_dir / "lutris",
shared.home / ".local" / "share" / "lutris",
locations = LutrisLocations(
Location(
schema_key="lutris-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "data" / "lutris",
shared.data_dir / "lutris",
shared.home / ".local" / "share" / "lutris",
),
paths={
"pga.db": LocationSubPath("pga.db"),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
),
paths={
"pga.db": (False, "pga.db"),
},
)
cache_location = Location(
schema_key="lutris-cache-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "cache" / "lutris",
shared.cache_dir / "lutris",
shared.home / ".cache" / "lutris",
Location(
schema_key="lutris-cache-location",
candidates=(
shared.flatpak_dir / "net.lutris.Lutris" / "cache" / "lutris",
shared.cache_dir / "lutris",
shared.home / ".cache" / "lutris",
),
paths={
"coverart": LocationSubPath("coverart", True),
},
invalid_subtitle=Location.CACHE_INVALID_SUBTITLE,
),
paths={
"coverart": (True, "coverart"),
},
)
@property
def game_id_format(self):
return self.id + "_{runner}_{game_id}"
return self.source_id + "_{runner}_{game_id}"

38
src/importer/sources/source.py
View File

@@ -19,8 +19,8 @@
import sys
from abc import abstractmethod
from collections.abc import Iterable, Iterator
from typing import Any, Generator, Optional
from collections.abc import Iterable
from typing import Any, Generator, Collection
from src.game import Game
from src.importer.sources.location import Location
@@ -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
@@ -60,13 +51,12 @@ class SourceIterator(Iterator):
class Source(Iterable):
"""Source of games. E.g an installed app with a config file that lists game directories"""
source_id: str
name: str
variant: str = None
available_on: set[str] = set()
data_location: Optional[Location] = None
cache_location: Optional[Location] = None
config_location: Optional[Location] = None
iterator_class: type[SourceIterator]
iterable_class: type[SourceIterable]
locations: Collection[Location]
@property
def full_name(self) -> str:
@@ -76,18 +66,10 @@ class Source(Iterable):
full_name_ += f" ({self.variant})"
return full_name_
@property
def id(self) -> str: # pylint: disable=invalid-name
"""The source's identifier"""
id_ = self.name.lower()
if self.variant is not None:
id_ += f"_{self.variant.lower()}"
return id_
@property
def game_id_format(self) -> str:
"""The string format used to construct game IDs"""
return self.id + "_{game_id}"
return self.source_id + "_{game_id}"
@property
def is_available(self):
@@ -98,7 +80,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 +90,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

60
src/importer/sources/steam_source.py
View File

@@ -18,28 +18,25 @@
#
# SPDX-License-Identifier: GPL-3.0-or-later
import logging
import re
from pathlib import Path
from time import time
from typing import Iterable
from typing import Iterable, NamedTuple
from src import shared
from src.game import Game
from src.importer.sources.source import (
SourceIterationResult,
SourceIterator,
URLExecutableSource,
)
from src.importer.sources.location import Location, LocationSubPath
from src.importer.sources.source import SourceIterable, URLExecutableSource
from src.utils.steam import SteamFileHelper, SteamInvalidManifestError
from src.importer.sources.location import Location
class SteamSourceIterator(SourceIterator):
class SteamSourceIterable(SourceIterable):
source: "SteamSource"
def get_manifest_dirs(self) -> Iterable[Path]:
"""Get dirs that contain steam app manifests"""
libraryfolders_path = self.source.data_location["libraryfolders.vdf"]
libraryfolders_path = self.source.locations.data["libraryfolders.vdf"]
with open(libraryfolders_path, "r", encoding="utf-8") as file:
contents = file.read()
return [
@@ -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)
@@ -92,7 +92,7 @@ class SteamSourceIterator(SourceIterator):
values = {
"added": added_time,
"name": local_data["name"],
"source": self.source.id,
"source": self.source.source_id,
"game_id": self.source.game_id_format.format(game_id=appid),
"executable": self.source.executable_format.format(game_id=appid),
}
@@ -100,7 +100,7 @@ class SteamSourceIterator(SourceIterator):
# Add official cover image
image_path = (
self.source.data_location["librarycache"]
self.source.locations.data["librarycache"]
/ f"{appid}_library_600x900.jpg"
)
additional_data = {"local_image_path": image_path, "steam_appid": appid}
@@ -109,22 +109,30 @@ class SteamSourceIterator(SourceIterator):
yield (game, additional_data)
class SteamLocations(NamedTuple):
data: Location
class SteamSource(URLExecutableSource):
source_id = "steam"
name = _("Steam")
available_on = {"linux", "win32"}
iterator_class = SteamSourceIterator
iterable_class = SteamSourceIterable
url_format = "steam://rungameid/{game_id}"
data_location = Location(
schema_key="steam-location",
candidates=(
shared.home / ".steam" / "steam",
shared.data_dir / "Steam",
shared.flatpak_dir / "com.valvesoftware.Steam" / "data" / "Steam",
shared.programfiles32_dir / "Steam",
),
paths={
"libraryfolders.vdf": (False, "steamapps/libraryfolders.vdf"),
"librarycache": (True, "appcache/librarycache"),
},
locations = SteamLocations(
Location(
schema_key="steam-location",
candidates=(
shared.home / ".steam" / "steam",
shared.data_dir / "Steam",
shared.flatpak_dir / "com.valvesoftware.Steam" / "data" / "Steam",
shared.programfiles32_dir / "Steam",
),
paths={
"libraryfolders.vdf": LocationSubPath("steamapps/libraryfolders.vdf"),
"librarycache": LocationSubPath("appcache/librarycache", True),
},
invalid_subtitle=Location.DATA_INVALID_SUBTITLE,
)
)