🚧 SGDB

src/importer/importer.py

@@ -2,7 +2,7 @@ from threading import Thread, Lock
 from gi.repository import Adw, Gtk, Gio
 
 from .game import Game
-from .steamgriddb import SGDBSave
+from .steamgriddb import SGDBHelper
 
 
 class Importer:
@@ -97,47 +97,20 @@ class Importer:
 
     def __import_from_source(self, *args, **kwargs):
         """Source import thread entry point"""
-        # TODO just get Game objects from the sources
         source, *rest = args
 
         iterator = source.__iter__()
-        for game_values in iterator:
-            game = Game(self.win, game_values)
-
+        for game in iterator:
             self.games_lock.acquire()
             self.games.add(game)
             self.games_lock.release()
 
+            # TODO SGDB image
+            # Who's in charge of image adding ?
+
             self.progress_lock.acquire()
             self.counts[source.id]["total"] = len(iterator)
             if not game.blacklisted:
                 self.counts[source.id]["done"] += 1
             self.update_progressbar()
             self.progress_lock.release()
-
-    # TODO remove after not needed
-    def save_game(self, values=None, cover_path=None):
-        if values:
-            game = Game(self.win, values)
-
-            if save_cover:
-                save_cover(self.win, game.game_id, resize_cover(self.win, cover_path))
-
-            self.games.add(game)
-
-            self.games_no += 1
-            if game.blacklisted:
-                self.games_no -= 1
-
-        self.queue -= 1
-        self.update_progressbar()
-
-        if self.queue == 0 and not self.blocker:
-            if self.games:
-                self.total_queue = len(self.games)
-                self.queue = len(self.games)
-                self.import_statuspage.set_title(_("Importing Covers…"))
-                self.update_progressbar()
-                SGDBSave(self.win, self.games, self)
-            else:
-                self.done()

src/importer/sources/lutris_source.py

@@ -1,6 +1,7 @@
 from functools import cached_property
 from sqlite3 import connect
 
+from src.game import Game
 from src.utils.save_cover import resize_cover, save_cover
 from src.importer.source import Source, SourceIterator
 from src.importer.decorators import replaced_by_schema_key, replaced_by_path
@@ -55,6 +56,7 @@ class LutrisSourceIterator(SourceIterator):
                 continue
 
             # Build basic game
+            # TODO decouple game creation from the window object (later)
             values = {
                 "hidden": row[4],
                 "name": row[1],
@@ -65,6 +67,7 @@ class LutrisSourceIterator(SourceIterator):
                 "executable": self.source.executable_format.format(game_id=row[2]),
                 "developer": None,  # TODO get developer metadata on Lutris
             }
+            game = Game(self.source.win, values)
 
             # Save official image
             image_path = self.source.cache_location / "coverart" / f"{row[2]}.jpg"
@@ -73,6 +76,7 @@ class LutrisSourceIterator(SourceIterator):
                 save_cover(self.source.win, values["game_id"], resized)
 
             # TODO Save SGDB
+            SGDBSave(self.win, self.games, self)
 
             return values
 

src/utils/steamgriddb.py

@@ -8,6 +8,85 @@ from .create_dialog import create_dialog
 from .save_cover import save_cover, resize_cover
 
 
+class SGDBError(Exception):
+    pass
+
+
+class SGDBHelper:
+    base_url = "https://www.steamgriddb.com/api/v2/"
+
+    win = None
+    importer = None
+    exception = None
+
+    def __init__(self, win, importer=None) -> None:
+        self.win = win
+        self.importer = importer
+
+    @property
+    def auth_header(self):
+        key = self.win.schema.get_string("sgdb-key")
+        headers = {"Authorization": f"Bearer {key}"}
+        return headers
+
+    # TODO delegate that to the app
+    def create_exception_dialog(self, exception):
+        dialog = create_dialog(
+            self.win,
+            _("Couldn't Connect to SteamGridDB"),
+            exception,
+            "open_preferences",
+            _("Preferences"),
+        )
+        dialog.connect("response", self.response)
+
+    # TODO same as create_exception_dialog
+    def on_exception_dialog_response(self, _widget, response):
+        if response == "open_preferences":
+            self.win.get_application().on_preferences_action(page_name="sgdb")
+
+    def get_game_id(self, game):
+        """Get grid results for a game. Can raise an exception."""
+
+        # Request
+        res = requests.get(
+            f"{self.base_url}search/autocomplete/{game.name}",
+            headers=self.auth_headers,
+            timeout=5,
+        )
+        if res.status_code == 200:
+            return res.json()["data"][0]["id"]
+
+        # HTTP error
+        res.raise_for_status()
+
+        # SGDB API error
+        res_json = res.json()
+        if "error" in tuple(res_json):
+            raise SGDBError(res_json["errors"])
+        else:
+            raise SGDBError(res.status_code)
+
+    def get_image_uri(self, game, animated=False):
+        """Get the image for a game"""
+        uri = f"{self.base_url}grids/game/{self.get_game_id(game)}?dimensions=600x900"
+        if animated:
+            uri += "&types=animated"
+        grid = requests.get(uri, headers=self.auth_header, timeout=5)
+        image_uri = grid.json()["data"][0]["url"]
+        return image_uri
+
+
+# Current steps to save image for N games
+# Create a task for every game
+# Call update_cover
+# If using sgdb and (prefer or no image) and not blacklisted
+# Search for game
+# Get image from sgdb (animated if preferred and found, or still)
+# Exit task and enter task_done
+# If error, create popup
+
+
 class SGDBSave:
     def __init__(self, games, importer=None):
         self.win = shared.win