做个笔记，ncm格式转mp3格式
参考：传送门
import os
import json
import base64
import struct
import logging
import binascii
from glob import glob
from tqdm.auto import tqdm
from textwrap import dedent
from Crypto.Cipher import AES
from multiprocessing import Poolclass TqdmLoggingHandler(logging.StreamHandler):"""Avoid tqdm progress bar interruption by logger's output to console"""# see logging.StreamHandler.eval method:# https://github.com/python/cpython/blob/d2e2534751fd675c4d5d3adc208bf4fc984da7bf/Lib/logging/__init__.py#L1082-L1091# and tqdm.write method:# https://github.com/tqdm/tqdm/blob/f86104a1f30c38e6f80bfd8fb16d5fcde1e7749f/tqdm/std.py#L614-L620def emit(self, record):try:msg = self.format(record)tqdm.write(msg, end=self.terminator)except RecursionError:raiseexcept Exception:self.handleError(record)log = logging.getLogger(__name__)
log.setLevel(logging.INFO)
handler = TqdmLoggingHandler()
fmt = '%(levelname)7s [%(asctime)s] %(message)s'
datefmt = '%Y-%m-%d %H:%M:%S'
handler.setFormatter(logging.Formatter(fmt, datefmt))
log.addHandler(handler)def dump_single_file(filepath):try:filename = filepath.split('/')[-1]if not filename.endswith('.ncm'): returnfilename = filename[:-4]for ftype in ['mp3', 'flac']:fname = f'{filename}.{ftype}'if os.path.isfile(fname):log.warning(f'Skipping "{filepath}" due to existing file "{fname}"')returnlog.info(f'Converting "{filepath}"')# hex to strcore_key = binascii.a2b_hex('687A4852416D736F356B496E62617857')meta_key = binascii.a2b_hex('2331346C6A6B5F215C5D2630553C2728')unpad = lambda s: s[0:-(s[-1] if isinstance(s[-1], int) else ord(s[-1]))]with open(filepath, 'rb') as f:header = f.read(8)# str to hexassert binascii.b2a_hex(header) == b'4354454e4644414d'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] ^= 0x64key_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 = 0last_byte = 0key_offset = 0for i in range(256):swap = key_box[i]c = (swap + last_byte + key_data[key_offset]) & 0xffkey_offset += 1if key_offset >= key_length:key_offset = 0key_box[i] = key_box[c]key_box[c] = swaplast_byte = cmeta_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] ^= 0x63meta_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]image_data = f.read(image_size)target_filename = filename + '.' + meta_data['format']with open(target_filename, 'wb') as m:chunk = bytearray()while True:chunk = bytearray(f.read(0x8000))chunk_length = len(chunk)if not chunk:breakfor i in range(1, chunk_length + 1):j = i & 0xffchunk[i - 1] ^= key_box[(key_box[j] + key_box[(key_box[j] + j) & 0xff]) & 0xff]m.write(chunk)log.info(f'Converted file saved at "{target_filename}"')return target_filenameexcept KeyboardInterrupt:log.warning('Aborted')quit()def list_filepaths(path):if os.path.isfile(path):return [path]elif os.path.isdir(path):return [fp for p in glob(f'{path}/*') for fp in list_filepaths(p)]else:raise ValueError(f'path not recognized: {path}')def dump(*paths, n_workers=None):header = dedent(r'''_ _ ___ __ __ ___ _ _ __ __ ____ __ _ _| \| |/ __| \/ | \| | | | \/ | _ \| '_ \ || | .` | (__| |\/| | |) | |_| | |\/| | _/| .__/\_, |_|\_|\___|_| |_|___/ \___/|_| |_|_| |_| |__/ pyNCMDUMP https://github.com/allenfrostline/pyNCMDUMP ''')for line in header.split('\n'):log.info(line)all_filepaths = [fp for p in paths for fp in list_filepaths(p)]if n_workers > 1:log.info(f'Running pyNCMDUMP with up to {n_workers} parallel workers')with Pool(processes=n_workers) as p:list(p.map(dump_single_file, all_filepaths))else:log.info('Running pyNCMDUMP on single-worker mode')for fp in tqdm(all_filepaths, leave=False): dump_single_file(fp)log.info('All finished')if __name__ == '__main__':from argparse import ArgumentParserparser = ArgumentParser(description='pyNCMDUMP command-line interface')parser.add_argument('paths',metavar='paths',type=str,nargs='+',help='one or more paths to source files')parser.add_argument('-w', '--workers',metavar='',type=int,help=f'parallel convertion when set to more than 1 workers (default: 1)',default=1)args = parser.parse_args()dump(*args.paths, n_workers=args.workers)
