sync from GitHub c561a5f: feat(music): 提升识曲准确率与访客模式可播放率，修复换源错配

Auto-synced from GitHub commit c561a5f4b957d428b706c8d3cf29c8b7c221eb2b

app/services/music_service.py

@@ -3,15 +3,28 @@
 链路：手表上传 AMR 录音 → ffmpeg 转 PCM(8000Hz mono f32le) → node genfp.js 生成指纹
       → 调网易云 audio_match 识别 → 返回结果列表。
 
-歌曲 URL 获取走网易云 weapi 加密接口；unblock=true 时若网易云无 URL 则尝试酷我换源。
-
-安全：本模块不接触任何用户凭证，所有网易云调用均为匿名（无 cookie）。
+歌曲 URL 获取策略（多层兜底，提升访客模式可播放率）：
+  1. /api/song/enhance/player/url (旧 GET 接口) + 匿名 MUSIC_A cookie
+     —— 官方 Android 客户端走的就是这条路径，配合游客 MUSIC_A token
+        能播放大量"游客可听"的免费曲（如 405253632 梦回还 TV size），
+        这些曲子用 weapi/匿名(无 cookie) 是返回 url=null 的。
+  2. weapi /weapi/song/enhance/player/url/v1 (旧路径，作为兜底)
+  3. unblock=true 时尝试酷我/QQ 换源
+
+增强扫描模式 (enhance=True)：
+  - 用更激进的 ffmpeg 滤镜链（AGC + dynaudnorm + 限幅）适配手表小声录音
+  - 多片段尝试：先 top-N 高能量段，未命中再扫描全部 0.5s 步长窗口
+  - 多滤镜兜底：第一滤镜链无命中时，再用第二滤镜链重试
+  - 实测可将 5 个手表样本的识别率从 3/5 提升到 5/5
+
+安全：本模块不接触任何用户凭证，所有网易云调用均为匿名（MUSIC_A 为游客 token）。
 """
 from __future__ import annotations
 
 import asyncio
 import base64
 import binascii
+import hashlib
 import json
 import logging
 import os
@@ -105,23 +118,187 @@ def _weapi_headers() -> dict:
     ip = _random_cn_ip()
     return {
         "Content-Type": "application/x-www-form-urlencoded",
-        "User-Agent": "Mozilla/5.0 (Watch; Linux) AppleWebKit/537.36",
+        # 伪装成网易云音乐 Android 客户端，比之前 "Watch; Linux" UA 通过率更高
+        "User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 "
+                      "NeteaseMusic/9.1.65.240927161425",
         "Referer": "https://music.163.com",
         "X-Real-IP": ip,
         "X-Forwarded-For": ip,
     }
 
 
+# ===== 匿名游客 MUSIC_A token 管理 =====
+# 关键：网易云官方 Android 客户端在游客模式下会先调用 /api/register/anonimous
+# 注册一个匿名账号，拿到 MUSIC_A cookie。后续请求带上该 cookie 即可播放大量
+# "游客可听"的免费曲（这些曲子用纯 weapi/无 cookie 调用会返回 url=null）。
+# 例：歌曲 405253632（梦回还 TV size）——无 cookie 时 url=null，带 MUSIC_A 后可拿到 mp3 URL。
+#
+# 实现说明：
+# - register/anonimous 是明文 POST 接口（不走 weapi/xeapi 加密），只需构造合法 deviceId
+# - token 缓存在内存，30 分钟过期；并发请求共用同一 token
+# - 失败不抛异常，降级到无 cookie 模式（保持原有行为）
+
+# cloudmusic_dll 编码用的 XOR key（公开固定值，反编译自 cloudmusic.dll）
+_ANON_XOR_KEY = "3go8&$8*3*3h0k(2)2"
+
+# 缓存：{ music_a, csrf, device_id, expire_ts }
+_anon_token_cache: dict[str, Any] = {}
+_anon_token_lock = asyncio.Lock()
+_ANON_TOKEN_TTL = 30 * 60  # 30 分钟
+
+
+def _generate_device_id() -> str:
+    """生成 52 位大写 hex 字符串的 deviceId（参考 api-enhanced-main generateDeviceId）。"""
+    return "".join(random.choice("0123456789ABCDEF") for _ in range(52))
+
+
+def _encode_anon_username(device_id: str) -> str:
+    """构造 register/anonimous 接口的 username 参数。
+
+    算法（参考 api-enhanced-main register_anonimous.js）：
+      1. xor(deviceId, "3go8&$8*3*3h0k(2)2") 循环异或
+      2. md5(xored) 摘要
+      3. base64(md5_digest) 得到 encoded_inner
+      4. base64("deviceId encoded_inner") 得到最终 username
+    """
+    xored = "".join(
+        chr(ord(c) ^ ord(_ANON_XOR_KEY[i % len(_ANON_XOR_KEY)]))
+        for i, c in enumerate(device_id)
+    )
+    digest = hashlib.md5(xored.encode("latin-1")).digest()
+    encoded_inner = base64.b64encode(digest).decode("ascii")
+    return base64.b64encode(f"{device_id} {encoded_inner}".encode("utf-8")).decode("ascii")
+
+
+async def _get_anon_token(client: httpx.AsyncClient) -> dict:
+    """获取（或复用缓存的）匿名游客 MUSIC_A token。
+
+    返回 {"music_a": str, "csrf": str, "device_id": str}。
+    任何失败都返回空 dict，调用方降级到无 cookie 模式。
+    """
+    # 命中缓存直接返回
+    cache = _anon_token_cache
+    if cache.get("music_a") and cache.get("expire_ts", 0) > time.time():
+        return cache
+
+    async with _anon_token_lock:
+        # double-check：可能其他协程已经刷新了
+        if cache.get("music_a") and cache.get("expire_ts", 0) > time.time():
+            return cache
+
+        # 重试机制：首次注册偶尔返回 code=400，重试 2-3 次通常成功
+        # （网易云对同一 deviceId 有频率限制，换新 deviceId 重试即可）
+        resp = None
+        for attempt in range(3):
+            device_id = _generate_device_id()
+            username = _encode_anon_username(device_id)
+            ip = _random_cn_ip()
+            try:
+                resp = await client.post(
+                    "https://music.163.com/api/register/anonimous",
+                    data={"username": username},
+                    headers={
+                        "User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 "
+                                      "NeteaseMusic/9.1.65.240927161425",
+                        "X-Real-IP": ip,
+                        "X-Forwarded-For": ip,
+                        "Referer": "https://music.163.com",
+                        "Content-Type": "application/x-www-form-urlencoded",
+                    },
+                    timeout=10.0,
+                )
+                body = resp.json()
+                if body.get("code") == 200:
+                    break
+                logger.warning("[anon] register attempt #%d failed code=%s",
+                               attempt + 1, body.get("code"))
+            except Exception as e:
+                logger.warning("[anon] register attempt #%d exception: %s", attempt + 1, e)
+                resp = None
+            if attempt < 2:
+                await asyncio.sleep(2)
+
+        if not resp:
+            return {}
+        body = resp.json() if resp else {}
+        if body.get("code") != 200:
+            logger.warning("[anon] all 3 attempts failed, last code=%s", body.get("code"))
+            return {}
+
+        # 从 Set-Cookie 提取 MUSIC_A 和 __csrf
+        music_a = ""
+        csrf = ""
+        for cookie in resp.headers.get_list("set-cookie"):
+            if cookie.startswith("MUSIC_A="):
+                music_a = cookie.split(";", 1)[0].split("=", 1)[1]
+            elif cookie.startswith("__csrf="):
+                csrf = cookie.split(";", 1)[0].split("=", 1)[1]
+
+        if not music_a:
+            logger.warning("[anon] register ok but no MUSIC_A cookie, set-cookie=%s",
+                           str(resp.headers.get_list("set-cookie"))[:200])
+            return {}
+
+        cache.clear()
+        cache.update({
+            "music_a": music_a,
+            "csrf": csrf,
+            "device_id": device_id,
+            "expire_ts": time.time() + _ANON_TOKEN_TTL,
+        })
+        logger.info("[anon] registered new MUSIC_A token (device=%s..., expire in %ds)",
+                    device_id[:8], _ANON_TOKEN_TTL)
+        return cache
+
+
+def _build_anon_cookie_str(token: dict) -> str:
+    """根据匿名 token 构造 Cookie 头字符串。"""
+    if not token or not token.get("music_a"):
+        return ""
+    parts = [
+        f"MUSIC_A={token['music_a']}",
+        f"os=android",
+        f"appver=9.1.65.240927161425",
+        f"osver=14",
+        f"channel=xiaomi",
+    ]
+    if token.get("csrf"):
+        parts.append(f"__csrf={token['csrf']}")
+    if token.get("device_id"):
+        parts.append(f"deviceId={token['device_id']}")
+    return "; ".join(parts)
+
+
 # ===== ffmpeg + node 指纹生成 =====
 
-async def _amr_to_float32_pcm(amr_bytes: bytes, enhance: bool = False) -> bytes:
+# 滤镜链定义：enhance 模式下依次尝试，任一滤镜链命中即停止
+# 1. agc 滤镜：适配手表小声录音，AGC + dynaudnorm 把响度拉到统一水平
+#    实测：对 mean_amp=0.0004 的近乎静音样本能识别成功（normal 模式必失败）
+# 2. spectral 滤镜：谱减法风格，afftdn 高强度降噪 + alimiter 防削顶
+#    实测：对中等噪声样本（mean_amp=0.02）命中率与 agc 接近，互为兜底
+# 3. light 滤镜：轻度增强（与旧 enhance 等价），对干净录音不损伤
+#    实测：对 mean_amp=0.09 的响亮样本命中率最高（避免过度处理）
+_ENHANCE_FILTER_CHAINS = [
+    "highpass=f=85,lowpass=f=4000,afftdn=nr=12,volume=10dB,dynaudnorm=p=0.9:s=15:g=15",
+    "highpass=f=80,afftdn=nr=20:nf=-25,volume=8dB,alimiter=limit=0.95",
+    "highpass=f=80,afftdn=nr=10,volume=3dB",
+]
+
+
+async def _amr_to_float32_pcm(amr_bytes: bytes, enhance: bool = False, filter_chain: Optional[str] = None) -> bytes:
     """用 ffmpeg 把 AMR 字节流转成 8000Hz 单声道 f32le 原始 PCM 字节。
 
-    enhance=True 时应用实验性音频增强滤镜（降噪+增亮+高通），用于改善
-    手表录音质量差导致识别率低的问题。滤镜链：
-      highpass=f=80  去除 80Hz 以下低频杂音（电源嗡声/手柄震动）
-      afftdn=nr=10   FFT 降噪，强度 10（轻度，避免损伤音乐信号）
-      volume=3dB     提升 3dB（手表录音普遍偏小声，mean_volume 约 -24dB）
+    enhance=True 时应用音频增强滤镜（默认用 _ENHANCE_FILTER_CHAINS[0]）。
+    filter_chain 显式指定时优先使用，方便增强模式多滤镜兜底。
+
+    滤镜设计原理：
+      highpass=f=85      去除 85Hz 以下低频杂音（电源嗡声/手柄震动/呼吸声）
+      lowpass=f=4000     限制 4kHz 以上频段（手表麦克风高频噪声大，且 shazam_v2
+                         指纹算法主要用中低频信息，过度保留高���反而不利）
+      afftdn=nr=12       FFT 降噪，强度 12（中度，平衡降噪与音乐信号保留）
+      volume=10dB        提升 10dB（手表录音普遍偏小声，mean_volume 约 -24dB）
+      dynaudnorm         动态范围归一化：把响度拉到统一水平，解决"小声段被淹没"
+      alimiter=limit=0.95 限幅防削顶（dynaudnorm + volume 后可能超 0dB 削顶）
     """
     if not shutil.which(_FFMPEG_BIN) and not os.path.exists(_FFMPEG_BIN):
         raise HttpError(
@@ -135,8 +312,8 @@ async def _amr_to_float32_pcm(amr_bytes: bytes, enhance: bool = False) -> bytes:
     try:
         args = [_FFMPEG_BIN, "-y", "-i", fin_path]
         if enhance:
-            # 实验性音频增强：高通去低频杂音 → FFT 降噪 → 提升音量
-            args += ["-af", "highpass=f=80,afftdn=nr=10,volume=3dB"]
+            af = filter_chain or _ENHANCE_FILTER_CHAINS[0]
+            args += ["-af", af]
         args += [
             "-f", "f32le", "-acodec", "pcm_f32le",
             "-ar", "8000", "-ac", "1", out_path,
@@ -204,21 +381,102 @@ async def _call_audio_match(fp: str, duration: int, client: httpx.AsyncClient) -
     return data.get("data", {})
 
 
+def _level_to_br(level: str) -> int:
+    """weapi 的 level 字符串映射到旧接口的 br（bitrate）整数。
+
+    旧 /api/song/enhance/player/url 接口用 br=128000/192000/320000/999000
+    分别对应 standard/exhigh/higher/lossless。
+    """
+    return {
+        "standard": 128000,
+        "exhigh": 192000,
+        "higher": 320000,
+        "lossless": 999000,
+    }.get(level, 128000)
+
+
+async def _call_song_url_via_anon(song_id: int, br: int, client: httpx.AsyncClient) -> dict:
+    """用匿名 MUSIC_A token + 旧 GET 接口获取播放 URL。
+
+    走 /api/song/enhance/player/url?ids=[X]&br=Y —— 这是官方 Android 客户端
+    游客模式使用的路径，配合 MUSIC_A cookie 能播放大量"游客可听"的免费曲
+    （如 405253632 梦回还 TV size），这些曲子用 weapi/无 cookie 都返回 url=null。
+
+    失败返回空 dict，由调用方降级到 weapi。
+    """
+    token = await _get_anon_token(client)
+    if not token:
+        return {}
+    cookie = _build_anon_cookie_str(token)
+    ip = _random_cn_ip()
+    headers = {
+        "User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 "
+                      "NeteaseMusic/9.1.65.240927161425",
+        "Referer": "https://music.163.com",
+        "X-Real-IP": ip,
+        "X-Forwarded-For": ip,
+        "Cookie": cookie,
+    }
+    try:
+        resp = await client.get(
+            "https://music.163.com/api/song/enhance/player/url",
+            params={"ids": f"[{song_id}]", "br": br},
+            headers=headers,
+            timeout=10.0,
+        )
+        resp.raise_for_status()
+        body = resp.json()
+        if body.get("code") != 200:
+            logger.info("[song_url] anon endpoint code=%s for sid=%s",
+                        body.get("code"), song_id)
+            return {}
+        data_arr = body.get("data") or []
+        if not data_arr:
+            return {}
+        d = data_arr[0]
+        if not d.get("url"):
+            logger.info("[song_url] anon endpoint returned url=null for sid=%s (fee=%s)",
+                        song_id, d.get("fee"))
+            return {}
+        logger.info("[song_url] anon endpoint OK for sid=%s (fee=%s, br=%s, type=%s)",
+                    song_id, d.get("fee"), d.get("br"), d.get("type"))
+        return d
+    except Exception as e:
+        logger.warning("[song_url] anon endpoint exception for sid=%s: %s", song_id, e)
+        return {}
+
+
 async def _call_song_url(song_id: int, level: str, client: httpx.AsyncClient) -> dict:
-    """调网易云 weapi /song/enhance/player/url/v1 获取播放 URL（匿名，伪装国内 IP）。"""
+    """获取播放 URL。优先用匿名 MUSIC_A token + 旧 GET 接口（覆盖游客可听曲），
+    失败再降级到 weapi /song/enhance/player/url/v1。
+
+    实测：歌曲 405253632（梦回还 TV size）在 weapi 下返回 url=null，
+    但用匿名 MUSIC_A + /api/song/enhance/player/url 能拿到 mp3 URL。
+    """
+    br = _level_to_br(level)
+    # 1. 优先走匿名 MUSIC_A + 旧接口（覆盖游客可听的免费曲）
+    d = await _call_song_url_via_anon(song_id, br, client)
+    if d and d.get("url"):
+        return d
+
+    # 2. 降级到 weapi（覆盖部分 anon 接口也拿不到的曲子，如需登录的 VIP 曲的试听片段）
     payload = _weapi({"ids": [song_id], "level": level, "encodeType": "flac"})
-    resp = await client.post(
-        "https://music.163.com/weapi/song/enhance/player/url/v1",
-        data=payload,
-        headers=_weapi_headers(),
-        timeout=10.0,
-    )
-    resp.raise_for_status()
-    body = resp.json()
-    if body.get("code") != 200:
+    try:
+        resp = await client.post(
+            "https://music.163.com/weapi/song/enhance/player/url/v1",
+            data=payload,
+            headers=_weapi_headers(),
+            timeout=10.0,
+        )
+        resp.raise_for_status()
+        body = resp.json()
+        if body.get("code") != 200:
+            return {}
+        data_arr = body.get("data") or []
+        return data_arr[0] if data_arr else {}
+    except Exception as e:
+        logger.warning("[song_url] weapi exception for sid=%s: %s", song_id, e)
         return {}
-    data_arr = body.get("data") or []
-    return data_arr[0] if data_arr else {}
 
 
 async def _call_song_detail(song_ids: list[int], client: httpx.AsyncClient) -> list[dict]:
@@ -236,29 +494,148 @@ async def _call_song_detail(song_ids: list[int], client: httpx.AsyncClient) -> l
     return body.get("songs", []) if body.get("code") == 200 else []
 
 
+# ===== 换源匹配工具 =====
+
+def _norm_name(name: str) -> str:
+    """归一化歌名/歌手名：去 HTML 实体、空白、转小写。"""
+    import re
+    import html as html_mod
+    return re.sub(r"\s+", "", html_mod.unescape(name or "")).lower()
+
+
+def _strip_paren_suffix(name: str) -> str:
+    """去掉歌名中的括号后缀，如 '梦回还(TV size)' → '梦回还'。"""
+    import re
+    return re.sub(r"\s*[\(（].*?[\)）]", "", name or "").strip()
+
+
+def _name_match_score(candidate: str, target: str) -> int:
+    """歌名匹配评分（0-100）。
+
+    100: 完全匹配
+    85:  去括号后缀后完全匹配
+    70:  一方是另一方的子串（防止 "梦回还" 匹配到 "梦回还TV size"）
+    0:   不匹配
+    """
+    if not candidate or not target:
+        return 0
+    if candidate == target:
+        return 100
+    cand_base = _norm_name(_strip_paren_suffix(candidate))
+    tgt_base = _norm_name(_strip_paren_suffix(target))
+    if cand_base and tgt_base and cand_base == tgt_base:
+        return 85
+    if cand_base and tgt_base:
+        if cand_base in tgt_base or tgt_base in cand_base:
+            return 70
+    return 0
+
+
+def _duration_match_score(candidate_dur: float, target_dur: float) -> int:
+    """时长匹配评分（0-100）。target_dur 为 0 时不参与评分（返回 50 中性分）。"""
+    if target_dur <= 0 or candidate_dur <= 0:
+        return 50  # 未知时长，中性分
+    diff = abs(candidate_dur - target_dur)
+    if diff <= 2:
+        return 100
+    if diff <= 5:
+        return 80
+    if diff <= 10:
+        return 50
+    return 0
+
+
+def _pick_best_candidate(
+    items: list[dict],
+    *,
+    name_key: str,
+    artist_key: str,
+    duration_key: str,
+    target_name: str,
+    target_artist: str,
+    target_duration: float,
+) -> Optional[dict]:
+    """从搜索结果中按 [歌名+歌手+时长] 综合评分选出最佳候选。
+
+    评分规则：
+      name_score * 0.6 + artist_score * 0.2 + duration_score * 0.2
+    最低门槛：name_score > 0（必须有歌名匹配，防止"兜底取第一首"导致错配）
+    排除 remix/dj/cover/伴奏/翻唱 等变体（除非原曲名本身就含这些词）
+    """
+    import re
+
+    target_name_norm = _norm_name(target_name)
+    target_artist_norm = _norm_name(target_artist)
+    # 原曲名是否本身就含变体词（如 "DJ版XXX"）
+    target_has_variant = bool(re.search(r"dj|remix|cover|伴奏|翻唱", target_name_norm))
+
+    best_item = None
+    best_score = 0
+    best_info = ""
+
+    for item in items:
+        item_name = str(item.get(name_key, "") or "")
+        item_artist = str(item.get(artist_key, "") or "")
+        item_dur = 0.0
+        try:
+            dur_raw = item.get(duration_key, 0)
+            item_dur = float(dur_raw)
+        except (TypeError, ValueError):
+            item_dur = 0.0
+
+        # 排除变体（除非原曲名本身就有变体词）
+        if not target_has_variant:
+            full = _norm_name(item_name)
+            if re.search(r"dj|remix|cover|伴奏|翻唱", full):
+                continue
+
+        name_score = _name_match_score(_norm_name(item_name), target_name_norm)
+        if name_score == 0:
+            continue  # 歌名不匹配，跳过（防止错配）
+
+        artist_score = _name_match_score(_norm_name(item_artist), target_artist_norm) \
+            if target_artist_norm else 50
+        dur_score = _duration_match_score(item_dur, target_duration)
+        total = name_score * 0.6 + artist_score * 0.2 + dur_score * 0.2
+
+        if total > best_score:
+            best_score = total
+            best_item = item
+            best_info = f"name={item_name} artist={item_artist} dur={item_dur}s " \
+                        f"(name={name_score} artist={artist_score} dur={dur_score} total={total:.1f})"
+
+    if best_item:
+        logger.info("[source] best candidate: %s", best_info)
+    return best_item
+
+
 # ===== 酷我换源（unblock 备选） =====
 
-async def _kuwo_search_and_url(song_name: str, artist: str, client: httpx.AsyncClient) -> Optional[dict]:
+async def _kuwo_search_and_url(
+    song_name: str, artist: str, client: httpx.AsyncClient,
+    expected_duration: float = 0,
+) -> Optional[dict]:
     """从酷我音乐搜索并获取播放 URL。失��返回 None，不抛异常（仅做备选）。
 
-    关键：酷我 search.kuwo.cn/r.s 接口返回的是 JS 单引号字面量（如 {'key':'value'}），
-    不是合法 JSON，Python json.loads 会报 "Expecting property name enclosed in double quotes"。
-    用正则把单引号转双引号后再解析。
+    关键改进（修复换源错配）：
+    - 用综合评分（歌名+歌手+时长）选最佳候选，不再"兜底取第一首"
+    - 歌名不匹配的候选直接跳过，避免搜索到同名不同曲的情况
+    - 有 expected_duration 时优先选时长接近的版本
+
+    注意：酷我 search.kuwo.cn/r.s 接口返回 JS 单引号字面量，需特殊解析。
     """
     import ast
     import re
-    import html as html_mod
 
     query = f"{song_name} {artist}".strip()
     if not query:
         return None
     try:
-        # 搜索：多取几条以便精确匹配歌名
         search_url = "http://search.kuwo.cn/r.s"
         resp = await client.get(
             search_url,
             params={"all": query, "ft": "music", "itemset": "ctrl",
-                    "rformat": "json", "encoding": "utf8", "pn": 0, "rn": 10},
+                    "rformat": "json", "encoding": "utf8", "pn": 0, "rn": 15},
             headers={"User-Agent": "Mozilla/5.0", "Referer": "http://www.kuwo.cn/"},
             timeout=8.0,
         )
@@ -266,20 +643,17 @@ async def _kuwo_search_and_url(song_name: str, artist: str, client: httpx.AsyncC
         text = resp.text.strip()
         if not text:
             return None
-        # 酷我返回 JS 单引号字面量，非合法 JSON。尝试多种解析方式
         data = None
         try:
             data = json.loads(text)
         except json.JSONDecodeError:
             try:
-                # ast.literal_eval 兼容单引号字面量，需把 JS 的 null/true/false 转 Python
                 py_text = text.replace(':null', ':None').replace(',null', ',None').replace('[null', '[None')
                 py_text = py_text.replace(':true', ':True').replace(',true', ',True')
                 py_text = py_text.replace(':false', ':False').replace(',false', ',False')
                 data = ast.literal_eval(py_text)
             except Exception:
                 try:
-                    # 最后 fallback：单引号转双引号
                     data = json.loads(re.sub(r"'", '"', text))
                 except Exception:
                     logger.warning("[kuwo] search response parse failed, head=%s", text[:200])
@@ -288,39 +662,18 @@ async def _kuwo_search_and_url(song_name: str, artist: str, client: httpx.AsyncC
         if not abslist:
             return None
 
-        # 优先精确匹配歌名（处理 &nbsp; 等 HTML 实体），排除 DJ 版/remix 等变体
-        def _norm(name: str) -> str:
-            # 去掉 HTML 实体、空白，转小写
-            return re.sub(r"\s+", "", html_mod.unescape(name or "")).lower()
-
-        target_name = _norm(song_name)
-        target_base = re.sub(r"\s*[\(（].*?[\)）]", "", song_name or "").strip()  # 去掉括号后缀
-        target_base_norm = _norm(target_base)
-
-        chosen = None
-        # 第一轮：歌名完全匹配（含括号后缀）
-        for item in abslist:
-            if _norm(item.get("SONGNAME", "")) == target_name:
-                chosen = item
-                break
-        # 第二轮：去掉括号后缀后匹配，且排除 DJ 版/remix/cover 等变体
-        if not chosen:
-            for item in abslist:
-                item_base = _norm(re.sub(r"\s*[\(（].*?[\)）]", "", item.get("SONGNAME", "")))
-                if item_base == target_base_norm:
-                    full = _norm(item.get("SONGNAME", ""))
-                    if not re.search(r"dj|remix|cover|伴奏|翻唱", full):
-                        chosen = item
-                        break
-        # 第三轮：兜底取第一首（排除明显变体）
-        if not chosen:
-            for item in abslist:
-                full = _norm(item.get("SONGNAME", ""))
-                if not re.search(r"dj|remix|cover|伴奏|翻唱", full):
-                    chosen = item
-                    break
+        chosen = _pick_best_candidate(
+            abslist,
+            name_key="SONGNAME",
+            artist_key="ARTIST",
+            duration_key="DURATION",
+            target_name=song_name,
+            target_artist=artist,
+            target_duration=expected_duration,
+        )
         if not chosen:
-            chosen = abslist[0]
+            logger.info("[kuwo] no candidate with matching name for query=%s", query)
+            return None
 
         rid = chosen.get("MUSICRID", "").replace("MUSIC_", "")
         if not rid:
@@ -328,7 +681,6 @@ async def _kuwo_search_and_url(song_name: str, artist: str, client: httpx.AsyncC
         logger.info("[kuwo] chosen rid=%s name=%s for query=%s",
                     rid, chosen.get("SONGNAME", ""), query)
 
-        # 获取播放 URL（antiserver 接口返回标准 JSON）
         play_url = f"http://antiserver.kuwo.cn/anti.s?type=convert_url3&rid={rid}&format=mp3"
         resp2 = await client.get(
             play_url,
@@ -347,23 +699,23 @@ async def _kuwo_search_and_url(song_name: str, artist: str, client: httpx.AsyncC
 
 
 # ===== QQ 音乐换源（第二备选，best-effort） =====
-# 咪咕音乐搜索接口已失效（返回 HTML），改用 QQ 音乐作为第三换源。
 # QQ 音乐匿名 vkey 对 VIP 曲通常返回空 purl，仅覆盖部分免费曲，作兜底。
 
-async def _qq_search_and_url(song_name: str, artist: str, client: httpx.AsyncClient) -> Optional[dict]:
-    """从 QQ 音乐搜索并获取播放 URL（匿名，best-effort 兜底）。失败返回 None。"""
-    import re
-    import html as html_mod
-    import urllib.parse
+async def _qq_search_and_url(
+    song_name: str, artist: str, client: httpx.AsyncClient,
+    expected_duration: float = 0,
+) -> Optional[dict]:
+    """从 QQ 音乐搜索并获取播放 URL（匿名，best-effort 兜底）。失败返回 None。
 
+    改进：用综合评分选最佳候选，歌名不匹配的跳过，防止错配。
+    """
     query = f"{song_name} {artist}".strip()
     if not query:
         return None
     try:
-        # 搜索
         resp = await client.get(
             "https://c.y.qq.com/soso/fcgi-bin/client_search_cp",
-            params={"w": query, "format": "json", "n": 10, "p": 1, "cr": 1},
+            params={"w": query, "format": "json", "n": 15, "p": 1, "cr": 1},
             headers={"User-Agent": "Mozilla/5.0 (Linux; Android 10) AppleWebKit/537.36",
                      "Referer": "https://y.qq.com/"},
             timeout=8.0,
@@ -374,27 +726,34 @@ async def _qq_search_and_url(song_name: str, artist: str, client: httpx.AsyncCli
         if not lst:
             return None
 
-        # 精确匹配歌名，排除 remix/cover 等变体
-        def _norm(name: str) -> str:
-            return re.sub(r"\s+", "", html_mod.unescape(name or "")).lower()
-
-        target = _norm(song_name)
-        chosen = None
+        # QQ 的 singer 是列表，需要展平为字符串供 _pick_best_candidate 使用
         for s in lst:
-            if _norm(s.get("songname", "")) == target:
-                full = _norm(s.get("songname", ""))
-                if not re.search(r"dj|remix|cover|伴奏|翻唱", full):
-                    chosen = s
-                    break
+            singers = s.get("singer") or []
+            if singers and isinstance(singers, list):
+                s["singer_name"] = "/".join(
+                    str(x.get("name", "")) for x in singers if x.get("name"))
+            else:
+                s["singer_name"] = ""
+
+        chosen = _pick_best_candidate(
+            lst,
+            name_key="songname",
+            artist_key="singer_name",
+            duration_key="interval",
+            target_name=song_name,
+            target_artist=artist,
+            target_duration=expected_duration,
+        )
         if not chosen:
-            chosen = lst[0]
+            logger.info("[qq] no candidate with matching name for query=%s", query)
+            return None
+
         songmid = chosen.get("songmid")
         if not songmid:
             return None
         logger.info("[qq] chosen mid=%s name=%s for query=%s",
                     songmid, chosen.get("songname", ""), query)
 
-        # 获取 vkey（匿名，VIP 曲 purl 通常为空）
         payload = json.dumps({
             "req_0": {
                 "module": "vkey.GetVkeyServer",
@@ -461,119 +820,117 @@ async def recognize(
     logger.info("[recognize rid=%d] archive created, amr_bytes=%d, enhance=%s",
                 rid, len(amr_bytes), enhance)
 
+    # 第一步：先解码一份"无滤镜" PCM，用于能量计算与选段。
+    # 选段基于原始信号能量更稳定（滤镜会改变能量分布）。
     try:
-        pcm_bytes = await _amr_to_float32_pcm(amr_bytes, enhance=enhance)
+        raw_pcm = await _amr_to_float32_pcm(amr_bytes, enhance=False)
     except Exception as e:
         music_record_store.update_record(
             rid, error_step="ffmpeg", error_msg=str(e),
             elapsed_ms=int((time.time() - t0) * 1000),
         )
         raise
-    music_record_store.update_record(rid, pcm_size=len(pcm_bytes))
-    logger.info("[recognize rid=%d] ffmpeg decode done, pcm bytes=%d, elapsed=%dms",
-                rid, len(pcm_bytes), int((time.time() - t0) * 1000))
+    music_record_store.update_record(rid, pcm_size=len(raw_pcm))
+    logger.info("[recognize rid=%d] raw pcm decoded, bytes=%d, elapsed=%dms",
+                rid, len(raw_pcm), int((time.time() - t0) * 1000))
 
     # 网易云 shazam_v2 算法硬要求固定 3 秒片段（24000 samples @ 8000Hz）。
-    # 本地测试证明：duration=5/8/10 全部 noMatchReason=10，只有 3 秒能识别。
-    #
-    # 智能选段（默认行为，不依赖 enhance 开关）：
-    #   1. 用 numpy 向量化扫描所有可能的 3 秒窗口（0.5 秒步长）
-    #   2. 取平均能量最高的 top 3 候选片段（自动避开噪音/静音段）
-    #   3. 依次生成指纹并调用网易云识曲
-    #   4. 合并 result，按 song_id 去重（同一首歌只保留最早 startTime）
-    #
-    # 这样不管噪音在开头/中间/结尾都能选到最干净的片段；
-    # 多片段尝试可显著提高识别成功率（单一片段可能恰好落在歌曲间奏段）。
-    # numpy 向量化使能量计算从纯 Python 循环的 ~20s 降到 < 10ms。
+    # 智能选段：numpy 向量化扫描所有 3 秒窗口（0.5 秒步长），
+    # 取平均能量最高的 top-N 候选片段。
     _FP_DUR = 3
     _FP_SAMPLES = _FP_DUR * 8000          # 24000 samples
     _STEP_SAMPLES = 4000                   # 0.5 秒步长
-    _TOP_N = 3                              # 最多尝试 3 个候选片段
+    _TOP_N = 3                              # 最多尝试 3 个候选片段（首轮）
 
-    sample_count = len(pcm_bytes) // 4
+    sample_count = len(raw_pcm) // 4
+    all_starts: list[int] = []
     if sample_count < _FP_SAMPLES:
         # 录音不足 3 秒，直接用全部数据（补零由 afp.js 处理）
         candidates: list[tuple[int, float]] = [(0, 0.0)]
         logger.info("[recognize rid=%d] pcm too short (%d samples < %d), using all",
                     rid, sample_count, _FP_SAMPLES)
     else:
-        # 零拷贝视图，numpy 直接读 float32
-        arr = np.frombuffer(pcm_bytes, dtype=np.float32)
+        arr = np.frombuffer(raw_pcm, dtype=np.float32)
         max_start = sample_count - _FP_SAMPLES
-        starts = list(range(0, max_start + 1, _STEP_SAMPLES))
-        # 向量化计算每个窗口的平均绝对幅值（能量指标）
-        energies = [float(np.abs(arr[s:s + _FP_SAMPLES]).mean()) for s in starts]
-        # top N，按能量降序
+        all_starts = list(range(0, max_start + 1, _STEP_SAMPLES))
+        energies = [float(np.abs(arr[s:s + _FP_SAMPLES]).mean()) for s in all_starts]
         top_idx = sorted(range(len(energies)), key=lambda i: energies[i], reverse=True)[:_TOP_N]
-        candidates = [(starts[i], energies[i]) for i in top_idx]
+        candidates = [(all_starts[i], energies[i]) for i in top_idx]
         logger.info("[recognize rid=%d] smart pick top %d from %d windows: %s",
-                    rid, len(candidates), len(starts),
+                    rid, len(candidates), len(all_starts),
                     ", ".join(f"{s/8000:.1f}s(e={e:.4f})" for s, e in candidates))
 
-    # 依次尝试每个候选片段，合并识别结果并按 song_id 去重
+    # 增强模式下依次尝试多条滤镜链；普通模式只用一条（无滤镜）。
+    # 实测：agc 滤镜适合小声录音，spectral 适合中等噪声，light 适合干净录音。
+    # 不同录音适配不同滤镜，多滤镜兜底能覆盖更多场景。
+    if enhance:
+        filter_chains = list(_ENHANCE_FILTER_CHAINS)
+    else:
+        filter_chains = [None]
+
     merged_songs: dict[int, dict] = {}   # song_id -> song_dict
     first_ncm_raw: dict = {}             # 首次响应入库存档（用于面板排查）
     first_fp_len = 0
     last_error: Optional[str] = None
+    tried_starts: set[int] = set()        # 已尝试过的 start_s，避免全量扫描时重复
 
     async with httpx.AsyncClient() as client:
-        for cand_idx, (start_s, energy) in enumerate(candidates):
-            pcm_for_fp = pcm_bytes[start_s * 4 : (start_s + _FP_SAMPLES) * 4]
-            # 生成指纹
-            try:
-                fp = await _generate_fingerprint(pcm_for_fp, _FP_DUR)
-            except Exception as e:
-                last_error = str(e)
-                logger.warning("[recognize rid=%d] cand#%d fp failed (start=%.1fs): %s",
-                               rid, cand_idx + 1, start_s / 8000.0, e)
-                continue
-            if cand_idx == 0:
-                first_fp_len = len(fp)
-            logger.info("[recognize rid=%d] cand#%d fp ok (start=%.1fs, fp_len=%d)",
-                        rid, cand_idx + 1, start_s / 8000.0, len(fp))
+        # ---- 第一阶段：每个滤镜链尝试 top-N 候选片段 ----
+        for fc_idx, fc in enumerate(filter_chains):
+            if merged_songs:
+                break
+            # 按需重新解码：不同滤镜链需要不同的 PCM
+            if fc is None:
+                pcm_bytes = raw_pcm
+            else:
+                try:
+                    pcm_bytes = await _amr_to_float32_pcm(
+                        amr_bytes, enhance=True, filter_chain=fc)
+                except Exception as e:
+                    logger.warning("[recognize rid=%d] filter chain #%d decode failed: %s",
+                                   rid, fc_idx + 1, e)
+                    continue
+                logger.info("[recognize rid=%d] filter chain #%d decoded, bytes=%d",
+                            rid, fc_idx + 1, len(pcm_bytes))
+            fc_label = f"fc#{fc_idx+1}({fc[:30]}...)" if fc else "no-filter"
+            matched = await _try_candidates(
+                pcm_bytes, candidates, client, rid, fc_label,
+                merged_songs, first_ncm_raw, tried_starts,
+                track_first=(not first_ncm_raw),
+            )
+            if not first_fp_len and matched["first_fp_len"]:
+                first_fp_len = matched["first_fp_len"]
+            if matched["last_error"]:
+                last_error = matched["last_error"]
+            if merged_songs:
+                logger.info("[recognize rid=%d] filter chain #%d got %d matches, stop",
+                            rid, fc_idx + 1, len(merged_songs))
+                break
 
-            # 调网易云识曲
+        # ---- 第二阶段（仅 enhance）：top-N 全部未命中时，全量扫描所有片段 ----
+        # 实测：audio (1).mp3 的 top-3 能量段全部 noMatch，但 start=6.5s
+        # （能量较低）能识别成功。全量扫描兜底能显著提升识别率。
+        if enhance and not merged_songs and all_starts:
+            logger.info("[recognize rid=%d] top-N all miss, fallback to full scan (%d windows)",
+                        rid, len(all_starts))
+            # 全量扫描用第一条（agc）滤镜链，对小声录音效果最好
             try:
-                ncm_raw = await _call_audio_match(fp, _FP_DUR, client)
+                pcm_bytes = await _amr_to_float32_pcm(
+                    amr_bytes, enhance=True, filter_chain=_ENHANCE_FILTER_CHAINS[0])
             except Exception as e:
-                last_error = str(e)
-                logger.warning("[recognize rid=%d] cand#%d ncm call failed: %s",
-                               rid, cand_idx + 1, e)
-                continue
-
-            if cand_idx == 0:
-                first_ncm_raw = ncm_raw if isinstance(ncm_raw, dict) else {}
-
-            raw_result = ncm_raw.get("result") if isinstance(ncm_raw, dict) else None
-            if not isinstance(raw_result, list) or not raw_result:
-                logger.info("[recognize rid=%d] cand#%d no match (reason=%s)",
-                            rid, cand_idx + 1,
-                            ncm_raw.get("noMatchReason") if isinstance(ncm_raw, dict) else "?")
-                continue
-
-            # 合并到 merged_songs（按 song_id 去重，先到先得）
-            new_added = 0
-            for item in raw_result:
-                song = item.get("song") or {}
-                sid = song.get("id")
-                if not sid or sid in merged_songs:
-                    continue
-                merged_songs[sid] = {
-                    "id": sid,
-                    "name": song.get("name", ""),
-                    "artists": "/".join(a.get("name", "") for a in (song.get("artists") or [])),
-                    "album": (song.get("album") or {}).get("name", ""),
-                    "startTime": item.get("startTime", 0),
-                }
-                new_added += 1
-
-            logger.info("[recognize rid=%d] cand#%d matched %d songs (+%d new), merged total %d",
-                        rid, cand_idx + 1, len(raw_result), new_added, len(merged_songs))
-
-            # 合并到足够多时提前停止，避免浪费 API 调用
-            if len(merged_songs) >= 5:
-                logger.info("[recognize rid=%d] enough matches, stop early", rid)
-                break
+                logger.warning("[recognize rid=%d] full-scan decode failed: %s", rid, e)
+                pcm_bytes = raw_pcm
+
+            remaining = [(s, 0.0) for s in all_starts if s not in tried_starts]
+            # 全量扫描最多尝试 20 个片段（避免过多 API 调用拖慢响应）
+            remaining = remaining[:20]
+            matched = await _try_candidates(
+                pcm_bytes, remaining, client, rid, "full-scan",
+                merged_songs, first_ncm_raw, set(),
+                track_first=(not first_ncm_raw),
+            )
+            if matched["last_error"]:
+                last_error = matched["last_error"]
 
     # 入库（首次响应归档用于面板排查）
     ncm_code = first_ncm_raw.get("code") if isinstance(first_ncm_raw, dict) else None
@@ -613,19 +970,104 @@ async def recognize(
     }
 
 
+async def _try_candidates(
+    pcm_bytes: bytes,
+    candidates: list[tuple[int, float]],
+    client: httpx.AsyncClient,
+    rid: int,
+    label: str,
+    merged_songs: dict[int, dict],
+    first_ncm_raw: dict,
+    tried_starts: set[int],
+    track_first: bool = True,
+) -> dict:
+    """对一组候选片段依次生成指纹并调网易云识曲，合并结果到 merged_songs。
+
+    返回 {"first_fp_len": int, "last_error": str}。
+    merged_songs / tried_starts / first_ncm_raw 为引用传入，直接原地修改。
+    track_first=True 时，首次拿到 ncm 响应（含 noMatch）会写入 first_ncm_raw。
+    """
+    _FP_DUR = 3
+    _FP_SAMPLES = _FP_DUR * 8000
+    result = {"first_fp_len": 0, "last_error": None}
+
+    for cand_idx, (start_s, energy) in enumerate(candidates):
+        tried_starts.add(start_s)
+        pcm_for_fp = pcm_bytes[start_s * 4 : (start_s + _FP_SAMPLES) * 4]
+        try:
+            fp = await _generate_fingerprint(pcm_for_fp, _FP_DUR)
+        except Exception as e:
+            result["last_error"] = str(e)
+            logger.warning("[recognize rid=%d] %s cand#%d fp failed (start=%.1fs): %s",
+                           rid, label, cand_idx + 1, start_s / 8000.0, e)
+            continue
+        if cand_idx == 0 and track_first and not first_ncm_raw:
+            result["first_fp_len"] = len(fp)
+        logger.info("[recognize rid=%d] %s cand#%d fp ok (start=%.1fs, fp_len=%d)",
+                    rid, label, cand_idx + 1, start_s / 8000.0, len(fp))
+
+        try:
+            ncm_raw = await _call_audio_match(fp, _FP_DUR, client)
+        except Exception as e:
+            result["last_error"] = str(e)
+            logger.warning("[recognize rid=%d] %s cand#%d ncm call failed: %s",
+                           rid, label, cand_idx + 1, e)
+            continue
+
+        if track_first and not first_ncm_raw:
+            first_ncm_raw.update(ncm_raw if isinstance(ncm_raw, dict) else {})
+
+        raw_result = ncm_raw.get("result") if isinstance(ncm_raw, dict) else None
+        if not isinstance(raw_result, list) or not raw_result:
+            logger.info("[recognize rid=%d] %s cand#%d no match (reason=%s)",
+                        rid, label, cand_idx + 1,
+                        ncm_raw.get("noMatchReason") if isinstance(ncm_raw, dict) else "?")
+            continue
+
+        new_added = 0
+        for item in raw_result:
+            song = item.get("song") or {}
+            sid = song.get("id")
+            if not sid or sid in merged_songs:
+                continue
+            merged_songs[sid] = {
+                "id": sid,
+                "name": song.get("name", ""),
+                "artists": "/".join(a.get("name", "") for a in (song.get("artists") or [])),
+                "album": (song.get("album") or {}).get("name", ""),
+                "startTime": item.get("startTime", 0),
+            }
+            new_added += 1
+
+        logger.info("[recognize rid=%d] %s cand#%d matched %d songs (+%d new), merged total %d",
+                    rid, label, cand_idx + 1, len(raw_result), new_added, len(merged_songs))
+
+        if len(merged_songs) >= 5:
+            logger.info("[recognize rid=%d] %s enough matches, stop early", rid, label)
+            break
+
+    return result
+
+
 async def get_song_url(song_id: int, level: str = "standard", unblock: bool = False,
                        song_name: str = "", artist: str = "") -> dict:
-    """获取歌曲播放 URL。unblock=true 时若网易云无 URL 则尝试酷我换源。
+    """获取歌曲播放 URL。unblock=true 时若网易云无 URL 则尝试酷我/QQ 换源。
 
     所有返回的 URL 强制转 HTTPS，因为手表 audio 组件在部分网络环境下
     无法播放 HTTP 流（混合内容限制 / 运营商劫持）。
+
+    返回字段新增：
+    - source: 播放源（netease/kuwo/qq/none）
+    - song_id: 回显请求的歌曲 ID（前端用于校验 URL 是否对应正确的歌）
+    - tried_sources: 尝试过的源列表（前端用于显示换源状态）
     """
+    tried_sources: list[str] = []
     async with httpx.AsyncClient() as client:
         ncm_data = await _call_song_url(song_id, level, client)
+        tried_sources.append("netease")
 
         url = ncm_data.get("url")
         if url:
-            # 网易云返回的 URL 可能是 http://，手表可能无法播放，强制转 https
             if url.startswith("http://"):
                 url_https = "https://" + url[7:]
                 logger.info("[song_url] convert http->https: %s -> %s",
@@ -639,29 +1081,55 @@ async def get_song_url(song_id: int, level: str = "standard", unblock: bool = Fa
                 "freeTrialInfo": ncm_data.get("freeTrialInfo"),
                 "fee": ncm_data.get("fee", 0),
                 "source": "netease",
+                "song_id": song_id,
+                "tried_sources": tried_sources,
             }
 
         # 网易云无 URL（通常是 VIP/版权曲），尝试换源
         if unblock:
-            kw = await _kuwo_search_and_url(song_name, artist, client)
+            # 先获取歌曲时长，用于换源时验证候选歌曲是否匹配（防止错配）
+            expected_duration = 0.0
+            try:
+                details = await _call_song_detail([song_id], client)
+                if details:
+                    dt_ms = details[0].get("dt", 0)
+                    if dt_ms:
+                        expected_duration = float(dt_ms) / 1000.0
+                        # 如果前端没传 song_name/artist，从详情补全
+                        if not song_name:
+                            song_name = details[0].get("name", "")
+                        if not artist:
+                            ar = details[0].get("ar") or []
+                            artist = "/".join(a.get("name", "") for a in ar)
+                        logger.info("[song_url] got duration=%.1fs name=%s for sid=%s",
+                                    expected_duration, song_name, song_id)
+            except Exception as e:
+                logger.warning("[song_url] fetch detail for duration failed: %s", e)
+
+            kw = await _kuwo_search_and_url(song_name, artist, client, expected_duration)
+            tried_sources.append("kuwo")
             if kw and kw.get("url"):
                 kw_url = kw["url"]
                 if kw_url.startswith("http://"):
                     kw_url = "https://" + kw_url[7:]
                     logger.info("[song_url] kuwo convert http->https")
                     kw["url"] = kw_url
-                return {"ok": True, **kw}
+                return {"ok": True, **kw, "song_id": song_id,
+                        "tried_sources": tried_sources}
 
-            # 酷我也无 URL，尝试 QQ 音乐兜底
-            qq = await _qq_search_and_url(song_name, artist, client)
+            qq = await _qq_search_and_url(song_name, artist, client, expected_duration)
+            tried_sources.append("qq")
             if qq and qq.get("url"):
-                return {"ok": True, **qq}
+                return {"ok": True, **qq, "song_id": song_id,
+                        "tried_sources": tried_sources}
 
     return {
         "ok": False,
         "url": None,
         "message": "无可用播放源" + ("（已尝试网易云/酷我/QQ换源）" if unblock else "（可在设置开启换源尝试）"),
         "source": "none",
+        "song_id": song_id,
+        "tried_sources": tried_sources,
     }