初始提交：vipsong 音乐加密格式转码工具

README.md

@@ -0,0 +1,59 @@
+# vipsong - 酷狗/网易云音乐加密格式转码工具
+
+CLI 工具，将酷狗加密格式 (.kgm/.kgma/.vpr) 和网易云加密格式 (.ncm) 解密转码为常见音频格式。
+
+## 安装依赖
+
+pip install -r requirements.txt
+
+确保系统已安装 FFmpeg（用于格式转换）。
+
+## 酷狗密钥
+
+首次使用前，需获取酷狗密钥文件 kugou_key.xz：
+
+curl -o kugou_key.xz https://raw.githubusercontent.com/HRuiCcc/music-geshizhuanhuan/master/assets/kugou_key.xz
+
+## 使用方法
+
+### 命令行模式
+
+python vipsong.py *.ncm
+python vipsong.py ./music/song.kgm
+python vipsong.py *.ncm -f flac -d
+python vipsong.py song.ncm -d
+python vipsong.py ./downloads/*.ncm ./music/*.kgm
+
+### 交互模式
+
+python vipsong.py
+
+vipsong> *.ncm              # 转换当前目录全部 ncm 文件
+vipsong> ./xx/xx/xx.kgm     # 转换指定 kgm 文件
+vipsong> song.ncm -d        # 转码后删除原 ncm 文件
+vipsong> /get               # 查看转码统计
+vipsong> /exit              # 退出
+
+## 命令说明
+
+| 命令/参数 | 说明 |
+|-----------|------|
+| *.ncm | 转换当前目录全部 ncm 文件 |
+| ./xx/xx/xx.kgm | 转换指定路径的文件 |
+| -f <格式> | 指定输出格式：mp3/flac/wav/ogg/m4a/original（默认 mp3）|
+| -d | 转码成功后删除原加密文件 |
+| /exit | 退出交互模式 |
+| /get | 显示转码统计（成功数、失败数、文件列表） |
+
+## 支持格式
+
+| 输入格式 | 说明 | 输出格式 |
+|----------|------|----------|
+| .ncm | 网易云音乐加密格式 | mp3/flac/wav/ogg/m4a |
+| .kgm/.kgma/.vpr | 酷狗音乐加密格式 (v3/v4) | mp3/flac/wav/ogg/m4a |
+
+## 注意
+
+- 转换后保留原始 ID3 标签（歌名、艺术家、专辑、封面）
+- KGM 格式需要 kugou_key.xz 密钥文件（存放在项目根目录）
+- 仅支持 KGM v3/v4，v5 需要客户端密钥暂不支持

requirements.txt

@@ -0,0 +1,3 @@
+pycryptodome>=3.20
+mutagen>=1.47
+pydub>=0.25.1

vipsong.py

@@ -0,0 +1,405 @@
+#!/usr/bin/env python3
+import os, sys, glob, struct, base64, json
+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_PREFIX = b"163 key(D't modify):"
+
+def _u32le(data, pos):
+    return struct.unpack(b"<I", data[pos:pos+4])[0]
+
+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_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()