ypshidifu
6676eb12ab
1. _process_patterns 新增防御性合并逻辑： - 检测 patterns 中是否包含 > 重定向标记 - 若存在则将 token 按 > 位置分割，合并为完整 pattern - 正确处理 after_gt 不以路径分隔符结尾的情况（自动补 /） - 解决 Linux bash 将 '路径/歌曲名.ncm>./1/2.flac' 切割成多个 token 的问题 2. convert_file 新增自动创建输出目录： - 在调用 _convert_audio 前自动 mkdir -p 父目录 - 目录创建失败时记录警告并标记为失败 示例: 输入: ['/www/wwwroot/vipsong/霓虹兔,Luv4', '-', 'MMook', 'JJi', 'BBa.ncm>./1', '2.flac'] 合并: ['/www/wwwroot/vipsong/霓虹兔,Luv4 - MMook JJi BBa.ncm>./1/2.flac']
648 lines
26 KiB
Python
648 lines
26 KiB
Python
#!/usr/bin/env python3
|
||
import os, sys, glob, struct, base64, json, io, re
|
||
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
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
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("<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 _parse_gt_output(input_path, output_spec):
|
||
"""解析 'input>output' 重定向语法，返回 (output_path, output_fmt)。"""
|
||
p = Path(input_path)
|
||
spec = output_spec.strip()
|
||
# 规范化路径分隔符后的空格（如 '../ file.mp3' -> '../file.mp3'），
|
||
# 保留文件名内部的空格
|
||
spec = re.sub(r'([/\\])\s+', r'\1', spec)
|
||
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("<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
|
||
if block_count > pub_len:
|
||
raise ValueError("Key file mismatch")
|
||
# Pre-compute all 17 phase tables; workers only read from cache
|
||
for phase in range(17):
|
||
self._phase_tables(phase, own_key, self.pub_key)
|
||
# Parallel block decryption: each block is independent
|
||
def _decrypt_block(block_idx):
|
||
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))
|
||
blk = bytearray(end - start)
|
||
for col in range(end - start):
|
||
blk[col] = tables[col][pub_value][audio[start + col]]
|
||
return start, bytes(blk)
|
||
max_workers = min(16, max(1, block_count))
|
||
with ThreadPoolExecutor(max_workers=max_workers) as executor:
|
||
for start, blk in executor.map(_decrypt_block, range(block_count)):
|
||
out[start:start + len(blk)] = blk
|
||
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:
|
||
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
|
||
except Exception as e:
|
||
print(" [WARN] 无法创建输出目录: " + str(Path(output_path).parent) + " -> " + str(e))
|
||
stats["failed"] += 1
|
||
stats["files"].append((file_path, False))
|
||
return
|
||
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):
|
||
"""处理文件模式列表，支持 > 重定向语法。
|
||
|
||
防御工程：Linux shell 会把含空格路径在 '>' 处切开的 token 拆成多个，
|
||
例如用户输入 `路径/歌曲名.mp3>./1/2.flac` 变成 ["...mp3>./1", "2.flac"]，
|
||
这里做后处理合并，把它们重新拼成一个完整的重定向 pattern。
|
||
"""
|
||
# 检测是否有重定向模式（包含 > 的 token）
|
||
has_redirect = any(">" in pat for pat in patterns)
|
||
if has_redirect and len(patterns) > 1:
|
||
# 找到包含 > 的位置
|
||
redirect_idx = next(i for i, p in enumerate(patterns) if ">" in p)
|
||
# input 部分：redirect_idx 之前的所有 token + redirect_idx token 中 > 前面的内容
|
||
before_redirect = " ".join(patterns[:redirect_idx])
|
||
first_with_gt = patterns[redirect_idx]
|
||
input_part, _, after_gt = first_with_gt.partition(">")
|
||
input_part = (before_redirect + " " + input_part).strip() if before_redirect else input_part
|
||
remaining_parts = patterns[redirect_idx + 1:]
|
||
# 合并输出路径
|
||
if after_gt and remaining_parts:
|
||
if not after_gt.endswith("/") and not after_gt.endswith("\\"):
|
||
output_spec = after_gt + "/" + "".join(remaining_parts)
|
||
else:
|
||
output_spec = after_gt + "".join(remaining_parts)
|
||
else:
|
||
output_spec = after_gt + "".join(remaining_parts)
|
||
patterns = [f"{input_part}>{output_spec}"]
|
||
elif not has_redirect and len(patterns) > 1:
|
||
# 多 token 但无重定向，检查是否是有空格的单路径
|
||
# 例如: ["path/song name.ncm"] 已经是单个 token，不需要合并
|
||
# 但如果用户忘了加引号: ["path/song", "name.ncm"]，尝试合并
|
||
combined = " ".join(patterns)
|
||
if Path(combined).exists() or glob.has_wildcard(combined):
|
||
patterns = [combined]
|
||
|
||
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):
|
||
"""按空白分割命令行，处理引号。支持 '>' 重定向语法：> 后的输出路径可含空格。
|
||
|
||
注意：'>' 后的内容（含空格）作为输出路径整体；参数请写在重定向前。
|
||
"""
|
||
args = []
|
||
current = []
|
||
in_single = False
|
||
in_double = False
|
||
after_redirect = 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 == '>' and not in_single and not in_double:
|
||
# 遇到 > 重定向：附加到当前 token，进入输出模式（保留空格）
|
||
after_redirect = True
|
||
current.append(ch)
|
||
elif ch in (' ', '\t') and not in_single and not in_double:
|
||
if after_redirect:
|
||
# 输出路径模式：保留空格（用于含空格的文件名）
|
||
current.append(ch)
|
||
else:
|
||
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(" <filepath> 转换文件或通配符模式 (如: *.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: <filepath> [-f format] [-d] 或 <filepath>>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()
|