danvics/soulsync
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions 1

Cover Art Filler: detect missing art ON DISK + actually write it to files

Browse source 
Previously the filler only flagged albums whose DB thumb_url was empty and, on
apply, only updated that DB thumb_url — so albums whose files had no embedded
art and no cover.jpg (but whose DB row had a URL) were never found, and even
'applying' art never touched the files. That's the reported 'doesn't scan all
albums' gap.

New core.metadata.art_apply (reuses the post-processing standard so the user's
album_art_order is honored):
- album_has_art_on_disk(): cheap-first check — folder cover.jpg/folder.jpg
  sidecar, then embedded art in a representative track (FLAC/ID3/MP4/Vorbis).
- apply_art_to_album_files(): embeds via embed_album_art_metadata + writes
  cover.jpg via download_cover_art; only ADDS art (never rewrites the user's
  tags); read-only/unwritable files are skipped + counted, never crash.

Scan now examines every titled album and flags it when art is missing in the DB
OR on disk. Apply embeds into the album's audio files + writes cover.jpg in
addition to the DB thumbnail (media-server-only albums fall back to DB-only).

Tests cover sidecar/embedded detection, the cheap-first short-circuit, and the
apply orchestration (embeds each file + cover.jpg; read-only failures counted).
This commit is contained in:
BoulderBadgeDad 2026-06-03 22:12:56 -07:00
parent 80828b86cf
commit 33965c7cbd
5 changed files with 382 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

147
core/metadata/art_apply.py Normal file
View file

@ -0,0 +1,147 @@
"""Apply album art to existing library files.
Two jobs, both reusing the post-processing standard so the user's
``album_art_order`` preference is honored and embedded art matches cover.jpg:
- Detect whether an album already has art ON DISK (embedded in the audio file
or a cover.jpg/folder.jpg sidecar) — the Cover Art Filler previously only
looked at the DB ``thumb_url``, so albums whose files were artless but whose
DB row had a URL were never flagged.
- Embed found art into the album's audio files (``embed_album_art_metadata``)
and write a cover.jpg (``download_cover_art``). Only ADDS art — it does not
clear or rewrite the user's existing tags.
"""
from __future__ import annotations
import contextlib
import os
from typing import Iterable
from core.metadata.artwork import download_cover_art, embed_album_art_metadata
from core.metadata.common import get_mutagen_symbols
from utils.logging_config import get_logger
logger = get_logger("metadata.art_apply")
# Folder-level cover files recognised across players (matches soulsync_client).
_COVER_SIDECARS = (
"cover.jpg", "cover.jpeg", "cover.png",
"folder.jpg", "folder.jpeg", "folder.png",
)
def folder_has_cover_sidecar(folder: str) -> bool:
"""True if the album folder already carries a cover.jpg/folder.jpg sidecar."""
if not folder:
return False
try:
for name in _COVER_SIDECARS:
if os.path.isfile(os.path.join(folder, name)):
return True
except OSError:
return False
return False
def file_has_embedded_art(file_path: str) -> bool:
"""True if the audio file already has embedded cover art (FLAC picture,
ID3 APIC, MP4 covr, or a Vorbis metadata_block_picture)."""
if not file_path or not os.path.isfile(file_path):
return False
symbols = get_mutagen_symbols()
if not symbols:
return False
try:
audio = symbols.File(file_path)
if audio is None:
return False
# FLAC / Ogg expose picture blocks directly.
if getattr(audio, "pictures", None):
return True
tags = getattr(audio, "tags", None)
if isinstance(audio, symbols.MP4):
return bool(audio.get("covr"))
if tags is None:
return False
with contextlib.suppress(Exception):
if isinstance(tags, symbols.ID3):
return bool(tags.getall("APIC"))
with contextlib.suppress(Exception):
if "metadata_block_picture" in tags:
return True
return False
except Exception as exc:
logger.debug("art presence check failed for %s: %s", file_path, exc)
return False
def album_has_art_on_disk(rep_file_path: str) -> bool:
"""Does this album have art on disk?
Checks the folder for a cover sidecar first (cheap stat) and only opens the
representative audio file when there's no sidecar. Returns True when there's
no local file to inspect (e.g. a media-server-only album) so such albums
aren't wrongly flagged as missing file art.
"""
if not rep_file_path:
return True
folder = os.path.dirname(rep_file_path)
if folder_has_cover_sidecar(folder):
return True
return file_has_embedded_art(rep_file_path)
def apply_art_to_album_files(
file_paths: Iterable[str],
metadata: dict,
album_info: dict,
folder: str = None,
context: dict = None,
) -> dict:
"""Embed art into each audio file + write cover.jpg, reusing the standard.
``metadata`` feeds ``embed_album_art_metadata`` (needs album_artist/artist/
album, optionally musicbrainz_release_id and album_art_url as the fallback
URL). ``album_info`` feeds ``download_cover_art`` (album_name/album_image_url/
musicbrainz_release_id). Existing tags are preserved — only art is added.
Returns counts; never raises (unwritable/read-only files are skipped).
"""
result = {"embedded": 0, "failed": 0, "skipped": 0, "cover_written": False}
symbols = get_mutagen_symbols()
paths = [p for p in (file_paths or []) if p]
if not symbols:
return result
for fp in paths:
if not os.path.isfile(fp):
result["skipped"] += 1
continue
try:
audio = symbols.File(fp)
if audio is None:
result["skipped"] += 1
continue
# ID3 needs a tag container before APIC can be added.
if getattr(audio, "tags", None) is None and hasattr(audio, "add_tags"):
with contextlib.suppress(Exception):
audio.add_tags()
if embed_album_art_metadata(audio, metadata):
audio.save()
result["embedded"] += 1
else:
result["failed"] += 1
except Exception as exc:
# Read-only mounts / permission errors land here — skip, don't crash.
logger.warning("Could not embed art into %s: %s", fp, exc)
result["failed"] += 1
target_dir = folder or (os.path.dirname(paths[0]) if paths else None)
if target_dir and os.path.isdir(target_dir):
try:
download_cover_art(album_info, target_dir, context)
result["cover_written"] = folder_has_cover_sidecar(target_dir)
except Exception as exc:
logger.warning("cover.jpg write failed for %s: %s", target_dir, exc)
return result

35
core/repair_jobs/missing_cover_art.py
View file

@ -1,7 +1,9 @@
"""Missing Cover Art Filler Job — finds albums without artwork and locates art from APIs."""
import os
import re
from core.metadata.art_apply import album_has_art_on_disk
from core.metadata_service import get_client_for_source, get_primary_source, get_source_priority
from core.repair_jobs import register_job
from core.repair_jobs.base import JobContext, JobResult, RepairJob
@ -88,6 +90,12 @@ class MissingCoverArtJob(RepairJob):
"al.spotify_album_id",
"al.thumb_url",
"ar.thumb_url",
# A representative local track path, so we can check whether the
# album actually has art ON DISK (embedded / cover.jpg) — not just
# whether the DB row has a thumb_url.
("(SELECT t.file_path FROM tracks t WHERE t.album_id = al.id "
"AND t.file_path IS NOT NULL AND t.file_path != '' "
"ORDER BY t.disc_number, t.track_number LIMIT 1) AS rep_path"),
]
column_map = [
("itunes_album_id", "al.itunes_album_id"),
@ -101,12 +109,14 @@ class MissingCoverArtJob(RepairJob):
column_index[alias] = len(select_cols)
select_cols.append(f"{column} AS {alias}")
# Scan every titled album — we decide per-album whether art is
# missing in the DB OR on disk (the file/cover.jpg). The on-disk
# check is cheap-first (a sidecar stat before opening any audio).
cursor.execute(f"""
SELECT {', '.join(select_cols)}
FROM albums al
LEFT JOIN artists ar ON ar.id = al.artist_id
WHERE (al.thumb_url IS NULL OR al.thumb_url = '')
AND al.title IS NOT NULL AND al.title != ''
WHERE al.title IS NOT NULL AND al.title != ''
""")
albums = cursor.fetchall()
except Exception as e:
@ -132,7 +142,7 @@ class MissingCoverArtJob(RepairJob):
if i % 10 == 0 and context.wait_if_paused():
return result
album_id, title, artist_name, spotify_album_id, _, artist_thumb = row[:6]
album_id, title, artist_name, spotify_album_id, album_thumb, artist_thumb, rep_path = row[:7]
source_album_ids = {
'spotify': spotify_album_id,
'itunes': row[column_index['itunes_album_id']] if 'itunes_album_id' in column_index else None,
@ -142,6 +152,14 @@ class MissingCoverArtJob(RepairJob):
}
result.scanned += 1
# Art can be missing in the DB (no thumb_url) and/or on disk (no
# embedded art and no cover.jpg). Skip albums that already have both.
db_missing = not (str(album_thumb).strip() if album_thumb else '')
disk_missing = bool(rep_path) and not album_has_art_on_disk(rep_path)
if not db_missing and not disk_missing:
result.skipped += 1
continue
if context.report_progress:
context.report_progress(
scanned=i + 1, total=total,
@ -183,6 +201,12 @@ class MissingCoverArtJob(RepairJob):
'found_artwork_url': artwork_url,
'spotify_album_id': spotify_album_id,
'artist_thumb_url': artist_thumb or None,
# Where the files live + what was missing, so the
# apply can embed into the audio + write cover.jpg.
'album_folder': os.path.dirname(rep_path) if rep_path else None,
'db_missing': db_missing,
'disk_missing': disk_missing,
'musicbrainz_release_id': None,
}
)
if inserted:
@ -328,10 +352,11 @@ class MissingCoverArtJob(RepairJob):
try:
conn = context.db._get_connection()
cursor = conn.cursor()
# Upper bound: every titled album is examined (the per-album DB/disk
# art check decides which actually need filling).
cursor.execute("""
SELECT COUNT(*) FROM albums
WHERE (thumb_url IS NULL OR thumb_url = '')
AND title IS NOT NULL AND title != ''
WHERE title IS NOT NULL AND title != ''
""")
row = cursor.fetchone()
return row[0] if row else 0

67
core/repair_worker.py
View file

@ -1277,28 +1277,85 @@ class RepairWorker:
return {'success': False, 'error': str(e)}
def _fix_missing_cover_art(self, entity_type, entity_id, file_path, details):
"""Update album thumbnail URL from the found artwork."""
"""Apply found artwork: update the DB thumbnail AND embed art into the
album's audio files + write cover.jpg (using the post-processing
standard, so the user's album_art_order preference is honored)."""
artwork_url = details.get('found_artwork_url')
if not artwork_url:
return {'success': False, 'error': 'No artwork URL found in finding details'}
album_id = details.get('album_id') or entity_id
if not album_id:
return {'success': False, 'error': 'No album ID associated with this finding'}
conn = None
track_paths = []
album_title = details.get('album_title')
artist_name = details.get('artist')
mbid = details.get('musicbrainz_release_id')
try:
conn = self.db._get_connection()
cursor = conn.cursor()
cursor.execute("UPDATE albums SET thumb_url = ?, updated_at = CURRENT_TIMESTAMP WHERE id = ?",
(artwork_url, album_id))
conn.commit()
if cursor.rowcount > 0:
return {'success': True, 'action': 'applied_cover_art',
'message': 'Applied cover art to album'}
return {'success': False, 'error': 'Album not found in database'}
if cursor.rowcount == 0:
return {'success': False, 'error': 'Album not found in database'}
# Pull album metadata + local track paths so we can write art to disk.
cursor.execute("""
SELECT al.title, ar.name, al.musicbrainz_release_id
FROM albums al LEFT JOIN artists ar ON ar.id = al.artist_id
WHERE al.id = ?
""", (album_id,))
meta_row = cursor.fetchone()
if meta_row:
album_title = album_title or meta_row[0]
artist_name = artist_name or meta_row[1]
mbid = mbid or meta_row[2]
cursor.execute("""
SELECT file_path FROM tracks
WHERE album_id = ? AND file_path IS NOT NULL AND file_path != ''
""", (album_id,))
track_paths = [r[0] for r in cursor.fetchall()]
finally:
if conn:
conn.close()
# Resolve container/host path mismatches, keep only files that exist.
download_folder = self._config_manager.get('soulseek.download_path', '') if self._config_manager else None
resolved = []
for p in track_paths:
rp = _resolve_file_path(p, self.transfer_folder, download_folder, config_manager=self._config_manager) or p
if os.path.isfile(rp):
resolved.append(rp)
if not resolved:
# Media-server-only album (no local files): DB thumbnail is all we can set.
return {'success': True, 'action': 'applied_cover_art',
'message': 'Applied cover art to album (database only — no local files found)'}
from core.metadata.art_apply import apply_art_to_album_files
metadata = {
'artist': artist_name, 'album_artist': artist_name,
'album': album_title, 'album_art_url': artwork_url,
'musicbrainz_release_id': mbid,
}
album_info = {
'album_name': album_title, 'album_image_url': artwork_url,
'musicbrainz_release_id': mbid,
}
folder = details.get('album_folder') or os.path.dirname(resolved[0])
art_result = apply_art_to_album_files(resolved, metadata, album_info, folder=folder)
embedded = art_result.get('embedded', 0)
msg = f'Applied cover art: embedded into {embedded}/{len(resolved)} file(s)'
if art_result.get('cover_written'):
msg += ' + wrote cover.jpg'
if embedded == 0 and not art_result.get('cover_written'):
# DB updated but nothing reached disk (e.g. read-only mount).
msg = 'Updated database thumbnail, but could not write art to files (read-only?)'
return {'success': True, 'action': 'applied_cover_art', 'message': msg, 'art_result': art_result}
def _fix_metadata_gap(self, entity_type, entity_id, file_path, details):
"""Apply found metadata fields to the track."""
found_fields = details.get('found_fields')

131
tests/test_art_apply.py Normal file
View file

@ -0,0 +1,131 @@
"""Tests for core.metadata.art_apply — on-disk art detection + applying art."""
import sys
import types
from types import SimpleNamespace
# Stubs so core.metadata.artwork (pulled in transitively) imports without the
# real Spotify / config dependencies (mirrors test_missing_cover_art.py).
if 'spotipy' not in sys.modules:
spotipy = types.ModuleType('spotipy')
oauth2 = types.ModuleType('spotipy.oauth2')
spotipy.Spotify = type('S', (), {})
oauth2.SpotifyOAuth = oauth2.SpotifyClientCredentials = type('O', (), {})
spotipy.oauth2 = oauth2
sys.modules['spotipy'] = spotipy
sys.modules['spotipy.oauth2'] = oauth2
if 'config.settings' not in sys.modules:
config_mod = types.ModuleType('config')
settings_mod = types.ModuleType('config.settings')
class _Cfg:
def get(self, key, default=None):
return default
def get_active_media_server(self):
return 'plex'
settings_mod.config_manager = _Cfg()
config_mod.settings = settings_mod
sys.modules['config'] = config_mod
sys.modules['config.settings'] = settings_mod
from core.metadata import art_apply as aa
# ── sidecar detection ──
def test_folder_has_cover_sidecar(tmp_path):
assert aa.folder_has_cover_sidecar(str(tmp_path)) is False
(tmp_path / 'cover.jpg').write_bytes(b'x')
assert aa.folder_has_cover_sidecar(str(tmp_path)) is True
def test_album_has_art_on_disk_no_local_file_is_true():
# No representative file (e.g. media-server-only album) → not flagged.
assert aa.album_has_art_on_disk('') is True
assert aa.album_has_art_on_disk(None) is True
def test_album_has_art_on_disk_sidecar_short_circuits(tmp_path, monkeypatch):
(tmp_path / 'cover.jpg').write_bytes(b'x')
track = tmp_path / '01 song.flac'
track.write_bytes(b'')
# Sidecar present → True without ever opening the audio file.
called = {'n': 0}
monkeypatch.setattr(aa, 'file_has_embedded_art', lambda p: called.__setitem__('n', called['n'] + 1) or False)
assert aa.album_has_art_on_disk(str(track)) is True
assert called['n'] == 0
def test_album_has_art_on_disk_no_sidecar_checks_file(tmp_path, monkeypatch):
track = tmp_path / '01 song.flac'
track.write_bytes(b'')
monkeypatch.setattr(aa, 'file_has_embedded_art', lambda p: False)
assert aa.album_has_art_on_disk(str(track)) is False
monkeypatch.setattr(aa, 'file_has_embedded_art', lambda p: True)
assert aa.album_has_art_on_disk(str(track)) is True
# ── embedded-art detection ──
def _fake_symbols(audio):
return SimpleNamespace(
File=lambda path: audio,
ID3=type('ID3', (), {}),
MP4=type('MP4', (), {}),
FLAC=type('FLAC', (), {}),
)
def test_file_has_embedded_art_flac_picture(tmp_path, monkeypatch):
f = tmp_path / 'a.flac'
f.write_bytes(b'')
audio = SimpleNamespace(pictures=['pic'], tags=None)
monkeypatch.setattr(aa, 'get_mutagen_symbols', lambda: _fake_symbols(audio))
assert aa.file_has_embedded_art(str(f)) is True
def test_file_has_embedded_art_none(tmp_path, monkeypatch):
f = tmp_path / 'a.flac'
f.write_bytes(b'')
audio = SimpleNamespace(pictures=[], tags=None)
monkeypatch.setattr(aa, 'get_mutagen_symbols', lambda: _fake_symbols(audio))
assert aa.file_has_embedded_art(str(f)) is False
# ── applying art ──
def test_apply_embeds_into_each_file_and_writes_cover(tmp_path, monkeypatch):
f1 = tmp_path / '01.flac'; f1.write_bytes(b'')
f2 = tmp_path / '02.flac'; f2.write_bytes(b'')
saved = []
audio = SimpleNamespace(tags=None, add_tags=lambda: None, save=lambda: saved.append(True))
monkeypatch.setattr(aa, 'get_mutagen_symbols', lambda: _fake_symbols(audio))
embed_calls = []
monkeypatch.setattr(aa, 'embed_album_art_metadata', lambda a, m: embed_calls.append(m) or True)
# download_cover_art is the standard cover.jpg writer — stub it to drop one.
monkeypatch.setattr(aa, 'download_cover_art',
lambda album_info, folder, ctx=None: open(f"{folder}/cover.jpg", 'wb').close())
meta = {'artist': 'A', 'album': 'B', 'album_art_url': 'http://x/y.jpg'}
res = aa.apply_art_to_album_files([str(f1), str(f2)], meta, {'album_name': 'B'}, folder=str(tmp_path))
assert res['embedded'] == 2
assert len(saved) == 2 # each file saved after embed
assert res['cover_written'] is True
def test_apply_counts_failures_without_raising(tmp_path, monkeypatch):
f1 = tmp_path / '01.flac'; f1.write_bytes(b'')
audio = SimpleNamespace(tags=object(), save=lambda: (_ for _ in ()).throw(OSError('read-only')))
monkeypatch.setattr(aa, 'get_mutagen_symbols', lambda: _fake_symbols(audio))
monkeypatch.setattr(aa, 'embed_album_art_metadata', lambda a, m: True)
monkeypatch.setattr(aa, 'download_cover_art', lambda *a, **k: None)
res = aa.apply_art_to_album_files([str(f1)], {'album': 'B'}, {'album_name': 'B'}, folder=str(tmp_path))
assert res['embedded'] == 0
assert res['failed'] == 1 # save() raised (read-only) — counted, not crashed

12
tests/test_missing_cover_art.py
View file

@ -93,6 +93,18 @@ def _make_db(album_row):
)
"""
)
# The scan now joins a representative track path to check art on disk.
cursor.execute(
"""
CREATE TABLE tracks (
id INTEGER PRIMARY KEY,
album_id INTEGER,
file_path TEXT,
disc_number INTEGER,
track_number INTEGER
)
"""
)
cursor.execute(
"INSERT INTO artists (id, name, thumb_url) VALUES (?, ?, ?)",
(1, 'Artist', 'https://artist/thumb'),