""" AMLL TTML 歌词搜索、录音匹配和下载模块。 """ import json import math import threading import time from datetime import datetime, timezone from email.utils import parsedate_to_datetime from typing import Any, Optional, cast from app.adapters.network.http import RequestUtils from app.domain.context import MusicInfo, MusicLyrics from app.domain.meta.metamusic import MetaMusic from app.modules import _ModuleBase from app.modules.amll.lyrics import parse_lyrics from app.modules.amll.matching import ( lyric_id, match_score, normalize_isrc, search_identity, text_values, ) from app.runtime.cache import Cache from app.runtime.log import logger from app.runtime.settings import get_runtime_setting from app.schemas.types import ModuleType, OtherModulesType class AmllModule(_ModuleBase): """ 聚合 AMLL 原生 API 的可信 TTML 候选，由宿主刮削链统一保存。 """ # 文本搜索仅检查首页，最多读取 20 条身份候选 _page_size = 20 # 一次搜索最多下载三个不同候选的完整歌词 _download_limit = 3 # 服务端 Retry-After 最多延长冷却 300 秒 _maximum_cooldown = 300.0 _cache_region = "app.modules.amll" def __init__(self) -> None: """ 建立实例级请求互斥和冷却状态，不在模块构造时请求网络。 """ super().__init__() self._request_lock = threading.Lock() self._cooldown_until = 0.0 def init_module(self) -> None: """ 来源配置变化时清理缓存与冷却，后续查询使用新的服务地址。 """ with self._request_lock: Cache(maxsize=1024, ttl=3600).clear(region=self._cache_region) self._cooldown_until = 0.0 def init_setting(self) -> None: """ 公共歌词服务无需密钥，请求由音乐歌词刮削策略控制。 """ def stop(self) -> None: """ 当前模块不持有后台任务或持久网络客户端。 """ def test(self) -> tuple[bool, str]: """ 用有界搜索请求检查原生歌词接口连通性。 """ response = self._search(self._base_url(), "test", "", "") return (True, "") if response is not None else (False, "AMLL TTML API 网络连接失败") @staticmethod def get_name() -> str: """ 返回歌词来源展示名称。 """ return "AMLL TTML" @staticmethod def get_type() -> ModuleType: """ 返回歌词模块所属的杂项能力分类。 """ return ModuleType.Other @staticmethod def get_subtype() -> OtherModulesType: """ 返回 AMLL 歌词来源的稳定模块标识。 """ return OtherModulesType.Amll @staticmethod def get_priority() -> int: """ 在候选聚合阶段按常规歌词来源顺序执行。 """ return 6 @staticmethod def _base_url() -> str: """ 解析当前配置地址，空地址表示本次不调用该来源。 """ return str(get_runtime_setting("AMLL_BASE_URL") or "").strip().rstrip("/") def music_lyrics_candidates(self, music: MetaMusic | MusicInfo) -> list[MusicLyrics]: """ 优先按 ISRC 查询，未获得有效歌词时回退严格的文本身份匹配。 文本搜索仅检查首页 20 条，去重后最多下载三个候选并再次核对身份。 请求在实例内串行执行，限流或服务繁忙时遵守有上限的冷却期。 :param music: 待查单曲的标题、艺术家、专辑和可选 ISRC :return: 最多三个已验证歌词候选；无可用来源配置或无匹配时为空列表 """ base_url = self._base_url() if not base_url: return [] isrc = normalize_isrc(music.isrc) if isrc: payload = self._lookup_isrc(base_url, isrc) if isinstance(payload, dict) and isrc in { normalize_isrc(value) for value in text_values(payload, "isrcs") }: lyrics = self._to_lyrics(music, payload, score=100) if lyrics: return [lyrics] return self._search_candidates(music, base_url) def _search_candidates(self, music: MetaMusic | MusicInfo, base_url: str) -> list[MusicLyrics]: """ 搜索仅提供候选身份，详情下载后再次核对身份与请求 ID。 """ title, artists, album = search_identity(music) if not title or not artists: return [] response = self._search(base_url, title, artists[0], album) items = response.get("items") if isinstance(response, dict) else None if not isinstance(items, list): return [] candidates: list[MusicLyrics] = [] requested: set[str] = set() for item in items[:self._page_size]: if not isinstance(item, dict) or not match_score(music, item): continue identifier = lyric_id(item) if not identifier or identifier in requested: continue requested.add(identifier) payload = self._get(base_url, identifier) if isinstance(payload, dict) and lyric_id(payload) == identifier: score = match_score(music, payload) lyrics = self._to_lyrics(music, payload, score=score) if score else None if lyrics: candidates.append(lyrics) if len(requested) >= self._download_limit: break return candidates @staticmethod def _to_lyrics( music: MetaMusic | MusicInfo, payload: dict[str, Any], score: int, ) -> Optional[MusicLyrics]: """ 将确认身份后的 TTML 正文交给来源解析器，保留逐词时间轴。 """ content = payload.get("lyrics") identifier = lyric_id(payload) if payload.get("format") != "ttml" or not identifier or not isinstance(content, str): return None title, artists, _album = search_identity(music) titles = text_values(payload, "musicNames") source_artists = text_values(payload, "artistNames") return parse_lyrics( content, title=title or next(iter(titles), ""), artist=" / ".join(artists or source_artists), provider_id=identifier, match_score=score, ) def _search( self, base_url: str, music_name: str, artist_name: str, album_name: str, ) -> Optional[dict[str, Any]]: """ 短期缓存动态搜索，控制单页候选数量并避免宽泛正文查询。 """ params = {"musicName": music_name, "page": 1, "pageSize": self._page_size} if artist_name: params["artistName"] = artist_name if album_name: params["albumName"] = album_name return self._cached_json("/v1/lyrics/search", params, base_url, ttl=3600) def _get(self, base_url: str, identifier: str) -> Optional[dict[str, Any]]: """ 特定歌词 ID 内容固定，成功下载后使用独立的较长缓存。 """ return self._cached_json("/v1/lyrics/get", {"id": identifier}, base_url, ttl=7 * 24 * 3600) def _lookup_isrc(self, base_url: str, isrc: str) -> Optional[dict[str, Any]]: """ ISRC 对应的最新版歌词会变化，因此使用短期缓存。 """ return self._cached_json("/v1/lyrics/get", {"isrc": isrc}, base_url, ttl=3600) def _cached_json( self, path: str, params: dict[str, Any], base_url: str, ttl: int, ) -> Optional[dict[str, Any]]: """ 按来源地址隔离缓存，未命中短期保存，网络失败保持可立即重试。 """ cache = Cache(maxsize=1024, ttl=3600) key = json.dumps([base_url, path, params], ensure_ascii=False, sort_keys=True) cached_value = cache.get(key, region=self._cache_region) if isinstance(cached_value, dict): return cast(dict[str, Any], cached_value) result = self._request_json(path, params, base_url) if result is not None: has_result = bool(result.get("items")) if path.endswith("/search") else bool(result) cache.set(key, result, ttl=ttl if has_result else 300, region=self._cache_region) return result def _request_json( self, path: str, params: dict[str, Any], base_url: str, ) -> Optional[dict[str, Any]]: """ 串行发起有限超时请求，限流期间跳过该来源并确保释放响应。 """ with self._request_lock: if time.monotonic() < self._cooldown_until: return None try: response = RequestUtils( headers={"Accept": "application/json", "User-Agent": get_runtime_setting("USER_AGENT")}, proxies=get_runtime_setting("PROXY"), timeout=10, ).get_res(f"{base_url}{path}", params=params) except (OSError, ValueError) as error: logger.warning(f"AMLL TTML API 网络请求失败：{error}") return None if response is None: return None try: if response.status_code == 404: return {} if response.status_code in (429, 503): delay = self._retry_after_seconds(response.headers.get("Retry-After")) self._cooldown_until = time.monotonic() + delay logger.warning(f"AMLL TTML API 进入冷却 {delay:g} 秒") return None if response.status_code != 200: logger.warning(f"AMLL TTML API 请求失败：HTTP {response.status_code}") return None payload = response.json() if not isinstance(payload, dict) or payload.get("status") != 200: return None data = payload.get("data") return cast(dict[str, Any], data) if isinstance(data, dict) else None except (TypeError, ValueError) as error: logger.warning(f"AMLL TTML API 响应解析失败：{error}") return None finally: response.close() @classmethod def _retry_after_seconds(cls, value: str | None) -> float: """ 解析秒数或 HTTP 日期，限制异常远期值并避免无限等待。 """ try: delay = float(value or "1") except ValueError: try: target = parsedate_to_datetime(value or "") delay = (target - datetime.now(timezone.utc)).total_seconds() except (TypeError, ValueError, OverflowError): delay = 1.0 if not math.isfinite(delay): return 1.0 return min(max(delay, 1.0), cls._maximum_cooldown)