feat(python): 完成 Python 全量迁移——剩余平台、登录通道与公共层修复

- 新增 5sing/apple/xm/me/qt/lr 平台包与取流实现(含 xm-sign、MissEvan 签名、蜻蜓 HMAC、Apple developer token)
- 新增 wy/kg/tx/lr 登录:扫码、手机号、账密(kg 20028 风控指纹)、cookie 绑定、kg/tx 续期
- request 层:保留多值 Set-Cookie;JSON body 紧凑化(修复 kg 网关验签);直连不经系统代理(对齐 Node)
- 登录路由统一 Body(embed=True),修复扫码轮询 422;lifespan 接入登录续期调度
- ruff/format/pyright 全绿;14 平台真实联网与 HTTP 路由验证通过
This commit is contained in:
snowy-yang committed 2026-10-23 21:55:45 +08:00
1 parent 800871ef6e
commit dfc06bf4de
182 files changed
+26620

No files matched your search

+80
View File
@@ -0,0 +1,80 @@
"""咪咕音乐(对应 sdk/mg/index.ts 的 default export)。
能力位与 TS 一致:搜索 / 专辑 / 歌单 / 链接解析,顶层挂 get_music_url(取流在 apis/mg.py)、
get_lyric、get_pic、get_music_info、get_music_detail_page_url。
TS 侧没有榜单、歌单广场、热搜、歌手、评论、个人歌单子模块,这里一律省略 ——
缺省即不支持,/sources 与路由据此返回 404。
"""
from __future__ import annotations
from typing import Any
from app.sdk.platforms._base import Platform
from app.sdk.platforms.mg import album, lyric, music_search, parse, playlist
from app.sdk.platforms.mg.lyric import content_id_of
from app.sdk.platforms.mg.util import (
fetch_song_detail,
first_non_empty,
migu_song_link,
normalize_migu_image_url,
pick_migu_image,
)
async def get_music_url(song_info: dict[str, Any], quality: str, retry_num: int = 0) -> dict[str, Any]:
"""取流入口。延迟导入 apis.mg:apis 模块反向依赖 platforms.mg.util,
模块级互相 import 会在「先导入 apis」的入口下拿到半初始化的模块。"""
from app.sdk.apis import mg as mg_api
return await mg_api.get_music_url(song_info, quality, retry_num)
async def get_pic(song_info: dict[str, Any]) -> str | None:
if song_info.get("img"):
return str(song_info["img"])
content_id = content_id_of(song_info)
if not content_id:
return None
item = await fetch_song_detail(content_id)
return (
normalize_migu_image_url(
first_non_empty(
pick_migu_image(item.get("imgItems")),
pick_migu_image(item.get("albumImgs")),
item.get("img1"),
item.get("img2"),
item.get("img3"),
)
)
or None
)
async def get_music_info(song_info: dict[str, Any]) -> dict[str, Any]:
content_id = content_id_of(song_info)
if not content_id:
raise RuntimeError("migu 歌曲 id 无效")
return await fetch_song_detail(content_id)
def get_music_detail_page_url(song_info: dict[str, Any]) -> str:
extra = song_info.get("extra") or {}
link = str(extra.get("link") or "") if isinstance(extra, dict) else ""
return link or migu_song_link(content_id_of(song_info))
MODULE = Platform(
id="mg",
name="咪咕音乐",
music_search=music_search,
album=album,
playlist=playlist,
parse=parse,
get_music_url=get_music_url,
get_lyric=lyric.get_lyric,
get_pic=get_pic,
get_music_info=get_music_info,
get_music_detail_page_url=get_music_detail_page_url,
)
+164
View File
@@ -0,0 +1,164 @@
"""咪咕专辑(对应 sdk/mg/album.ts)。
搜索(search_all.do 的 album 开关)+ 曲目(queryAlbumSong,分页聚合去重)+
详情(resourceinfo 2003 的元数据)。上游对应 music-lib/migu/album.go、
migu.go 的 fetchAlbumDetail / fetchAlbumSongs。
"""
from __future__ import annotations
import re
from typing import Any
from app.sdk.platforms.mg.music_search import search_all
from app.sdk.platforms.mg.util import (
build_url,
convert_item_to_song,
first_non_empty,
migu_album_link,
migu_get,
pick_migu_image,
to_number,
)
QUERY_ALBUM_SONG_URL = "https://app.c.nf.migu.cn/MIGUM2.0/v1.0/content/queryAlbumSong"
RESOURCE_INFO_URL = "https://app.c.nf.migu.cn/MIGUM2.0/v1.0/content/resourceinfo.do"
_ALBUM_PATTERNS = (
re.compile(r"music\.migu\.cn/(?:v3|v5)/music/album/(\d+)"),
re.compile(r"albumId=(\d+)"),
re.compile(r"resourceId=(\d+)"),
)
async def search_album(keyword: str, page: int = 1, limit: int = 10) -> dict[str, Any]:
body = await search_all(keyword, "album", page, limit)
data = body.get("albumResultData") if isinstance(body, dict) else None
result = (data or {}).get("result") or []
album_list: list[dict[str, Any]] = []
for item in result:
album_id = str(item.get("id") or "").strip()
if not album_id:
continue
album_list.append(
{
"id": album_id,
"name": str(item.get("name") or "").strip(),
"cover": pick_migu_image(item.get("imgItems")),
"creator": str(item.get("singer") or "").strip(),
"description": first_non_empty(item.get("desc"), item.get("publishDate")),
"link": migu_album_link(album_id),
"source": "mg",
}
)
return {"list": album_list, "source": "mg", "page": page, "limit": limit, "total": len(album_list)}
async def fetch_album_songs(album_id: str) -> dict[str, Any]:
"""专辑曲目(每页 50 全量翻页 + 去重;上游 fetchAlbumSongs)。"""
album = str(album_id or "").strip()
if not album:
raise RuntimeError("migu 专辑 id 为空")
page_size = 50
seen: set[str] = set()
songs: list[dict[str, Any]] = []
total_count = 0
page_no = 1
while True:
body = await migu_get(
build_url(QUERY_ALBUM_SONG_URL, {"albumId": album, "pageNo": page_no, "pageSize": page_size})
)
body = body if isinstance(body, dict) else {}
if body.get("code") and body["code"] != "000000":
raise RuntimeError(f"migu api error: {body.get('info')} (code {body.get('code')})")
data = body.get("data") or {}
song_list = data.get("songList") or []
if not total_count:
total_count = int(to_number(data.get("totalCount")))
if not song_list:
break
before = len(songs)
for item in song_list:
song = convert_item_to_song(item, True)
if song is None:
continue
key = (
first_non_empty(
song["extra"].get("content_id"), item.get("contentId"), song["songmid"], item.get("copyrightId")
)
or f"{song['name']}|{song['singer']}"
)
if key in seen:
continue
seen.add(key)
songs.append(song)
if len(song_list) < page_size:
break
if total_count > 0 and len(songs) >= total_count:
break
if len(songs) == before:
break
page_no += 1
if not songs:
raise RuntimeError("migu 专辑没有可播放歌曲")
return {"songs": songs, "total": total_count or len(songs)}
async def get_album_detail(album_id: str, page: int = 1, limit: int = 50) -> dict[str, Any]:
"""专辑详情(曲目 + resourceinfo 2003 的元数据),返回形状对齐 kg/kw 的专辑详情。"""
album = str(album_id or "").strip()
if not album:
raise RuntimeError("migu 专辑 id 为空")
fetched = await fetch_album_songs(album)
songs: list[dict[str, Any]] = fetched["songs"]
total: int = fetched["total"]
info: dict[str, str] = {"name": album, "img": "", "desc": "", "author": ""}
try:
body = await migu_get(
build_url(RESOURCE_INFO_URL, {"needSimple": "00", "resourceType": "2003", "resourceId": album})
)
body = body if isinstance(body, dict) else {}
if body.get("code") and body["code"] != "000000":
raise RuntimeError(str(body.get("info")))
resource = body.get("resource") or []
res = resource[0] if resource and isinstance(resource[0], dict) else None
if res:
info = {
"name": str(res.get("title") or "").strip() or album,
"img": pick_migu_image(res.get("imgItems")),
"desc": str(res.get("summary") or "").strip(),
"author": str(res.get("singer") or "").strip(),
}
if not info["img"] and songs[0].get("img"):
info["img"] = str(songs[0]["img"])
except Exception: # noqa: BLE001 - 元数据是锦上添花,失败仍返回曲目
if songs[0].get("img"):
info["img"] = str(songs[0]["img"])
start = (page - 1) * limit
return {
"source": "mg",
"info": info,
"list": songs[start : start + limit],
"page": page,
"limit": limit,
"total": total or len(songs),
}
async def get_album_songs(album_id: str) -> list[dict[str, Any]]:
return (await fetch_album_songs(album_id))["songs"]
async def parse_album_link(link: str) -> dict[str, Any]:
raw = str(link or "")
for pattern in _ALBUM_PATTERNS:
match = pattern.search(raw)
if match:
return await get_album_detail(match.group(1))
raise RuntimeError("无效的咪咕专辑链接")
+49
View File
@@ -0,0 +1,49 @@
"""咪咕歌词(对应 sdk/mg/lyric.ts)。
resourceinfo.do 取 lrcUrl/lyricUrl 再下载文本;歌词 CDN 只认 y.migu.cn 的 Referer,
与资源接口的 music.migu.cn Referer 不同,所以单独发一次请求。
上游对应 music-lib/migu/lyric.go。
"""
from __future__ import annotations
import re
from typing import Any
from app.sdk.common.request import http_fetch
from app.sdk.platforms.mg.util import SONG_DETAIL_URL, build_url, migu_get
LRC_REFERER = "https://y.migu.cn/"
LRC_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/143.0.0.0 Safari/537.36"
)
_HTTP_PREFIX = re.compile(r"^http://")
def content_id_of(song_info: dict[str, Any]) -> str:
"""从 extra.content_id 或 songmid(contentId|resourceType|formatType)取 contentId。"""
extra = song_info.get("extra") or {}
if isinstance(extra, dict) and extra.get("content_id"):
return str(extra["content_id"])
raw_id = str(song_info.get("songmid") or song_info.get("id") or "")
return raw_id.split("|")[0]
async def get_lyric(song_info: dict[str, Any]) -> dict[str, str]:
content_id = content_id_of(song_info)
if not content_id:
raise RuntimeError("migu 歌曲 id 无效")
body = await migu_get(build_url(SONG_DETAIL_URL, {"resourceId": content_id, "resourceType": "2"}))
resource = body.get("resource") if isinstance(body, dict) else None
if not resource:
raise RuntimeError("migu 歌词资源不存在")
first = resource[0] if isinstance(resource[0], dict) else {}
lrc_url = str(first.get("lrcUrl") or "") or str(first.get("lyricUrl") or "")
if not lrc_url:
return {"lyric": "", "tlyric": ""}
lrc_url = _HTTP_PREFIX.sub("https://", lrc_url)
resp = await http_fetch(lrc_url, headers={"User-Agent": LRC_UA, "Referer": LRC_REFERER}, timeout=20_000)
if resp.status_code != 200:
raise RuntimeError(f"migu 歌词下载失败 {resp.status_code}")
return {"lyric": resp.body if isinstance(resp.body, str) else "", "tlyric": ""}
+60
View File
@@ -0,0 +1,60 @@
"""咪咕音乐搜索(对应 sdk/mg/musicSearch.ts)。
search_all.do(Android_migu 参数),searchSwitch 控制返回类别 —— 歌曲/专辑/歌单三个
子模块共用这一个接口,只切换 searchSwitch。
上游对应 music-lib/migu/migu.go 的 (*Migu).Search。
"""
from __future__ import annotations
import json
import math
from typing import Any
from app.sdk.platforms.mg.util import build_url, convert_item_to_song, migu_get, parse_int
SEARCH_ALL_URL = "http://pd.musicapp.migu.cn/MIGUM2.0/v1.0/content/search_all.do"
def search_switch(kind: str) -> str:
"""搜索开关:只开某一类(song/album/songlist),bestShow 保留最优推荐。"""
payload = {
"song": 1 if kind == "song" else 0,
"album": 1 if kind == "album" else 0,
"singer": 0,
"tagSong": 0,
"mvSong": 0,
"songlist": 1 if kind == "songlist" else 0,
"bestShow": 1,
}
return json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
async def search_all(text: str, kind: str, page: int = 1, limit: int = 20) -> Any:
url = build_url(
SEARCH_ALL_URL,
{
"ua": "Android_migu",
"version": "5.0.1",
"text": text,
"pageNo": page,
"pageSize": limit,
"searchSwitch": search_switch(kind),
},
)
return await migu_get(url)
async def search(keyword: str, page: int = 1, limit: int = 20, retry_num: int = 0) -> dict[str, Any]:
body = await search_all(keyword, "song", page, limit)
data = body.get("songResultData") if isinstance(body, dict) else None
result = (data or {}).get("result") or []
songs = [song for item in result if (song := convert_item_to_song(item, True)) is not None]
total = parse_int((data or {}).get("totalCount")) or len(songs)
return {
"list": songs,
"allPage": max(1, math.ceil(total / limit)),
"total": total,
"limit": limit,
"source": "mg",
}
+44
View File
@@ -0,0 +1,44 @@
"""咪咕链接解析(对应 sdk/mg/index.ts 的 parseSong)。
歌曲:`https://music.migu.cn/v3/music/song/<contentId>` 或裸 contentId(6 位以上数字)。
专辑/歌单分别复用 album / playlist 的链接解析。
"""
from __future__ import annotations
import re
from typing import Any
from app.sdk.platforms.mg.album import parse_album_link
from app.sdk.platforms.mg.playlist import parse_playlist_link
from app.sdk.platforms.mg.util import convert_item_to_song, fetch_song_detail
_SONG_PATTERNS = (
re.compile(r"music\.migu\.cn/v3/music/song/(\d+)"),
re.compile(r"^(\d{6,})$"),
)
async def parse_song(link: str) -> dict[str, Any]:
raw = str(link or "")
content_id = ""
for pattern in _SONG_PATTERNS:
match = pattern.search(raw)
if match:
content_id = match.group(1)
break
if not content_id:
raise RuntimeError("无效的咪咕歌曲链接")
item = await fetch_song_detail(content_id)
song = convert_item_to_song(item, True)
if song is None:
raise RuntimeError("该歌曲没有可用格式")
return song
async def parse_album(link: str) -> dict[str, Any]:
return await parse_album_link(link)
async def parse_playlist(link: str) -> dict[str, Any]:
return await parse_playlist_link(link)
+187
View File
@@ -0,0 +1,187 @@
"""咪咕歌单(对应 sdk/mg/playlist.ts)。
搜索(search_all.do 的 songlist 开关)、曲目(MIGUM3.0 playlist/song/v2.0)、
分类(静态关键词表,上游同款,平台未提供分类接口)、链接解析。
上游对应 music-lib/migu/playlist.go。
"""
from __future__ import annotations
import re
from typing import Any
from app.sdk.platforms.mg.music_search import search_all
from app.sdk.platforms.mg.util import (
build_url,
convert_item_to_song,
first_non_empty,
migu_get,
migu_playlist_link,
parse_int,
pick_migu_image,
to_number,
)
PLAYLIST_SONG_URL = "https://app.c.nf.migu.cn/MIGUM3.0/resource/playlist/song/v2.0"
_PLAYLIST_PATTERNS = (
re.compile(r"playlistId=(\d+)"),
re.compile(r"musicListId=(\d+)"),
re.compile(r"(?:playlist|songlist)/(\d+)"),
)
# 静态分类(上游 miguPlaylistCategoryKeywords)
CATEGORY_KEYWORDS: list[dict[str, Any]] = [
{"name": "华语", "group": "语种", "hot": True},
{"name": "欧美", "group": "语种", "hot": True},
{"name": "日语", "group": "语种"},
{"name": "韩语", "group": "语种"},
{"name": "粤语", "group": "语种"},
{"name": "流行", "group": "风格", "hot": True},
{"name": "摇滚", "group": "风格", "hot": True},
{"name": "民谣", "group": "风格", "hot": True},
{"name": "电子", "group": "风格"},
{"name": "说唱", "group": "风格"},
{"name": "古风", "group": "风格"},
{"name": "轻音乐", "group": "风格"},
{"name": "影视", "group": "场景", "hot": True},
{"name": "ACG", "group": "场景"},
{"name": "治愈", "group": "场景"},
{"name": "运动", "group": "场景"},
{"name": "学习", "group": "场景"},
{"name": "睡前", "group": "场景"},
]
def _to_playlist(item: dict[str, Any], category_id: str | None = None) -> dict[str, Any] | None:
playlist_id = str(item.get("id") or "").strip()
name = str(item.get("name") or "").strip()
if not playlist_id or not name:
return None
entry: dict[str, Any] = {
"id": playlist_id,
"name": name,
"cover": first_non_empty(item.get("musicListPicUrl"), pick_migu_image(item.get("imgItems"))),
"trackCount": parse_int(item.get("musicNum")),
"playCount": parse_int(item.get("playNum")),
"creator": first_non_empty(item.get("userName"), item.get("ownerName")),
"link": migu_playlist_link(playlist_id),
"source": "mg",
}
if category_id:
entry["categoryId"] = category_id
return entry
async def search_playlist(keyword: str, page: int = 1, limit: int = 20) -> dict[str, Any]:
"""歌单搜索;分类歌单也用同一接口,关键词即分类名。"""
body = await search_all(keyword, "songlist", page, limit)
data = body.get("songListResultData") if isinstance(body, dict) else None
result = (data or {}).get("result") or []
playlists = [entry for item in result if (entry := _to_playlist(item)) is not None]
return {"list": playlists, "source": "mg", "page": page, "limit": limit, "total": len(playlists)}
async def get_categories() -> dict[str, Any]:
return {
"source": "mg",
"list": [
{"id": "", "name": "全部", "group": "全部"},
*(
{"id": item["name"], "name": item["name"], "group": item["group"], "hot": bool(item.get("hot"))}
for item in CATEGORY_KEYWORDS
),
],
}
async def get_category_playlists(category_id: str, page: int = 1, limit: int = 20) -> dict[str, Any]:
keyword = str(category_id or "").strip() or "华语"
found = await search_playlist(keyword, page, limit)
playlists = [entry for item in found["list"] if (entry := _to_playlist(item, keyword)) is not None]
return {"list": playlists, "source": "mg", "page": page, "limit": limit}
async def fetch_playlist_songs(playlist_id: str) -> dict[str, Any]:
"""歌单曲目(每页 50 全量翻页 + 去重;上游 GetPlaylistSongs)。"""
playlist = str(playlist_id or "").strip()
if not playlist:
raise RuntimeError("migu 歌单 id 为空")
page_size = 50
seen: set[str] = set()
songs: list[dict[str, Any]] = []
total_count = 0
page_no = 1
while True:
body = await migu_get(
build_url(PLAYLIST_SONG_URL, {"pageNo": page_no, "pageSize": page_size, "playlistId": playlist})
)
body = body if isinstance(body, dict) else {}
if body.get("code") and body["code"] != "000000":
raise RuntimeError(f"migu api error: {body.get('info')} (code {body.get('code')})")
data = body.get("data") or {}
song_list = data.get("songList") or []
if not total_count:
total_count = int(to_number(data.get("totalCount")))
if not song_list:
break
before = len(songs)
for item in song_list:
song = convert_item_to_song(item, True)
if song is None:
continue
key = (
first_non_empty(
song["extra"].get("content_id"), item.get("contentId"), song["songmid"], item.get("copyrightId")
)
or f"{song['name']}|{song['singer']}"
)
if key in seen:
continue
seen.add(key)
songs.append(song)
if len(song_list) < page_size:
break
if total_count > 0 and len(songs) >= total_count:
break
if len(songs) == before:
break
page_no += 1
if not songs:
raise RuntimeError("migu 歌单没有可播放歌曲")
return {"songs": songs, "total": total_count or len(songs)}
async def get_playlist_detail(playlist_id: str, page: int = 1, limit: int = 50) -> dict[str, Any]:
playlist = str(playlist_id or "").strip()
fetched = await fetch_playlist_songs(playlist)
songs: list[dict[str, Any]] = fetched["songs"]
total: int = fetched["total"]
start = (page - 1) * limit
return {
"source": "mg",
"info": {"name": playlist, "img": songs[0].get("img") or "", "desc": "", "author": ""},
"list": songs[start : start + limit],
"page": page,
"limit": limit,
"total": total or len(songs),
}
async def get_playlist_songs(playlist_id: str) -> list[dict[str, Any]]:
return (await fetch_playlist_songs(playlist_id))["songs"]
async def parse_playlist_link(link: str) -> dict[str, Any]:
raw = str(link or "")
for pattern in _PLAYLIST_PATTERNS:
match = pattern.search(raw)
if match:
return await get_playlist_detail(match.group(1))
if raw and "/" not in raw:
return await get_playlist_detail(raw)
raise RuntimeError("无效的咪咕歌单链接")
+460
View File
@@ -0,0 +1,460 @@
"""咪咕音乐公共层(对应 sdk/mg/util.ts)。
常量、请求封装、条目 → 歌曲的统一转换、档位/扩展名工具与 Z3D 解密。
上游对应 music-lib/migu/migu.go(convertItemToSong*、collectMiguFormats、pickMiguImage 等)。
匿名态 listen-url 仅放行 PQ(128k mp3);HQ/SQ/ZQ 各档需携带登录 Cookie
(接口返回「登录后免费试听」)。
Z3D 解密与 `app/http/mg_stream.py` 的流式实现同源:那边服务 /stream/mg 代理,
这里供取流探测阶段校验候选是否真能解出音频,两处算法必须逐字一致。
"""
from __future__ import annotations
import hashlib
import math
import re
from typing import Any
from urllib.parse import urlencode
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from app.sdk.auth_store import get_cookie
from app.sdk.common.request import http_fetch
# ---------------- 常量 ----------------
#: 咪咕 toneFlag → 平台音质标签(types 展示与取流返回共用)
TONE_QUALITY: dict[str, str] = {
"LQ": "64k",
"PQ": "128k",
"HQ": "320k",
"SQ": "flac",
"ZQ": "flac24bit",
"ZQ24": "flac24bit",
"ZQ32": "wav",
"Z3D": "wav",
"I3D": "flac24bit",
"3D60": "wav",
}
#: tone → 明文容器扩展名(取流侧判定档位与扩展名共用)
TONE_EXT: dict[str, str] = {
"LQ": "mp3",
"PQ": "mp3",
"HQ": "mp3",
"SQ": "flac",
"ZQ": "flac",
"ZQ24": "flac",
"ZQ32": "wav",
"Z3D": "wav",
"I3D": "m4a",
"3D60": "wav",
}
USER_AGENT = (
"Mozilla/5.0 (iPhone; CPU iPhone OS 9_1 like Mac OS X) AppleWebKit/601.1.46 "
"(KHTML, like Gecko) Version/9.0 Mobile/13B143 Safari/601.1"
)
REFERER = "http://music.migu.cn/"
#: listen-url 走 Android 客户端 UA/头,与网页通道不同 UA
ANDROID_UA = "Android_migu/7.41.13 okhttp/3.12.13"
ANDROID_CHANNEL = "0146832"
ANDROID_VERSION = "7.41.13"
SONG_DETAIL_URL = "http://c.musicapp.migu.cn/MIGUM2.0/v1.0/content/resourceinfo.do"
# Z3D 加密流的 app code 密文(AES-128-CBC,key 为 AppKey 前 16 字符)
_ANDROID_APP_KEY = b"308202333082019c"
_ENCRYPTED_APP_CODE = bytes(
[
0xAE, 0xEA, 0xC7, 0x17, 0x05, 0xA6, 0x89, 0x55,
0xA9, 0x31, 0xA6, 0xFE, 0x01, 0xE7, 0x28, 0xFF,
0xF6, 0x49, 0x71, 0xE6, 0x1C, 0xEA, 0xF0, 0x0A,
0xE3, 0xEA, 0x43, 0xC3, 0xE0, 0x58, 0x33, 0x83,
]
) # fmt: skip
_Z3D_IV = bytes(range(16))
_INT_PREFIX = re.compile(r"[+-]?\d+")
# ---------------- 请求 ----------------
def get_migu_cookie() -> str:
return get_cookie("mg").strip()
async def migu_get(url: str, extra_headers: dict[str, str] | None = None) -> Any:
"""带 Referer/UA/Cookie 的 GET,返回解析后的 body(JSON 已解析)。"""
headers: dict[str, str] = {"User-Agent": USER_AGENT, "Referer": REFERER, **(extra_headers or {})}
cookie = get_migu_cookie()
if cookie:
headers["Cookie"] = cookie
resp = await http_fetch(url, headers=headers, timeout=20_000)
if resp.status_code != 200:
raise RuntimeError(f"migu http {resp.status_code}")
return resp.body
def build_url(base: str, params: dict[str, Any]) -> str:
"""URLSearchParams 语义:值统一 String 后 form-urlencoded。"""
return f"{base}?{urlencode([(key, str(value)) for key, value in params.items()])}"
# ---------------- 工具函数(1:1 对应上游) ----------------
def parse_int(value: Any) -> int:
"""JS parseInt 语义:取前导整数,解析不出返回 0。"""
match = _INT_PREFIX.match(str(value if value is not None else "").strip())
return int(match.group()) if match else 0
def to_number(value: Any) -> float:
"""JS Number() 语义:不可解析或非有限值归零(分页 totalCount / 体积等字段用)。"""
try:
number = float(str(value if value is not None else "").strip())
except (TypeError, ValueError):
return 0.0
return number if math.isfinite(number) else 0.0
def first_non_empty(*values: Any) -> str:
for value in values:
if value is None:
continue
if str(value).strip():
return str(value)
return ""
def first_non_zero_string(*values: Any) -> str:
for value in values:
text = str(value if value is not None else "").strip()
if text and text != "0":
return text
return ""
def pick_migu_image(items: Any) -> str:
"""按 02/01/03 的尺寸优先级取图,都没有则取第一张非空。"""
if not isinstance(items, list):
return ""
for preferred in ("02", "01", "03"):
for item in items:
entry = item if isinstance(item, dict) else {}
if entry.get("imgSizeType") == preferred and str(entry.get("img") or "").strip():
return str(entry.get("img"))
for item in items:
entry = item if isinstance(item, dict) else {}
if str(entry.get("img") or "").strip():
return str(entry.get("img"))
return ""
def normalize_migu_image_url(image: Any) -> str:
url = str(image if image is not None else "").strip()
if not url:
return ""
if url.startswith(("http://", "https://")):
return url
if url.startswith("//"):
return f"https:{url}"
if url.startswith("/"):
return f"https://d.musicapp.migu.cn{url}"
return url
def collect_migu_artist_names(item: dict[str, Any]) -> list[str]:
names: list[str] = []
seen: set[str] = set()
def append(name: Any) -> None:
text = str(name if name is not None else "").strip()
if not text or text in seen:
return
seen.add(text)
names.append(text)
for group in ("singers", "singerList", "artists"):
for singer in item.get(group) or []:
append((singer or {}).get("name") if isinstance(singer, dict) else None)
if not names:
for name in str(item.get("singer") or "").split("|"):
append(name)
return names
_Z3D_CODE_KEYS = (
"formatType",
"resourceType",
"size",
"androidSize",
"isize",
"asize",
"fileType",
"androidFileType",
"iformat",
"aformat",
"price",
)
def collect_migu_formats(item: dict[str, Any]) -> list[dict[str, Any]]:
"""汇总三组 rateFormats + z3dCode,按 formatType|resourceType 去重。"""
groups: list[list[dict[str, Any]]] = []
for key in ("newRateFormats", "rateFormats", "audioFormats"):
group = item.get(key)
if isinstance(group, list):
groups.append([entry for entry in group if isinstance(entry, dict)])
code = item.get("z3dCode")
if isinstance(code, dict) and code.get("formatType"):
groups.append([{key: code.get(key) for key in _Z3D_CODE_KEYS}])
formats: list[dict[str, Any]] = []
seen: set[str] = set()
for group in groups:
for fmt in group:
key = f"{str(fmt.get('formatType') or '').strip().upper()}|{str(fmt.get('resourceType') or '').strip()}"
if key == "|" or key in seen:
continue
seen.add(key)
formats.append(fmt)
return formats
_FORMAT_RANK: dict[str, int] = {
"ZQ32": 80,
"ZQ": 75,
"ZQ24": 75,
"SQ": 70,
"HQ": 60,
"I3D": 50,
"Z3D": 50,
"3D60": 40,
"PQ": 30,
"LQ": 20,
}
def migu_format_rank(format_type: Any = "") -> int:
return _FORMAT_RANK.get(str(format_type or "").strip().upper(), 10)
def migu_format_ext(format_type: Any = "", format_code: Any = "") -> str:
tone = str(format_type or "").strip().upper()
code = str(format_code or "").strip()
if tone in ("SQ", "ZQ", "ZQ24"):
return "flac"
if tone in ("ZQ32", "Z3D", "3D60"):
return "wav"
if tone == "I3D":
return "m4a"
if tone in ("PQ", "HQ", "LQ"):
return "mp3"
return "flac" if code.startswith("011") else "mp3"
def normalize_tone(value: Any = "") -> str:
"""取流档位归一:去掉前导点、大写,只认 TONE_EXT 里已知的 toneFlag。"""
text = str(value if value is not None else "").strip()
if text.startswith("."):
text = text[1:]
text = text.upper()
return text if text in TONE_EXT else ""
def migu_album_link(album_id: str) -> str:
return f"https://music.migu.cn/v3/music/album/{album_id}"
def migu_playlist_link(playlist_id: str) -> str:
return f"https://music.migu.cn/v5/#/playlist?playlistId={playlist_id}&playlistType=ordinary"
def migu_song_link(content_id: str) -> str:
return f"https://music.migu.cn/v3/music/song/{content_id}"
def format_seconds(seconds: float) -> str:
minutes = math.floor(seconds / 60)
secs = math.floor(seconds % 60)
return f"{minutes:02d}:{secs:02d}"
# ---------------- 条目 → 歌曲(上游 convertItemToSongWithOption) ----------------
def _seconds_from_size(size: int, bitrate: int) -> int:
return math.floor(size * 8 / bitrate) if bitrate > 0 else 0
def convert_item_to_song(item: dict[str, Any], allow_paid: bool = True) -> dict[str, Any] | None:
artist_names = collect_migu_artist_names(item)
song_name = first_non_empty(item.get("name"), item.get("songName"))
album_name = str(item.get("album") or "").strip()
albums = item.get("albums") or []
if isinstance(albums, list) and albums:
first_album = albums[0] if isinstance(albums[0], dict) else {}
if str(first_album.get("name") or "").strip():
album_name = str(first_album.get("name")).strip()
formats = collect_migu_formats(item)
if not formats:
return None
duration = to_number(item.get("duration"))
download_tags = [str(tag or "").strip().lower() for tag in (item.get("downloadTags") or [])]
candidates: list[dict[str, Any]] = []
for index, fmt in enumerate(formats):
size_text = first_non_zero_string(fmt.get("androidSize"), fmt.get("asize"), fmt.get("size"), fmt.get("isize"))
size_value = parse_int(size_text)
ext = first_non_empty(fmt.get("androidFileType"), fmt.get("fileType"))
if not ext:
ext = migu_format_ext(fmt.get("formatType"), first_non_empty(fmt.get("aformat"), fmt.get("iformat")))
if not duration and size_value > 0:
bitrate = {"PQ": 128_000, "HQ": 320_000, "LQ": 64_000}.get(str(fmt.get("formatType") or ""), 0)
if bitrate > 0:
duration = _seconds_from_size(size_value, bitrate)
price_value = parse_int(fmt.get("price"))
show_tags = fmt.get("showTag")
tags = show_tags if isinstance(show_tags, list) and show_tags else (fmt.get("showTags") or [])
is_vip_tag = any(str(tag or "").strip().lower() == "vip" for tag in tags)
if not is_vip_tag:
is_vip_tag = "vip" in download_tags
hidden_paid = item.get("chargeAuditions") == "1" and price_value >= 200
if allow_paid or (not is_vip_tag and not hidden_paid):
candidates.append(
{
"index": index,
"size": size_value,
"ext": ext,
"rank": migu_format_rank(fmt.get("formatType")),
"vip": is_vip_tag or hidden_paid,
"tone": str(fmt.get("formatType") or ""),
}
)
if not candidates:
return None
candidates.sort(key=lambda entry: (-entry["rank"], -entry["size"]))
best = candidates[0]
best_format = formats[best["index"]]
cover_url = normalize_migu_image_url(
first_non_empty(
pick_migu_image(item.get("imgItems")),
pick_migu_image(item.get("albumImgs")),
item.get("img1"),
item.get("img2"),
item.get("img3"),
)
)
# albums 数组是搜索响应的专辑来源(顶层 albumId 常为空),与上游 Extra 的 album_id 同义
album_id = first_non_empty(item.get("albumId"), (albums[0] if albums else {}).get("id"))
link_id = first_non_empty(item.get("contentId"), item.get("copyrightId"))
resource_type = best_format.get("resourceType")
format_type = best_format.get("formatType")
extra: dict[str, str] = {
"content_id": str(item.get("contentId") or ""),
# 注意:extra 用空串兜底,songmid 用 '2' 兜底(与原文两处写法不同)
"resource_type": str(resource_type if resource_type is not None else ""),
"format_type": str(format_type if format_type is not None else ""),
}
if item.get("copyrightId"):
extra["copyright_id"] = str(item["copyrightId"])
if item.get("songId"):
extra["song_id"] = str(item["songId"])
if album_id:
extra["album_id"] = album_id
is_vip = item.get("chargeAuditions") == "1" or bool(best["vip"])
songmid = f"{item.get('contentId') or ''}|{resource_type if resource_type is not None else '2'}|{format_type or ''}"
# 各档位可用性(供前端展示);实际可取档位由取流接口的探测结果决定
types = [{"type": TONE_QUALITY.get(entry["tone"], "128k"), "size": str(entry["size"])} for entry in candidates]
return {
"songmid": songmid,
"name": song_name,
"singer": " / ".join(artist_names),
"albumId": album_id,
"albumName": album_name,
"interval": format_seconds(duration) if duration else "0",
"img": cover_url or None,
"lrc": None,
"types": types,
"_types": {},
"payPlay": is_vip,
# 路由/取流需要 content_id|resource_type|format_type 三元组与版权号,放在 extra 里
"extra": {**extra, "link": migu_song_link(link_id)},
"source": "mg",
}
# ---------------- 歌曲详情 ----------------
async def fetch_song_detail(content_id: str) -> dict[str, Any]:
url = build_url(SONG_DETAIL_URL, {"resourceType": "2", "resourceId": content_id})
body = await migu_get(url)
resource = body.get("resource") if isinstance(body, dict) else None
if not resource:
raise RuntimeError("migu 歌曲详情为空")
return resource[0]
# ---------------- Z3D 解密(上游 DecryptAudio / miguZ3DStreamKey) ----------------
def is_migu_audio_magic(data: bytes) -> bool:
"""判断是否明文音频:fLaC / RIFF-WAVE / OggS / ID3 / MP4(ftyp) / MP3 帧同步。"""
if not data or len(data) < 4:
return False
if data[:4] in (b"fLaC", b"OggS"):
return True
if data[:4] == b"RIFF" and len(data) >= 12 and data[8:12] == b"WAVE":
return True
if data[:3] == b"ID3":
return True
if len(data) >= 12 and data[4:8] == b"ftyp":
return True
return data[0] == 0xFF and (data[1] & 0xE0) == 0xE0
def migu_z3d_stream_key(file_key: str) -> bytes:
"""streamKey = MD5(解密后的 appCode + fileKey) 的大写十六进制(32 字节 ASCII)。"""
key = str(file_key or "").strip()
if not key:
raise RuntimeError("空的 z3d fileKey")
# Go 的 CryptBlocks 按块解密不做 PKCS#7 去填充,这里同样禁用填充,否则尾块校验失败
decryptor = Cipher(algorithms.AES(_ANDROID_APP_KEY), modes.CBC(_Z3D_IV)).decryptor()
app_code = decryptor.update(_ENCRYPTED_APP_CODE) + decryptor.finalize()
digest = hashlib.md5(app_code + key.encode("utf-8")).hexdigest().upper()
return digest.encode("ascii")
def decrypt_z3d(data: bytes, file_key: str) -> bytes:
"""Z3D 流解密:逐字节 `plain[i] = cipher[i] - key[i % 32]`;已是明文则原样返回。"""
if not data:
raise RuntimeError("空的 z3d 数据")
if is_migu_audio_magic(data):
return data
payload = data[3:] if data[:3] == b"Z3D" else data
if not payload:
raise RuntimeError("剥离 Z3D 魔数后无数据")
stream_key = migu_z3d_stream_key(file_key)
plain = bytes((byte - stream_key[index & 31]) & 0xFF for index, byte in enumerate(payload))
if not is_migu_audio_magic(plain):
raise RuntimeError("z3d 解密后不是有效音频")
return plain