feat: add qqmusic fetcher

2026-03-31 00:43:37 +02:00
parent d51a3eda1d
commit a0f988d0d8
4 changed files with 220 additions and 10 deletions
@@ -55,6 +55,9 @@ NETEASE_LYRIC_URL = "https://interface3.music.163.com/api/song/lyric"
 LRCLIB_API_URL = "https://lrclib.net/api/get"
 LRCLIB_SEARCH_URL = "https://lrclib.net/api/search"

+# QQ Music API (self-hosted proxy)
+QQ_MUSIC_API_URL = os.environ.get("QQ_MUSIC_API_URL", "").rstrip("/")
+
 # User-Agents
 UA_BROWSER = "Mozilla/5.0 (X11; Linux x86_64; rv:148.0) Gecko/20100101 Firefox/148.0"
 UA_LRCFETCH = "LRCFetch (https://github.com/Uyanide/lrcfetch)"
@@ -17,6 +17,7 @@ from loguru import logger
 from typing import Literal

 from .fetchers.netease import NeteaseFetcher
+from .fetchers.qqmusic import QQMusicFetcher
 from .fetchers.lrclib_search import LrclibSearchFetcher
 from .fetchers.lrclib import LrclibFetcher
 from .fetchers.spotify import SpotifyFetcher
@@ -28,9 +29,23 @@ from .lrc import LRC_LINE_RE, normalize_tags
 from .config import TTL_SYNCED, TTL_UNSYNCED, TTL_NOT_FOUND, TTL_NETWORK_ERROR
 from .models import TrackMeta, LyricResult, CacheStatus

-METHODS = ("local", "cache-search", "spotify", "lrclib", "lrclib-search", "netease")
+METHODS = (
+    "local",
+    "cache-search",
+    "spotify",
+    "lrclib",
+    "lrclib-search",
+    "netease",
+    "qqmusic",
+)
 FetcherMethodType = Literal[
-    "local", "cache-search", "spotify", "lrclib", "lrclib-search", "netease"
+    "local",
+    "cache-search",
+    "spotify",
+    "lrclib",
+    "lrclib-search",
+    "netease",
+    "qqmusic",
 ]


@@ -78,6 +93,7 @@ class LrcManager:
            "lrclib": LrclibFetcher(),
            "lrclib-search": LrclibSearchFetcher(),
            "netease": NeteaseFetcher(),
+            "qqmusic": QQMusicFetcher(),
        }
        assert set(self.fetchers) == set(METHODS), (
            f"METHODS and fetchers out of sync: {set(METHODS) ^ set(self.fetchers)}"
@@ -105,6 +121,7 @@ class LrcManager:
        if track.title:
            sequence.append(self.fetchers["lrclib-search"])
        sequence.append(self.fetchers["netease"])
+        sequence.append(self.fetchers["qqmusic"])

        logger.debug(f"Fallback sequence: {[f.source_name for f in sequence]}")
        return sequence
@@ -0,0 +1,173 @@
+"""
+Description: QQ Music fetcher via self-hosted API proxy
+"""
+
+"""
+Requires a running qq-music-api instance (Cloudflare Worker).
+The base URL is read from the QQ_MUSIC_API_URL environment variable.
+
+Search → pick best match by duration → fetch LRC lyrics.
+"""
+
+from typing import Optional
+import httpx
+from loguru import logger
+
+from .base import BaseFetcher
+from ..models import TrackMeta, LyricResult, CacheStatus
+from ..lrc import is_synced
+from ..config import (
+    HTTP_TIMEOUT,
+    TTL_NOT_FOUND,
+    TTL_NETWORK_ERROR,
+    DURATION_TOLERANCE_MS,
+    QQ_MUSIC_API_URL,
+)
+
+
+class QQMusicFetcher(BaseFetcher):
+    @property
+    def source_name(self) -> str:
+        return "qqmusic"
+
+    def _search(self, track: TrackMeta, limit: int = 10) -> Optional[str]:
+        """Search QQ Music and return the best-matching song MID."""
+        query = f"{track.artist or ''} {track.title or ''}".strip()
+        if not query:
+            return None
+
+        logger.debug(f"QQMusic: searching for '{query}' (limit={limit})")
+
+        try:
+            with httpx.Client(timeout=HTTP_TIMEOUT) as client:
+                resp = client.get(
+                    f"{QQ_MUSIC_API_URL}/api/search",
+                    params={"keyword": query, "type": "song", "num": limit},
+                )
+                resp.raise_for_status()
+                data = resp.json()
+
+            if data.get("code") != 0:
+                logger.error(f"QQMusic: search API error: {data}")
+                return None
+
+            songs = data.get("data", {}).get("list", [])
+            if not songs:
+                logger.debug("QQMusic: search returned 0 results")
+                return None
+
+            logger.debug(f"QQMusic: search returned {len(songs)} candidates")
+
+            # Duration-based best-match selection
+            if track.length is not None:
+                track_ms = track.length
+                best_mid: Optional[str] = None
+                best_diff = float("inf")
+
+                for song in songs:
+                    if not isinstance(song, dict):
+                        continue
+                    mid = song.get("mid")
+                    name = song.get("name", "?")
+                    # interval is in seconds
+                    interval = song.get("interval")
+                    if not isinstance(interval, int):
+                        logger.debug(
+                            f"  candidate {mid} '{name}': no duration, skipped"
+                        )
+                        continue
+                    duration_ms = interval * 1000
+                    diff = abs(duration_ms - track_ms)
+                    logger.debug(
+                        f"  candidate {mid} '{name}': "
+                        f"duration={duration_ms}ms, diff={diff}ms"
+                    )
+                    if diff < best_diff:
+                        best_diff = diff
+                        best_mid = mid
+
+                if best_mid is not None and best_diff <= DURATION_TOLERANCE_MS:
+                    logger.debug(
+                        f"QQMusic: selected mid={best_mid} (diff={best_diff}ms)"
+                    )
+                    return best_mid
+
+                logger.debug(
+                    f"QQMusic: no candidate within {DURATION_TOLERANCE_MS}ms "
+                    f"(best diff={best_diff}ms)"
+                )
+                return None
+
+            # No duration info — take the first result
+            first = songs[0]
+            if not isinstance(first, dict) or "mid" not in first:
+                logger.error("QQMusic: first search result has no 'mid'")
+                return None
+            logger.debug(
+                f"QQMusic: no duration available, using first result "
+                f"mid={first['mid']} '{first.get('name', '?')}'"
+            )
+            return first["mid"]
+
+        except Exception as e:
+            logger.error(f"QQMusic: search failed: {e}")
+            return None
+
+    def _get_lyric(self, mid: str) -> Optional[LyricResult]:
+        """Fetch lyrics for a given QQ Music song MID."""
+        logger.debug(f"QQMusic: fetching lyrics for mid={mid}")
+
+        try:
+            with httpx.Client(timeout=HTTP_TIMEOUT) as client:
+                resp = client.get(
+                    f"{QQ_MUSIC_API_URL}/api/lyric",
+                    params={"mid": mid},
+                )
+                resp.raise_for_status()
+                data = resp.json()
+
+            if data.get("code") != 0:
+                logger.error(f"QQMusic: lyric API error: {data}")
+                return LyricResult(
+                    status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR
+                )
+
+            lrc = data.get("data", {}).get("lyric", "")
+            if not isinstance(lrc, str) or not lrc.strip():
+                logger.debug(f"QQMusic: empty lyrics for mid={mid}")
+                return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
+
+            synced = is_synced(lrc)
+            status = (
+                CacheStatus.SUCCESS_SYNCED if synced else CacheStatus.SUCCESS_UNSYNCED
+            )
+            logger.info(
+                f"QQMusic: got {status.value} lyrics for mid={mid} "
+                f"({len(lrc.splitlines())} lines)"
+            )
+            return LyricResult(
+                status=status, lyrics=lrc.strip(), source=self.source_name
+            )
+
+        except Exception as e:
+            logger.error(f"QQMusic: lyric fetch failed for mid={mid}: {e}")
+            return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)
+
+    def fetch(self, track: TrackMeta) -> Optional[LyricResult]:
+        """Search for the track and fetch its lyrics."""
+        if not QQ_MUSIC_API_URL:
+            logger.debug("QQMusic: skipped — QQ_MUSIC_API_URL not configured")
+            return None
+
+        query = f"{track.artist or ''} {track.title or ''}".strip()
+        if not query:
+            logger.debug("QQMusic: skipped — insufficient metadata")
+            return None
+
+        logger.info(f"QQMusic: fetching lyrics for {track.display_name()}")
+        mid = self._search(track)
+        if not mid:
+            logger.debug(f"QQMusic: no match found for {track.display_name()}")
+            return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)
+
+        return self._get_lyric(mid)