Tighten metadata helper boundaries
- remove stale wrapper helpers from web_server and metadata_common - import provider helpers directly in metadata_source - keep the metadata modules' public surface explicit
This commit is contained in:
Antti Kettunen
parent
edd9048f86
commit
b9269b4f16
7 changed files with 131 additions and 154 deletions
|
|
@ -51,18 +51,16 @@ from core.import_runtime_state import (
|
|||
_processed_download_ids,
|
||||
tasks_lock,
|
||||
)
|
||||
from core.metadata_enrichment import (
|
||||
download_cover_art,
|
||||
enhance_file_metadata,
|
||||
generate_lrc_file,
|
||||
wipe_source_tags,
|
||||
)
|
||||
from core.metadata_artwork import download_cover_art
|
||||
from core.metadata_common import wipe_source_tags
|
||||
from core.metadata_enrichment import enhance_file_metadata
|
||||
from core.import_paths import (
|
||||
build_final_path_for_track,
|
||||
build_simple_download_destination,
|
||||
docker_resolve_path,
|
||||
)
|
||||
from core.import_album_naming import resolve_album_group
|
||||
from core.metadata_lyrics import generate_lrc_file
|
||||
from database.music_database import get_database
|
||||
from utils.logging_config import get_logger
|
||||
|
||||
|
|
|
|||
|
|
@ -8,17 +8,22 @@ import urllib.request
|
|||
|
||||
from core.import_context import get_import_context_album
|
||||
from core.metadata_common import (
|
||||
_get_config_manager,
|
||||
_get_image_dimensions,
|
||||
_get_logger,
|
||||
_get_mutagen_symbols,
|
||||
get_config_manager,
|
||||
get_image_dimensions,
|
||||
get_logger,
|
||||
get_mutagen_symbols,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"embed_album_art_metadata",
|
||||
"download_cover_art",
|
||||
]
|
||||
|
||||
|
||||
def embed_album_art_metadata(audio_file, metadata: dict):
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
symbols = _get_mutagen_symbols()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
return
|
||||
|
||||
|
|
@ -59,7 +64,7 @@ def embed_album_art_metadata(audio_file, metadata: dict):
|
|||
picture.data = image_data
|
||||
picture.type = 3
|
||||
picture.mime = mime_type
|
||||
width, height = _get_image_dimensions(image_data)
|
||||
width, height = get_image_dimensions(image_data)
|
||||
picture.width = width or 640
|
||||
picture.height = height or 640
|
||||
picture.depth = 24
|
||||
|
|
@ -74,8 +79,8 @@ def embed_album_art_metadata(audio_file, metadata: dict):
|
|||
|
||||
|
||||
def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
if cfg.get("metadata_enhancement.cover_art_download", True) is False:
|
||||
return
|
||||
|
||||
|
|
|
|||
|
|
@ -7,10 +7,24 @@ import threading
|
|||
from types import SimpleNamespace
|
||||
from typing import Any, Dict
|
||||
|
||||
from utils.logging_config import get_logger
|
||||
from utils.logging_config import get_logger as _create_logger
|
||||
|
||||
|
||||
logger = get_logger("metadata_enrichment")
|
||||
logger = _create_logger("metadata_common")
|
||||
|
||||
__all__ = [
|
||||
"get_logger",
|
||||
"get_config_manager",
|
||||
"get_mutagen_symbols",
|
||||
"get_file_lock",
|
||||
"is_ogg_opus",
|
||||
"is_vorbis_like",
|
||||
"save_audio_file",
|
||||
"get_image_dimensions",
|
||||
"strip_all_non_audio_tags",
|
||||
"verify_metadata_written",
|
||||
"wipe_source_tags",
|
||||
]
|
||||
|
||||
_FILE_LOCKS: Dict[str, threading.Lock] = {}
|
||||
_FILE_LOCKS_LOCK = threading.Lock()
|
||||
|
|
@ -21,11 +35,11 @@ class _NullConfigManager:
|
|||
return default
|
||||
|
||||
|
||||
def _get_logger():
|
||||
def get_logger():
|
||||
return logger
|
||||
|
||||
|
||||
def _get_config_manager():
|
||||
def get_config_manager():
|
||||
try:
|
||||
from config.settings import config_manager as settings_config_manager
|
||||
|
||||
|
|
@ -34,33 +48,7 @@ def _get_config_manager():
|
|||
return _NullConfigManager()
|
||||
|
||||
|
||||
def _get_database():
|
||||
try:
|
||||
from database.music_database import get_database
|
||||
|
||||
return get_database()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _get_itunes_client():
|
||||
try:
|
||||
from core.metadata_service import get_itunes_client
|
||||
|
||||
return get_itunes_client()
|
||||
except Exception:
|
||||
return None
|
||||
|
||||
|
||||
def _extract_artist_name(artist: Any) -> str:
|
||||
if isinstance(artist, dict):
|
||||
return str(artist.get("name", "") or "")
|
||||
if hasattr(artist, "name"):
|
||||
return str(getattr(artist, "name") or "")
|
||||
return str(artist) if artist else ""
|
||||
|
||||
|
||||
def _get_mutagen_symbols():
|
||||
def get_mutagen_symbols():
|
||||
"""Lazy mutagen import so tests can monkeypatch this without the package installed."""
|
||||
try:
|
||||
from mutagen import File as MutagenFile
|
||||
|
|
@ -128,7 +116,7 @@ def _get_mutagen_symbols():
|
|||
)
|
||||
|
||||
|
||||
def _get_file_lock(file_path: str) -> threading.Lock:
|
||||
def get_file_lock(file_path: str) -> threading.Lock:
|
||||
with _FILE_LOCKS_LOCK:
|
||||
lock = _FILE_LOCKS.get(file_path)
|
||||
if lock is None:
|
||||
|
|
@ -137,21 +125,21 @@ def _get_file_lock(file_path: str) -> threading.Lock:
|
|||
return lock
|
||||
|
||||
|
||||
def _is_ogg_opus(audio_file: Any) -> bool:
|
||||
def is_ogg_opus(audio_file: Any) -> bool:
|
||||
return type(audio_file).__name__ == "OggOpus"
|
||||
|
||||
|
||||
def _is_vorbis_like(audio_file: Any, symbols: Any) -> bool:
|
||||
def is_vorbis_like(audio_file: Any, symbols: Any) -> bool:
|
||||
vorbis_classes = tuple(
|
||||
cls for cls in (
|
||||
getattr(symbols, "FLAC", None),
|
||||
getattr(symbols, "OggVorbis", None),
|
||||
) if cls is not None
|
||||
)
|
||||
return bool(vorbis_classes) and isinstance(audio_file, vorbis_classes) or _is_ogg_opus(audio_file)
|
||||
return bool(vorbis_classes) and isinstance(audio_file, vorbis_classes) or is_ogg_opus(audio_file)
|
||||
|
||||
|
||||
def _save_audio_file(audio_file: Any, symbols: Any) -> None:
|
||||
def save_audio_file(audio_file: Any, symbols: Any) -> None:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.save(v1=0, v2_version=4)
|
||||
elif isinstance(audio_file, symbols.FLAC):
|
||||
|
|
@ -160,7 +148,7 @@ def _save_audio_file(audio_file: Any, symbols: Any) -> None:
|
|||
audio_file.save()
|
||||
|
||||
|
||||
def _get_image_dimensions(data: bytes):
|
||||
def get_image_dimensions(data: bytes):
|
||||
try:
|
||||
if data[:8] == b"\x89PNG\r\n\x1a\n":
|
||||
import struct
|
||||
|
|
@ -185,12 +173,12 @@ def _get_image_dimensions(data: bytes):
|
|||
return None, None
|
||||
|
||||
|
||||
def _strip_all_non_audio_tags(file_path: str) -> dict:
|
||||
def strip_all_non_audio_tags(file_path: str) -> dict:
|
||||
summary = {"apev2_stripped": False, "apev2_tag_count": 0}
|
||||
if os.path.splitext(file_path)[1].lower() != ".mp3":
|
||||
return summary
|
||||
|
||||
symbols = _get_mutagen_symbols()
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
return summary
|
||||
|
||||
|
|
@ -209,8 +197,8 @@ def _strip_all_non_audio_tags(file_path: str) -> dict:
|
|||
return summary
|
||||
|
||||
|
||||
def _verify_metadata_written(file_path: str) -> bool:
|
||||
symbols = _get_mutagen_symbols()
|
||||
def verify_metadata_written(file_path: str) -> bool:
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
return False
|
||||
|
||||
|
|
@ -231,7 +219,7 @@ def _verify_metadata_written(file_path: str) -> bool:
|
|||
return False
|
||||
except symbols.APENoHeaderError:
|
||||
pass
|
||||
elif _is_vorbis_like(check, symbols):
|
||||
elif is_vorbis_like(check, symbols):
|
||||
title_found = bool(check.get("title"))
|
||||
artist_found = bool(check.get("artist"))
|
||||
elif isinstance(check, symbols.MP4):
|
||||
|
|
@ -251,8 +239,8 @@ def _verify_metadata_written(file_path: str) -> bool:
|
|||
|
||||
def wipe_source_tags(file_path: str) -> bool:
|
||||
try:
|
||||
_strip_all_non_audio_tags(file_path)
|
||||
symbols = _get_mutagen_symbols()
|
||||
strip_all_non_audio_tags(file_path)
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
return False
|
||||
|
||||
|
|
@ -267,7 +255,7 @@ def wipe_source_tags(file_path: str) -> bool:
|
|||
else:
|
||||
audio.add_tags()
|
||||
tag_count = 0
|
||||
_save_audio_file(audio, symbols)
|
||||
save_audio_file(audio, symbols)
|
||||
if tag_count > 0:
|
||||
logger.info("[Tag Wipe] Stripped %s source tags from: %s", tag_count, os.path.basename(file_path))
|
||||
return True
|
||||
|
|
|
|||
|
|
@ -4,53 +4,30 @@ from __future__ import annotations
|
|||
|
||||
import os
|
||||
|
||||
from core.metadata_artwork import (
|
||||
download_cover_art as _download_cover_art_impl,
|
||||
embed_album_art_metadata,
|
||||
)
|
||||
from core.metadata_artwork import embed_album_art_metadata
|
||||
from core.metadata_common import (
|
||||
_get_config_manager,
|
||||
_get_file_lock,
|
||||
_get_image_dimensions as _common_get_image_dimensions,
|
||||
_get_logger,
|
||||
_get_mutagen_symbols,
|
||||
_is_ogg_opus as _common_is_ogg_opus,
|
||||
_is_vorbis_like,
|
||||
_save_audio_file,
|
||||
_strip_all_non_audio_tags,
|
||||
_verify_metadata_written,
|
||||
wipe_source_tags as _common_wipe_source_tags,
|
||||
get_config_manager,
|
||||
get_file_lock,
|
||||
get_logger,
|
||||
get_mutagen_symbols,
|
||||
is_vorbis_like,
|
||||
save_audio_file,
|
||||
strip_all_non_audio_tags,
|
||||
verify_metadata_written,
|
||||
)
|
||||
from core.metadata_lyrics import generate_lrc_file as _generate_lrc_file_impl
|
||||
from core.metadata_source import embed_source_ids, extract_source_metadata
|
||||
|
||||
|
||||
logger = _get_logger()
|
||||
|
||||
|
||||
def _get_image_dimensions(data: bytes):
|
||||
return _common_get_image_dimensions(data)
|
||||
|
||||
|
||||
def _is_ogg_opus(audio_file):
|
||||
return _common_is_ogg_opus(audio_file)
|
||||
|
||||
|
||||
def wipe_source_tags(file_path: str) -> bool:
|
||||
return _common_wipe_source_tags(file_path)
|
||||
|
||||
|
||||
def generate_lrc_file(file_path: str, context: dict, artist: dict, album_info: dict) -> bool:
|
||||
return _generate_lrc_file_impl(file_path, context, artist, album_info)
|
||||
|
||||
|
||||
def download_cover_art(album_info: dict, target_dir: str, context: dict = None):
|
||||
return _download_cover_art_impl(album_info, target_dir, context)
|
||||
__all__ = [
|
||||
"enhance_file_metadata",
|
||||
"extract_source_metadata",
|
||||
"embed_source_ids",
|
||||
]
|
||||
|
||||
|
||||
def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_info: dict) -> bool:
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
if cfg.get("metadata_enhancement.enabled", True) is False:
|
||||
logger_.warning("Metadata enhancement disabled in config.")
|
||||
return True
|
||||
|
|
@ -58,16 +35,16 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
|
|||
if album_info is None:
|
||||
album_info = {}
|
||||
|
||||
symbols = _get_mutagen_symbols()
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
logger_.error("Mutagen is unavailable, cannot enhance metadata.")
|
||||
return False
|
||||
|
||||
file_lock = _get_file_lock(file_path)
|
||||
file_lock = get_file_lock(file_path)
|
||||
with file_lock:
|
||||
logger_.info("Enhancing metadata for: %s", os.path.basename(file_path))
|
||||
try:
|
||||
_strip_all_non_audio_tags(file_path)
|
||||
strip_all_non_audio_tags(file_path)
|
||||
audio_file = symbols.File(file_path)
|
||||
if audio_file is None:
|
||||
logger_.error("Could not load audio file with Mutagen: %s", file_path)
|
||||
|
|
@ -84,12 +61,12 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
|
|||
else:
|
||||
audio_file.add_tags()
|
||||
|
||||
_save_audio_file(audio_file, symbols)
|
||||
save_audio_file(audio_file, symbols)
|
||||
|
||||
metadata = extract_source_metadata(context, artist, album_info)
|
||||
if not metadata:
|
||||
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
|
||||
|
||||
track_num_str = f"{metadata.get('track_number', 1)}/{metadata.get('total_tracks', 1)}"
|
||||
|
|
@ -114,7 +91,7 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
|
|||
audio_file.tags.add(symbols.TRCK(encoding=3, text=[track_num_str]))
|
||||
if metadata.get("disc_number"):
|
||||
audio_file.tags.add(symbols.TPOS(encoding=3, text=[str(metadata["disc_number"])]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
if metadata.get("title"):
|
||||
audio_file["title"] = [metadata["title"]]
|
||||
if metadata.get("artist"):
|
||||
|
|
@ -161,14 +138,14 @@ def enhance_file_metadata(file_path: str, context: dict, artist: dict, album_inf
|
|||
if quality and cfg.get("metadata_enhancement.tags.quality_tag", True) is not False:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TXXX(encoding=3, desc="QUALITY", text=[quality]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["quality"] = [quality]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:QUALITY"] = [symbols.MP4FreeForm(quality.encode("utf-8"))]
|
||||
|
||||
_save_audio_file(audio_file, symbols)
|
||||
save_audio_file(audio_file, symbols)
|
||||
|
||||
verified = _verify_metadata_written(file_path)
|
||||
verified = verify_metadata_written(file_path)
|
||||
if verified:
|
||||
logger_.info("Metadata enhanced successfully.")
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -9,12 +9,16 @@ from core.import_context import (
|
|||
get_import_original_search,
|
||||
normalize_import_context,
|
||||
)
|
||||
from core.metadata_common import _get_config_manager, _get_logger
|
||||
from core.metadata_common import get_config_manager, get_logger
|
||||
|
||||
__all__ = [
|
||||
"generate_lrc_file",
|
||||
]
|
||||
|
||||
|
||||
def generate_lrc_file(file_path: str, context: dict, artist: dict, album_info: dict) -> bool:
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
if cfg.get("metadata_enhancement.lrclib_enabled", True) is False:
|
||||
return False
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import time
|
|||
from typing import Any, Dict
|
||||
|
||||
from core.import_context import (
|
||||
extract_artist_name,
|
||||
get_import_clean_artist,
|
||||
get_import_clean_title,
|
||||
get_import_context_album,
|
||||
|
|
@ -18,16 +19,20 @@ from core.import_context import (
|
|||
get_source_tag_names,
|
||||
normalize_import_context,
|
||||
)
|
||||
from core.metadata_service import get_itunes_client
|
||||
from database.music_database import get_database
|
||||
from core.metadata_common import (
|
||||
_extract_artist_name,
|
||||
_get_config_manager,
|
||||
_get_database,
|
||||
_get_itunes_client,
|
||||
_get_logger,
|
||||
_get_mutagen_symbols,
|
||||
_is_vorbis_like,
|
||||
get_config_manager,
|
||||
get_logger,
|
||||
get_mutagen_symbols,
|
||||
is_vorbis_like,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"extract_source_metadata",
|
||||
"embed_source_ids",
|
||||
]
|
||||
|
||||
|
||||
_MB_RELEASE_CACHE: Dict[tuple, str] = {}
|
||||
_MB_RELEASE_CACHE_LOCK = threading.RLock()
|
||||
|
|
@ -48,7 +53,7 @@ _EDITION_BARE_RE = re.compile(
|
|||
)
|
||||
|
||||
|
||||
def _normalize_album_cache_key(album_name: str) -> str:
|
||||
def normalize_album_cache_key(album_name: str) -> str:
|
||||
result = _EDITION_PAREN_RE.sub("", album_name or "")
|
||||
result = _EDITION_BARE_RE.sub("", result)
|
||||
return result.lower().strip()
|
||||
|
|
@ -58,8 +63,8 @@ def extract_source_metadata(context: dict, artist: dict, album_info: dict) -> di
|
|||
if album_info is None:
|
||||
album_info = {}
|
||||
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
context = normalize_import_context(context)
|
||||
original_search = get_import_original_search(context)
|
||||
album_ctx = get_import_context_album(context)
|
||||
|
|
@ -68,7 +73,7 @@ def extract_source_metadata(context: dict, artist: dict, album_info: dict) -> di
|
|||
source_ids = get_import_source_ids(context)
|
||||
|
||||
artist_dict = artist if isinstance(artist, dict) else {
|
||||
"name": _extract_artist_name(artist),
|
||||
"name": extract_artist_name(artist),
|
||||
"id": getattr(artist, "id", ""),
|
||||
"genres": list(getattr(artist, "genres", []) or []),
|
||||
}
|
||||
|
|
@ -135,7 +140,7 @@ def extract_source_metadata(context: dict, artist: dict, album_info: dict) -> di
|
|||
artist_id = str(artist_dict.get("id", ""))
|
||||
if source == "itunes" and artist_id.isdigit():
|
||||
try:
|
||||
itunes_client = _get_itunes_client()
|
||||
itunes_client = get_itunes_client()
|
||||
if itunes_client and hasattr(itunes_client, "resolve_primary_artist"):
|
||||
resolved = itunes_client.resolve_primary_artist(artist_id)
|
||||
if resolved and resolved != raw_album_artist:
|
||||
|
|
@ -199,9 +204,9 @@ def extract_source_metadata(context: dict, artist: dict, album_info: dict) -> di
|
|||
|
||||
|
||||
def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=None):
|
||||
cfg = _get_config_manager()
|
||||
logger_ = _get_logger()
|
||||
symbols = _get_mutagen_symbols()
|
||||
cfg = get_config_manager()
|
||||
logger_ = get_logger()
|
||||
symbols = get_mutagen_symbols()
|
||||
if not symbols:
|
||||
return
|
||||
|
||||
|
|
@ -323,7 +328,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if not isinstance(source_order, list) or not source_order:
|
||||
source_order = ["musicbrainz", "deezer", "audiodb", "tidal", "qobuz", "lastfm", "genius"]
|
||||
|
||||
db = _get_database()
|
||||
db = get_database()
|
||||
|
||||
for source_name in source_order:
|
||||
if source_name == "musicbrainz":
|
||||
|
|
@ -723,7 +728,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
key = f"----:com.apple.iTunes:{mp4_tag_map.get(tag_name, tag_name)}"
|
||||
audio_file[key] = [symbols.MP4FreeForm(str(value).encode("utf-8"))]
|
||||
written.append(key)
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
for tag_name, value in filtered_tags.items():
|
||||
audio_file[vorbis_tag_map.get(tag_name, tag_name)] = [str(value)]
|
||||
written.append(vorbis_tag_map.get(tag_name, tag_name))
|
||||
|
|
@ -737,7 +742,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
metadata["date"] = release_year
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TDRC(encoding=3, text=[release_year]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["date"] = [release_year]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["\xa9day"] = [release_year]
|
||||
|
|
@ -747,7 +752,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
bpm_int = int(pp["deezer_bpm"])
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TBPM(encoding=3, text=[str(bpm_int)]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["BPM"] = [str(bpm_int)]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["tmpo"] = [bpm_int]
|
||||
|
|
@ -756,7 +761,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if _tag_enabled("audiodb.tags.mood") and pp["audiodb_mood"]:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TXXX(encoding=3, desc="MOOD", text=[pp["audiodb_mood"]]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["MOOD"] = [pp["audiodb_mood"]]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:MOOD"] = [symbols.MP4FreeForm(pp["audiodb_mood"].encode("utf-8"))]
|
||||
|
|
@ -764,7 +769,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if _tag_enabled("audiodb.tags.style") and pp["audiodb_style"]:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TXXX(encoding=3, desc="STYLE", text=[pp["audiodb_style"]]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["STYLE"] = [pp["audiodb_style"]]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:STYLE"] = [symbols.MP4FreeForm(pp["audiodb_style"].encode("utf-8"))]
|
||||
|
|
@ -795,7 +800,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
genre_string = ", ".join(merged)
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TCON(encoding=3, text=[genre_string]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["GENRE"] = [genre_string]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["\xa9gen"] = [genre_string]
|
||||
|
|
@ -814,7 +819,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
isrc_source, final_isrc = isrc_candidates[0]
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TSRC(encoding=3, text=[final_isrc]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["ISRC"] = [final_isrc]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:ISRC"] = [symbols.MP4FreeForm(final_isrc.encode("utf-8"))]
|
||||
|
|
@ -829,7 +834,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
copyright_source, final_copyright = copyright_candidates[0]
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TCOP(encoding=3, text=[final_copyright]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["COPYRIGHT"] = [final_copyright]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["cprt"] = [final_copyright]
|
||||
|
|
@ -838,7 +843,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if _tag_enabled("qobuz.tags.label") and pp["qobuz_label"]:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TPUB(encoding=3, text=[pp["qobuz_label"]]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["LABEL"] = [pp["qobuz_label"]]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:LABEL"] = [symbols.MP4FreeForm(pp["qobuz_label"].encode("utf-8"))]
|
||||
|
|
@ -846,7 +851,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if _tag_enabled("lastfm.tags.url") and pp["lastfm_url"]:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TXXX(encoding=3, desc="LASTFM_URL", text=[pp["lastfm_url"]]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["LASTFM_URL"] = [pp["lastfm_url"]]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:LASTFM_URL"] = [symbols.MP4FreeForm(pp["lastfm_url"].encode("utf-8"))]
|
||||
|
|
@ -854,7 +859,7 @@ def embed_source_ids(audio_file, metadata: dict, context: dict = None, runtime=N
|
|||
if _tag_enabled("genius.tags.url") and pp["genius_url"]:
|
||||
if isinstance(audio_file.tags, symbols.ID3):
|
||||
audio_file.tags.add(symbols.TXXX(encoding=3, desc="GENIUS_URL", text=[pp["genius_url"]]))
|
||||
elif _is_vorbis_like(audio_file, symbols):
|
||||
elif is_vorbis_like(audio_file, symbols):
|
||||
audio_file["GENIUS_URL"] = [pp["genius_url"]]
|
||||
elif isinstance(audio_file, symbols.MP4):
|
||||
audio_file["----:com.apple.iTunes:GENIUS_URL"] = [symbols.MP4FreeForm(pp["genius_url"].encode("utf-8"))]
|
||||
|
|
|
|||
|
|
@ -110,8 +110,8 @@ def _fake_symbols(audio):
|
|||
|
||||
|
||||
def test_extract_source_metadata_keeps_neutral_fields_and_skips_itunes_fallback_for_non_itunes_sources(monkeypatch):
|
||||
monkeypatch.setattr(ms, "_get_config_manager", lambda: _Config({"file_organization.collab_artist_mode": "first"}))
|
||||
monkeypatch.setattr(ms, "_get_itunes_client", lambda: (_ for _ in ()).throw(AssertionError("itunes fallback should not run for non-itunes sources")))
|
||||
monkeypatch.setattr(ms, "get_config_manager", lambda: _Config({"file_organization.collab_artist_mode": "first"}))
|
||||
monkeypatch.setattr(ms, "get_itunes_client", lambda: (_ for _ in ()).throw(AssertionError("itunes fallback should not run for non-itunes sources")))
|
||||
|
||||
context = {
|
||||
"source": "spotify",
|
||||
|
|
@ -160,9 +160,9 @@ def test_extract_source_metadata_keeps_neutral_fields_and_skips_itunes_fallback_
|
|||
def test_embed_source_ids_uses_current_source_ids_and_legacy_fallback(monkeypatch):
|
||||
audio = _FakeAudio()
|
||||
symbols = _fake_symbols(audio)
|
||||
monkeypatch.setattr(ms, "_get_config_manager", lambda: _Config())
|
||||
monkeypatch.setattr(ms, "_get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "_get_database", lambda: None)
|
||||
monkeypatch.setattr(ms, "get_config_manager", lambda: _Config())
|
||||
monkeypatch.setattr(ms, "get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "get_database", lambda: None)
|
||||
|
||||
current_metadata = {
|
||||
"source": "deezer",
|
||||
|
|
@ -182,7 +182,7 @@ def test_embed_source_ids_uses_current_source_ids_and_legacy_fallback(monkeypatc
|
|||
|
||||
audio = _FakeAudio()
|
||||
symbols = _fake_symbols(audio)
|
||||
monkeypatch.setattr(ms, "_get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "get_mutagen_symbols", lambda: symbols)
|
||||
|
||||
legacy_metadata = {
|
||||
"source": "",
|
||||
|
|
@ -214,18 +214,18 @@ def test_enhance_file_metadata_writes_tags_and_propagates_release_id(monkeypatch
|
|||
strip_calls = []
|
||||
verify_calls = []
|
||||
|
||||
monkeypatch.setattr(me, "_get_config_manager", lambda: _Config(
|
||||
monkeypatch.setattr(me, "get_config_manager", lambda: _Config(
|
||||
{
|
||||
"metadata_enhancement.enabled": True,
|
||||
"metadata_enhancement.embed_album_art": False,
|
||||
"metadata_enhancement.tags.write_multi_artist": False,
|
||||
}
|
||||
))
|
||||
monkeypatch.setattr(ms, "_get_config_manager", lambda: _Config())
|
||||
monkeypatch.setattr(me, "_get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "_get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "_get_database", lambda: None)
|
||||
monkeypatch.setattr(me, "_strip_all_non_audio_tags", lambda file_path: strip_calls.append(file_path) or {"apev2_stripped": False, "apev2_tag_count": 0})
|
||||
monkeypatch.setattr(ms, "get_config_manager", lambda: _Config())
|
||||
monkeypatch.setattr(me, "get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "get_mutagen_symbols", lambda: symbols)
|
||||
monkeypatch.setattr(ms, "get_database", lambda: None)
|
||||
monkeypatch.setattr(me, "strip_all_non_audio_tags", lambda file_path: strip_calls.append(file_path) or {"apev2_stripped": False, "apev2_tag_count": 0})
|
||||
monkeypatch.setattr(
|
||||
me,
|
||||
"extract_source_metadata",
|
||||
|
|
@ -247,7 +247,7 @@ def test_enhance_file_metadata_writes_tags_and_propagates_release_id(monkeypatch
|
|||
},
|
||||
)
|
||||
monkeypatch.setattr(me, "embed_album_art_metadata", lambda *args, **kwargs: None)
|
||||
monkeypatch.setattr(me, "_verify_metadata_written", lambda file_path: verify_calls.append(file_path) or True)
|
||||
monkeypatch.setattr(me, "verify_metadata_written", lambda file_path: verify_calls.append(file_path) or True)
|
||||
|
||||
album_info = {}
|
||||
result = me.enhance_file_metadata(
|
||||
|
|
@ -269,7 +269,7 @@ def test_enhance_file_metadata_writes_tags_and_propagates_release_id(monkeypatch
|
|||
|
||||
|
||||
def test_download_cover_art_uses_album_context_image_url(tmp_path, monkeypatch):
|
||||
monkeypatch.setattr(ma, "_get_config_manager", lambda: _Config(
|
||||
monkeypatch.setattr(ma, "get_config_manager", lambda: _Config(
|
||||
{
|
||||
"metadata_enhancement.cover_art_download": True,
|
||||
"metadata_enhancement.prefer_caa_art": False,
|
||||
|
|
@ -281,7 +281,7 @@ def test_download_cover_art_uses_album_context_image_url(tmp_path, monkeypatch):
|
|||
target_dir = tmp_path / "Album One"
|
||||
target_dir.mkdir()
|
||||
|
||||
me.download_cover_art(
|
||||
ma.download_cover_art(
|
||||
{},
|
||||
str(target_dir),
|
||||
{"album": {"image_url": "https://img.example/album.jpg"}},
|
||||
|
|
|
|||
Loading…
Reference in a new issue