#!/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", } _EXT_TO_FMT = { "mp3": "mp3", "flac": "flac", "wav": "wav", "ogg": "ogg", "oga": "ogg", "m4a": "mp4", "mp4": "mp4", } stats = {"success": 0, "failed": 0, "files": []} kgm_decoder = None DEBUG = bool(os.environ.get("VIPSONG_DEBUG", "")) def _normalize_format(fmt, default="mp3"): """规范化格式字符串,确保非空且可识别。""" if not fmt or not isinstance(fmt, str) or not fmt.strip(): return _FORMAT_MAP.get(default.lower(), "mp3") fmt = fmt.strip().lower() if fmt in _EXT_TO_FMT: return _EXT_TO_FMT[fmt] if fmt in _FORMAT_MAP: return _FORMAT_MAP[fmt] return default def _u32le(data, pos): return struct.unpack("= 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 _parse_gt_output(input_path, output_spec): """解析 'input>output' 重定向语法,返回 (output_path, output_fmt)。""" p = Path(input_path) spec = output_spec.strip() if not spec: # 空输出规格(如 'song.ncm>'),使用默认 mp3 格式 out_path = p.with_suffix(".mp3") return str(out_path), "mp3" # 如果包含路径分隔符,判断是目录+格式还是具体文件路径 if "/" in spec or "\\" in spec: last_sep = max(spec.rfind("/"), spec.rfind("\\")) remainder = spec[last_sep + 1:].strip() if not remainder: # 仅目录(如 '/path/'),使用原文件名 + mp3 扩展名 out_path = Path(spec) / (p.stem + ".mp3") elif remainder.startswith(".") and remainder[1:].lower() in _EXT_TO_FMT: # 目录 + 格式扩展(如 '../.ogg'),输出到该目录 ext = remainder[1:].lower() dir_part = spec[:last_sep + 1] out_path = Path(dir_part) / (p.stem + "." + ext) elif remainder.lower() in _EXT_TO_FMT: # 目录 + 裸格式名(如 '/path/ogg'),输出到该目录 ext = _extension_for(_EXT_TO_FMT[remainder.lower()]) dir_part = spec[:last_sep + 1] out_path = Path(dir_part) / (p.stem + "." + ext) else: # 具体文件路径(如 '/path/yy.mp3') out_path = Path(spec) elif "." in spec: # 仅文件名,放在输入文件同目录 out_path = p.parent / spec else: # 裸格式名,仅修改扩展名 out_path = p.with_suffix("." + spec) # 相对路径相对于输入文件的目录解析(而非 cwd),避免从不同目录运行时输出路径错误 if not out_path.is_absolute(): out_path = p.parent / out_path ext = out_path.suffix.lower().lstrip(".") out_fmt = _EXT_TO_FMT.get(ext, ext) if not out_fmt: out_fmt = "mp3" return str(out_path), out_fmt 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, "little") ^ int.from_bytes(stream, "little")).to_bytes(len(data), "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("= 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) if not src_fmt: src_fmt = "mp3" dst_fmt = _FORMAT_MAP.get(out_fmt, out_fmt) if not dst_fmt: dst_fmt = "mp3" if DEBUG: print(" [DEBUG] container=" + str(container) + " out_fmt=" + str(out_fmt) + " src_fmt=" + str(src_fmt) + " dst_fmt=" + str(dst_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="mp3", delete_flag=False, output_path=None): 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 if not out_fmt: out_fmt = "mp3" if output_path is None: 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 _process_patterns(patterns, output_fmt, delete_flag): """处理文件模式列表,支持 > 重定向语法。""" for pat in patterns: if ">" in pat: input_spec, output_spec = pat.split(">", 1) input_spec = input_spec.strip() output_spec = output_spec.strip() if not input_spec: print(" [WARN] > 重定向语法缺少输入文件: " + pat) continue if not output_spec: print(" [WARN] > 重定向语法缺少输出: " + pat + " (使用默认格式: mp3)") files = expand_files([input_spec]) for f in files: out_path, out_fmt = _parse_gt_output(f, output_spec) if not out_fmt: out_fmt = output_fmt convert_file(f, out_fmt, delete_flag, output_path=out_path) else: files = expand_files([pat]) for f in files: convert_file(f, output_fmt, delete_flag) def print_stats(): print() print("=== 转码统计 ===") print(" 成功: " + str(stats["success"]) + " | 失败: " + str(stats["failed"])) if stats["files"]: print() print(" 文件:") for fp, ok in stats["files"]: status = "成功" if ok else "失败" print(" [" + status + "] " + fp) print() def _split_line(line): """按空白分割命令行,处理引号但不处理反斜杠转义(适配 Windows 路径)。""" args = [] current = [] in_single = False in_double = False for ch in line: if ch == '"' and not in_single: in_double = not in_double elif ch == "'" and not in_double: in_single = not in_single elif ch in (' ', '\t') and not in_single and not in_double: if current: args.append(''.join(current)) current = [] else: current.append(ch) if current: args.append(''.join(current)) return args 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 = _normalize_format(argv[i + 1], "mp3") 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("用法: vipsong.py [选项] <文件...>") print(" 选项: -f <格式> 输出格式 (mp3/flac/wav/ogg/m4a/original)") print(" -d 转码后删除原文件") print(" 示例:") print(" vipsong.py *.ncm") print(" vipsong.py -f flac -d ./music/*.ncm") print(" vipsong.py ./song.kgm") sys.exit(1) _process_patterns(patterns, output_fmt, delete_flag) print_stats() return print("=" * 50) print(" vipsong - 音乐加密格式转码工具") print(" 支持: .ncm (网易云) .kgm/.vpr (酷狗)") print(" 输出: mp3 / flac / wav / ogg / m4a") print("=" * 50) print("命令:") print(" 转换文件或通配符模式 (如: *.ncm)") print(" -f <格式> 输出格式 (默认: mp3)") print(" -d 转码后删除原文件") print(" xx>yy.mp3 输出到指定文件名 (同目录下)") print(" xx>wav 转换格式,保留原名 (xx.wav)") print(" xx>/path/yy.mp3 输出到指定目录") print(" /exit 或 exit 退出程序") print(" /get 或 get 显示转码统计") print() while True: try: line = input("vipsong> ").strip() except (EOFError, KeyboardInterrupt): print() print("再见!") break if not line: continue parts = _split_line(line) if not parts: continue # 检查命令(支持带/或不带/前缀),避免指令被当作文件路径处理 cmd = parts[0].lstrip("/") if cmd == "exit": print("再见!") break if cmd == "get": print_stats() continue # 解析参数和文件模式(从 i=0 开始,parts[0] 可能是文件路径) patterns = [] output_fmt = "mp3" delete_flag = False i = 0 while i < len(parts): if parts[i] == "-f" and i + 1 < len(parts): output_fmt = _normalize_format(parts[i + 1], "mp3") i += 2 elif parts[i] == "-d": delete_flag = True i += 1 elif parts[i].startswith("-") and len(parts[i]) > 1 and not Path(parts[i]).exists(): print(" [WARN] Unknown option: " + parts[i]) i += 1 else: patterns.append(parts[i]) i += 1 if not patterns: print(" Usage: [-f format] [-d] 或 >output.xxx") print(" Examples: *.ncm") print(" ./xx/xx.kgm") print(" song.ncm -d") print(" *.ncm -f flac") print(" xx.ncm>yy.mp3 (重命名输出)") print(" xx.kgm>wav (转换格式)") print(" xx.ncm>/xx/xx.mp3 (指定输出目录)") continue _process_patterns(patterns, output_fmt, delete_flag) print() print(" 当前状态: 成功 " + str(stats["success"]) + " | 失败 " + str(stats["failed"])) print() if __name__ == "__main__": main()