mirror of https://github.com/Nezreka/SoulSync.git
Merge pull request #574 from Nezreka/feature/write-artist-jpg-to-folder-for-navidrome
Write artist.jpg to artist folder so Navidrome shows real photospull/575/head
commit
63f313d0c2
@ -0,0 +1,161 @@
|
||||
"""Write `artist.jpg` to the artist's folder on disk.
|
||||
|
||||
Navidrome has no API for setting an artist image — it reads
|
||||
`artist.jpg` (or `artist.png` / `folder.jpg`) directly from the
|
||||
artist's folder during library scans. Plex and Jellyfin have API
|
||||
uploads (already implemented elsewhere), but their `read_from_disk`
|
||||
behavior also picks up `artist.jpg` as a fallback, so writing the
|
||||
file to disk is a portable mechanism that works for every server.
|
||||
|
||||
Pre-existing reference: issue #572 (rhwc) — Navidrome users only
|
||||
saw album-art-derived artist thumbnails. SoulSync's
|
||||
`update_artist_poster()` for Navidrome at `core/navidrome_client.py`
|
||||
was a NO-OP (returned True without doing anything).
|
||||
|
||||
This module is the pure helpers backing the new endpoint. No
|
||||
network, no DB, no Flask. Each function is testable in isolation
|
||||
with `tmp_path` fixtures.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from typing import Optional, Tuple
|
||||
|
||||
import requests
|
||||
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
|
||||
logger = get_logger("library.artist_image")
|
||||
|
||||
|
||||
_ARTIST_IMAGE_FILENAME = "artist.jpg"
|
||||
|
||||
# Reasonable timeout for the image download. Artist images from
|
||||
# Spotify/Deezer are typically 100-500KB so a generous timeout still
|
||||
# completes in a few seconds on a slow connection.
|
||||
_DEFAULT_IMAGE_DOWNLOAD_TIMEOUT = 30
|
||||
|
||||
|
||||
def derive_artist_folder(album_folder: str) -> str:
|
||||
"""Derive the artist's folder from an album's folder.
|
||||
|
||||
Standard SoulSync path templates produce
|
||||
``<library_root>/<artist>/<album>/...`` — so the artist folder is
|
||||
one level up from the album folder. Returns empty string for
|
||||
empty input; preserves the platform's path separator.
|
||||
|
||||
Doesn't validate that the result exists on disk. Caller checks.
|
||||
"""
|
||||
if not album_folder or not isinstance(album_folder, str):
|
||||
return ""
|
||||
# Trim trailing separator so dirname doesn't return the album
|
||||
# folder unchanged on inputs like "Music/Drake/Views/".
|
||||
trimmed = album_folder.rstrip("/").rstrip("\\")
|
||||
parent = os.path.dirname(trimmed)
|
||||
return parent or ""
|
||||
|
||||
|
||||
def pick_artist_image_url(artist_obj) -> Optional[str]:
|
||||
"""Return the URL to use for the artist image, if any.
|
||||
|
||||
Reads the `image_url` attribute from a typed Artist dataclass
|
||||
(Spotify / Deezer / Discogs / etc — every typed Artist exposes
|
||||
this). Source converters already pick the largest variant the
|
||||
provider returns (Spotify upgrades to 640+, Deezer uses
|
||||
`picture_xl` at ~1000px) so we don't need to re-rank here.
|
||||
|
||||
Returns None when the attribute is missing or empty.
|
||||
"""
|
||||
if artist_obj is None:
|
||||
return None
|
||||
image_url = getattr(artist_obj, "image_url", "")
|
||||
if not image_url or not isinstance(image_url, str):
|
||||
return None
|
||||
image_url = image_url.strip()
|
||||
return image_url or None
|
||||
|
||||
|
||||
def download_image_bytes(url: str, timeout: int = _DEFAULT_IMAGE_DOWNLOAD_TIMEOUT) -> Optional[bytes]:
|
||||
"""Fetch image bytes from a URL.
|
||||
|
||||
Returns None on any failure (HTTP error, timeout, non-image
|
||||
content-type, empty body). Caller surfaces a user-facing error.
|
||||
Doesn't raise.
|
||||
"""
|
||||
if not url or not isinstance(url, str):
|
||||
return None
|
||||
try:
|
||||
resp = requests.get(url, timeout=timeout, stream=True)
|
||||
except Exception as exc:
|
||||
logger.debug("artist image fetch failed for %s: %s", url, exc)
|
||||
return None
|
||||
if resp.status_code != 200:
|
||||
logger.debug("artist image fetch %s returned status %s", url, resp.status_code)
|
||||
return None
|
||||
content_type = (resp.headers.get("Content-Type") or "").lower()
|
||||
if "image" not in content_type:
|
||||
logger.debug("artist image URL %s returned non-image content-type %s", url, content_type)
|
||||
return None
|
||||
try:
|
||||
body = resp.content
|
||||
except Exception as exc:
|
||||
logger.debug("artist image read failed for %s: %s", url, exc)
|
||||
return None
|
||||
if not body:
|
||||
return None
|
||||
return body
|
||||
|
||||
|
||||
def write_artist_jpg(
|
||||
folder: str,
|
||||
image_bytes: bytes,
|
||||
*,
|
||||
overwrite: bool = False,
|
||||
) -> Tuple[bool, str]:
|
||||
"""Write `artist.jpg` to the given folder.
|
||||
|
||||
Returns ``(True, written_path)`` on success or ``(False, reason)``
|
||||
on failure. Atomic write via `<filename>.tmp` + os.replace so a
|
||||
partial write never leaves a corrupt file on disk.
|
||||
|
||||
When `overwrite=False` and the target file already exists,
|
||||
returns ``(False, 'file exists')`` without touching anything —
|
||||
respects user-supplied artist images.
|
||||
"""
|
||||
if not folder or not isinstance(folder, str):
|
||||
return False, "no folder provided"
|
||||
if not image_bytes:
|
||||
return False, "no image bytes"
|
||||
if not os.path.isdir(folder):
|
||||
return False, f"folder does not exist: {folder}"
|
||||
|
||||
target = os.path.join(folder, _ARTIST_IMAGE_FILENAME)
|
||||
if os.path.exists(target) and not overwrite:
|
||||
return False, "artist.jpg already exists; pass overwrite=True to replace"
|
||||
|
||||
tmp = target + ".tmp"
|
||||
try:
|
||||
with open(tmp, "wb") as f:
|
||||
f.write(image_bytes)
|
||||
os.replace(tmp, target)
|
||||
except Exception as exc:
|
||||
# Best-effort cleanup of the partial temp file. Not worth
|
||||
# propagating any error here — primary write already failed.
|
||||
try:
|
||||
if os.path.exists(tmp):
|
||||
os.remove(tmp)
|
||||
except Exception: # noqa: S110 — cleanup, not critical
|
||||
pass
|
||||
return False, f"write failed: {exc}"
|
||||
|
||||
return True, target
|
||||
|
||||
|
||||
__all__ = [
|
||||
"derive_artist_folder",
|
||||
"pick_artist_image_url",
|
||||
"download_image_bytes",
|
||||
"write_artist_jpg",
|
||||
]
|
||||
@ -0,0 +1,228 @@
|
||||
"""Pin the pure helpers in `core/library/artist_image.py`.
|
||||
|
||||
These back the new artist-image-to-disk feature added for issue
|
||||
#572 (Navidrome can't show real artist photos because Navidrome has
|
||||
no API for setting them — only reads `artist.jpg` from the artist
|
||||
folder on disk).
|
||||
|
||||
Tests are intentionally fixture-driven (tmp_path) so they actually
|
||||
exercise the filesystem code (atomic replace, overwrite guard,
|
||||
missing folder), not just mock interactions.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from types import SimpleNamespace
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# derive_artist_folder
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestDeriveArtistFolder:
|
||||
def test_one_level_up_from_album(self):
|
||||
from core.library.artist_image import derive_artist_folder
|
||||
# POSIX path
|
||||
assert derive_artist_folder("/music/Drake/Views") == "/music/Drake"
|
||||
|
||||
def test_handles_trailing_slash(self):
|
||||
"""Caller might pass an album folder with a trailing slash.
|
||||
Without trimming, `os.path.dirname` returns the input unchanged
|
||||
— silently breaks the up-one-level contract."""
|
||||
from core.library.artist_image import derive_artist_folder
|
||||
assert derive_artist_folder("/music/Drake/Views/") == "/music/Drake"
|
||||
|
||||
def test_empty_string_returns_empty(self):
|
||||
from core.library.artist_image import derive_artist_folder
|
||||
assert derive_artist_folder("") == ""
|
||||
|
||||
def test_none_returns_empty(self):
|
||||
from core.library.artist_image import derive_artist_folder
|
||||
assert derive_artist_folder(None) == ""
|
||||
|
||||
def test_non_string_returns_empty(self):
|
||||
"""Defensive — caller might hand us a Path object or similar.
|
||||
Currently we require str; return empty rather than raise."""
|
||||
from core.library.artist_image import derive_artist_folder
|
||||
assert derive_artist_folder(42) == ""
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# pick_artist_image_url
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestPickArtistImageUrl:
|
||||
def test_returns_image_url_when_set(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
artist = SimpleNamespace(image_url="https://example.com/drake.jpg")
|
||||
assert pick_artist_image_url(artist) == "https://example.com/drake.jpg"
|
||||
|
||||
def test_returns_none_when_empty_string(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
assert pick_artist_image_url(SimpleNamespace(image_url="")) is None
|
||||
|
||||
def test_returns_none_when_attribute_missing(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
assert pick_artist_image_url(SimpleNamespace()) is None
|
||||
|
||||
def test_returns_none_when_artist_is_none(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
assert pick_artist_image_url(None) is None
|
||||
|
||||
def test_strips_whitespace(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
artist = SimpleNamespace(image_url=" https://example.com/drake.jpg ")
|
||||
assert pick_artist_image_url(artist) == "https://example.com/drake.jpg"
|
||||
|
||||
def test_returns_none_when_non_string(self):
|
||||
from core.library.artist_image import pick_artist_image_url
|
||||
# int / list / dict would all hit the `isinstance(..., str)` guard
|
||||
assert pick_artist_image_url(SimpleNamespace(image_url=42)) is None
|
||||
assert pick_artist_image_url(SimpleNamespace(image_url=["url"])) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# download_image_bytes
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _fake_response(status_code=200, content_type="image/jpeg", body=b"\x89PNG..."):
|
||||
resp = MagicMock()
|
||||
resp.status_code = status_code
|
||||
resp.headers = {"Content-Type": content_type}
|
||||
resp.content = body
|
||||
return resp
|
||||
|
||||
|
||||
class TestDownloadImageBytes:
|
||||
def test_returns_bytes_on_success(self):
|
||||
from core.library import artist_image as ai
|
||||
fake = _fake_response(body=b"image-data-here")
|
||||
with patch.object(ai, "requests") as r:
|
||||
r.get.return_value = fake
|
||||
result = ai.download_image_bytes("https://example.com/x.jpg")
|
||||
assert result == b"image-data-here"
|
||||
|
||||
def test_returns_none_on_404(self):
|
||||
from core.library import artist_image as ai
|
||||
fake = _fake_response(status_code=404)
|
||||
with patch.object(ai, "requests") as r:
|
||||
r.get.return_value = fake
|
||||
assert ai.download_image_bytes("https://example.com/x.jpg") is None
|
||||
|
||||
def test_returns_none_on_non_image_content_type(self):
|
||||
"""Defensive: if a URL returns HTML or JSON (e.g. an error page),
|
||||
don't try to write it as artist.jpg."""
|
||||
from core.library import artist_image as ai
|
||||
fake = _fake_response(content_type="text/html")
|
||||
with patch.object(ai, "requests") as r:
|
||||
r.get.return_value = fake
|
||||
assert ai.download_image_bytes("https://example.com/x.jpg") is None
|
||||
|
||||
def test_returns_none_on_empty_body(self):
|
||||
from core.library import artist_image as ai
|
||||
fake = _fake_response(body=b"")
|
||||
with patch.object(ai, "requests") as r:
|
||||
r.get.return_value = fake
|
||||
assert ai.download_image_bytes("https://example.com/x.jpg") is None
|
||||
|
||||
def test_returns_none_on_exception(self):
|
||||
"""Network timeout / DNS failure / etc shouldn't raise to
|
||||
the caller — caller just sees None and surfaces a generic
|
||||
'image fetch failed' error to the user."""
|
||||
from core.library import artist_image as ai
|
||||
with patch.object(ai, "requests") as r:
|
||||
r.get.side_effect = RuntimeError("network down")
|
||||
assert ai.download_image_bytes("https://example.com/x.jpg") is None
|
||||
|
||||
def test_returns_none_for_empty_url(self):
|
||||
from core.library.artist_image import download_image_bytes
|
||||
assert download_image_bytes("") is None
|
||||
assert download_image_bytes(None) is None
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# write_artist_jpg
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestWriteArtistJpg:
|
||||
def test_writes_file_on_success(self, tmp_path):
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
success, path = write_artist_jpg(str(tmp_path), b"image-bytes")
|
||||
assert success is True
|
||||
assert os.path.exists(path)
|
||||
assert open(path, "rb").read() == b"image-bytes"
|
||||
|
||||
def test_returns_failure_when_folder_missing(self, tmp_path):
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
missing = str(tmp_path / "does-not-exist")
|
||||
success, reason = write_artist_jpg(missing, b"image-bytes")
|
||||
assert success is False
|
||||
assert "does not exist" in reason
|
||||
|
||||
def test_returns_failure_for_empty_bytes(self, tmp_path):
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
success, reason = write_artist_jpg(str(tmp_path), b"")
|
||||
assert success is False
|
||||
assert "image bytes" in reason
|
||||
|
||||
def test_returns_failure_for_empty_folder(self):
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
success, reason = write_artist_jpg("", b"image-bytes")
|
||||
assert success is False
|
||||
assert "folder" in reason
|
||||
|
||||
def test_respects_existing_file_without_overwrite(self, tmp_path):
|
||||
"""Default overwrite=False protects user-supplied artist.jpg
|
||||
from being clobbered by a programmatic update. User must opt
|
||||
in to overwrite."""
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
target = tmp_path / "artist.jpg"
|
||||
target.write_bytes(b"user-supplied")
|
||||
|
||||
success, reason = write_artist_jpg(str(tmp_path), b"new-bytes")
|
||||
assert success is False
|
||||
assert "already exists" in reason
|
||||
# Existing file must be untouched.
|
||||
assert target.read_bytes() == b"user-supplied"
|
||||
|
||||
def test_overwrites_when_flag_set(self, tmp_path):
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
target = tmp_path / "artist.jpg"
|
||||
target.write_bytes(b"old-bytes")
|
||||
|
||||
success, path = write_artist_jpg(str(tmp_path), b"new-bytes", overwrite=True)
|
||||
assert success is True
|
||||
assert target.read_bytes() == b"new-bytes"
|
||||
|
||||
def test_atomic_write_no_temp_left_on_success(self, tmp_path):
|
||||
"""`.tmp` artifact must be cleaned up by `os.replace`. Don't
|
||||
leave litter behind for the next backup / sync run to puzzle
|
||||
over."""
|
||||
from core.library.artist_image import write_artist_jpg
|
||||
success, _ = write_artist_jpg(str(tmp_path), b"image-bytes")
|
||||
assert success is True
|
||||
assert not (tmp_path / "artist.jpg.tmp").exists()
|
||||
|
||||
def test_atomic_write_cleans_temp_on_failure(self, tmp_path, monkeypatch):
|
||||
"""If `os.replace` fails (permission, cross-device, etc),
|
||||
the helper should remove the temp file rather than leaving
|
||||
a half-written `.tmp` on disk."""
|
||||
from core.library import artist_image as ai
|
||||
|
||||
def _failing_replace(src, dst):
|
||||
raise OSError("simulated replace failure")
|
||||
|
||||
monkeypatch.setattr(os, "replace", _failing_replace)
|
||||
success, reason = ai.write_artist_jpg(str(tmp_path), b"image-bytes")
|
||||
assert success is False
|
||||
assert "write failed" in reason
|
||||
# Temp must not be left behind
|
||||
assert not (tmp_path / "artist.jpg.tmp").exists()
|
||||
Loading…
Reference in new issue