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

Move shared metadata helpers into package

Browse source 
- Relocate the shared metadata helper module from core/metadata_common.py into core/metadata/common.py.
- Update the new metadata package, the import pipeline, and the web entrypoint to use the package-scoped helper.
- Keep the shared config, mutagen, file-lock, and tag-writing helpers centralized without touching unrelated files.
This commit is contained in:
Antti Kettunen 2026-04-25 14:32:17 +03:00
parent 9656dbd46a
commit 9e496397da
No known key found for this signature in database
GPG key ID: C6B2A3D250359BD7
7 changed files with 45 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
core/imports/pipeline.py
View file

@ -52,7 +52,7 @@ from core.runtime_state import (
tasks_lock, tasks_lock,
) )
from core.metadata.artwork import download_cover_art from core.metadata.artwork import download_cover_art
from core.metadata_common import wipe_source_tags from core.metadata.common import wipe_source_tags
from core.metadata.enrichment import enhance_file_metadata from core.metadata.enrichment import enhance_file_metadata
from core.imports.paths import ( from core.imports.paths import (
build_final_path_for_track, build_final_path_for_track,

27
core/metadata/artwork.py
View file

@ -7,12 +7,12 @@ import re
import urllib.request import urllib.request
from core.imports.context import get_import_context_album from core.imports.context import get_import_context_album
from core.metadata_common import ( from core.metadata.common import (
get_config_manager, get_config_manager,
get_image_dimensions, get_image_dimensions,
get_logger,
get_mutagen_symbols, get_mutagen_symbols,
) )
from utils.logging_config import get_logger as _create_logger
__all__ = [ __all__ = [
"embed_album_art_metadata", "embed_album_art_metadata",
@ -20,9 +20,11 @@ __all__ = [
] ]
logger = _create_logger("metadata.artwork")
def embed_album_art_metadata(audio_file, metadata: dict): def embed_album_art_metadata(audio_file, metadata: dict):
cfg = get_config_manager() cfg = get_config_manager()
logger_ = get_logger()
symbols = get_mutagen_symbols() symbols = get_mutagen_symbols()
if not symbols: if not symbols:
return return
@ -47,14 +49,14 @@ def embed_album_art_metadata(audio_file, metadata: dict):
if not image_data: if not image_data:
art_url = metadata.get("album_art_url") art_url = metadata.get("album_art_url")
if not art_url: if not art_url:
logger_.warning("No album art URL available for embedding.") logger.warning("No album art URL available for embedding.")
return return
with urllib.request.urlopen(art_url, timeout=10) as response: with urllib.request.urlopen(art_url, timeout=10) as response:
image_data = response.read() image_data = response.read()
mime_type = response.info().get_content_type() or "image/jpeg" mime_type = response.info().get_content_type() or "image/jpeg"
if not image_data: if not image_data:
logger_.error("Failed to download album art data.") logger.error("Failed to download album art data.")
return return
if isinstance(audio_file.tags, symbols.ID3): if isinstance(audio_file.tags, symbols.ID3):
@ -73,14 +75,13 @@ def embed_album_art_metadata(audio_file, metadata: dict):
fmt = symbols.MP4Cover.FORMAT_JPEG if "jpeg" in mime_type else symbols.MP4Cover.FORMAT_PNG fmt = symbols.MP4Cover.FORMAT_JPEG if "jpeg" in mime_type else symbols.MP4Cover.FORMAT_PNG
audio_file["covr"] = [symbols.MP4Cover(image_data, imageformat=fmt)] audio_file["covr"] = [symbols.MP4Cover(image_data, imageformat=fmt)]
logger_.info("Album art successfully embedded.") logger.info("Album art successfully embedded.")
except Exception as exc: except Exception as exc:
logger_.error("Error embedding album art: %s", exc) logger.error("Error embedding album art: %s", exc)
def download_cover_art(album_info: dict, target_dir: str, context: dict = None): def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
cfg = get_config_manager() cfg = get_config_manager()
logger_ = get_logger()
if cfg.get("metadata_enhancement.cover_art_download", True) is False: if cfg.get("metadata_enhancement.cover_art_download", True) is False:
return return
@ -117,7 +118,7 @@ def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
image_data = None image_data = None
if is_upgrade and not image_data: if is_upgrade and not image_data:
logger_.error("CAA upgrade failed - keeping existing cover.jpg") logger.error("CAA upgrade failed - keeping existing cover.jpg")
return return
if not image_data: if not image_data:
@ -130,7 +131,7 @@ def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
if images and isinstance(images[0], dict): if images and isinstance(images[0], dict):
art_url = images[0].get("url", "") art_url = images[0].get("url", "")
if art_url: if art_url:
logger_.info("Using cover art URL from album context") logger.info("Using cover art URL from album context")
if art_url and "i.scdn.co" in art_url: if art_url and "i.scdn.co" in art_url:
try: try:
from core.spotify_client import _upgrade_spotify_image_url from core.spotify_client import _upgrade_spotify_image_url
@ -141,7 +142,7 @@ def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
elif art_url and "mzstatic.com" in art_url: elif art_url and "mzstatic.com" in art_url:
art_url = re.sub(r"\d+x\d+bb", "3000x3000bb", art_url) art_url = re.sub(r"\d+x\d+bb", "3000x3000bb", art_url)
if not art_url: if not art_url:
logger_.warning("No cover art URL available for download.") logger.warning("No cover art URL available for download.")
return return
with urllib.request.urlopen(art_url, timeout=10) as response: with urllib.request.urlopen(art_url, timeout=10) as response:
image_data = response.read() image_data = response.read()
@ -151,6 +152,6 @@ def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
with open(cover_path, "wb") as handle: with open(cover_path, "wb") as handle:
handle.write(image_data) handle.write(image_data)
logger_.info("Cover art downloaded to: %s", cover_path) logger.info("Cover art downloaded to: %s", cover_path)
except Exception as exc: except Exception as exc:
logger_.error("Error downloading cover.jpg: %s", exc) logger.error("Error downloading cover.jpg: %s", exc)

2
core/metadata_common.py → core/metadata/common.py
View file

@ -10,7 +10,7 @@ from typing import Any, Dict
from utils.logging_config import get_logger as _create_logger from utils.logging_config import get_logger as _create_logger
logger = _create_logger("metadata_common") logger = _create_logger("metadata.common")
__all__ = [ __all__ = [
"get_logger", "get_logger",

36
core/metadata/enrichment.py
View file

@ -5,10 +5,9 @@ from __future__ import annotations
import os import os
from core.metadata.artwork import embed_album_art_metadata from core.metadata.artwork import embed_album_art_metadata
from core.metadata_common import ( from core.metadata.common import (
get_config_manager, get_config_manager,
get_file_lock, get_file_lock,
get_logger,
get_mutagen_symbols, get_mutagen_symbols,
is_vorbis_like, is_vorbis_like,
save_audio_file, save_audio_file,
@ -16,6 +15,7 @@ from core.metadata_common import (
verify_metadata_written, verify_metadata_written,
) )
from core.metadata.source import embed_source_ids, extract_source_metadata from core.metadata.source import embed_source_ids, extract_source_metadata
from utils.logging_config import get_logger as _create_logger
__all__ = [ __all__ = [
@ -25,11 +25,13 @@ __all__ = [
] ]
logger = _create_logger("metadata.enrichment")
def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_info: dict, runtime=None) -> bool: def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_info: dict, runtime=None) -> bool:
cfg = get_config_manager() cfg = get_config_manager()
logger_ = get_logger()
if cfg.get("metadata_enhancement.enabled", True) is False: if cfg.get("metadata_enhancement.enabled", True) is False:
logger_.warning("Metadata enhancement disabled in config.") logger.warning("Metadata enhancement disabled in config.")
return True return True
if album_info is None: if album_info is None:
@ -37,17 +39,17 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
symbols = get_mutagen_symbols() symbols = get_mutagen_symbols()
if not symbols: if not symbols:
logger_.error("Mutagen is unavailable, cannot enhance metadata.") logger.error("Mutagen is unavailable, cannot enhance metadata.")
return False return False
file_lock = get_file_lock(file_path) file_lock = get_file_lock(file_path)
with file_lock: with file_lock:
logger_.info("Enhancing metadata for: %s", os.path.basename(file_path)) logger.info("Enhancing metadata for: %s", os.path.basename(file_path))
try: try:
strip_all_non_audio_tags(file_path) strip_all_non_audio_tags(file_path)
audio_file = symbols.File(file_path) audio_file = symbols.File(file_path)
if audio_file is None: if audio_file is None:
logger_.error("Could not load audio file with Mutagen: %s", file_path) logger.error("Could not load audio file with Mutagen: %s", file_path)
return False return False
if hasattr(audio_file, "clear_pictures"): if hasattr(audio_file, "clear_pictures"):
@ -56,7 +58,7 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
if audio_file.tags is not None: if audio_file.tags is not None:
if len(audio_file.tags) > 0: if len(audio_file.tags) > 0:
tag_keys = list(audio_file.tags.keys())[:15] tag_keys = list(audio_file.tags.keys())[:15]
logger_.info("Clearing %s existing tags: %s", len(audio_file.tags), ", ".join(str(k) for k in tag_keys)) logger.info("Clearing %s existing tags: %s", len(audio_file.tags), ", ".join(str(k) for k in tag_keys))
audio_file.tags.clear() audio_file.tags.clear()
else: else:
audio_file.add_tags() audio_file.add_tags()
@ -65,7 +67,7 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
metadata = extract_source_metadata(context, artist, album_info) metadata = extract_source_metadata(context, artist, album_info)
if not metadata: if not metadata:
logger_.error("Could not extract source metadata, saving with cleared tags.") logger.error("Could not extract source metadata, saving with cleared tags.")
save_audio_file(audio_file, symbols) save_audio_file(audio_file, symbols)
return True return True
@ -147,17 +149,17 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
verified = verify_metadata_written(file_path) verified = verify_metadata_written(file_path)
if verified: if verified:
logger_.info("Metadata enhanced successfully.") logger.info("Metadata enhanced successfully.")
else: else:
logger_.info("Metadata saved but verification found issues (see above).") logger.info("Metadata saved but verification found issues (see above).")
return True return True
except Exception as exc: except Exception as exc:
import traceback import traceback
logger_.error("Error enhancing metadata for %s: %s", file_path, exc) logger.error("Error enhancing metadata for %s: %s", file_path, exc)
logger_.error("[Metadata Debug] Exception type: %s", type(exc).__name__) logger.error("[Metadata Debug] Exception type: %s", type(exc).__name__)
logger_.info("[Metadata Debug] File exists: %s", os.path.exists(file_path)) logger.info("[Metadata Debug] File exists: %s", os.path.exists(file_path))
logger_.warning("[Metadata Debug] Artist: %s", artist.get("name", "MISSING") if artist else "None") logger.warning("[Metadata Debug] Artist: %s", artist.get("name", "MISSING") if artist else "None")
logger_.warning("[Metadata Debug] Album info: %s", album_info.get("album_name", "MISSING") if album_info else "None") logger.warning("[Metadata Debug] Album info: %s", album_info.get("album_name", "MISSING") if album_info else "None")
logger_.error("[Metadata Debug] Traceback:\n%s", traceback.format_exc()) logger.error("[Metadata Debug] Traceback:\n%s", traceback.format_exc())
return False return False

13
core/metadata/lyrics.py
View file

@ -9,16 +9,19 @@ from core.imports.context import (
get_import_original_search, get_import_original_search,
normalize_import_context, normalize_import_context,
) )
from core.metadata_common import get_config_manager, get_logger from core.metadata.common import get_config_manager
from utils.logging_config import get_logger as _create_logger
__all__ = [ __all__ = [
"generate_lrc_file", "generate_lrc_file",
] ]
logger = _create_logger("metadata.lyrics")
def generate_lrc_file(file_path: str, context: dict, artist: dict, album_info: dict) -> bool: def generate_lrc_file(file_path: str, context: dict, artist: dict, album_info: dict) -> bool:
cfg = get_config_manager() cfg = get_config_manager()
logger_ = get_logger()
if cfg.get("metadata_enhancement.lrclib_enabled", True) is False: if cfg.get("metadata_enhancement.lrclib_enabled", True) is False:
return False return False
@ -58,10 +61,10 @@ def generate_lrc_file(file_path: str, context: dict, artist: dict, album_info: d
) )
if success: if success:
logger_.info("LRC file generated for: %s", track_name) logger.info("LRC file generated for: %s", track_name)
else: else:
logger_.warning("No lyrics found for: %s", track_name) logger.warning("No lyrics found for: %s", track_name)
return success return success
except Exception as exc: except Exception as exc:
logger_.error("Error generating LRC file for %s: %s", file_path, exc) logger.error("Error generating LRC file for %s: %s", file_path, exc)
return False return False

2
core/metadata/source.py
View file

@ -21,7 +21,7 @@ from core.imports.context import (
) )
from core.metadata_service import get_itunes_client from core.metadata_service import get_itunes_client
from database.music_database import get_database from database.music_database import get_database
from core.metadata_common import ( from core.metadata.common import (
get_config_manager, get_config_manager,
get_mutagen_symbols, get_mutagen_symbols,
is_vorbis_like, is_vorbis_like,

2
web_server.py
View file

@ -132,7 +132,7 @@ from core.imports.staging import (
start_import_suggestions_cache, start_import_suggestions_cache,
) )
from core.imports.paths import build_final_path_for_track as _build_final_path_for_track from core.imports.paths import build_final_path_for_track as _build_final_path_for_track
from core.metadata_common import get_file_lock from core.metadata.common import get_file_lock
from core.metadata.source import ( from core.metadata.source import (
mb_release_cache, mb_release_cache,
mb_release_cache_lock, mb_release_cache_lock,