ypshidifu
3af7501b47
1. 修复 Line 277 语法错误：output_path")" 多余引号导致的 SyntaxError
2. 添加缺失的 META_KEY 常量（NCM 元数据 AES 密钥，16 字节）
3. 添加缺失的 KUGOU_KEY_PATH 常量（指向 kugou_key.xz）
4. 添加缺失的 stats 和 kgm_decoder 模块级初始化
5. 添加缺失的 import io（用于 _convert_audio 的 BytesIO）
6. 添加缺失的 _sniff_container 函数（音频容器魔术字节检测）
7. 添加缺失的 _extension_for 函数（格式到扩展名映射）
8. 添加缺失的 _convert_audio 函数（原始音频转码）
9. 修复 print("\n") 为 print()（恢复更清洁的输出格式）
458 lines
17 KiB
Python
458 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
import os, sys, glob, struct, base64, json, io
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
from Crypto.Cipher import AES
|
|
from Crypto.Util.Padding import unpad
|
|
from mutagen.mp3 import MP3
|
|
from mutagen.flac import FLAC
|
|
from mutagen.oggvorbis import OggVorbis
|
|
from mutagen.id3 import ID3, TIT2, TPE1, TALB, APIC
|
|
from pydub import AudioSegment
|
|
|
|
NCM_MAGIC = b"CTENFDAM"
|
|
CORE_KEY = b"hzHRAmso5kInbaxW"
|
|
META_KEY = bytes.fromhex("2331346C6A6B5F215C5D2630553C2728")
|
|
META_PREFIX = b"163 key(D't modify):"
|
|
KUGOU_KEY_PATH = Path(__file__).parent / "kugou_key.xz"
|
|
|
|
_FORMAT_MAP = {
|
|
"mp3": "mp3",
|
|
"flac": "flac",
|
|
"wav": "wav",
|
|
"ogg": "ogg",
|
|
"m4a": "mp4",
|
|
"mp4": "mp4",
|
|
}
|
|
|
|
stats = {"success": 0, "failed": 0, "files": []}
|
|
kgm_decoder = None
|
|
|
|
def _u32le(data, pos):
|
|
return struct.unpack(b"<I", data[pos:pos+4])[0]
|
|
|
|
def _sniff_container(header):
|
|
"""从魔术字节检测音频容器类型。"""
|
|
if len(header) >= 3 and header[:3] == b"ID3":
|
|
return "mp3"
|
|
if len(header) >= 2 and header[0] == 0xFF and (header[1] & 0xE0) == 0xE0:
|
|
return "mp3"
|
|
if len(header) >= 4 and header[:4] == b"fLaC":
|
|
return "flac"
|
|
if len(header) >= 4 and header[:4] == b"OggS":
|
|
return "ogg"
|
|
if len(header) >= 12 and header[:4] == b"RIFF" and header[8:12] == b"WAVE":
|
|
return "wav"
|
|
if len(header) >= 12 and header[4:8] == b"ftyp":
|
|
return "mp4"
|
|
return None
|
|
|
|
def _extension_for(fmt):
|
|
"""将格式名称映射为文件后缀。"""
|
|
mapping = {
|
|
"mp3": "mp3",
|
|
"flac": "flac",
|
|
"wav": "wav",
|
|
"ogg": "ogg",
|
|
"m4a": "m4a",
|
|
"mp4": "m4a",
|
|
}
|
|
return mapping.get(fmt, "mp3")
|
|
|
|
def ncm_stream_bytes(key):
|
|
sbox = bytearray(range(256))
|
|
j = 0
|
|
klen = len(key)
|
|
for i in range(256):
|
|
j = (j + sbox[i] + key[i % klen]) & 0xFF
|
|
sbox[i], sbox[j] = sbox[j], sbox[i]
|
|
box = bytearray(256)
|
|
for i in range(256):
|
|
box[i] = sbox[(sbox[i] + sbox[(i + sbox[i]) & 0xFF]) & 0xFF]
|
|
return bytes(box)
|
|
|
|
def ncm_decrypt_audio(data, box):
|
|
reps = (len(data) + 256) // 256 + 1
|
|
stream = (box * reps)[1:1 + len(data)]
|
|
return (int.from_bytes(data, b"little") ^ int.from_bytes(stream, b"little")).to_bytes(len(data), b"little")
|
|
|
|
def decrypt_ncm(file_path):
|
|
data = Path(file_path).read_bytes()
|
|
if len(data) < 0x2C or data[:8] != NCM_MAGIC:
|
|
raise ValueError("Not a valid NCM file")
|
|
pos = 10
|
|
key_len = _u32le(data, pos); pos += 4
|
|
key_blob = bytearray(data[pos:pos+key_len])
|
|
pos += key_len
|
|
for i in range(len(key_blob)):
|
|
key_blob[i] ^= 0x64
|
|
key_plain = unpad(AES.new(CORE_KEY, AES.MODE_ECB).decrypt(bytes(key_blob)), 16)
|
|
rc4_key = key_plain[17:]
|
|
meta_len = _u32le(data, pos); pos += 4
|
|
meta = {}
|
|
if meta_len:
|
|
meta_xor = bytes(b ^ 0x63 for b in data[pos:pos+meta_len])
|
|
pos += meta_len
|
|
if meta_xor.startswith(META_PREFIX):
|
|
payload = base64.b64decode(meta_xor[len(META_PREFIX):])
|
|
plain = unpad(AES.new(META_KEY, AES.MODE_ECB).decrypt(payload), 16)
|
|
meta = json.loads(plain[6:].decode("utf-8", "replace"))
|
|
pos += 5
|
|
cover = None
|
|
if pos + 8 <= len(data):
|
|
image_space = _u32le(data, pos); pos += 4
|
|
image_size = _u32le(data, pos); pos += 4
|
|
if image_size and pos + image_size <= len(data):
|
|
cover = data[pos:pos+image_size]
|
|
if pos + image_space <= len(data):
|
|
pos += image_space
|
|
else:
|
|
pos = len(data)
|
|
box = ncm_stream_bytes(rc4_key)
|
|
audio = ncm_decrypt_audio(data[pos:], box)
|
|
container = _sniff_container(audio[:64])
|
|
if not container:
|
|
container = "mp3"
|
|
tags = {}
|
|
if meta:
|
|
if meta.get("musicName"):
|
|
tags["title"] = str(meta["musicName"])
|
|
artists = meta.get("artist") or []
|
|
if artists:
|
|
if isinstance(artists[0], (list, tuple)):
|
|
tags["artist"] = str(artists[0][0])
|
|
else:
|
|
tags["artist"] = str(artists[0])
|
|
if meta.get("album"):
|
|
tags["album"] = str(meta["album"])
|
|
return audio, container, tags, cover
|
|
|
|
KGM_MAGIC = bytes([0x7C, 0xD5, 0x32, 0xEB, 0x86, 0x02, 0x7F, 0x4B,
|
|
0xA8, 0xAF, 0xA6, 0x8E, 0x0F, 0xFF, 0x99, 0x14])
|
|
VPR_MAGIC = bytes([0x05, 0x28, 0xBC, 0x96, 0xE9, 0xE4, 0x5A, 0x43,
|
|
0x91, 0xAA, 0xBD, 0xD0, 0x7A, 0xF5, 0x36, 0x31])
|
|
HEADER_LEN = 1024
|
|
OWN_KEY_LEN = 17
|
|
BLOCK = 16
|
|
MEND_TABLE = bytes([
|
|
0xB8, 0xD5, 0x3D, 0xB2, 0xE9, 0xAF, 0x78, 0x8C, 0x83, 0x33, 0x71, 0x51, 0x76, 0xA0, 0xCD, 0x37,
|
|
0x2F, 0x3E, 0x35, 0x8D, 0xA9, 0xBE, 0x98, 0xB7, 0xE7, 0x8C, 0x22, 0xCE, 0x5A, 0x61, 0xDF, 0x68,
|
|
0x69, 0x89, 0xFE, 0xA5, 0xB6, 0xDE, 0xA9, 0x77, 0xFC, 0xC8, 0xBD, 0xBD, 0xE5, 0x6D, 0x3E, 0x5A,
|
|
0x36, 0xEF, 0x69, 0x4E, 0xBE, 0xE1, 0xE9, 0x66, 0x1C, 0xF3, 0xD9, 0x02, 0xB6, 0xF2, 0x12, 0x9B,
|
|
0x44, 0xD0, 0x6F, 0xB9, 0x35, 0x89, 0xB6, 0x46, 0x6D, 0x73, 0x82, 0x06, 0x69, 0xC1, 0xED, 0xD7,
|
|
0x85, 0xC2, 0x30, 0xDF, 0xA2, 0x62, 0xBE, 0x79, 0x2D, 0x62, 0x62, 0x3D, 0x0D, 0x7E, 0xBE, 0x48,
|
|
0x89, 0x23, 0x02, 0xA0, 0xE4, 0xD5, 0x75, 0x51, 0x32, 0x02, 0x53, 0xFD, 0x16, 0x3A, 0x21, 0x3B,
|
|
0x16, 0x0F, 0xC3, 0xB2, 0xBB, 0xB3, 0xE2, 0xBA, 0x3A, 0x3D, 0x13, 0xEC, 0xF6, 0x01, 0x45, 0x84,
|
|
0xA5, 0x70, 0x0F, 0x93, 0x49, 0x0C, 0x64, 0xCD, 0x31, 0xD5, 0xCC, 0x4C, 0x07, 0x01, 0x9E, 0x00,
|
|
0x1A, 0x23, 0x90, 0xBF, 0x88, 0x1E, 0x3B, 0xAB, 0xA6, 0x3E, 0xC4, 0x73, 0x47, 0x10, 0x7E, 0x3B,
|
|
0x5E, 0xBC, 0xE3, 0x00, 0x84, 0xFF, 0x09, 0xD4, 0xE0, 0x89, 0x0F, 0x5B, 0x58, 0x70, 0x4F, 0xFB,
|
|
0x65, 0xD8, 0x5C, 0x53, 0x1B, 0xD3, 0xC8, 0xC6, 0xBF, 0xEF, 0x98, 0xB0, 0x50, 0x4F, 0x0F, 0xEA,
|
|
0xE5, 0x83, 0x58, 0x8C, 0x28, 0x2C, 0x84, 0x67, 0xCD, 0xD0, 0x9E, 0x47, 0xDB, 0x27, 0x50, 0xCA,
|
|
0xF4, 0x63, 0x63, 0xE8, 0x97, 0x7F, 0x1B, 0x4B, 0x0C, 0xC2, 0xC1, 0x21, 0x4C, 0xCC, 0x58, 0xF5,
|
|
0x94, 0x52, 0xA3, 0xF3, 0xD3, 0xE0, 0x68, 0xF4, 0x00, 0x23, 0xF3, 0x5E, 0x0A, 0x7B, 0x93, 0xDD,
|
|
0xAB, 0x12, 0xB2, 0x13, 0xE8, 0x84, 0xD7, 0xA7, 0x9F, 0x0F, 0x32, 0x4C, 0x55, 0x1D, 0x04, 0x36,
|
|
0x52, 0xDC, 0x03, 0xF3, 0xF9, 0x4E, 0x42, 0xE9, 0x3D, 0x61, 0xEF, 0x7C, 0xB6, 0xB3, 0x93, 0x50,
|
|
])
|
|
|
|
def _scramble(value):
|
|
return value ^ ((value & 0x0F) << 4)
|
|
|
|
class KgmDecoder:
|
|
def __init__(self, key_path):
|
|
import lzma
|
|
self.pub_key = lzma.decompress(Path(key_path).read_bytes())
|
|
self._phase_cache = {}
|
|
def _phase_tables(self, phase, own_key, pub_key):
|
|
cache_key = (phase, own_key, pub_key)
|
|
if cache_key in self._phase_cache:
|
|
return self._phase_cache[cache_key]
|
|
mend = len(MEND_TABLE)
|
|
columns = []
|
|
for col in range(BLOCK):
|
|
own_idx = (phase * BLOCK + col) % OWN_KEY_LEN
|
|
pub_idx = (phase * BLOCK + col) % mend
|
|
table = []
|
|
for pub_value in range(256):
|
|
xormask = _scramble(pub_value ^ MEND_TABLE[pub_idx])
|
|
table.append(bytes(_scramble(v ^ own_key[own_idx]) ^ xormask for v in range(256)))
|
|
columns.append(table)
|
|
tables = tuple(columns)
|
|
self._phase_cache[cache_key] = tables
|
|
return tables
|
|
def decode(self, file_path):
|
|
data = Path(file_path).read_bytes()
|
|
if len(data) < HEADER_LEN + 16:
|
|
raise ValueError("File too short for KGM/VPR")
|
|
header = data[:HEADER_LEN]
|
|
magic = header[:16]
|
|
if magic != KGM_MAGIC and magic != VPR_MAGIC:
|
|
raise ValueError("Not a valid KGM/VPR file")
|
|
audio_offset = struct.unpack("<I", header[0x10:0x14])[0]
|
|
crypto_version = struct.unpack("<I", header[0x14:0x18])[0]
|
|
crypto_test = header[0x1C:0x2C]
|
|
if crypto_version == 5:
|
|
raise ValueError("KGG v5 not supported")
|
|
own_key = crypto_test + b"\x00"
|
|
audio = bytearray(data[audio_offset:])
|
|
pub_len = len(self.pub_key)
|
|
out = bytearray(len(audio))
|
|
block_count = (len(audio) + BLOCK - 1) // BLOCK
|
|
for block_idx in range(block_count):
|
|
if block_idx >= pub_len:
|
|
raise ValueError("Key file mismatch")
|
|
phase = block_idx % 17
|
|
tables = self._phase_tables(phase, own_key, self.pub_key)
|
|
pub_value = self.pub_key[block_idx]
|
|
start = block_idx * BLOCK
|
|
end = min(start + BLOCK, len(audio))
|
|
for col, offset in enumerate(range(start, end)):
|
|
out[offset] = tables[col][pub_value][audio[offset]]
|
|
payload = bytes(out)
|
|
container = _sniff_container(payload[:64])
|
|
return payload, container, {}, None
|
|
|
|
def _resolve_output(file_path, output_fmt):
|
|
p = Path(file_path)
|
|
ext = _extension_for(output_fmt) if output_fmt != "original" else p.suffix.lstrip(".") or "mp3"
|
|
return str(p.with_suffix("." + ext))
|
|
|
|
def _add_tags(output_path, tags, cover=None):
|
|
p = Path(output_path)
|
|
ext = p.suffix.lower()
|
|
try:
|
|
if ext == ".mp3":
|
|
audio = MP3(output_path, ID3=ID3)
|
|
if "title" in tags:
|
|
audio["TIT2"] = TIT2(encoding=3, text=tags["title"])
|
|
if "artist" in tags:
|
|
audio["TPE1"] = TPE1(encoding=3, text=tags["artist"])
|
|
if "album" in tags:
|
|
audio["TALB"] = TALB(encoding=3, text=tags["album"])
|
|
if cover:
|
|
mime = "image/jpeg" if cover[:3] == b"\xff\xd8\xff" else "image/png"
|
|
audio["APIC"] = APIC(encoding=3, mime=mime, type=3, desc="cover", data=cover)
|
|
audio.save()
|
|
elif ext == ".flac":
|
|
if p.exists():
|
|
audio = FLAC(output_path)
|
|
else:
|
|
audio = FLAC()
|
|
if "title" in tags: audio["title"] = tags["title"]
|
|
if "artist" in tags: audio["artist"] = tags["artist"]
|
|
if "album" in tags: audio["album"] = tags["album"]
|
|
if cover:
|
|
audio["metadata_block_picture"] = base64.b64encode(cover).decode()
|
|
audio.save()
|
|
elif ext == ".ogg":
|
|
audio = OggVorbis(output_path)
|
|
if "title" in tags: audio["TITLE"] = tags["title"]
|
|
if "artist" in tags: audio["ARTIST"] = tags["artist"]
|
|
if "album" in tags: audio["ALBUM"] = tags["album"]
|
|
audio.save()
|
|
except Exception:
|
|
pass
|
|
|
|
def _convert_audio(raw, container, out_fmt, output_path):
|
|
"""将解密后的原始音频转码为目标格式。"""
|
|
src_fmt = _FORMAT_MAP.get(container, container)
|
|
dst_fmt = _FORMAT_MAP.get(out_fmt, out_fmt)
|
|
if src_fmt == dst_fmt:
|
|
Path(output_path).write_bytes(raw)
|
|
else:
|
|
audio = AudioSegment.from_file(io.BytesIO(raw), format=src_fmt)
|
|
audio.export(output_path, format=dst_fmt)
|
|
|
|
def convert_file(file_path, output_fmt, delete_flag):
|
|
global stats, kgm_decoder
|
|
file_path = str(Path(file_path).resolve())
|
|
if not Path(file_path).exists():
|
|
print(" [FAIL] File not found: " + file_path)
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
p = Path(file_path)
|
|
suffix = p.suffix.lower()
|
|
if suffix == ".ncm":
|
|
try:
|
|
raw, container, tags, cover = decrypt_ncm(file_path)
|
|
except Exception as e:
|
|
print(" [FAIL] NCM decrypt failed: " + file_path + " -> " + str(e))
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
elif suffix in (".kgm", ".kgma", ".vpr", ".kgm.flac", ".vpr.flac"):
|
|
if kgm_decoder is None:
|
|
if not KUGOU_KEY_PATH.exists():
|
|
print(" [FAIL] Missing kugou key file: " + str(KUGOU_KEY_PATH))
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
try:
|
|
kgm_decoder = KgmDecoder(str(KUGOU_KEY_PATH))
|
|
except Exception as e:
|
|
print(" [FAIL] Failed to load kugou key: " + str(e))
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
try:
|
|
raw, container, tags, cover = kgm_decoder.decode(file_path)
|
|
except Exception as e:
|
|
print(" [FAIL] KGM decrypt failed: " + file_path + " -> " + str(e))
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
else:
|
|
print(" [SKIP] Unsupported format: " + suffix)
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
out_fmt = container if output_fmt == "original" else output_fmt
|
|
output_path = _resolve_output(file_path, out_fmt)
|
|
try:
|
|
_convert_audio(raw, container, out_fmt, output_path)
|
|
except Exception as e:
|
|
print(" [FAIL] Conversion failed: " + file_path + " -> " + str(e))
|
|
stats["failed"] += 1
|
|
stats["files"].append((file_path, False))
|
|
return
|
|
try:
|
|
_add_tags(output_path, tags, cover)
|
|
except Exception as e:
|
|
print(" [WARN] Tag write failed: " + str(e))
|
|
if delete_flag:
|
|
try:
|
|
os.remove(file_path)
|
|
print(" [OK] " + file_path + " -> " + output_path + " (original deleted)")
|
|
except Exception as e:
|
|
print(" [OK] " + file_path + " -> " + output_path + " (delete failed: " + str(e) + ")")
|
|
else:
|
|
print(" [OK] " + file_path + " -> " + output_path)
|
|
stats["success"] += 1
|
|
stats["files"].append((file_path, True))
|
|
|
|
def expand_files(patterns):
|
|
result = []
|
|
for pat in patterns:
|
|
matched = glob.glob(pat, recursive=True)
|
|
if matched:
|
|
result.extend(matched)
|
|
else:
|
|
p = Path(pat)
|
|
if p.exists():
|
|
result.append(str(p.resolve()))
|
|
else:
|
|
result.append(pat)
|
|
return result
|
|
|
|
def print_stats():
|
|
print()
|
|
print("=== Conversion Stats ===")
|
|
print(" Success: " + str(stats["success"]) + " | Failed: " + str(stats["failed"]))
|
|
if stats["files"]:
|
|
print()
|
|
print(" Files:")
|
|
for fp, ok in stats["files"]:
|
|
status = "OK" if ok else "FAIL"
|
|
print(" [" + status + "] " + fp)
|
|
print()
|
|
|
|
def main():
|
|
global stats
|
|
if len(sys.argv) > 1:
|
|
argv = sys.argv[1:]
|
|
patterns = []
|
|
output_fmt = "mp3"
|
|
delete_flag = False
|
|
i = 0
|
|
while i < len(argv):
|
|
arg = argv[i]
|
|
if arg == "-f" and i + 1 < len(argv):
|
|
output_fmt = argv[i + 1].lower()
|
|
i += 2
|
|
elif arg == "-d":
|
|
delete_flag = True
|
|
i += 1
|
|
elif arg.startswith("-"):
|
|
i += 1
|
|
else:
|
|
patterns.append(arg)
|
|
i += 1
|
|
if not patterns:
|
|
print("Usage: vipsong.py [options] <files...>")
|
|
print(" Options: -f <format> Output format (mp3/flac/wav/ogg/m4a/original)")
|
|
print(" -d Delete original file after conversion")
|
|
print(" Examples:")
|
|
print(" vipsong.py *.ncm")
|
|
print(" vipsong.py -f flac -d ./music/*.ncm")
|
|
print(" vipsong.py ./song.kgm")
|
|
sys.exit(1)
|
|
files = expand_files(patterns)
|
|
for f in files:
|
|
convert_file(f, output_fmt, delete_flag)
|
|
print_stats()
|
|
return
|
|
print("=" * 50)
|
|
print(" vipsong - Music Format Converter")
|
|
print(" Support: .ncm (NetEase) .kgm/.vpr (KuGou)")
|
|
print(" Output: mp3 / flac / wav / ogg / m4a")
|
|
print("=" * 50)
|
|
print("Commands:")
|
|
print(" <filepath> Convert file or glob pattern (e.g. *.ncm)")
|
|
print(" -f <format> Output format (default: mp3)")
|
|
print(" -d Delete original after conversion")
|
|
print(" /exit Exit")
|
|
print(" /get Show stats")
|
|
print()
|
|
while True:
|
|
try:
|
|
line = input("vipsong> ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print()
|
|
print("Bye!")
|
|
break
|
|
if not line:
|
|
continue
|
|
parts = line.split()
|
|
if not parts:
|
|
continue
|
|
cmd = parts[0]
|
|
if cmd == "/exit":
|
|
print("Bye!")
|
|
break
|
|
elif cmd == "/get":
|
|
print_stats()
|
|
continue
|
|
patterns = []
|
|
output_fmt = "mp3"
|
|
delete_flag = False
|
|
i = 1
|
|
while i < len(parts):
|
|
if parts[i] == "-f" and i + 1 < len(parts):
|
|
output_fmt = parts[i + 1].lower()
|
|
i += 2
|
|
elif parts[i] == "-d":
|
|
delete_flag = True
|
|
i += 1
|
|
elif parts[i].startswith("-"):
|
|
print(" [WARN] Unknown option: " + parts[i])
|
|
i += 1
|
|
else:
|
|
patterns.append(parts[i])
|
|
i += 1
|
|
if not patterns:
|
|
print(" Usage: <filepath> [-f format] [-d]")
|
|
print(" Examples: *.ncm")
|
|
print(" ./xx/xx.kgm")
|
|
print(" song.ncm -d")
|
|
print(" *.ncm -f flac")
|
|
continue
|
|
files = expand_files(patterns)
|
|
for f in files:
|
|
convert_file(f, output_fmt, delete_flag)
|
|
print()
|
|
print(" Current: Success " + str(stats["success"]) + " | Failed " + str(stats["failed"]))
|
|
print()
|
|
|
|
if __name__ == "__main__":
|
|
main()
|