lrx-cli/lrcfetch/fetchers/netease.py

"""Netease Cloud Music fetcher.

Uses the public cloudsearch API for searching and the song/lyric API for
retrieving lyrics. No authentication required.

Search results are filtered by duration when the track has a known length
to avoid returning lyrics for the wrong version of a song.
"""

import httpx
from typing import Optional
from loguru import logger
from lrcfetch.models import TrackMeta, LyricResult, CacheStatus
from lrcfetch.fetchers.base import BaseFetcher
from lrcfetch.lrc import is_synced
from lrcfetch.config import (
    HTTP_TIMEOUT,
    TTL_NOT_FOUND,
    TTL_NETWORK_ERROR,
    DURATION_TOLERANCE_MS,
    NETEASE_SEARCH_URL,
    NETEASE_LYRIC_URL,
    UA_BROWSER,
)

_HEADERS = {
    "User-Agent": UA_BROWSER,
    "Referer": "https://music.163.com/",
}


class NeteaseFetcher(BaseFetcher):
    @property
    def source_name(self) -> str:
        return "netease"

    def _search(self, track: TrackMeta, limit: int = 10) -> Optional[int]:
        """Search Netease and return the best-matching song ID.

        When ``track.length`` is available, candidates are ranked by duration
        difference and only accepted if within ``DURATION_TOLERANCE_MS``.
        """
        query = f"{track.artist or ''} {track.title or ''}".strip()
        if not query:
            return None

        logger.debug(f"Netease: searching for '{query}' (limit={limit})")

        try:
            with httpx.Client(timeout=HTTP_TIMEOUT) as client:
                resp = client.post(
                    NETEASE_SEARCH_URL,
                    headers=_HEADERS,
                    data={"s": query, "type": "1", "limit": str(limit), "offset": "0"},
                )
                resp.raise_for_status()
                result = resp.json()

            # Validate response
            if not isinstance(result, dict):
                logger.error(f"Netease: search returned non-dict: {type(result).__name__}")
                return None

            result_body = result.get("result")
            if not isinstance(result_body, dict):
                logger.debug(f"Netease: search 'result' field missing or invalid")
                return None

            songs = result_body.get("songs")
            if not isinstance(songs, list) or len(songs) == 0:
                logger.debug("Netease: search returned 0 results")
                return None

            logger.debug(f"Netease: search returned {len(songs)} candidates")

            # Duration-based best-match selection
            if track.length is not None:
                track_ms = track.length
                best_id: Optional[int] = None
                best_diff = float("inf")

                for song in songs:
                    if not isinstance(song, dict):
                        continue
                    sid = song.get("id")
                    name = song.get("name", "?")
                    duration = song.get("dt")  # milliseconds
                    if not isinstance(duration, int):
                        logger.debug(f"  candidate {sid} '{name}': no duration, skipped")
                        continue
                    diff = abs(duration - track_ms)
                    logger.debug(
                        f"  candidate {sid} '{name}': "
                        f"duration={duration}ms, diff={diff}ms"
                    )
                    if diff < best_diff:
                        best_diff = diff
                        best_id = sid

                if best_id is not None and best_diff <= DURATION_TOLERANCE_MS:
                    logger.debug(
                        f"Netease: selected id={best_id} (diff={best_diff}ms)"
                    )
                    return best_id

                logger.debug(
                    f"Netease: 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 "id" not in first:
                logger.error("Netease: first search result has no 'id'")
                return None
            logger.debug(
                f"Netease: no duration available, using first result "
                f"id={first['id']} '{first.get('name', '?')}'"
            )
            return first["id"]

        except Exception as e:
            logger.error(f"Netease: search failed: {e}")
            return None

    def _get_lyric(self, song_id: int) -> Optional[LyricResult]:
        """Fetch lyrics for a given Netease song ID."""
        logger.debug(f"Netease: fetching lyrics for song_id={song_id}")

        try:
            with httpx.Client(timeout=HTTP_TIMEOUT) as client:
                resp = client.post(
                    NETEASE_LYRIC_URL,
                    headers=_HEADERS,
                    data={
                        "id": str(song_id),
                        "cp": "false",
                        "tv": "0",
                        "lv": "0",
                        "rv": "0",
                        "kv": "0",
                        "yv": "0",
                        "ytv": "0",
                        "yrv": "0",
                    },
                )
                resp.raise_for_status()
                data = resp.json()

            # Validate response
            if not isinstance(data, dict):
                logger.error(f"Netease: lyric response is not dict: {type(data).__name__}")
                return LyricResult(status=CacheStatus.NETWORK_ERROR, ttl=TTL_NETWORK_ERROR)

            lrc_obj = data.get("lrc")
            if not isinstance(lrc_obj, dict):
                logger.debug(f"Netease: no 'lrc' object in response for song_id={song_id}")
                return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)

            lrc: str = lrc_obj.get("lyric", "")
            if not isinstance(lrc, str) or not lrc.strip():
                logger.debug(f"Netease: empty lyrics for song_id={song_id}")
                return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)

            # Determine sync status
            synced = is_synced(lrc)
            status = CacheStatus.SUCCESS_SYNCED if synced else CacheStatus.SUCCESS_UNSYNCED
            logger.info(
                f"Netease: got {status.value} lyrics for song_id={song_id} "
                f"({len(lrc.splitlines())} lines)"
            )
            return LyricResult(
                status=status, lyrics=lrc.strip(), source=self.source_name
            )

        except Exception as e:
            logger.error(f"Netease: lyric fetch failed for song_id={song_id}: {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."""
        query = f"{track.artist or ''} {track.title or ''}".strip()
        if not query:
            logger.debug("Netease: skipped — insufficient metadata")
            return None

        logger.info(f"Netease: fetching lyrics for {track.display_name()}")
        song_id = self._search(track)
        if not song_id:
            logger.debug(f"Netease: no match found for {track.display_name()}")
            return LyricResult(status=CacheStatus.NOT_FOUND, ttl=TTL_NOT_FOUND)

        return self._get_lyric(song_id)