🚧 Base Legendary source

src/importer/sources/legendary_source.py

@@ -0,0 +1,93 @@
+import logging
+from pathlib import Path
+from typing import Generator
+import json
+from json import JSONDecodeError
+from time import time
+
+from src import shared
+from src.game import Game
+from src.importer.sources.source import (
+    LinuxSource,
+    Source,
+    SourceIterationResult,
+    SourceIterator,
+)
+from src.utils.decorators import replaced_by_env_path, replaced_by_path
+
+
+class LegendarySourceIterator(SourceIterator):
+    source: "LegendarySource"
+
+    def game_from_library_entry(self, entry: dict) -> SourceIterationResult:
+        # Skip non-games
+        if entry["is_dlc"]:
+            return None
+
+        # Build game
+        app_name = entry["app_name"]
+        values = {
+            "version": shared.SPEC_VERSION,
+            "added": int(time()),
+            "source": self.source.id,
+            "name": entry["title"],
+            "game_id": self.source.game_id_format.format(game_id=app_name),
+            "executable": self.source.game_id_format.format(app_name=app_name),
+        }
+        data = {}
+
+        # Get additional metadata from file (optional)
+        metadata_file = self.source.location / "metadata" / f"{app_name}.json"
+        try:
+            metadata = json.load(metadata_file.open())
+            values["developer"] = metadata["metadata"]["developer"]
+            for image_entry in metadata["metadata"]["keyImages"]:
+                if image_entry["type"] == "DieselGameBoxTall":
+                    data["online_cover_url"] = image_entry["url"]
+                    break
+        except (JSONDecodeError, OSError, KeyError):
+            pass
+
+        game = Game(values, allow_side_effects=False)
+        return (game, data)
+
+    def generator_builder(self) -> Generator[SourceIterationResult, None, None]:
+        # Open library
+        file = self.source.location / "installed.json"
+        try:
+            library = json.load(file.open())
+        except (JSONDecodeError, OSError):
+            logging.warning("Couldn't open Legendary file: %s", str(file))
+            return
+        # Generate games from library
+        for entry in library:
+            try:
+                result = self.game_from_library_entry(entry)
+            except KeyError:
+                # Skip invalid games
+                logging.warning("Invalid Legendary game skipped in %s", str(file))
+                continue
+            yield result
+
+
+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}"
+
+    @property
+    @LegendarySource.replaced_by_schema_key()
+    @replaced_by_env_path("XDG_CONFIG_HOME", "legendary/")
+    @replaced_by_path("~/.config/legendary/")
+    def location(self) -> Path:
+        raise FileNotFoundError()

src/store/managers/online_cover_manager.py

@@ -0,0 +1,31 @@
+from pathlib import Path
+
+import requests
+from gi.repository import Gio
+from requests import HTTPError
+from urllib3.exceptions import SSLError
+
+from src.game import Game
+from src.store.managers.manager import Manager
+from src.store.managers.local_cover_manager import LocalCoverManager
+from src.utils.save_cover import resize_cover, save_cover
+
+
+class OnlineCoverManager(Manager):
+    """Manager that downloads game covers from URLs"""
+
+    run_after = set((LocalCoverManager,))
+    retryable_on = set((HTTPError, SSLError))
+
+    def manager_logic(self, game: Game, additional_data: dict) -> None:
+        # Ensure that we have a cover to download
+        cover_url = additional_data.get("online_cover_url", None)
+        if not cover_url:
+            return
+        # Download cover
+        tmp_file = Gio.File.new_tmp()[0]
+        with requests.get(cover_url, timeout=5) as cover:
+            cover.raise_for_status()
+            Path(tmp_file.get_path()).write_bytes(cover.content)
+        # Resize and save
+        save_cover(game.game_id, resize_cover(tmp_file.get_path()))