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

Lift _execute_retag to core/library/retag.py

Browse source 
Pulls the 258-line retag worker out of `web_server.py` into a new
`core/library/` package. Pure 1:1 lift — wrapper keeps the original
entry-point name so the retag-trigger endpoint continues to work
without changes.

What `execute_retag` does:

1. Fetch album + track metadata for the new `album_id` (Spotify or
   iTunes — the Spotify client transparently falls back).
2. Load existing files in the retag group from the DB.
3. Match each existing track to a new Spotify track:
   - Priority 1: same disc + track number.
   - Priority 2: title similarity >= 0.6 (SequenceMatcher).
4. For each matched pair:
   - Re-write metadata tags via `_enhance_file_metadata`.
   - Compute the new path via `_build_final_path_for_track` and move
     the audio file (plus .lrc / .txt sidecars) if the path changes.
   - Drop an orphaned cover.jpg if it's left in an empty directory.
   - Clean up empty parent directories left behind.
   - Download the new cover art into the new album dir.
5. Update the retag group record with new artist / album / image /
   total_tracks / release_date and the appropriate Spotify-or-iTunes
   album ID (numeric → iTunes, alphanumeric → Spotify).
6. Mark the retag state 'finished' (or 'error' on exception).

Strict 1:1 byte parity:
The original mutated `retag_state` as a module global (the function
declared `global retag_state` even though it only mutates in place).
Here `retag_state` is exposed through the `RetagDeps` proxy as a Python
property so the lifted body keeps `name[key] = value` /
`name.update(...)` syntax. The property setter rebinds the
web_server.py reference if the function ever reassigns it (currently
it doesn't, but the setter is wired for parity with the watchlist lift).

Diff vs original after `deps.X` → global X normalization is **zero
differences** apart from the dropped `global retag_state` decl and the
inline `from database.music_database import get_database` (replaced by
deps.get_database()). 258 lines orig = 258 lines lifted, byte-identical
body otherwise.

Dependencies injected via `RetagDeps` (13 fields) — config_manager,
retag_lock, spotify_client, plus 8 callable helpers
(get_audio_quality_string, enhance_file_metadata,
build_final_path_for_track, safe_move_file, cleanup_empty_directories,
download_cover_art, docker_resolve_path, get_database) and 2 property
delegates (_get_retag_state / _set_retag_state).

Tests: 11 new under tests/library/test_retag.py covering setup error
paths (no album data, no album tracks, no existing tracks),
track-number priority match, title-similarity fallback, no-match skip,
missing file skip, file move when path changes, group record update
(spotify vs iTunes ID branching by alphanumeric vs numeric album_id),
multi-disc total_discs computation.

Full suite: 1330 passing (was 1319). Ruff clean.
This commit is contained in:
Broque Thomas 2026-04-29 09:03:42 -07:00
parent 12b9ca48df
commit 3a6597561a
5 changed files with 683 additions and 255 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

0
core/library/__init__.py Normal file
View file

329
core/library/retag.py Normal file
View file

@ -0,0 +1,329 @@
"""Library retag worker.
`execute_retag(group_id, album_id, deps)` rewrites tags + filenames for a
group of audio files when the user has matched them to a different
album. The worker:
1. Fetches album + track metadata for the new `album_id` (Spotify or
iTunes — Spotify client transparently falls back).
2. Loads existing files in the retag group from the DB.
3. Matches each existing track to a new Spotify track:
- Priority 1: same disc + track number.
- Priority 2: title similarity >= 0.6 (SequenceMatcher).
4. For each matched pair:
- Re-write metadata tags via `_enhance_file_metadata`.
- Compute the new path via `_build_final_path_for_track` and move
the audio file (plus .lrc / .txt sidecars) if the path changes.
- Drop an orphaned cover.jpg if it's left in an empty directory.
- Clean up empty parent directories left behind.
- Download the new cover art into the new album dir.
5. Update the retag group record with the new artist / album / image /
total_tracks / release_date and the appropriate Spotify-or-iTunes
album ID.
6. Mark the retag state 'finished' (or 'error' on exception).
The original mutated `retag_state` as a module global. Here it's exposed
through the `RetagDeps` proxy as a Python property so the lifted body
keeps the same `name[key] = value` syntax. The property setter rebinds
the web_server.py reference if needed (currently the function only
mutates in place via .update() and key assignment, so the setter never
fires).
"""
from __future__ import annotations
import logging
import os
import traceback
from dataclasses import dataclass
from difflib import SequenceMatcher
from typing import Any, Callable
logger = logging.getLogger(__name__)
@dataclass
class RetagDeps:
"""Bundle of cross-cutting deps the retag worker needs.
`retag_state` is exposed as a property so the lifted body keeps
`name[key] = value` / `name.update(...)` syntax.
"""
config_manager: Any
retag_lock: Any # threading.Lock
spotify_client: Any
get_audio_quality_string: Callable[[str], str]
enhance_file_metadata: Callable
build_final_path_for_track: Callable
safe_move_file: Callable
cleanup_empty_directories: Callable
download_cover_art: Callable
docker_resolve_path: Callable[[str], str]
_get_retag_state: Callable[[], dict]
_set_retag_state: Callable[[dict], None]
get_database: Callable[[], Any]
@property
def retag_state(self) -> dict:
return self._get_retag_state()
@retag_state.setter
def retag_state(self, value: dict) -> None:
self._set_retag_state(value)
def execute_retag(group_id, album_id, deps: RetagDeps):
"""Execute a retag operation: re-tag files in a group with metadata from a new album match."""
try:
with deps.retag_lock:
deps.retag_state.update({
"status": "running",
"phase": "Fetching album metadata...",
"progress": 0,
"current_track": "",
"total_tracks": 0,
"processed": 0,
"error_message": ""
})
# 1. Fetch new album metadata from Spotify/iTunes
album_data = deps.spotify_client.get_album(album_id)
if not album_data:
raise ValueError(f"Could not fetch album data for ID: {album_id}")
album_tracks_response = deps.spotify_client.get_album_tracks(album_id)
if not album_tracks_response:
raise ValueError(f"Could not fetch album tracks for ID: {album_id}")
album_tracks_items = album_tracks_response.get('items', [])
# Extract artist info
album_artists = album_data.get('artists', [])
new_artist = album_artists[0] if album_artists else {'name': 'Unknown Artist', 'id': ''}
# Ensure artist is a dict with expected fields
if not isinstance(new_artist, dict):
new_artist = {'name': str(new_artist), 'id': ''}
new_album_name = album_data.get('name', 'Unknown Album')
new_images = album_data.get('images', [])
new_image_url = new_images[0]['url'] if new_images else None
new_release_date = album_data.get('release_date', '')
total_tracks = album_data.get('total_tracks', len(album_tracks_items))
# Build spotify track list
spotify_tracks = []
for item in album_tracks_items:
track_artists = item.get('artists', [])
spotify_tracks.append({
'name': item.get('name', ''),
'track_number': item.get('track_number', 1),
'disc_number': item.get('disc_number', 1),
'id': item.get('id', ''),
'artists': track_artists,
'duration_ms': item.get('duration_ms', 0)
})
total_discs = max((t['disc_number'] for t in spotify_tracks), default=1)
# 2. Load existing tracks for this group
db = deps.get_database()
existing_tracks = db.get_retag_tracks(group_id)
if not existing_tracks:
raise ValueError(f"No tracks found for retag group {group_id}")
with deps.retag_lock:
deps.retag_state['total_tracks'] = len(existing_tracks)
deps.retag_state['phase'] = "Matching tracks..."
# 3. Match existing files to new tracklist
matched_pairs = []
for existing_track in existing_tracks:
best_match = None
best_score = 0
# Priority 1: Match by track number
for st in spotify_tracks:
if (st['track_number'] == existing_track.get('track_number') and
st['disc_number'] == existing_track.get('disc_number', 1)):
best_match = st
best_score = 1.0
break
# Priority 2: Match by title similarity
if not best_match:
from difflib import SequenceMatcher
existing_title = (existing_track.get('title') or '').lower().strip()
for st in spotify_tracks:
st_title = (st.get('name') or '').lower().strip()
score = SequenceMatcher(None, existing_title, st_title).ratio()
if score > best_score and score > 0.6:
best_score = score
best_match = st
if best_match:
matched_pairs.append((existing_track, best_match))
else:
logger.warning(f"[Retag] No match found for track: '{existing_track.get('title')}'")
matched_pairs.append((existing_track, None))
with deps.retag_lock:
deps.retag_state['phase'] = "Retagging files..."
# 4. Retag each matched track
for existing_track, matched_spotify in matched_pairs:
current_file_path = existing_track.get('file_path', '')
track_title = matched_spotify['name'] if matched_spotify else existing_track.get('title', 'Unknown')
with deps.retag_lock:
deps.retag_state['current_track'] = track_title
if not matched_spotify:
with deps.retag_lock:
deps.retag_state['processed'] += 1
deps.retag_state['progress'] = int(deps.retag_state['processed'] / deps.retag_state['total_tracks'] * 100)
continue
# Verify file exists
if not os.path.exists(current_file_path):
logger.warning(f"[Retag] File not found, skipping: {current_file_path}")
with deps.retag_lock:
deps.retag_state['processed'] += 1
deps.retag_state['progress'] = int(deps.retag_state['processed'] / deps.retag_state['total_tracks'] * 100)
continue
# Build synthetic context for _enhance_file_metadata
track_artists = matched_spotify.get('artists', [])
context = {
'original_search_result': {
'spotify_clean_title': matched_spotify['name'],
'spotify_clean_album': new_album_name,
'track_number': matched_spotify['track_number'],
'disc_number': matched_spotify.get('disc_number', 1),
'artists': track_artists,
'title': matched_spotify['name']
},
'spotify_album': {
'id': album_id,
'name': new_album_name,
'release_date': new_release_date,
'total_tracks': total_tracks,
'image_url': new_image_url,
'total_discs': total_discs
},
'track_info': {'id': matched_spotify['id']},
'spotify_artist': new_artist,
'_audio_quality': deps.get_audio_quality_string(current_file_path) or ''
}
album_info = {
'is_album': total_tracks > 1,
'album_name': new_album_name,
'track_number': matched_spotify['track_number'],
'disc_number': matched_spotify.get('disc_number', 1),
'clean_track_name': matched_spotify['name'],
'album_image_url': new_image_url
}
# Re-write metadata tags
try:
deps.enhance_file_metadata(current_file_path, context, new_artist, album_info)
logger.info(f"[Retag] Re-tagged: '{track_title}'")
except Exception as meta_err:
logger.error(f"[Retag] Metadata write failed for '{track_title}': {meta_err}")
# Compute new path and move if different
file_ext = os.path.splitext(current_file_path)[1]
try:
new_path, _ = deps.build_final_path_for_track(context, new_artist, album_info, file_ext)
if os.path.normpath(current_file_path) != os.path.normpath(new_path):
logger.info(f"[Retag] Moving '{os.path.basename(current_file_path)}' -> '{new_path}'")
old_dir = os.path.dirname(current_file_path)
os.makedirs(os.path.dirname(new_path), exist_ok=True)
deps.safe_move_file(current_file_path, new_path)
# Move lyrics sidecar file alongside audio file if it exists
for lyrics_ext in ('.lrc', '.txt'):
old_lyrics = os.path.splitext(current_file_path)[0] + lyrics_ext
if os.path.exists(old_lyrics):
new_lyrics = os.path.splitext(new_path)[0] + lyrics_ext
try:
deps.safe_move_file(old_lyrics, new_lyrics)
logger.info(f"[Retag] Moved {lyrics_ext} file alongside audio")
except Exception as lrc_err:
logger.error(f"[Retag] Failed to move {lyrics_ext} file: {lrc_err}")
# Remove old cover.jpg if directory changed and old dir is now empty of audio
new_dir = os.path.dirname(new_path)
if os.path.normpath(old_dir) != os.path.normpath(new_dir):
old_cover = os.path.join(old_dir, 'cover.jpg')
if os.path.exists(old_cover):
# Check if any audio files remain in old directory
audio_exts = {'.flac', '.mp3', '.m4a', '.ogg', '.opus', '.wav', '.aac'}
remaining_audio = [f for f in os.listdir(old_dir)
if os.path.splitext(f)[1].lower() in audio_exts]
if not remaining_audio:
try:
os.remove(old_cover)
logger.warning("[Retag] Removed orphaned cover.jpg from old directory")
except Exception:
pass
# Cleanup old empty directories
transfer_dir = deps.docker_resolve_path(deps.config_manager.get('soulseek.transfer_path', './Transfer'))
deps.cleanup_empty_directories(transfer_dir, current_file_path)
# Update DB record
db.update_retag_track_path(existing_track['id'], str(new_path))
current_file_path = new_path
else:
logger.warning(f"[Retag] Path unchanged for '{track_title}', no move needed")
except Exception as move_err:
logger.error(f"[Retag] Path/move failed for '{track_title}': {move_err}")
# Download cover art to album directory
try:
deps.download_cover_art(album_info, os.path.dirname(current_file_path), context)
except Exception as cover_err:
logger.error(f"[Retag] Cover art download failed: {cover_err}")
with deps.retag_lock:
deps.retag_state['processed'] += 1
deps.retag_state['progress'] = int(deps.retag_state['processed'] / deps.retag_state['total_tracks'] * 100)
# 5. Update the retag group record with new metadata
update_kwargs = {
'artist_name': new_artist.get('name', 'Unknown Artist'),
'album_name': new_album_name,
'image_url': new_image_url,
'total_tracks': total_tracks,
'release_date': new_release_date
}
# Set the correct ID field based on Spotify vs iTunes
if str(album_id).isdigit():
update_kwargs['itunes_album_id'] = album_id
update_kwargs['spotify_album_id'] = None
else:
update_kwargs['spotify_album_id'] = album_id
update_kwargs['itunes_album_id'] = None
db.update_retag_group(group_id, **update_kwargs)
with deps.retag_lock:
deps.retag_state.update({
"status": "finished",
"phase": "Retag complete!",
"progress": 100,
"current_track": ""
})
logger.info(f"[Retag] Retag operation complete for group {group_id}")
except Exception as e:
import traceback
logger.error(f"[Retag] Error during retag: {e}")
logger.error(traceback.format_exc())
with deps.retag_lock:
deps.retag_state.update({
"status": "error",
"phase": "Error",
"error_message": str(e)
})

0
tests/library/__init__.py Normal file
View file

321
tests/library/test_retag.py Normal file
View file

@ -0,0 +1,321 @@
"""Tests for core/library/retag.py — retag worker."""
from __future__ import annotations
import threading
from dataclasses import dataclass
import pytest
from core.library import retag as ret
# ---------------------------------------------------------------------------
# Fakes
# ---------------------------------------------------------------------------
class _FakeSpotify:
def __init__(self, album=None, tracks=None):
self._album = album
self._tracks = tracks
def get_album(self, album_id):
return self._album
def get_album_tracks(self, album_id):
return self._tracks
class _FakeDB:
def __init__(self, retag_tracks=None):
self._tracks = retag_tracks or []
self.path_updates = []
self.group_updates = []
def get_retag_tracks(self, group_id):
return self._tracks
def update_retag_track_path(self, track_id, new_path):
self.path_updates.append((track_id, new_path))
def update_retag_group(self, group_id, **kwargs):
self.group_updates.append((group_id, kwargs))
def _build_deps(
*,
spotify_album=None,
spotify_tracks=None,
retag_tracks=None,
state=None,
enhance_calls=None,
move_calls=None,
cover_calls=None,
build_path_result=None,
):
state = state if state is not None else {}
enhance_calls = enhance_calls if enhance_calls is not None else []
move_calls = move_calls if move_calls is not None else []
cover_calls = cover_calls if cover_calls is not None else []
db = _FakeDB(retag_tracks=retag_tracks or [])
deps = ret.RetagDeps(
config_manager=type('C', (), {'get': lambda self, k, d=None: d})(),
retag_lock=threading.Lock(),
spotify_client=_FakeSpotify(album=spotify_album, tracks=spotify_tracks),
get_audio_quality_string=lambda fp: 'FLAC 16bit',
enhance_file_metadata=lambda fp, ctx, artist, ai: enhance_calls.append((fp, ctx, artist, ai)),
build_final_path_for_track=lambda ctx, artist, ai, ext: (
(build_path_result if build_path_result is not None else ctx['original_search_result']['title'] + ext),
True,
),
safe_move_file=lambda src, dst: move_calls.append((src, dst)),
cleanup_empty_directories=lambda transfer_dir, file_path: None,
download_cover_art=lambda ai, dest_dir, ctx: cover_calls.append((ai, dest_dir)),
docker_resolve_path=lambda p: p,
_get_retag_state=lambda: state,
_set_retag_state=lambda v: state.clear() or state.update(v),
get_database=lambda: db,
)
deps._db = db
deps._state = state
deps._enhance_calls = enhance_calls
deps._move_calls = move_calls
deps._cover_calls = cover_calls
return deps
# ---------------------------------------------------------------------------
# Setup error paths
# ---------------------------------------------------------------------------
def test_no_album_data_marks_state_error(tmp_path):
"""spotify.get_album returning None → state.error_message set, state status='error'."""
deps = _build_deps(spotify_album=None)
ret.execute_retag('g1', 'alb-1', deps)
assert deps._state['status'] == 'error'
assert 'Could not fetch album' in deps._state['error_message']
def test_no_album_tracks_marks_state_error():
"""spotify.get_album_tracks returning None → error state."""
deps = _build_deps(spotify_album={'name': 'A', 'artists': []}, spotify_tracks=None)
ret.execute_retag('g1', 'alb-1', deps)
assert deps._state['status'] == 'error'
def test_no_existing_tracks_marks_state_error():
"""retag_group has no tracks → error state."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [], 'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'T1', 'track_number': 1, 'disc_number': 1, 'id': 'sp1', 'artists': [], 'duration_ms': 1000}]},
retag_tracks=[],
)
ret.execute_retag('g1', 'alb-1', deps)
assert deps._state['status'] == 'error'
assert 'No tracks found' in deps._state['error_message']
# ---------------------------------------------------------------------------
# Successful retag — track-number match
# ---------------------------------------------------------------------------
def test_track_number_match_priority_1(tmp_path):
"""Existing track with matching track+disc number → matched even if title differs."""
src_file = tmp_path / 'old.flac'
src_file.touch()
deps = _build_deps(
spotify_album={'name': 'New Album', 'artists': [{'name': 'Artist A', 'id': 'a1'}],
'images': [{'url': 'http://img'}], 'release_date': '2024-01-01', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'Brand New Title', 'track_number': 5,
'disc_number': 1, 'id': 'sp5', 'artists': [{'name': 'X'}],
'duration_ms': 1000}]},
retag_tracks=[{
'id': 1,
'title': 'Completely Unrelated Old Name',
'track_number': 5,
'disc_number': 1,
'file_path': str(src_file),
}],
build_path_result=str(tmp_path / 'new.flac'),
)
ret.execute_retag('g1', 'alb-x', deps)
# Match found via priority 1 (track number) — enhance_file_metadata called
assert len(deps._enhance_calls) == 1
fp, ctx, artist, _ai = deps._enhance_calls[0]
assert fp == str(src_file)
assert ctx['original_search_result']['spotify_clean_title'] == 'Brand New Title'
# State marks finished
assert deps._state['status'] == 'finished'
assert deps._state['progress'] == 100
# ---------------------------------------------------------------------------
# Title-similarity fallback (priority 2)
# ---------------------------------------------------------------------------
def test_title_similarity_fallback_when_no_track_number_match():
"""No track-number match → falls back to fuzzy title match."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'Hello World', 'track_number': 99,
'disc_number': 99, 'id': 'sp1', 'artists': [],
'duration_ms': 1000}]},
retag_tracks=[{
'id': 1,
'title': 'Hello World', # title matches
'track_number': 1, 'disc_number': 1, # but numbers don't
'file_path': '/nonexistent/old.flac',
}],
)
ret.execute_retag('g1', 'alb-x', deps)
# File doesn't exist so enhance is skipped, but match was made
# (state.processed == 1 confirms loop iterated)
assert deps._state['processed'] == 1
def test_no_match_skips_track():
"""No track-number AND title similarity below 0.6 → no match, no enhance call."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'Completely Different', 'track_number': 99,
'disc_number': 99, 'id': 'sp1', 'artists': [],
'duration_ms': 1000}]},
retag_tracks=[{
'id': 1,
'title': 'Hello World',
'track_number': 1, 'disc_number': 1,
'file_path': '/nonexistent/old.flac',
}],
)
ret.execute_retag('g1', 'alb-x', deps)
# No match, no enhance call
assert deps._enhance_calls == []
assert deps._state['processed'] == 1
# ---------------------------------------------------------------------------
# Missing file
# ---------------------------------------------------------------------------
def test_missing_file_skipped(tmp_path):
"""If the audio file doesn't exist, enhance_file_metadata is NOT called."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'T1', 'track_number': 1, 'disc_number': 1,
'id': 'sp1', 'artists': [], 'duration_ms': 1000}]},
retag_tracks=[{
'id': 1, 'title': 'T1', 'track_number': 1, 'disc_number': 1,
'file_path': '/this/path/does/not/exist.flac',
}],
)
ret.execute_retag('g1', 'alb-x', deps)
assert deps._enhance_calls == []
assert deps._state['status'] == 'finished'
# ---------------------------------------------------------------------------
# Path move
# ---------------------------------------------------------------------------
def test_file_moved_when_path_changes(tmp_path):
"""When build_final_path_for_track returns a different path, file is moved."""
src_file = tmp_path / 'old.flac'
src_file.touch()
new_path = str(tmp_path / 'subdir' / 'new.flac')
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'T1', 'track_number': 1, 'disc_number': 1,
'id': 'sp1', 'artists': [], 'duration_ms': 1000}]},
retag_tracks=[{
'id': 1, 'title': 'T1', 'track_number': 1, 'disc_number': 1,
'file_path': str(src_file),
}],
build_path_result=new_path,
)
ret.execute_retag('g1', 'alb-x', deps)
assert len(deps._move_calls) == 1
assert deps._move_calls[0] == (str(src_file), new_path)
assert (1, new_path) in deps._db.path_updates
# ---------------------------------------------------------------------------
# Group record update
# ---------------------------------------------------------------------------
def test_spotify_album_id_used_for_alphanumeric_id():
"""Non-numeric album IDs → spotify_album_id set, itunes_album_id None."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'T1', 'track_number': 1, 'disc_number': 1,
'id': 'sp1', 'artists': [], 'duration_ms': 1000}]},
retag_tracks=[{
'id': 1, 'title': 'T1', 'track_number': 1, 'disc_number': 1,
'file_path': '/missing.flac',
}],
)
ret.execute_retag('g1', 'spotify_alpha_id_xyz', deps)
assert len(deps._db.group_updates) == 1
_gid, kwargs = deps._db.group_updates[0]
assert kwargs['spotify_album_id'] == 'spotify_alpha_id_xyz'
assert kwargs['itunes_album_id'] is None
def test_itunes_album_id_used_for_numeric_id():
"""Numeric album IDs → itunes_album_id set, spotify_album_id None."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 1},
spotify_tracks={'items': [{'name': 'T1', 'track_number': 1, 'disc_number': 1,
'id': 'sp1', 'artists': [], 'duration_ms': 1000}]},
retag_tracks=[{
'id': 1, 'title': 'T1', 'track_number': 1, 'disc_number': 1,
'file_path': '/missing.flac',
}],
)
ret.execute_retag('g1', '987654321', deps)
_gid, kwargs = deps._db.group_updates[0]
assert kwargs['itunes_album_id'] == '987654321'
assert kwargs['spotify_album_id'] is None
# ---------------------------------------------------------------------------
# Multi-disc detection
# ---------------------------------------------------------------------------
def test_multi_disc_total_discs_computed():
"""total_discs derived from max disc_number across all spotify tracks."""
deps = _build_deps(
spotify_album={'name': 'A', 'artists': [{'name': 'X', 'id': '1'}], 'images': [],
'release_date': '', 'total_tracks': 3},
spotify_tracks={'items': [
{'name': 'T1', 'track_number': 1, 'disc_number': 1, 'id': 'sp1', 'artists': [], 'duration_ms': 1000},
{'name': 'T2', 'track_number': 1, 'disc_number': 2, 'id': 'sp2', 'artists': [], 'duration_ms': 1000},
{'name': 'T3', 'track_number': 1, 'disc_number': 3, 'id': 'sp3', 'artists': [], 'duration_ms': 1000},
]},
retag_tracks=[{
'id': 1, 'title': 'T1', 'track_number': 1, 'disc_number': 1,
'file_path': '/missing.flac',
}],
)
ret.execute_retag('g1', 'alb-x', deps)
# Verify multi-disc reflected in retag — group update has total_tracks
# (total_discs not stored on group; check via state instead)
assert deps._state['status'] == 'finished'

288
web_server.py
View file

@ -16878,264 +16878,42 @@ _download_retry_attempts = {} # {context_key: {'count': N, 'first_attempt': tim
_download_retry_max = 10 # Max retries before giving up (10 seconds with 1s poll interval)
_download_retry_lock = threading.Lock()
# Retag worker logic lives in core/library/retag.py.
from core.library import retag as _library_retag
def _build_retag_deps():
"""Build the RetagDeps bundle from web_server.py globals on each call."""
from database.music_database import get_database as _get_db
def _get_state():
return retag_state
def _set_state(value):
global retag_state
retag_state = value
return _library_retag.RetagDeps(
config_manager=config_manager,
retag_lock=retag_lock,
spotify_client=spotify_client,
get_audio_quality_string=_get_audio_quality_string,
enhance_file_metadata=_enhance_file_metadata,
build_final_path_for_track=_build_final_path_for_track,
safe_move_file=_safe_move_file,
cleanup_empty_directories=_cleanup_empty_directories,
download_cover_art=_download_cover_art,
docker_resolve_path=docker_resolve_path,
_get_retag_state=_get_state,
_set_retag_state=_set_state,
get_database=_get_db,
)
def _execute_retag(group_id, album_id):
"""Execute a retag operation: re-tag files in a group with metadata from a new album match."""
global retag_state
from database.music_database import get_database
return _library_retag.execute_retag(group_id, album_id, _build_retag_deps())
try:
with retag_lock:
retag_state.update({
"status": "running",
"phase": "Fetching album metadata...",
"progress": 0,
"current_track": "",
"total_tracks": 0,
"processed": 0,
"error_message": ""
})
# 1. Fetch new album metadata from Spotify/iTunes
album_data = spotify_client.get_album(album_id)
if not album_data:
raise ValueError(f"Could not fetch album data for ID: {album_id}")
album_tracks_response = spotify_client.get_album_tracks(album_id)
if not album_tracks_response:
raise ValueError(f"Could not fetch album tracks for ID: {album_id}")
album_tracks_items = album_tracks_response.get('items', [])
# Extract artist info
album_artists = album_data.get('artists', [])
new_artist = album_artists[0] if album_artists else {'name': 'Unknown Artist', 'id': ''}
# Ensure artist is a dict with expected fields
if not isinstance(new_artist, dict):
new_artist = {'name': str(new_artist), 'id': ''}
new_album_name = album_data.get('name', 'Unknown Album')
new_images = album_data.get('images', [])
new_image_url = new_images[0]['url'] if new_images else None
new_release_date = album_data.get('release_date', '')
total_tracks = album_data.get('total_tracks', len(album_tracks_items))
# Build spotify track list
spotify_tracks = []
for item in album_tracks_items:
track_artists = item.get('artists', [])
spotify_tracks.append({
'name': item.get('name', ''),
'track_number': item.get('track_number', 1),
'disc_number': item.get('disc_number', 1),
'id': item.get('id', ''),
'artists': track_artists,
'duration_ms': item.get('duration_ms', 0)
})
total_discs = max((t['disc_number'] for t in spotify_tracks), default=1)
# 2. Load existing tracks for this group
db = get_database()
existing_tracks = db.get_retag_tracks(group_id)
if not existing_tracks:
raise ValueError(f"No tracks found for retag group {group_id}")
with retag_lock:
retag_state['total_tracks'] = len(existing_tracks)
retag_state['phase'] = "Matching tracks..."
# 3. Match existing files to new tracklist
matched_pairs = []
for existing_track in existing_tracks:
best_match = None
best_score = 0
# Priority 1: Match by track number
for st in spotify_tracks:
if (st['track_number'] == existing_track.get('track_number') and
st['disc_number'] == existing_track.get('disc_number', 1)):
best_match = st
best_score = 1.0
break
# Priority 2: Match by title similarity
if not best_match:
from difflib import SequenceMatcher
existing_title = (existing_track.get('title') or '').lower().strip()
for st in spotify_tracks:
st_title = (st.get('name') or '').lower().strip()
score = SequenceMatcher(None, existing_title, st_title).ratio()
if score > best_score and score > 0.6:
best_score = score
best_match = st
if best_match:
matched_pairs.append((existing_track, best_match))
else:
logger.warning(f"[Retag] No match found for track: '{existing_track.get('title')}'")
matched_pairs.append((existing_track, None))
with retag_lock:
retag_state['phase'] = "Retagging files..."
# 4. Retag each matched track
for existing_track, matched_spotify in matched_pairs:
current_file_path = existing_track.get('file_path', '')
track_title = matched_spotify['name'] if matched_spotify else existing_track.get('title', 'Unknown')
with retag_lock:
retag_state['current_track'] = track_title
if not matched_spotify:
with retag_lock:
retag_state['processed'] += 1
retag_state['progress'] = int(retag_state['processed'] / retag_state['total_tracks'] * 100)
continue
# Verify file exists
if not os.path.exists(current_file_path):
logger.warning(f"[Retag] File not found, skipping: {current_file_path}")
with retag_lock:
retag_state['processed'] += 1
retag_state['progress'] = int(retag_state['processed'] / retag_state['total_tracks'] * 100)
continue
# Build synthetic context for _enhance_file_metadata
track_artists = matched_spotify.get('artists', [])
context = {
'original_search_result': {
'spotify_clean_title': matched_spotify['name'],
'spotify_clean_album': new_album_name,
'track_number': matched_spotify['track_number'],
'disc_number': matched_spotify.get('disc_number', 1),
'artists': track_artists,
'title': matched_spotify['name']
},
'spotify_album': {
'id': album_id,
'name': new_album_name,
'release_date': new_release_date,
'total_tracks': total_tracks,
'image_url': new_image_url,
'total_discs': total_discs
},
'track_info': {'id': matched_spotify['id']},
'spotify_artist': new_artist,
'_audio_quality': _get_audio_quality_string(current_file_path) or ''
}
album_info = {
'is_album': total_tracks > 1,
'album_name': new_album_name,
'track_number': matched_spotify['track_number'],
'disc_number': matched_spotify.get('disc_number', 1),
'clean_track_name': matched_spotify['name'],
'album_image_url': new_image_url
}
# Re-write metadata tags
try:
_enhance_file_metadata(current_file_path, context, new_artist, album_info)
logger.info(f"[Retag] Re-tagged: '{track_title}'")
except Exception as meta_err:
logger.error(f"[Retag] Metadata write failed for '{track_title}': {meta_err}")
# Compute new path and move if different
file_ext = os.path.splitext(current_file_path)[1]
try:
new_path, _ = _build_final_path_for_track(context, new_artist, album_info, file_ext)
if os.path.normpath(current_file_path) != os.path.normpath(new_path):
logger.info(f"[Retag] Moving '{os.path.basename(current_file_path)}' -> '{new_path}'")
old_dir = os.path.dirname(current_file_path)
os.makedirs(os.path.dirname(new_path), exist_ok=True)
_safe_move_file(current_file_path, new_path)
# Move lyrics sidecar file alongside audio file if it exists
for lyrics_ext in ('.lrc', '.txt'):
old_lyrics = os.path.splitext(current_file_path)[0] + lyrics_ext
if os.path.exists(old_lyrics):
new_lyrics = os.path.splitext(new_path)[0] + lyrics_ext
try:
_safe_move_file(old_lyrics, new_lyrics)
logger.info(f"[Retag] Moved {lyrics_ext} file alongside audio")
except Exception as lrc_err:
logger.error(f"[Retag] Failed to move {lyrics_ext} file: {lrc_err}")
# Remove old cover.jpg if directory changed and old dir is now empty of audio
new_dir = os.path.dirname(new_path)
if os.path.normpath(old_dir) != os.path.normpath(new_dir):
old_cover = os.path.join(old_dir, 'cover.jpg')
if os.path.exists(old_cover):
# Check if any audio files remain in old directory
audio_exts = {'.flac', '.mp3', '.m4a', '.ogg', '.opus', '.wav', '.aac'}
remaining_audio = [f for f in os.listdir(old_dir)
if os.path.splitext(f)[1].lower() in audio_exts]
if not remaining_audio:
try:
os.remove(old_cover)
logger.warning("[Retag] Removed orphaned cover.jpg from old directory")
except Exception:
pass
# Cleanup old empty directories
transfer_dir = docker_resolve_path(config_manager.get('soulseek.transfer_path', './Transfer'))
_cleanup_empty_directories(transfer_dir, current_file_path)
# Update DB record
db.update_retag_track_path(existing_track['id'], str(new_path))
current_file_path = new_path
else:
logger.warning(f"[Retag] Path unchanged for '{track_title}', no move needed")
except Exception as move_err:
logger.error(f"[Retag] Path/move failed for '{track_title}': {move_err}")
# Download cover art to album directory
try:
_download_cover_art(album_info, os.path.dirname(current_file_path), context)
except Exception as cover_err:
logger.error(f"[Retag] Cover art download failed: {cover_err}")
with retag_lock:
retag_state['processed'] += 1
retag_state['progress'] = int(retag_state['processed'] / retag_state['total_tracks'] * 100)
# 5. Update the retag group record with new metadata
update_kwargs = {
'artist_name': new_artist.get('name', 'Unknown Artist'),
'album_name': new_album_name,
'image_url': new_image_url,
'total_tracks': total_tracks,
'release_date': new_release_date
}
# Set the correct ID field based on Spotify vs iTunes
if str(album_id).isdigit():
update_kwargs['itunes_album_id'] = album_id
update_kwargs['spotify_album_id'] = None
else:
update_kwargs['spotify_album_id'] = album_id
update_kwargs['itunes_album_id'] = None
db.update_retag_group(group_id, **update_kwargs)
with retag_lock:
retag_state.update({
"status": "finished",
"phase": "Retag complete!",
"progress": 100,
"current_track": ""
})
logger.info(f"[Retag] Retag operation complete for group {group_id}")
except Exception as e:
import traceback
logger.error(f"[Retag] Error during retag: {e}")
logger.error(traceback.format_exc())
with retag_lock:
retag_state.update({
"status": "error",
"phase": "Error",
"error_message": str(e)
})
def _automatic_wishlist_cleanup_after_db_update():
"""Automatic wishlist cleanup that runs after database updates."""
return _cleanup_wishlist_after_db_update(logger=logger)