added loading of existing SACDs

audiotagloader/app.py

@@ -3,11 +3,11 @@
 
 from .config import DISCOGS_TOKEN, MAX_PROPOSED_LEN, IMAGE_SIZE_STUB
 
-from .models import Artist, Album, Image, Tracklist, Track, ReleaseType
+from .models import Artist, Album, Image, Tracklist, Track, ReleaseType, ReleaseList
 
 import re
 
-from .output import track_tags_to_output
+from .output import track_tags_to_output, releases_to_output
 
 from .cache import cache
 
@@ -180,6 +180,8 @@ def get_track_tags_by_artist(
         image = self._get_cover_image(current_album.id)
         tracks = self._get_tracklist(current_album.id)
 
+        self._get_releases_by_master_id(current_album.id)
+
         return (current_album, image, tracks)
 
     @track_tags_to_output
@@ -190,6 +192,8 @@ def get_track_tags_by_master_id(
         image = self._get_cover_image(current_album.id)
         tracks = self._get_tracklist(current_album.id)
 
+        self._get_releases_by_master_id(master_id)
+
         return (current_album, image, tracks)
 
     @track_tags_to_output
@@ -201,3 +205,24 @@ def get_track_tags_by_release_id(
         tracks = self._get_tracklist(current_album.id, ReleaseType.RELEASE)
 
         return (current_album, image, tracks)
+
+    @releases_to_output
+    @cache
+    def _get_releases_by_master_id(self, master_id: int) -> ReleaseList:
+        res = self._client.master(master_id).versions
+
+        res.per_page = 100
+        releases = ReleaseList()
+        for i in range(res.pages):
+            for release in res.page(i):
+                if (id := release.data.get("id", None)) is not None:
+                    if "CD" in release.data.get("major_formats", None):
+                        releases.cd_flag = True
+                    if "SACD" in release.data.get("major_formats", None):
+                        releases.add_sacd(
+                            id=id,
+                            label=release.data.get("label", None),
+                            country=release.data.get("country", None),
+                            year=release.year,
+                        )
+        return releases

audiotagloader/cache.py

@@ -39,18 +39,19 @@ def wrapper(*args, **kwargs) -> Any:
                 data = redis_client.get(key)
 
                 if data is not None:
+                    typer.secho(
+                        "Load from Redis cache", fg=typer.colors.GREEN, bold=True
+                    )
                     if issubclass(return_type, BaseModel):
                         return return_type.model_validate_json(data.decode())
                     else:
                         arg_type = get_args(return_type)
                         current_type = arg_type[0]
-                        typer.secho(
-                            "Load from Redis cache", fg=typer.colors.GREEN, bold=True
-                        )
                         return [
                             current_type.model_validate(item)
                             for item in json.loads(data.decode())
                         ]
+
             except Exception as e:
                 raise e
 

audiotagloader/models.py

@@ -93,3 +93,43 @@ def convert_track_position(cls, tracks: list[Track]) -> list[Track]:
         for i in range(len(tracks)):
             tracks[i].position = i + 1
         return tracks
+
+
+class SacdRelease(BaseModel):
+    id: int
+    label: str = Field(default="")
+    country: str = Field(default="")
+    year: int = Field(default=0)
+
+    def __str__(self):
+        return f"({self.year} {self.label} {self.country}) - {self.id}"
+
+    def __hash__(self):
+        return hash(self.id)
+
+    def __eq__(self, other):
+        if not isinstance(other, Album):
+            return False
+        return self.id == other.id
+
+
+class ReleaseList(BaseModel):
+    sacds: list[SacdRelease] = Field(default_factory=lambda: list())
+    cd_flag: bool = Field(default=False)
+
+    def add_sacd(
+        self, id: int, label: str | None, country: str | None, year: int | str | None
+    ) -> None:
+        label = label if label is not None else ""
+        country = country if country is not None else ""
+        year = int(year) if year is not None else 0
+
+        self.sacds.append(SacdRelease(id=id, label=label, country=country, year=year))
+
+    @field_validator("sacds", mode="before")
+    @classmethod
+    def normalize_sacds(cls, value: list[SacdRelease] | None) -> list[SacdRelease]:
+        if value is None:
+            res: list[SacdRelease] = []
+            return res
+        return value

audiotagloader/output.py

@@ -1,7 +1,7 @@
 from functools import wraps
 from typing import Callable
 
-from .models import Album, Image, Tracklist
+from .models import Album, Image, Tracklist, ReleaseList
 
 import typer
 
@@ -47,3 +47,22 @@ def wrapper(*args, **kwargs) -> tuple[Album, Image, Tracklist]:
         return res
 
     return wrapper
+
+
+def releases_to_output(func) -> Callable:
+    @wraps(func)
+    def wrapper(*args, **kwargs) -> ReleaseList:
+        res: ReleaseList = func(*args, **kwargs)
+        if len(res.sacds) > 0:
+            typer.secho("SACDs:", fg=typer.colors.GREEN, bold=True)
+            for release in res.sacds:
+                print(release)
+        else:
+            typer.secho("no SACDs", fg=typer.colors.RED, bold=True)
+        if res.cd_flag:
+            typer.secho("CDs exist", fg=typer.colors.GREEN, bold=True)
+        else:
+            typer.secho("no CDs", fg=typer.colors.RED, bold=True)
+        return res
+
+    return wrapper