mirror of
https://github.com/jxxghp/MoviePilot.git
synced 2026-11-01 21:07:01 +08:00
6d4ce83837
严格检查全部宿主源码，清零类型错误预算并移除按模块类型豁免。 按原因、所属方法、错误码及语法指纹锁定必要兼容例外和站点类型桩， CI 阻断新增忽略、检查范围退化、本地桩替代真实接口及新的类型错误。 修复空值、异步和供应方类型边界，保持插件与 API 兼容合同， 并补充存储、订阅、通知、Agent 和工作流边界的回归覆盖。
327 lines
12 KiB
Python
327 lines
12 KiB
Python
"""
|
||
LRCLIB 歌词检索、严格候选匹配与请求缓存。
|
||
"""
|
||
|
||
import re
|
||
import threading
|
||
import time
|
||
from typing import Any, Optional, Tuple, Union, cast
|
||
|
||
from app.adapters.network.http import RequestUtils
|
||
from app.domain.context import MusicInfo, MusicLyrics
|
||
from app.domain.meta.metamusic import MetaMusic
|
||
from app.foundation.text import convert as zhconv_convert
|
||
from app.modules import _ModuleBase
|
||
from app.runtime.cache import CacheControls, cached
|
||
from app.runtime.log import logger
|
||
from app.runtime.settings import get_runtime_setting
|
||
from app.schemas.types import ModuleType, OtherModulesType
|
||
|
||
|
||
class LrclibModule(_ModuleBase):
|
||
"""
|
||
通过 LRCLIB 获取与单个音轨匹配的同步歌词或纯文本歌词。
|
||
"""
|
||
|
||
_source = "lrclib"
|
||
_request_interval = 0.3
|
||
_request_lock = threading.Lock()
|
||
_last_request_at = 0.0
|
||
_cooldown_until = 0.0
|
||
_match_pattern = re.compile(r"[^\w]+", flags=re.UNICODE)
|
||
|
||
def init_module(self) -> None:
|
||
"""
|
||
配置变化后清理跨实例请求缓存和供应商冷却状态。
|
||
"""
|
||
cast(CacheControls, self._request_json).cache_clear()
|
||
type(self)._cooldown_until = 0.0
|
||
|
||
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]:
|
||
"""
|
||
按单曲信息查询 LRCLIB，精确接口无结果时回退候选搜索。
|
||
|
||
艺术家优先取首位歌曲艺术家，缺失时使用专辑艺术家。
|
||
搜索回退校验标题、艺术家和可用时长，时长相差超过两秒的候选不接受。
|
||
|
||
:param music: 标题、艺术家、专辑和秒级时长线索，至少需要标题与艺术家
|
||
:return: 同步歌词、纯文本或纯音乐标记；线索不足、请求失败或无匹配时返回 None
|
||
"""
|
||
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
|
||
|
||
query_title = self._query_text(title)
|
||
query_artist = self._query_text(artist)
|
||
query_album = self._query_text(album)
|
||
exact_params: dict[str, Any] = {
|
||
"track_name": query_title,
|
||
"artist_name": query_artist,
|
||
}
|
||
if query_album:
|
||
exact_params["album_name"] = query_album
|
||
if duration:
|
||
exact_params["duration"] = duration
|
||
payload = self._request_json("/api/get", params=exact_params)
|
||
match_score = 100
|
||
if not payload:
|
||
results = self._request_json(
|
||
"/api/search",
|
||
params={
|
||
"track_name": query_title,
|
||
"artist_name": query_artist,
|
||
**({"album_name": query_album} if query_album else {}),
|
||
},
|
||
)
|
||
payload = self._select_result(
|
||
results if isinstance(results, list) else [],
|
||
title=query_title,
|
||
artist=query_artist,
|
||
album=query_album,
|
||
duration=duration,
|
||
)
|
||
match_score = 90
|
||
return self._to_lyrics(payload, match_score=match_score)
|
||
|
||
def music_lyrics_candidates(
|
||
self,
|
||
music: Union[MetaMusic, MusicInfo],
|
||
) -> list[MusicLyrics]:
|
||
"""
|
||
把旧单结果歌词接口包装为通用候选列表。
|
||
|
||
:param music: 待查单曲信息，匹配规则与 music_lyrics 一致
|
||
:return: 包含一个候选的列表；没有可用歌词时为空列表
|
||
"""
|
||
lyrics = self.music_lyrics(music)
|
||
return [lyrics] if lyrics else []
|
||
|
||
@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:
|
||
"""
|
||
移除大小写、标点和空白差异，生成歌词匹配文本。
|
||
"""
|
||
simplified = zhconv_convert(str(value or "").casefold(), "zh-hans")
|
||
return cls._match_pattern.sub("", str(simplified))
|
||
|
||
@staticmethod
|
||
def _query_text(value: str) -> str:
|
||
"""
|
||
将歌词查询字段统一为简体，适配 LRCLIB 的主流索引写法。
|
||
"""
|
||
return str(zhconv_convert(value, "zh-hans")).strip()
|
||
|
||
@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, match_score: int = 90) -> 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
|
||
lyricsfile = str(payload.get("lyricsfile") or "").strip() or None
|
||
instrumental = bool(payload.get("instrumental"))
|
||
if not instrumental and not plain_lyrics and not synced_lyrics and not lyricsfile:
|
||
return None
|
||
return MusicLyrics(
|
||
provider=cls._source,
|
||
provider_id=str(payload["id"]),
|
||
instrumental=instrumental,
|
||
plain_lyrics=plain_lyrics,
|
||
synced_lyrics=synced_lyrics,
|
||
lyricsfile=lyricsfile,
|
||
match_score=match_score,
|
||
provider_priority=20,
|
||
)
|
||
|
||
@classmethod
|
||
def _request_once(
|
||
cls,
|
||
path: str,
|
||
params: Optional[dict[str, Any]],
|
||
base_url: Optional[str] = None,
|
||
) -> Any:
|
||
"""
|
||
串行执行一次 LRCLIB 请求，确保批量专辑刮削遵守最小请求间隔。
|
||
"""
|
||
with cls._request_lock:
|
||
if time.monotonic() < cls._cooldown_until:
|
||
return None
|
||
delay = cls._request_interval - (time.monotonic() - cls._last_request_at)
|
||
if delay > 0:
|
||
time.sleep(delay)
|
||
response = RequestUtils(
|
||
headers={
|
||
"User-Agent": f"{get_runtime_setting('USER_AGENT')} (https://github.com/jxxghp/MoviePilot)",
|
||
"Accept": "application/json",
|
||
},
|
||
proxies=get_runtime_setting('PROXY'),
|
||
timeout=20,
|
||
).get_res(
|
||
f"{(base_url or str(get_runtime_setting('LRCLIB_BASE_URL'))).rstrip('/')}{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,
|
||
base_url: Optional[str] = None,
|
||
) -> Any:
|
||
"""
|
||
请求 LRCLIB JSON 接口，缓存命中与未命中结果并按 Retry-After 重试一次。
|
||
"""
|
||
response = cls._request_once(path, params, base_url)
|
||
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()
|
||
max_wait = max(int(get_runtime_setting('LYRICS_PROVIDER_RETRY_MAX_WAIT')), 0)
|
||
if retry_after > max_wait:
|
||
cls._cooldown_until = time.monotonic() + retry_after
|
||
logger.warning(f"LRCLIB 进入冷却 {retry_after:g} 秒，跳过当前批次后续请求")
|
||
response = None
|
||
return None
|
||
time.sleep(retry_after)
|
||
response = cls._request_once(path, params, base_url)
|
||
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
|