from pathlib import Path from typing import Any, Optional, Union from uuid import UUID from mutagen import File as MutagenFile from mutagen.flac import FLAC, Picture from mutagen.id3 import APIC from mutagen.mp4 import MP4, MP4Cover from app.domain.context import MusicInfo from app.domain.meta.metamusic import MetaMusic from app.runtime.log import logger from app.schemas.types import MUSIC_ENTITY_RECORDING, MediaSource class AudioMetadataHelper: """读取和写入音频标签,并转换为标准音乐元数据。""" @classmethod def read(cls, path: Path) -> MetaMusic: """读取本地音频标签,并以完整文件名模式和目录线索补充缺失字段。""" tag_meta = cls.read_tags(path) if tag_meta: return tag_meta.apply_path_context(path) return cls.read_filename(path) @classmethod def read_evidence( cls, path: Path, ) -> tuple[MetaMusic, Optional[MetaMusic], MetaMusic]: """分别返回合并元数据、纯标签元数据和纯文件名元数据。""" filename_meta = cls.read_filename(path) tag_meta = cls.read_tags(path) if path.exists() and path.is_file() else None if not tag_meta: return filename_meta, None, filename_meta merged_meta = MetaMusic.from_dict(tag_meta.to_dict()).apply_path_context(path) return merged_meta, tag_meta, filename_meta @classmethod def read_many(cls, paths: list[Path]) -> list[MetaMusic]: """批量读取一组音频路径的标签与文件名元数据。""" return [cls.read(path) for path in paths] @classmethod def read_tags(cls, path: Path) -> Optional[MetaMusic]: """只读取本地音频标签和流参数,不使用文件名或目录补齐。""" try: audio = MutagenFile(path, easy=True) except Exception as err: logger.warning(f"读取音频标签失败:{path} - {err}") return None if not audio: return None tags = audio.tags or {} track_number, total_tracks = cls._number_pair(cls._first(tags, "tracknumber")) disc_number, total_discs = cls._number_pair(cls._first(tags, "discnumber")) musicbrainz_id = cls._normalize_musicbrainz_id( cls._first_of( tags, "musicbrainz_trackid", "musicbrainz_recordingid", ) ) info = getattr(audio, "info", None) return MetaMusic( org_string=path.name, title=cls._first(tags, "title"), artists=cls._values(tags, "artist"), album=cls._first(tags, "album"), album_artist=cls._first(tags, "albumartist"), year=cls._year(cls._first(tags, "date") or cls._first(tags, "originaldate")), disc_number=disc_number, track_number=track_number, total_discs=total_discs, total_tracks=total_tracks, version=cls._first(tags, "version") or cls._first(tags, "subtitle"), audio_format=cls._audio_format(path, info), bit_depth=cls._optional_int(getattr(info, "bits_per_sample", None)), sample_rate=cls._optional_int(getattr(info, "sample_rate", None)), bitrate=cls._optional_int(getattr(info, "bitrate", None)), duration=round(info.length) if info and getattr(info, "length", None) else None, isrc=cls._first(tags, "isrc"), media_source=MediaSource.MusicBrainz if musicbrainz_id else None, media_id=musicbrainz_id, ) @staticmethod def read_filename(path: Path) -> MetaMusic: """只从文件名和目录结构解析音乐元数据。""" return MetaMusic( org_string=path.name, title=path.stem, audio_format=path.suffix.lstrip(".").upper() or None, ).apply_path_context(path) @classmethod def write( cls, path: Path, music: Union[MetaMusic, MusicInfo], cover_data: Optional[bytes] = None, cover_mime: str = "image/jpeg", overwrite: bool = True, write_tags: bool = True, cover_overwrite: Optional[bool] = None, ) -> bool: """按独立策略写入标准音乐标签,并为常见格式嵌入专辑封面。""" try: audio = MutagenFile(path, easy=True) if not audio: logger.warning(f"无法写入音频标签:{path}") return False if write_tags: if audio.tags is None: audio.add_tags() for key, value in cls._tag_values(music).items(): if value in (None, "", []): continue if not overwrite and audio.tags.get(key): continue try: audio[key] = value if isinstance(value, list) else [str(value)] except (KeyError, TypeError, ValueError) as err: logger.debug(f"音频格式不支持标签 {key}:{path} - {err}") audio.save() if cover_data: cls._write_cover( path=path, cover_data=cover_data, cover_mime=cover_mime, overwrite=( overwrite if cover_overwrite is None else cover_overwrite ), ) return True except Exception as err: logger.warning(f"写入音频标签失败:{path} - {err}") return False @classmethod def _tag_values(cls, music: Union[MetaMusic, MusicInfo]) -> dict[str, Any]: """把标准音乐对象转换为 Mutagen Easy 标签字典。""" track_number = cls._number_text( getattr(music, "track_number", None), getattr(music, "total_tracks", None), ) disc_number = cls._number_text( getattr(music, "disc_number", None), getattr(music, "total_discs", None), ) return { "title": getattr(music, "title", None), "artist": list(getattr(music, "artists", None) or []), "album": getattr(music, "album", None), "albumartist": getattr(music, "album_artist", None), "date": getattr(music, "year", None), "tracknumber": track_number, "discnumber": disc_number, "isrc": getattr(music, "isrc", None), "musicbrainz_trackid": cls._musicbrainz_recording_id(music), } @staticmethod def _musicbrainz_recording_id( music: Union[MetaMusic, MusicInfo], ) -> Optional[str]: """仅将 MusicBrainz 单曲身份写入 recording 标签,避免误写专辑 ID。""" if ( getattr(music, "media_source", None) == MediaSource.MusicBrainz and getattr(music, "music_type", MUSIC_ENTITY_RECORDING) == MUSIC_ENTITY_RECORDING ): media_id = getattr(music, "media_id", None) return str(media_id) if media_id else None return None @staticmethod def _number_text(current: Optional[int], total: Optional[int]) -> Optional[str]: """把曲序或碟号转换为常见的 current/total 标签文本。""" if current is None: return None return f"{current}/{total}" if total else str(current) @staticmethod def _write_cover( path: Path, cover_data: bytes, cover_mime: str, overwrite: bool, ) -> None: """为 MP3、FLAC 和 MP4/M4A 写入内嵌封面,其它格式保留标签写入结果。""" audio = MutagenFile(path) if isinstance(audio, FLAC): if audio.pictures and not overwrite: return picture = Picture() picture.type = 3 picture.mime = cover_mime picture.desc = "Cover" picture.data = cover_data if overwrite: audio.clear_pictures() audio.add_picture(picture) audio.save() return if isinstance(audio, MP4): if audio.tags is None: audio.add_tags() if audio.tags.get("covr") and not overwrite: return image_format = ( MP4Cover.FORMAT_PNG if cover_mime == "image/png" else MP4Cover.FORMAT_JPEG ) audio.tags["covr"] = [MP4Cover(cover_data, imageformat=image_format)] audio.save() return tags = getattr(audio, "tags", None) if tags is not None and hasattr(tags, "add"): if tags.getall("APIC") and not overwrite: return if overwrite: tags.delall("APIC") tags.add( APIC( encoding=3, mime=cover_mime, type=3, desc="Cover", data=cover_data, ) ) audio.save() @staticmethod def _values(tags: Any, key: str) -> list[str]: """从 Mutagen Easy 标签中提取非空字符串列表。""" value = tags.get(key) if hasattr(tags, "get") else None if value is None: return [] if isinstance(value, (list, tuple)): return [str(item).strip() for item in value if str(item).strip()] return [str(value).strip()] if str(value).strip() else [] @classmethod def _first(cls, tags: Any, key: str) -> Optional[str]: """返回指定音频标签的第一个非空值。""" values = cls._values(tags, key) return values[0] if values else None @classmethod def _first_of(cls, tags: Any, *keys: str) -> Optional[str]: """按顺序返回多个音频标签中的第一个非空值。""" for key in keys: if value := cls._first(tags, key): return value return None @staticmethod def _normalize_musicbrainz_id(value: Optional[str]) -> Optional[str]: """校验并规范化音频标签中的 MusicBrainz UUID。""" try: return str(UUID(str(value))) except (AttributeError, TypeError, ValueError): return None @staticmethod def _number_pair(value: Optional[str]) -> tuple[Optional[int], Optional[int]]: """解析 track/disc 标签中的当前编号和总数。""" if not value: return None, None parts = str(value).split("/", 1) current = AudioMetadataHelper._optional_int(parts[0]) total = AudioMetadataHelper._optional_int(parts[1]) if len(parts) > 1 else None return current, total @staticmethod def _year(value: Optional[str]) -> Optional[int]: """从完整或不完整日期标签中提取四位年份。""" if not value: return None return AudioMetadataHelper._optional_int(str(value)[:4]) @staticmethod def _audio_format(path: Path, info: Any) -> Optional[str]: """结合扩展名与流编码识别音频格式,区分同为 M4A 容器的 AAC 和 ALAC。""" codec_text = " ".join( str(value or "") for value in ( getattr(info, "codec", None), getattr(info, "codec_description", None), ) ).casefold() codec_formats = ( (("alac", "apple lossless"), "ALAC"), (("aac", "mp4a"), "AAC"), (("opus",), "OPUS"), (("vorbis",), "OGG"), (("flac",), "FLAC"), ) for markers, audio_format in codec_formats: if any(marker in codec_text for marker in markers): return audio_format return path.suffix.lstrip(".").upper() or None @staticmethod def _optional_int(value: Any) -> Optional[int]: """将音频技术参数安全转换为整数。""" try: return int(value) if value is not None and str(value).strip() else None except (TypeError, ValueError): return None