diff --git a/app/chain/media.py b/app/chain/media.py index a611d6a56..714dfbc74 100644 --- a/app/chain/media.py +++ b/app/chain/media.py @@ -1,9 +1,11 @@ import os +import re from copy import deepcopy +from dataclasses import dataclass from pathlib import Path from tempfile import NamedTemporaryFile, TemporaryDirectory from threading import Lock -from typing import Optional, List, Tuple, Union +from typing import TYPE_CHECKING, List, Optional, Tuple, Union from app import schemas from app.chain import ChainBase @@ -14,7 +16,9 @@ from app.core.context import ( MUSIC_ENTITY_RECORDING, Context, MediaInfo, + MusicAlbumInfo, MusicInfo, + MusicLyrics, ) from app.core.event import eventmanager, Event from app.core.meta import MetaBase, MetaMusic @@ -37,6 +41,9 @@ from app.utils.mixins import ConfigReloadMixin from app.utils.singleton import Singleton from app.utils.string import StringUtils +if TYPE_CHECKING: + from app.chain.music import MusicChain + recognize_lock = Lock() scraping_lock = Lock() @@ -44,6 +51,14 @@ current_umask = os.umask(0) os.umask(current_umask) +@dataclass +class _MusicScrapeFileResult: + """记录单个音轨的标签刮削结果和歌词处理状态。""" + + metadata_success: bool = True + lyrics_status: str = "disabled" + + class ScrapingOption: """刮削选项""" @@ -147,7 +162,7 @@ class ScrapingConfig: ("tv", ["nfo", "poster", "backdrop", "logo", "banner", "thumb", "clearart", "landscape"]), ("season", ["nfo", "poster", "backdrop", "banner", "thumb", "landscape"]), ("episode", ["nfo", "thumb"]), - ("music", ["nfo", "poster"]), + ("music", ["nfo", "poster", "lyrics"]), ] for md in mds ] @@ -182,6 +197,12 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): "landscape": ["thumb"], } + MUSIC_LYRICS_EXTENSIONS = (".lrc", ".txt") + _music_track_prefix_pattern = re.compile( + r"^\s*(?:(?:cd|disc)\s*\d+\s*[-_. ]+)?(?:\d+\s*[-_. ]+)+", + flags=re.IGNORECASE, + ) + def __init__(self): super().__init__() self.storagechain = StorageChain() @@ -1076,7 +1097,7 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): _, message = self.scrape_music_metadata( fileitem=fileitem, mediainfo=mediainfo, - overwrite=overwrite or self.scraping_policies.option("music", "nfo").is_overwrite, + overwrite=overwrite, ) if message: logger.info(f"音乐刮削：{message}") @@ -1420,11 +1441,12 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): ): return False, "单曲 MusicBrainz ID 仅支持刮削单个音频文件，整目录请选择专辑" - # 读取音乐刮削策略：music_nfo 控制标签写入，music_poster 控制封面嵌入 + # 三类音乐产物使用独立策略，允许只下载歌词而不改写音频标签。 nfo_option = self.scraping_policies.option("music", "nfo") poster_option = self.scraping_policies.option("music", "poster") - if nfo_option.is_skip: - return False, "音乐标签刮削策略为跳过，请先在高级设置中开启" + lyrics_option = self.scraping_policies.option("music", "lyrics") + if nfo_option.is_skip and poster_option.is_skip and lyrics_option.is_skip: + return False, "音乐标签、封面和歌词刮削策略均为跳过" with_cover = not poster_option.is_skip shared_cover = ( @@ -1432,19 +1454,71 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): if mediainfo and with_cover else None ) - failures: list[str] = [] - for audio_item in files: - if not self._scrape_music_file( - audio_item, - mediainfo, - overwrite=overwrite or nfo_option.is_overwrite, - with_cover=with_cover, - cover=shared_cover, + music_chain = None + album_info = None + if not lyrics_option.is_skip: + # 延迟导入避免 MediaChain 与 MusicChain 在模块加载阶段形成双向依赖。 + from app.chain.music import MusicChain + + music_chain = MusicChain() + if ( + mediainfo + and mediainfo.music_type == MUSIC_ENTITY_ALBUM + and mediainfo.source + and mediainfo.media_id ): - failures.append(f"{audio_item.name or audio_item.path} 标签写入失败") + album_info = music_chain.album( + source=mediainfo.source, + media_id=mediainfo.media_id, + ) + + failures: list[str] = [] + lyrics_counts = { + "saved": 0, + "existing": 0, + "missing": 0, + "failed": 0, + } + metadata_failure_label = ( + "音乐标签和封面" + if not nfo_option.is_skip and with_cover + else "音乐标签" if not nfo_option.is_skip else "封面" + ) + for audio_item in files: + result = self._scrape_music_file( + audio_item, + mediainfo, + write_tags=not nfo_option.is_skip, + tag_overwrite=overwrite or nfo_option.is_overwrite, + with_cover=with_cover, + cover_overwrite=overwrite or poster_option.is_overwrite, + cover=shared_cover, + lyrics_option=lyrics_option, + lyrics_overwrite=overwrite or lyrics_option.is_overwrite, + music_chain=music_chain, + album_info=album_info, + ) + if not result.metadata_success: + failures.append( + f"{audio_item.name or audio_item.path} {metadata_failure_label}写入失败" + ) + if result.lyrics_status in lyrics_counts: + lyrics_counts[result.lyrics_status] += 1 + if result.lyrics_status == "failed": + failures.append(f"{audio_item.name or audio_item.path} 歌词保存失败") + + message = f"已刮削 {len(files)} 个音频文件" + if not lyrics_option.is_skip: + message += ( + f"，歌词新增 {lyrics_counts['saved']} 首" + f"、已存在 {lyrics_counts['existing']} 首" + f"、未匹配 {lyrics_counts['missing']} 首" + ) + if lyrics_counts["failed"]: + message += f"、失败 {lyrics_counts['failed']} 首" if failures: - return False, "；".join(failures[:3]) - return True, f"已刮削 {len(files)} 个音频文件" + return False, f"{message}；{'；'.join(failures[:3])}" + return True, message @staticmethod def _download_music_cover(url: Optional[str]) -> tuple[Optional[bytes], str]: @@ -1486,46 +1560,120 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): self, fileitem: schemas.FileItem, mediainfo: Optional[MusicInfo], - overwrite: bool, + write_tags: bool, + tag_overwrite: bool, with_cover: bool = True, + cover_overwrite: bool = True, cover: Optional[tuple[Optional[bytes], str]] = None, - ) -> bool: - """下载单个音频文件、写入标签，并在远端存储场景上传覆盖原文件。""" - storage = StorageChain() + lyrics_option: Optional[ScrapingOption] = None, + lyrics_overwrite: bool = False, + music_chain: Optional["MusicChain"] = None, + album_info: Optional[MusicAlbumInfo] = None, + ) -> _MusicScrapeFileResult: + """下载单个音轨并执行标签、封面和歌词刮削，远端产物写回原目录。""" + storage = self.storagechain + download_failure = _MusicScrapeFileResult( + metadata_success=not (write_tags or with_cover), + lyrics_status=( + "failed" + if lyrics_option and not lyrics_option.is_skip and music_chain + else "disabled" + ), + ) if fileitem.storage == "local": local_path = storage.download_file(fileitem) - return bool( - local_path - and self._write_music_metadata( - local_path, - mediainfo, - overwrite=overwrite, - with_cover=with_cover, - cover=cover, - ) + if not local_path: + return download_failure + return self._apply_music_file_scrape( + fileitem=fileitem, + local_path=local_path, + mediainfo=mediainfo, + write_tags=write_tags, + tag_overwrite=tag_overwrite, + with_cover=with_cover, + cover_overwrite=cover_overwrite, + cover=cover, + lyrics_option=lyrics_option, + lyrics_overwrite=lyrics_overwrite, + music_chain=music_chain, + album_info=album_info, ) with TemporaryDirectory(prefix="moviepilot-music-scrape-") as temp_dir: local_path = storage.download_file(fileitem, path=Path(temp_dir)) - if not local_path or not self._write_music_metadata( - local_path, - mediainfo, - overwrite=overwrite, - with_cover=with_cover, - cover=cover, - ): - return False - parent = storage.get_parent_item(fileitem) + if not local_path: + return download_failure + return self._apply_music_file_scrape( + fileitem=fileitem, + local_path=local_path, + mediainfo=mediainfo, + write_tags=write_tags, + tag_overwrite=tag_overwrite, + with_cover=with_cover, + cover_overwrite=cover_overwrite, + cover=cover, + lyrics_option=lyrics_option, + lyrics_overwrite=lyrics_overwrite, + music_chain=music_chain, + album_info=album_info, + ) + + def _apply_music_file_scrape( + self, + fileitem: schemas.FileItem, + local_path: Path, + mediainfo: Optional[MusicInfo], + write_tags: bool, + tag_overwrite: bool, + with_cover: bool, + cover_overwrite: bool, + cover: Optional[tuple[Optional[bytes], str]], + lyrics_option: Optional[ScrapingOption], + lyrics_overwrite: bool, + music_chain: Optional["MusicChain"], + album_info: Optional[MusicAlbumInfo], + ) -> _MusicScrapeFileResult: + """在本地音轨副本上执行刮削，并将变更后的音频和歌词写回目标存储。""" + scrape_info = self._resolve_music_scrape_info(local_path, mediainfo) + metadata_requested = write_tags or with_cover + metadata_success = True + if metadata_requested: + metadata_success = self._write_music_metadata( + local_path=local_path, + mediainfo=mediainfo, + tag_overwrite=tag_overwrite, + write_tags=write_tags, + with_cover=with_cover, + cover_overwrite=cover_overwrite, + cover=cover, + scrape_info=scrape_info, + ) + + lyrics_status = self._scrape_music_lyrics( + fileitem=fileitem, + local_path=local_path, + scrape_info=scrape_info, + lyrics_option=lyrics_option, + overwrite=lyrics_overwrite, + music_chain=music_chain, + album_info=album_info, + ) + + if fileitem.storage != "local" and metadata_requested and metadata_success: + parent = self.storagechain.get_parent_item(fileitem) if not parent: logger.warning(f"无法获取远端音频父目录：{fileitem.path}") - return False - return bool( - storage.upload_file( + metadata_success = False + elif not self.storagechain.upload_file( parent, local_path, new_name=fileitem.name or local_path.name, - ) - ) + ): + metadata_success = False + return _MusicScrapeFileResult( + metadata_success=metadata_success, + lyrics_status=lyrics_status, + ) @staticmethod def _merge_music_album_metadata(local_meta: MetaMusic, album: MusicInfo) -> MetaMusic: @@ -1540,6 +1688,66 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): merged.media_id = album.media_id or local_meta.media_id return merged + @classmethod + def _match_music_album_track( + cls, + local_meta: MetaMusic, + album_info: Optional[MusicAlbumInfo], + ) -> Optional[MusicInfo]: + """按碟号、曲序、标题、艺术家和时长为本地文件匹配专辑中的单个音轨。""" + if not album_info or not album_info.tracks: + return None + local_title = cls._normalize_music_track_title(local_meta.title) + local_artists = { + cls._normalize_music_track_title(artist) + for artist in local_meta.artists + if artist + } + ranked: list[tuple[int, MusicInfo]] = [] + for track in album_info.tracks: + score = 0 + if local_meta.track_number and track.track_number: + if local_meta.track_number == track.track_number: + score += 6 + else: + continue + if local_meta.disc_number and track.disc_number: + if local_meta.disc_number == track.disc_number: + score += 3 + else: + continue + if local_title and local_title == cls._normalize_music_track_title(track.title): + score += 8 + track_artists = { + cls._normalize_music_track_title(artist) + for artist in track.artists + if artist + } + if local_artists and track_artists and local_artists.intersection(track_artists): + score += 3 + if local_meta.duration and track.duration: + duration_delta = abs(local_meta.duration - track.duration) + if duration_delta <= 2: + score += 4 + elif duration_delta > 5: + score -= 3 + if score >= 8: + ranked.append((score, track)) + if not ranked: + return None + ranked.sort(key=lambda pair: pair[0], reverse=True) + if len(ranked) > 1 and ranked[0][0] == ranked[1][0]: + return None + matched = deepcopy(ranked[0][1]) + matched.duration = matched.duration or local_meta.duration + return matched + + @classmethod + def _normalize_music_track_title(cls, value: Optional[str]) -> str: + """规范化音轨标题并移除常见文件名前置碟号和曲序。""" + title = cls._music_track_prefix_pattern.sub("", str(value or "").strip()) + return re.sub(r"[^\w]+", "", title.casefold(), flags=re.UNICODE) + @classmethod def _resolve_music_scrape_info( cls, @@ -1566,12 +1774,15 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): self, local_path: Path, mediainfo: Optional[MusicInfo], - overwrite: bool, + tag_overwrite: bool, + write_tags: bool, with_cover: bool, + cover_overwrite: bool, cover: Optional[tuple[Optional[bytes], str]] = None, + scrape_info: Optional[MetaMusic | MusicInfo] = None, ) -> bool: - """解析单个本地音轨并写入标签，显式专辑刮削可复用一次下载的封面。""" - scrape_info = self._resolve_music_scrape_info(local_path, mediainfo) + """解析单个本地音轨并按独立策略写入标签和封面。""" + scrape_info = scrape_info or self._resolve_music_scrape_info(local_path, mediainfo) if not scrape_info or not scrape_info.title: logger.warning(f"无法识别音乐信息：{local_path}") return False @@ -1587,9 +1798,131 @@ class MediaChain(ChainBase, ConfigReloadMixin, metaclass=Singleton): scrape_info, cover_data=cover_data, cover_mime=cover_mime, - overwrite=overwrite, + overwrite=tag_overwrite, + write_tags=write_tags, + cover_overwrite=cover_overwrite, ) + def _scrape_music_lyrics( + self, + fileitem: schemas.FileItem, + local_path: Path, + scrape_info: Optional[MetaMusic | MusicInfo], + lyrics_option: Optional[ScrapingOption], + overwrite: bool, + music_chain: Optional["MusicChain"], + album_info: Optional[MusicAlbumInfo], + ) -> str: + """按歌词策略查询单个音轨并保存同名旁挂歌词文件。""" + if not lyrics_option or lyrics_option.is_skip or not music_chain: + return "disabled" + existing = self._find_music_lyrics_sidecar(fileitem) + if existing and not overwrite: + return "existing" + if not scrape_info: + return "missing" + + lookup_info: MetaMusic | MusicInfo = scrape_info + if album_info: + local_meta = AudioMetadataHelper.read(local_path) + lookup_info = self._match_music_album_track(local_meta, album_info) or scrape_info + lyrics = music_chain.lyrics(lookup_info) + if not lyrics or lyrics.instrumental or not lyrics.content or not lyrics.extension: + return "missing" + return ( + "saved" + if self._write_music_lyrics_sidecar( + fileitem=fileitem, + local_path=local_path, + lyrics=lyrics, + overwrite=overwrite, + ) + else "failed" + ) + + def _find_music_lyrics_sidecar( + self, + fileitem: schemas.FileItem, + ) -> Optional[schemas.FileItem]: + """查找音轨旁已存在的同步或纯文本歌词文件。""" + audio_path = Path(fileitem.path) + for extension in self.MUSIC_LYRICS_EXTENSIONS: + item = self.storagechain.get_file_item( + storage=fileitem.storage, + path=audio_path.with_suffix(extension), + ) + if item: + return item + return None + + def _write_music_lyrics_sidecar( + self, + fileitem: schemas.FileItem, + local_path: Path, + lyrics: MusicLyrics, + overwrite: bool, + ) -> bool: + """原子写入本地歌词或上传远端歌词，并在覆盖时清理旧格式旁挂文件。""" + extension = lyrics.extension + content = lyrics.content + if not extension or not content: + return False + target_path = Path(fileitem.path).with_suffix(extension) + target_name = target_path.name + temp_path: Optional[Path] = None + try: + if fileitem.storage == "local": + with NamedTemporaryFile( + mode="w", + encoding="utf-8", + dir=target_path.parent, + prefix=f".{target_name}.", + delete=False, + ) as temp_file: + temp_file.write(f"{content.rstrip()}\n") + temp_path = Path(temp_file.name) + temp_path.replace(target_path) + else: + parent = self.storagechain.get_parent_item(fileitem) + if not parent: + logger.warning(f"无法获取远端歌词父目录：{fileitem.path}") + return False + temp_path = local_path.with_suffix(extension) + temp_path.write_text(f"{content.rstrip()}\n", encoding="utf-8") + if not self.storagechain.upload_file( + parent, + temp_path, + new_name=target_name, + ): + return False + + if overwrite: + self._remove_alternate_music_lyrics(fileitem, keep_extension=extension) + return True + except OSError as err: + logger.warning(f"保存音乐歌词失败：{target_path} - {err}") + return False + finally: + if temp_path and temp_path.exists() and temp_path != target_path: + self._cleanup_temp_file(temp_path) + + def _remove_alternate_music_lyrics( + self, + fileitem: schemas.FileItem, + keep_extension: str, + ) -> None: + """覆盖歌词格式后删除同音轨的旧扩展名文件，避免播放器优先读取过期内容。""" + audio_path = Path(fileitem.path) + for extension in self.MUSIC_LYRICS_EXTENSIONS: + if extension == keep_extension: + continue + item = self.storagechain.get_file_item( + storage=fileitem.storage, + path=audio_path.with_suffix(extension), + ) + if item and not self.storagechain.delete_file(item): + logger.warning(f"删除旧歌词文件失败：{item.path}") + def _handle_movie_scraping( self, fileitem: schemas.FileItem, diff --git a/app/chain/music.py b/app/chain/music.py index 6e0815a7e..9d75f5f91 100644 --- a/app/chain/music.py +++ b/app/chain/music.py @@ -13,6 +13,7 @@ from app.core.context import ( MusicAlbumInfo, MusicArtistInfo, MusicInfo, + MusicLyrics, ) from app.core.meta import MetaMusic from app.helper.audio import AudioMetadataHelper @@ -185,6 +186,28 @@ class MusicChain(ChainBase): return MusicAlbumInfo.from_dict(result) return None + def album(self, source: str, media_id: str) -> Optional[MusicAlbumInfo]: + """同步按来源和专辑 ID 获取标准化专辑详情及曲目。""" + result = self.run_module( + "music_album", + source=source, + media_id=media_id, + ) + if isinstance(result, MusicAlbumInfo): + return result + if isinstance(result, dict): + return MusicAlbumInfo.from_dict(result) + return None + + def lyrics(self, music: MetaMusic | MusicInfo) -> Optional[MusicLyrics]: + """按单曲元数据调用已启用的歌词模块并返回标准歌词。""" + result = self.run_module("music_lyrics", music=music) + if isinstance(result, MusicLyrics): + return result + if isinstance(result, dict): + return MusicLyrics.from_dict(result) + return None + async def async_artist(self, source: str, media_id: str) -> Optional[MusicArtistInfo]: """异步按来源和艺术家 ID 获取标准化艺术家详情。""" result = await self.async_run_module( diff --git a/app/core/context.py b/app/core/context.py index ddea5e74d..3c51044b1 100644 --- a/app/core/context.py +++ b/app/core/context.py @@ -79,6 +79,44 @@ def _music_init_values(model: type, data: dict[str, Any]) -> dict[str, Any]: return {key: value for key, value in data.items() if key in init_names} +@dataclass +class MusicLyrics: + """标准化单曲歌词，区分同步歌词、纯文本歌词和纯音乐结果。""" + + provider: str + provider_id: str | None = None + instrumental: bool = False + plain_lyrics: str | None = None + synced_lyrics: str | None = None + + @property + def content(self) -> str | None: + """优先返回同步歌词，不存在时回退到纯文本歌词。""" + return self.synced_lyrics or self.plain_lyrics + + @property + def extension(self) -> str | None: + """根据歌词内容返回适合播放器扫描的旁挂文件扩展名。""" + if self.synced_lyrics: + return ".lrc" + if self.plain_lyrics: + return ".txt" + return None + + @classmethod + def from_dict(cls, data: dict[str, Any]) -> Self: + """从模块或插件返回字典恢复标准歌词对象。""" + values = _music_init_values(cls, data) + values["provider"] = str(values.get("provider") or "") + values["provider_id"] = ( + str(values["provider_id"]) + if values.get("provider_id") is not None + else None + ) + values["instrumental"] = bool(values.get("instrumental")) + return cls(**values) + + @dataclass class MusicInfo: """标准化音乐元数据信息。""" diff --git a/app/helper/audio.py b/app/helper/audio.py index 1e67a9555..e54cdf032 100644 --- a/app/helper/audio.py +++ b/app/helper/audio.py @@ -62,31 +62,38 @@ class AudioMetadataHelper: 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 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 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, + overwrite=( + overwrite + if cover_overwrite is None + else cover_overwrite + ), ) return True except Exception as err: diff --git a/app/modules/lrclib/__init__.py b/app/modules/lrclib/__init__.py new file mode 100644 index 000000000..d3e4205b0 --- /dev/null +++ b/app/modules/lrclib/__init__.py @@ -0,0 +1,234 @@ +import re +import threading +import time +from typing import Any, Optional, Tuple, Union + +from app.core.cache import cached +from app.core.config import settings +from app.core.context import MusicInfo, MusicLyrics +from app.core.meta import MetaMusic +from app.log import logger +from app.modules import _ModuleBase +from app.schemas.types import ModuleType, OtherModulesType +from app.utils.http import RequestUtils + + +class LrclibModule(_ModuleBase): + """通过 LRCLIB 获取与单个音轨匹配的同步歌词或纯文本歌词。""" + + _base_url = "https://lrclib.net" + _source = "lrclib" + _request_interval = 0.3 + _request_lock = threading.Lock() + _last_request_at = 0.0 + _match_pattern = re.compile(r"[^\w]+", flags=re.UNICODE) + + def init_module(self) -> None: + """初始化无状态的 LRCLIB 歌词模块。""" + + def init_setting(self) -> Optional[Tuple[str, Union[str, bool]]]: + """LRCLIB 无需密钥，是否请求由音乐歌词刮削策略控制。""" + return None + + def stop(self) -> None: + """停止模块；当前实现没有需要释放的持久资源。""" + + def test(self) -> Tuple[bool, str]: + """测试 LRCLIB 搜索接口连通性。""" + result = self._request_json("/api/search", params={"track_name": "test"}) + return (True, "") if result is not None else (False, "LRCLIB 网络连接失败") + + @staticmethod + def get_name() -> str: + """返回模块展示名称。""" + return "LRCLIB" + + @staticmethod + def get_type() -> ModuleType: + """返回模块所属的其它能力类型。""" + return ModuleType.Other + + @staticmethod + def get_subtype() -> OtherModulesType: + """返回 LRCLIB 模块子类型。""" + return OtherModulesType.Lrclib + + @staticmethod + def get_priority() -> int: + """返回歌词模块执行优先级。""" + return 5 + + def music_lyrics(self, music: Union[MetaMusic, MusicInfo]) -> Optional[MusicLyrics]: + """按标题、艺术家、专辑和时长查询单曲歌词，并对搜索回退结果严格匹配。""" + title = str(getattr(music, "title", None) or "").strip() + artists = list(getattr(music, "artists", None) or []) + artist = str((artists[0] if artists else getattr(music, "album_artist", None)) or "").strip() + album = str(getattr(music, "album", None) or "").strip() + duration = self._optional_int(getattr(music, "duration", None)) + if not title or not artist: + return None + + exact_params: dict[str, Any] = { + "track_name": title, + "artist_name": artist, + } + if album: + exact_params["album_name"] = album + if duration: + exact_params["duration"] = duration + payload = self._request_json("/api/get", params=exact_params) + if not payload: + results = self._request_json( + "/api/search", + params={ + "track_name": title, + "artist_name": artist, + **({"album_name": album} if album else {}), + }, + ) + payload = self._select_result( + results if isinstance(results, list) else [], + title=title, + artist=artist, + album=album, + duration=duration, + ) + return self._to_lyrics(payload) + + @classmethod + def _select_result( + cls, + results: list[dict[str, Any]], + title: str, + artist: str, + album: str, + duration: Optional[int], + ) -> Optional[dict[str, Any]]: + """从模糊搜索结果中选择标题和艺术家一致且时长可信的歌词。""" + expected_title = cls._normalize_text(title) + expected_artist = cls._normalize_text(artist) + expected_album = cls._normalize_text(album) + ranked: list[tuple[int, dict[str, Any]]] = [] + for item in results: + if cls._normalize_text(item.get("trackName")) != expected_title: + continue + candidate_artist = cls._normalize_text(item.get("artistName")) + if not cls._compatible_text(expected_artist, candidate_artist): + continue + candidate_duration = cls._optional_int(item.get("duration")) + if duration and candidate_duration and abs(duration - candidate_duration) > 2: + continue + score = 4 + if candidate_artist == expected_artist: + score += 3 + if expected_album and cls._normalize_text(item.get("albumName")) == expected_album: + score += 2 + if duration and candidate_duration and abs(duration - candidate_duration) <= 2: + score += 3 + ranked.append((score, item)) + if not ranked: + return None + ranked.sort(key=lambda pair: pair[0], reverse=True) + return ranked[0][1] + + @classmethod + def _normalize_text(cls, value: Any) -> str: + """移除大小写、标点和空白差异，生成歌词匹配文本。""" + return cls._match_pattern.sub("", str(value or "").casefold()) + + @staticmethod + def _compatible_text(expected: str, candidate: str) -> bool: + """允许合作艺人字符串互相包含，同时拒绝完全无关的艺术家。""" + return bool(expected and candidate and (expected in candidate or candidate in expected)) + + @staticmethod + def _optional_int(value: Any) -> Optional[int]: + """把歌词源返回的时长安全转换为整数秒。""" + try: + return round(float(value)) if value not in (None, "") else None + except (TypeError, ValueError): + return None + + @classmethod + def _to_lyrics(cls, payload: Any) -> Optional[MusicLyrics]: + """把 LRCLIB 响应转换为标准歌词对象。""" + if not isinstance(payload, dict) or payload.get("id") is None: + return None + plain_lyrics = str(payload.get("plainLyrics") or "").strip() or None + synced_lyrics = str(payload.get("syncedLyrics") or "").strip() or None + instrumental = bool(payload.get("instrumental")) + if not instrumental and not plain_lyrics and not synced_lyrics: + return None + return MusicLyrics( + provider=cls._source, + provider_id=str(payload["id"]), + instrumental=instrumental, + plain_lyrics=plain_lyrics, + synced_lyrics=synced_lyrics, + ) + + @classmethod + def _request_once( + cls, + path: str, + params: Optional[dict[str, Any]], + ) -> Any: + """串行执行一次 LRCLIB 请求，确保批量专辑刮削遵守最小请求间隔。""" + with cls._request_lock: + delay = cls._request_interval - (time.monotonic() - cls._last_request_at) + if delay > 0: + time.sleep(delay) + response = RequestUtils( + headers={ + "User-Agent": f"{settings.USER_AGENT} (https://github.com/jxxghp/MoviePilot)", + "Accept": "application/json", + }, + proxies=settings.PROXY, + timeout=20, + ).get_res(f"{cls._base_url}{path}", params=params) + cls._last_request_at = time.monotonic() + return response + + @classmethod + @cached(maxsize=1024, ttl=7 * 24 * 60 * 60, skip_none=True) + def _request_json( + cls, + path: str, + params: Optional[dict[str, Any]] = None, + ) -> Any: + """请求 LRCLIB JSON 接口，缓存命中与未命中结果并按 Retry-After 重试一次。""" + response = cls._request_once(path, params) + if response is None: + return None + try: + if response.status_code == 404: + return {} + if response.status_code in (429, 503): + retry_after = cls._retry_after_seconds(response.headers.get("Retry-After")) + response.close() + time.sleep(retry_after) + response = cls._request_once(path, params) + if response is None: + return None + if response.status_code == 404: + return {} + if response.status_code != 200: + logger.warning( + f"LRCLIB 请求失败：{response.status_code} {response.text[:200]}" + ) + return None + return response.json() + except (TypeError, ValueError) as err: + logger.warning(f"LRCLIB 响应解析失败：{err}") + return None + finally: + if response is not None: + response.close() + + @staticmethod + def _retry_after_seconds(value: Any) -> float: + """解析 LRCLIB 限流等待秒数，异常值回退到一秒。""" + try: + return max(float(value), 0.0) + except (TypeError, ValueError): + return 1.0 diff --git a/app/schemas/types.py b/app/schemas/types.py index c0cc0a8d1..e9decf2b1 100644 --- a/app/schemas/types.py +++ b/app/schemas/types.py @@ -464,6 +464,8 @@ class OtherModulesType(Enum): Redis = "Redis" # ListenBrainz ListenBrainz = "ListenBrainz" + # LRCLIB 歌词 + Lrclib = "LRCLIB" class NameValueEnum(Enum): @@ -505,3 +507,4 @@ class ScrapingMetadata(NameValueEnum): DISC = "光盘图" CLEARART = "透明艺术图" LANDSCAPE = "横版缩略图" + LYRICS = "歌词" diff --git a/tests/test_audio_metadata.py b/tests/test_audio_metadata.py index b25b06b34..d2433163b 100644 --- a/tests/test_audio_metadata.py +++ b/tests/test_audio_metadata.py @@ -1,5 +1,6 @@ from pathlib import Path from types import SimpleNamespace +from unittest.mock import Mock from app.core.context import MusicInfo from app.helper.audio import AudioMetadataHelper @@ -86,3 +87,28 @@ def test_write_audio_metadata_maps_music_info_to_easy_tags(monkeypatch): assert audio.tags["title"] == ["Get Lucky"] assert audio.tags["artist"] == ["Daft Punk", "Pharrell Williams"] assert audio.tags["tracknumber"] == ["8/13"] + + +def test_write_audio_metadata_can_embed_cover_without_rewriting_tags(monkeypatch): + """音乐封面策略应能在标签策略关闭时独立执行。""" + audio = SimpleNamespace(tags={"title": ["Original"]}) + monkeypatch.setattr("app.helper.audio.MutagenFile", lambda *_args, **_kwargs: audio) + write_cover = Mock() + monkeypatch.setattr(AudioMetadataHelper, "_write_cover", write_cover) + + success = AudioMetadataHelper.write( + Path("/music/track.flac"), + MusicInfo(title="Changed"), + cover_data=b"cover", + write_tags=False, + cover_overwrite=False, + ) + + assert success is True + assert audio.tags == {"title": ["Original"]} + write_cover.assert_called_once_with( + path=Path("/music/track.flac"), + cover_data=b"cover", + cover_mime="image/jpeg", + overwrite=False, + ) diff --git a/tests/test_lrclib_module.py b/tests/test_lrclib_module.py new file mode 100644 index 000000000..a2a8302d9 --- /dev/null +++ b/tests/test_lrclib_module.py @@ -0,0 +1,203 @@ +from app.core.context import MusicInfo +from app.modules.lrclib import LrclibModule + + +class _FakeLrclibResponse: + """模拟 LRCLIB HTTP 响应，供解析、缓存和限流测试复用。""" + + def __init__(self, payload, status_code=200, headers=None): + self._payload = payload + self.status_code = status_code + self.headers = headers or {} + self.text = "" + self.closed = False + + def json(self): + """返回预设 JSON 负载。""" + return self._payload + + def close(self): + """记录响应已关闭。""" + self.closed = True + + def __bool__(self): + """复现 requests.Response 对 4xx/5xx 响应返回 False 的行为。""" + return self.status_code < 400 + + +def test_music_lyrics_prefers_exact_signature_and_synced_lyrics(monkeypatch) -> None: + """元数据完整时应调用精确接口并优先返回同步歌词。""" + module = LrclibModule() + requested = [] + + def fake_request(path, params=None): + """记录精确查询参数并返回同步歌词。""" + requested.append((path, params)) + return { + "id": 3396226, + "instrumental": False, + "plainLyrics": "plain", + "syncedLyrics": "[00:01.00]synced", + } + + monkeypatch.setattr(module, "_request_json", fake_request) + + lyrics = module.music_lyrics( + MusicInfo( + title="I Want to Live", + artists=["Borislav Slavov"], + album="Baldur's Gate 3", + duration=233, + ) + ) + + assert lyrics is not None + assert lyrics.provider == "lrclib" + assert lyrics.provider_id == "3396226" + assert lyrics.content == "[00:01.00]synced" + assert lyrics.extension == ".lrc" + assert requested == [ + ( + "/api/get", + { + "track_name": "I Want to Live", + "artist_name": "Borislav Slavov", + "album_name": "Baldur's Gate 3", + "duration": 233, + }, + ) + ] + + +def test_music_lyrics_search_fallback_rejects_wrong_duration(monkeypatch) -> None: + """精确接口未命中后只能选取标题、艺术家和时长均可信的搜索结果。""" + module = LrclibModule() + + def fake_request(path, params=None): + """精确查询返回未命中，搜索返回一条错误版本和一条正确版本。""" + if path == "/api/get": + return {} + return [ + { + "id": 1, + "trackName": "晴天", + "artistName": "周杰伦", + "albumName": "演唱会", + "duration": 310, + "plainLyrics": "wrong", + }, + { + "id": 2, + "trackName": "晴天", + "artistName": "周杰伦", + "albumName": "叶惠美", + "duration": 269, + "plainLyrics": "correct", + }, + ] + + monkeypatch.setattr(module, "_request_json", fake_request) + + lyrics = module.music_lyrics( + MusicInfo( + title="晴天", + artists=["周杰伦"], + album="叶惠美", + duration=270, + ) + ) + + assert lyrics is not None + assert lyrics.provider_id == "2" + assert lyrics.content == "correct" + assert lyrics.extension == ".txt" + + +def test_request_json_honors_retry_after_once(monkeypatch) -> None: + """LRCLIB 返回带 Retry-After 的过载响应时应等待并串行重试一次。""" + responses = iter( + [ + _FakeLrclibResponse(None, status_code=503, headers={"Retry-After": "2"}), + _FakeLrclibResponse([{"id": 1}], status_code=200), + ] + ) + sleeps = [] + monkeypatch.setattr(LrclibModule, "_request_once", lambda *_args, **_kwargs: next(responses)) + monkeypatch.setattr("app.modules.lrclib.time.sleep", lambda seconds: sleeps.append(seconds)) + LrclibModule._request_json.cache_clear() + + result = LrclibModule._request_json( + "/api/search", + params={"track_name": "晴天", "artist_name": "周杰伦"}, + ) + + assert result == [{"id": 1}] + assert sleeps == [2.0] + + +def test_request_json_retry_network_failure_is_not_cached(monkeypatch) -> None: + """过载重试遇到网络失败时应安全返回且不把失败写入缓存。""" + responses = iter( + [ + _FakeLrclibResponse(None, status_code=429, headers={"Retry-After": "0"}), + None, + _FakeLrclibResponse([{"id": 3}], status_code=200), + ] + ) + monkeypatch.setattr(LrclibModule, "_request_once", lambda *_args, **_kwargs: next(responses)) + monkeypatch.setattr("app.modules.lrclib.time.sleep", lambda _seconds: None) + LrclibModule._request_json.cache_clear() + params = {"track_name": "retry"} + + first = LrclibModule._request_json("/api/search", params=params) + second = LrclibModule._request_json("/api/search", params=params) + + assert first is None + assert second == [{"id": 3}] + + +def test_request_json_caches_not_found_response(monkeypatch) -> None: + """未匹配结果也应缓存，避免整库重复刮削持续请求同一首歌。""" + calls = {"count": 0} + + def fake_request(*_args, **_kwargs): + """记录请求次数并返回 404。""" + calls["count"] += 1 + return _FakeLrclibResponse(None, status_code=404) + + monkeypatch.setattr(LrclibModule, "_request_once", fake_request) + LrclibModule._request_json.cache_clear() + + first = LrclibModule._request_json("/api/get", params={"track_name": "missing"}) + second = LrclibModule._request_json("/api/get", params={"track_name": "missing"}) + + assert first == second == {} + assert calls["count"] == 1 + + +def test_request_json_caches_success_response(monkeypatch) -> None: + """成功歌词响应应进入有界 TLRU 缓存，相同签名只访问一次外部 API。""" + calls = {"count": 0} + + def fake_request(*_args, **_kwargs): + """记录请求次数并返回固定歌词。""" + calls["count"] += 1 + return _FakeLrclibResponse( + {"id": 2, "syncedLyrics": "[00:01.00]晴天"}, + status_code=200, + ) + + monkeypatch.setattr(LrclibModule, "_request_once", fake_request) + LrclibModule._request_json.cache_clear() + params = { + "track_name": "晴天", + "artist_name": "周杰伦", + "album_name": "叶惠美", + "duration": 269, + } + + first = LrclibModule._request_json("/api/get", params=params) + second = LrclibModule._request_json("/api/get", params=params) + + assert first == second == {"id": 2, "syncedLyrics": "[00:01.00]晴天"} + assert calls["count"] == 1 diff --git a/tests/test_music_scrape.py b/tests/test_music_scrape.py index 6ed9ada2f..b3bc1e79a 100644 --- a/tests/test_music_scrape.py +++ b/tests/test_music_scrape.py @@ -1,10 +1,12 @@ from types import SimpleNamespace -from unittest.mock import Mock +from unittest.mock import Mock, patch -from app.chain.media import MediaChain -from app.core.context import MUSIC_ENTITY_ALBUM, MusicInfo +from app.chain.media import MediaChain, ScrapingConfig, _MusicScrapeFileResult +from app.core.context import MUSIC_ENTITY_ALBUM, MusicAlbumInfo, MusicInfo, MusicLyrics +from app.core.event import Event from app.core.meta import MetaMusic from app.schemas import FileItem +from app.schemas.types import EventType, ScrapingPolicy def _media_chain() -> MediaChain: @@ -58,17 +60,24 @@ def test_album_directory_scrape_processes_each_track_and_reuses_cover() -> None: chain = _media_chain() chain.storagechain = Mock() chain.scraping_policies = Mock() - chain.scraping_policies.option.return_value = SimpleNamespace( - is_skip=False, - is_overwrite=False, - ) + + def scraping_option(_target, metadata): + """保持原测试只验证标签和封面，歌词由独立用例覆盖。""" + return SimpleNamespace( + is_skip=metadata == "lyrics", + is_overwrite=False, + ) + + chain.scraping_policies.option.side_effect = scraping_option audio_files = [ FileItem(storage="local", path="/music/叶惠美/01.flac", type="file", name="01.flac"), FileItem(storage="local", path="/music/叶惠美/02.m4a", type="file", name="02.m4a"), ] chain.storagechain.list_files.return_value = audio_files chain._download_music_cover = Mock(return_value=(b"cover", "image/jpeg")) - chain._scrape_music_file = Mock(return_value=True) + chain._scrape_music_file = Mock( + return_value=_MusicScrapeFileResult(metadata_success=True) + ) album = _album_info() success, message = chain.scrape_music_metadata( @@ -102,3 +111,244 @@ def test_recording_identity_rejects_multi_track_directory_scrape() -> None: assert success is False assert message == "单曲 MusicBrainz ID 仅支持刮削单个音频文件，整目录请选择专辑" + + +def test_default_scraping_config_enables_missing_only_music_lyrics() -> None: + """新安装和未保存过该字段的用户应默认仅在缺失时下载歌词。""" + assert ScrapingConfig.get_default_config()["music_lyrics"] == ScrapingPolicy.MISSINGONLY + + +def test_album_track_match_uses_disc_track_title_and_duration() -> None: + """整张专辑刮削时应把本地音轨绑定到对应 Recording，不能复用专辑级身份。""" + album = MusicAlbumInfo( + source="musicbrainz", + media_id="album-1", + title="叶惠美", + tracks=[ + MusicInfo( + media_id="recording-1", + title="以父之名", + artists=["周杰伦"], + disc_number=1, + track_number=1, + duration=342, + ), + MusicInfo( + media_id="recording-3", + title="晴天", + artists=["周杰伦"], + disc_number=1, + track_number=3, + duration=269, + ), + ], + ) + + matched = MediaChain._match_music_album_track( + MetaMusic( + title="03 - 晴天", + artists=["周杰伦"], + disc_number=1, + track_number=3, + duration=270, + ), + album, + ) + + assert matched is not None + assert matched.media_id == "recording-3" + assert matched.title == "晴天" + + +def test_music_scrape_can_run_lyrics_without_tags_or_cover() -> None: + """标签和封面关闭时，歌词开关仍应独立驱动逐曲处理。""" + chain = _media_chain() + chain.storagechain = Mock() + chain.scraping_policies = Mock() + + def scraping_option(_target, metadata): + """仅开启歌词的缺失刮削策略。""" + return SimpleNamespace( + is_skip=metadata != "lyrics", + is_overwrite=False, + ) + + chain.scraping_policies.option.side_effect = scraping_option + chain._scrape_music_file = Mock( + return_value=_MusicScrapeFileResult( + metadata_success=True, + lyrics_status="saved", + ) + ) + music_chain = Mock() + + with patch("app.chain.music.MusicChain", return_value=music_chain): + success, message = chain.scrape_music_metadata( + FileItem( + storage="local", + path="/music/晴天.flac", + type="file", + name="晴天.flac", + ), + mediainfo=MusicInfo(title="晴天", artists=["周杰伦"]), + overwrite=False, + ) + + assert success is True + assert message == "已刮削 1 个音频文件，歌词新增 1 首、已存在 0 首、未匹配 0 首" + call = chain._scrape_music_file.call_args + assert call.kwargs["write_tags"] is False + assert call.kwargs["with_cover"] is False + assert call.kwargs["music_chain"] is music_chain + + +def test_write_music_lyrics_sidecar_creates_same_name_lrc(tmp_path) -> None: + """同步歌词应以 UTF-8 同名 LRC 文件写入音轨所在目录。""" + chain = _media_chain() + chain.storagechain = Mock() + audio_path = tmp_path / "晴天.flac" + audio_path.write_bytes(b"audio") + + success = chain._write_music_lyrics_sidecar( + fileitem=FileItem( + storage="local", + path=audio_path.as_posix(), + type="file", + name=audio_path.name, + ), + local_path=audio_path, + lyrics=MusicLyrics( + provider="lrclib", + provider_id="1", + synced_lyrics="[00:01.00]故事的小黄花", + ), + overwrite=False, + ) + + lyric_path = audio_path.with_suffix(".lrc") + assert success is True + assert lyric_path.read_text(encoding="utf-8") == "[00:01.00]故事的小黄花\n" + + +def test_missing_only_lyrics_skips_existing_sidecar(tmp_path) -> None: + """仅缺失策略发现同名歌词后不得再次请求歌词源。""" + chain = _media_chain() + existing = FileItem(storage="local", path=(tmp_path / "晴天.lrc").as_posix(), type="file") + chain.storagechain = Mock() + chain.storagechain.get_file_item.return_value = existing + music_chain = Mock() + lyrics_option = SimpleNamespace(is_skip=False) + + status = chain._scrape_music_lyrics( + fileitem=FileItem(storage="local", path=(tmp_path / "晴天.flac").as_posix(), type="file"), + local_path=tmp_path / "晴天.flac", + scrape_info=MetaMusic(title="晴天", artists=["周杰伦"]), + lyrics_option=lyrics_option, + overwrite=False, + music_chain=music_chain, + album_info=None, + ) + + assert status == "existing" + music_chain.lyrics.assert_not_called() + + +def test_write_music_lyrics_sidecar_uploads_to_remote_audio_directory(tmp_path) -> None: + """远端存储歌词应使用音轨父目录和同名文件上传。""" + chain = _media_chain() + chain.storagechain = Mock() + parent = FileItem(storage="u115", path="/Music/叶惠美", type="dir") + chain.storagechain.get_parent_item.return_value = parent + chain.storagechain.upload_file.return_value = FileItem( + storage="u115", + path="/Music/叶惠美/晴天.lrc", + type="file", + ) + local_path = tmp_path / "晴天.flac" + local_path.write_bytes(b"audio") + fileitem = FileItem( + storage="u115", + path="/Music/叶惠美/晴天.flac", + type="file", + name="晴天.flac", + ) + + success = chain._write_music_lyrics_sidecar( + fileitem=fileitem, + local_path=local_path, + lyrics=MusicLyrics(provider="lrclib", synced_lyrics="[00:01.00]晴天"), + overwrite=False, + ) + + assert success is True + upload = chain.storagechain.upload_file.call_args + assert upload.args[0] is parent + assert upload.kwargs["new_name"] == "晴天.lrc" + + +def test_music_scrape_event_preserves_independent_policy_overwrite() -> None: + """标签覆盖策略不得升级为全局覆盖并误覆盖缺失模式下的歌词。""" + chain = _media_chain() + chain.scrape_music_metadata = Mock(return_value=(True, "done")) + fileitem = FileItem(storage="local", path="/music/叶惠美", type="dir") + mediainfo = MusicInfo( + source="musicbrainz", + media_id="album-1", + music_type=MUSIC_ENTITY_ALBUM, + title="叶惠美", + ) + + chain.scrape_metadata_event( + Event( + event_type=EventType.MetadataScrape, + event_data={ + "fileitem": fileitem, + "mediainfo": mediainfo, + "overwrite": False, + }, + ) + ) + + chain.scrape_music_metadata.assert_called_once_with( + fileitem=fileitem, + mediainfo=mediainfo, + overwrite=False, + ) + + +def test_music_download_failure_is_attributed_only_to_enabled_outputs() -> None: + """音频下载失败时只标记实际启用的标签、封面或歌词任务。""" + chain = _media_chain() + chain.storagechain = Mock() + chain.storagechain.download_file.return_value = None + fileitem = FileItem( + storage="local", + path="/music/晴天.flac", + type="file", + name="晴天.flac", + ) + lyrics_option = SimpleNamespace(is_skip=False) + + lyrics_only = chain._scrape_music_file( + fileitem=fileitem, + mediainfo=MusicInfo(title="晴天"), + write_tags=False, + tag_overwrite=False, + with_cover=False, + lyrics_option=lyrics_option, + music_chain=Mock(), + ) + metadata_only = chain._scrape_music_file( + fileitem=fileitem, + mediainfo=MusicInfo(title="晴天"), + write_tags=True, + tag_overwrite=False, + with_cover=False, + lyrics_option=SimpleNamespace(is_skip=True), + music_chain=None, + ) + + assert lyrics_only.metadata_success is True + assert lyrics_only.lyrics_status == "failed" + assert metadata_only.metadata_success is False + assert metadata_only.lyrics_status == "disabled"