[python]批量转换ncm格式文件

前言

最近想换用本地其它播放器听音乐，但网易云音乐下载下来的文件格式是.ncm，不兼容其它播放器。网上找了下方案，参考网易云音乐ncm格式分析以及ncm与mp3格式转换实现了基本功能，在此基础上加了个多进程同时转换，以及通过命令行传一些参数，比如并发执行数、输入输出目录路径。

示例代码

其中有个非标准库依赖需要安装

python -m pip install pycryptodome

基本逻辑是参考原文的，修改处是添加了多进程执行、路径操作改用pathlib、支持命令行传参

import binascii
import struct
import base64
import json
import os
from Crypto.Cipher import AES
from pathlib import Path
from concurrent.futures import ProcessPoolExecutor
from time import time
import argparse


def get_args():
    parser = argparse.ArgumentParser(description="ncm file convertor")
    parser.add_argument("-j", type=int, default=None, help="Maximum number of concurrency, default is cpu cores count")
    parser.add_argument("-i", type=str, default=".", help="input dir path, default is current dir")
    parser.add_argument("-o", type=str, default=None, help="output dir path")
    return parser.parse_args()

def get_savepath(output_dir: str, file_name: str) -> Path:
    base_dir = Path(output_dir)
    if not base_dir.exists():
        base_dir.mkdir(parents=True, exist_ok=True)
    p = base_dir / file_name
    return p

def process(file_path: Path, output_dir: str):
    start = time()
    print(f"start convert {file_path}")
    core_key = binascii.a2b_hex("687A4852416D736F356B496E62617857")
    meta_key = binascii.a2b_hex("2331346C6A6B5F215C5D2630553C2728")
    unpad = lambda s: s[0 : -(s[-1] if type(s[-1]) == int else ord(s[-1]))]
    f = open(file_path, "rb")
    header = f.read(8)
    if binascii.b2a_hex(header) != b"4354454e4644414d":
        print("incorrect file header, skiped ...")
        return
    f.seek(2, 1)
    key_length = f.read(4)
    key_length = struct.unpack("<I", bytes(key_length))[0]
    key_data = f.read(key_length)
    key_data_array = bytearray(key_data)
    for i in range(0, len(key_data_array)):
        key_data_array[i] ^= 0x64
    key_data = bytes(key_data_array)
    cryptor = AES.new(core_key, AES.MODE_ECB)
    key_data = unpad(cryptor.decrypt(key_data))[17:]
    key_length = len(key_data)
    key_data = bytearray(key_data)
    key_box = bytearray(range(256))
    c = 0
    last_byte = 0
    key_offset = 0
    for i in range(256):
        swap = key_box[i]
        c = (swap + last_byte + key_data[key_offset]) & 0xFF
        key_offset += 1
        if key_offset >= key_length:
            key_offset = 0
        key_box[i] = key_box[c]
        key_box[c] = swap
        last_byte = c
    meta_length = f.read(4)
    meta_length = struct.unpack("<I", bytes(meta_length))[0]
    meta_data = f.read(meta_length)
    meta_data_array = bytearray(meta_data)
    for i in range(0, len(meta_data_array)):
        meta_data_array[i] ^= 0x63
    meta_data = bytes(meta_data_array)
    meta_data = base64.b64decode(meta_data[22:])
    cryptor = AES.new(meta_key, AES.MODE_ECB)
    meta_data = unpad(cryptor.decrypt(meta_data)).decode("utf-8")[6:]
    meta_data = json.loads(meta_data)
    crc32 = f.read(4)
    crc32 = struct.unpack("<I", bytes(crc32))[0]
    f.seek(5, 1)
    image_size = f.read(4)
    image_size = struct.unpack("<I", bytes(image_size))[0]

    file_name = f"{file_path.stem}.{meta_data['format']}"
    file_name = get_savepath(output_dir, file_name)
    m = open(file_name, "wb")
    chunk = bytearray()
    while True:
        chunk = bytearray(f.read(0x8000))
        chunk_length = len(chunk)
        if not chunk:
            break
        for i in range(1, chunk_length + 1):
            j = i & 0xFF
            chunk[i - 1] ^= key_box[
                (key_box[j] + key_box[(key_box[j] + j) & 0xFF]) & 0xFF
            ]
        m.write(chunk)
    m.close()
    f.close()
    end = time()
    print(f"convert {file_path.name} elapsed {end - start:.4f} seconds")


def create_task(input_dir: str, output_dir: str, jobs: int):
    current_dir = Path(input_dir)
    if not current_dir.exists():
        raise FileNotFoundError(f"{current_dir} not found")
    ncmfiles = current_dir.rglob("*.ncm")
    futures = []
    print(f"Maximum number of concurrency: {jobs}")
    with ProcessPoolExecutor(max_workers=jobs) as executor:
        for i in ncmfiles:
            futures.append(executor.submit(process, i, output_dir))


if __name__ == "__main__":
    args = get_args()
    if args.j is None or args.j < 1:
        jobs = os.cpu_count()
    else:
        jobs = args.j
    
    input_dir = args.i

    if args.o is None:
        output_dir = "output"
    else:
        output_dir = args.o
    try:
        create_task(input_dir, output_dir, jobs)
    except Exception as e:
        print(e)

使用。假设代码文件名为demo.py

# 指定最大进程数为 4, 源文件路径为 music, 转换后的文件目录为 result
python demo.py -j 4 -i music -o result

来源链接：https://www.cnblogs.com/XY-Heruo/p/18592412