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

Library re-tag: standard dry-run pattern (shows the Dry Run tag, opt-in auto-apply)

Browse source 
The job was the odd one out — auto_fix=False, no dry_run setting, so it never
showed the 'Dry Run' badge the other jobs do (the badge keys off
settings.dry_run === true). Aligned it to the standard pattern:
- auto_fix=True + dry_run setting defaulting True. Default behavior is unchanged
  (findings only, nothing written) AND it now shows the Dry Run badge.
- Turning dry_run off makes the scan auto-apply in place (result.auto_fixed),
  no finding — the opt-in 'just retag it' mode.
- Extracted a shared apply_track_plans() used by both the scan auto-apply and
  the repair_worker fix handler (handler now resolves Docker paths then
  delegates — one code path, no duplication).
Tests: dry_run=False auto-applies + writes + no finding; existing dry-run
finding/skip/apply tests still green. 410 passing.
This commit is contained in:
BoulderBadgeDad 2026-06-04 10:06:34 -07:00
parent 2328e159a1
commit 0a4c3d7dc8
3 changed files with 101 additions and 56 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

69
core/repair_jobs/library_retag.py
View file

@ -45,6 +45,55 @@ def _read_current_tags(file_path):
return {}
def apply_track_plans(track_plans, cover_action=None, cover_url=None) -> dict:
"""Write each plan's tags in place (+ optionally embed/refresh cover art),
reusing tag_writer.write_tags_to_file. ``file_path`` on each plan must be a
real, reachable path (caller resolves Docker paths). Shared by the dry-run=
False auto-apply and the repair_worker fix handler. Never raises.
"""
import os as _os
result = {'written': 0, 'failed': 0, 'skipped': 0, 'cover_written': False}
embed_cover = bool(cover_action and cover_url)
cover_data = None
if embed_cover:
try:
from core.tag_writer import download_cover_art
cover_data = download_cover_art(cover_url)
except Exception as e:
logger.debug("retag cover download failed: %s", e)
embed_cover = embed_cover and cover_data is not None
from core.tag_writer import write_tags_to_file
last_dir = None
for tp in track_plans or []:
fp = tp.get('file_path')
db_data = tp.get('db_data') or {}
if not fp or not _os.path.isfile(fp) or (not db_data and not embed_cover):
result['skipped'] += 1
continue
try:
res = write_tags_to_file(fp, db_data, embed_cover=embed_cover, cover_data=cover_data)
if res.get('success'):
result['written'] += 1
last_dir = _os.path.dirname(fp)
else:
result['failed'] += 1
except Exception as e:
logger.warning("retag write failed for %s: %s", fp, e)
result['failed'] += 1
if cover_action and cover_data and last_dir:
try:
cover_path = _os.path.join(last_dir, 'cover.jpg')
if cover_action == 'replace' or not _os.path.exists(cover_path):
with open(cover_path, 'wb') as fh:
fh.write(cover_data[0])
result['cover_written'] = True
except Exception as e:
logger.debug("retag cover.jpg write failed: %s", e)
return result
def _add_source_ids(db_data, source, album_source_id, source_track):
"""Stamp the album/track source IDs onto the write payload so the canonical
writer embeds them too (Spotify / iTunes / MusicBrainz)."""
@ -95,6 +144,8 @@ class LibraryRetagJob(RepairJob):
'comes from. Each finding lists every tag that would change (old -> new) per '
'track so you can review before applying — nothing is written until you do.\n\n'
'Settings:\n'
'- Dry run (default ON): only create findings to review; nothing is written. '
'Turn it off to auto-apply on scan.\n'
'- Mode: "overwrite" rewrites every field the source provides; "fill_missing" '
'only fills blank tags (keeps your existing values).\n'
'- Cover art: replace / fill-missing / skip.\n'
@ -104,6 +155,7 @@ class LibraryRetagJob(RepairJob):
default_enabled = False
default_interval_hours = 168
default_settings = {
'dry_run': True,
'mode': MODE_OVERWRITE,
'cover_art': 'replace',
'source': '',
@ -113,7 +165,7 @@ class LibraryRetagJob(RepairJob):
'cover_art': ['replace', 'fill_missing', 'skip'],
'source': ['', 'spotify', 'itunes', 'deezer', 'musicbrainz'],
}
auto_fix = False
auto_fix = True
def _get_settings(self, context: JobContext) -> dict:
merged = dict(self.default_settings)
@ -133,6 +185,7 @@ class LibraryRetagJob(RepairJob):
settings = self._get_settings(context)
mode = settings.get('mode', MODE_OVERWRITE)
cover_mode = settings.get('cover_art', 'replace')
dry_run = settings.get('dry_run', True)
source_order = self._source_order(settings)
if not source_order:
logger.warning("Library re-tag: no usable metadata sources configured")
@ -180,7 +233,7 @@ class LibraryRetagJob(RepairJob):
try:
self._scan_album(context, result, album_id, album_title, artist_name,
source, album_source_id, mode, cover_mode)
source, album_source_id, mode, cover_mode, dry_run)
except Exception as e:
logger.debug("Library re-tag: album %s failed: %s", album_id, e)
result.errors += 1
@ -192,7 +245,7 @@ class LibraryRetagJob(RepairJob):
return result
def _scan_album(self, context, result, album_id, album_title, artist_name,
source, album_source_id, mode, cover_mode):
source, album_source_id, mode, cover_mode, dry_run=True):
# Local tracks for this album.
with context.db._get_connection() as conn:
cur = conn.cursor()
@ -257,6 +310,16 @@ class LibraryRetagJob(RepairJob):
result.skipped += 1
return
# Not dry-run: apply the tags in place now (the track paths were already
# isfile-checked above) and count it as an auto-fix — no finding.
if not dry_run:
res = apply_track_plans(track_plans, cover_action, cover_url)
if res['written'] or res['cover_written']:
result.auto_fixed += 1
else:
result.errors += 1
return
total_changes = sum(len(tp['changes']) for tp in track_plans)
summary_bits = []
if tag_change_tracks:

68
core/repair_worker.py
View file

@ -1365,68 +1365,30 @@ class RepairWorker:
if not tracks:
return {'success': False, 'error': 'No tracks to re-tag in finding'}
cover_action = details.get('cover_action')
cover_url = details.get('cover_url')
embed_cover = bool(cover_action and cover_url)
# Resolve container/host path mismatches, then delegate to the shared
# apply path the job's auto-fix mode also uses.
download_folder = self._config_manager.get('soulseek.download_path', '') if self._config_manager else None
# Download the cover once for the whole album (batch embed).
cover_data = None
if embed_cover:
try:
from core.tag_writer import download_cover_art
cover_data = download_cover_art(cover_url)
except Exception as e:
logger.debug("library_retag: cover download failed: %s", e)
embed_cover = embed_cover and cover_data is not None
from core.tag_writer import write_tags_to_file
written = failed = skipped = 0
last_dir = None
resolved_plans = []
for t in tracks:
raw = t.get('file_path')
db_data = t.get('db_data') or {}
if not raw or (not db_data and not embed_cover):
skipped += 1
if not raw:
continue
resolved = _resolve_file_path(raw, self.transfer_folder, download_folder,
config_manager=self._config_manager) or raw
if not os.path.isfile(resolved):
skipped += 1
continue
try:
res = write_tags_to_file(resolved, db_data, embed_cover=embed_cover, cover_data=cover_data)
if res.get('success'):
written += 1
last_dir = os.path.dirname(resolved)
else:
failed += 1
except Exception as e:
logger.warning("library_retag write failed for %s: %s", resolved, e)
failed += 1
rp = _resolve_file_path(raw, self.transfer_folder, download_folder,
config_manager=self._config_manager) or raw
resolved_plans.append({'file_path': rp, 'db_data': t.get('db_data') or {}})
# Refresh the cover.jpg sidecar to match (replace, or fill when missing).
cover_written = False
if cover_action and cover_data and last_dir:
try:
cover_path = os.path.join(last_dir, 'cover.jpg')
if cover_action == 'replace' or not os.path.exists(cover_path):
with open(cover_path, 'wb') as fh:
fh.write(cover_data[0])
cover_written = True
except Exception as e:
logger.debug("library_retag: cover.jpg write failed: %s", e)
from core.repair_jobs.library_retag import apply_track_plans
res = apply_track_plans(resolved_plans, details.get('cover_action'), details.get('cover_url'))
if written == 0 and not cover_written:
if res['written'] == 0 and not res['cover_written']:
return {'success': False,
'error': 'Nothing could be written — files unreachable or read-only?'}
msg = f'Re-tagged {written} track(s)'
if failed:
msg += f' ({failed} failed)'
if cover_written:
msg = f"Re-tagged {res['written']} track(s)"
if res['failed']:
msg += f" ({res['failed']} failed)"
if res['cover_written']:
msg += ' + refreshed cover.jpg'
return {'success': True, 'action': 'library_retag', 'message': msg,
'written': written, 'failed': failed, 'skipped': skipped}
return {'success': True, 'action': 'library_retag', 'message': msg, **res}
def _fix_metadata_gap(self, entity_type, entity_id, file_path, details):
"""Apply found metadata fields to the track."""

20
tests/test_library_retag_job.py
View file

@ -91,6 +91,26 @@ def test_scan_creates_detailed_finding(tmp_path, monkeypatch):
assert tp['db_data']['title'] == 'Real Title'
def test_scan_dry_run_off_auto_applies_no_finding(tmp_path, monkeypatch):
"""dry_run=False: scan applies in place (auto_fixed) and creates NO finding."""
track = tmp_path / 'track.flac'; track.write_bytes(b'')
conn = _db_with_album(str(tmp_path / 'm.db'), str(track), current_title='Old Title')
ctx = _context(conn, {'mode': 'overwrite', 'cover_art': 'skip', 'source': 'spotify', 'dry_run': False})
_patch_source(monkeypatch, {
'title': 'Old Title', 'album_artist': 'Real Artist', 'album': 'Real Album',
'year': '2021', 'genre': 'Rock', 'track_number': 1, 'disc_number': 1,
})
writes = []
monkeypatch.setattr('core.tag_writer.write_tags_to_file',
lambda fp, db_data, **k: writes.append(db_data) or {'success': True})
result = lr.LibraryRetagJob().scan(ctx)
assert ctx.findings == [] # no finding in apply mode
assert result.auto_fixed == 1
assert writes and writes[0]['title'] == 'Real Title' # actually wrote
def test_scan_skips_album_already_correct(tmp_path, monkeypatch):
track = tmp_path / 'track.flac'; track.write_bytes(b'')
conn = _db_with_album(str(tmp_path / 'm.db'), str(track), current_title='Real Title')