youpaishidifu/vipsong
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Compare commits

merge into: youpaishidifu/vipsong:main
Branches Tags
youpaishidifu/vipsong:main youpaishidifu/vipsong:master
..
pull from: youpaishidifu/vipsong:master
Branches Tags
youpaishidifu/vipsong:main youpaishidifu/vipsong:master
1 Commits
main .. master
Author SHA1 Message Date
ypshidifu 6f7e011686 Initial commit: vipsong music format converter 2026-09-17 12:55:30 +08:00
5 changed files with 47 additions and 692 deletions
Expand all files Collapse all files
.gitignore
BIN
View File
Binary file not shown.
app.py
-139
View File
@@ -1,139 +0,0 @@
#!/usr/bin/env python3
"""vipsong Web — 音乐加密格式转码工具 (Web 版)。
并发安全说明:
- vipsong.convert_file() 会修改模块级全局 stats 字典，因此所有转码调用
被互斥锁 (_convert_lock) 串行化，避免多线程/协程并发时统计数据错乱
或输出路径冲突。
- 每隔 10 分钟自动清扫 static/uploads 与 static/outputs 目录，释放磁盘。
"""
import shutil
import threading
from contextlib import asynccontextmanager
from pathlib import Path
from fastapi import FastAPI, File, UploadFile, Form, HTTPException
from fastapi.responses import HTMLResponse, FileResponse
from fastapi.staticfiles import StaticFiles
import vipsong
# --- 并发隔离 ---------------------------------------------
# vipsong.convert_file() 内部使用模块全局 stats, 需加锁串行化
_convert_lock = threading.Lock()
# --- 定期清理 -----------------------------------------------
_CLEANUP_INTERVAL = 10 * 60 # 10 分钟
_cleanup_stop = threading.Event()
def _purge_directory(directory: Path):
"""删除目录内所有文件（保留目录本身）。"""
if not directory.exists():
return
for entry in directory.iterdir():
try:
if entry.is_file() or entry.is_symlink():
entry.unlink(missing_ok=True)
elif entry.is_dir():
shutil.rmtree(entry, ignore_errors=True)
except OSError:
pass
def _cleanup_loop():
"""后台线程：每隔 _CLEANUP_INTERVAL 清扫上传和输出目录。"""
while not _cleanup_stop.wait(_CLEANUP_INTERVAL):
_purge_directory(UPLOAD_DIR)
_purge_directory(OUTPUT_DIR)
@asynccontextmanager
async def lifespan(app: FastAPI):
# 启动清理线程
t = threading.Thread(target=_cleanup_loop, daemon=True, name="vipsong-cleanup")
t.start()
yield
# 关闭时停止清理线程
_cleanup_stop.set()
t.join(timeout=5)
app = FastAPI(title="vipsong Web", description="音乐加密格式转码工具", lifespan=lifespan)
BASE_DIR = Path(__file__).parent
UPLOAD_DIR = BASE_DIR / "static" / "uploads"
OUTPUT_DIR = BASE_DIR / "static" / "outputs"
UPLOAD_DIR.mkdir(parents=True, exist_ok=True)
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
app.mount("/static", StaticFiles(directory=str(BASE_DIR / "static")), name="static")
SUPPORTED_EXTENSIONS = [".ncm", ".kgm", ".kgma", ".vpr"]
SUPPORTED_FORMATS = ["mp3", "flac", "wav", "ogg", "m4a"]
def is_supported(filename: str) -> bool:
name = filename.lower()
return any(name.endswith(ext) for ext in SUPPORTED_EXTENSIONS)
@app.get("/", response_class=HTMLResponse)
async def index():
html = (BASE_DIR / "templates" / "index.html").read_text(encoding="utf-8")
return HTMLResponse(html)
@app.post("/convert")
async def convert(
file: UploadFile = File(...),
format: str = Form("mp3"),
):
"""接收单文件上传，转码后返回下载链接。"""
if not is_supported(file.filename):
raise HTTPException(status_code=400, detail=f"不支持的文件格式: {file.filename}")
if format not in SUPPORTED_FORMATS:
raise HTTPException(status_code=400, detail=f"不支持的输出格式: {format}")
# 清理文件名中的危险字符（保留扩展名中的点）
safe_name = "".join(c for c in file.filename if c not in "/\\<>:\"|?*" and ord(c) > 31)
upload_path = UPLOAD_DIR / safe_name
with open(upload_path, "wb") as f:
shutil.copyfileobj(file.file, f)
output_name = Path(safe_name).stem + "." + format
output_path = OUTPUT_DIR / output_name
try:
# 加锁串行化: vipsong.convert_file() 修改全局 stats, 多线程下需隔离
with _convert_lock:
vipsong.convert_file(
str(upload_path),
format,
delete_flag=False,
output_path=str(output_path),
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"转码失败: {e}")
finally:
# 清理上传的临时文件
if upload_path.exists():
upload_path.unlink(missing_ok=True)
return {
"original_name": file.filename,
"output_name": output_name,
"download_url": f"/download/{output_name}",
}
@app.get("/download/{filename}")
async def download(filename: str):
file_path = OUTPUT_DIR / filename
if not file_path.exists():
raise HTTPException(status_code=404, detail="文件不存在")
return FileResponse(path=str(file_path), filename=filename)
@app.get("/health")
async def health():
return {"status": "ok"}
requirements.txt
-3
View File
@@ -1,6 +1,3 @@
pycryptodome>=3.20
mutagen>=1.47
pydub>=0.25.1
fastapi>=0.115
uvicorn>=0.30
python-multipart>=0.0.9
templates/index.html
-261
View File
@@ -1,261 +0,0 @@
<!DOCTYPE html>
<html lang="zh-CN">
<head>
<meta charset="UTF-8" />
<meta name="viewport" content="width=device-width, initial-scale=1.0"/>
<title>vipsong Web — 音乐加密格式转码</title>
<style>
* { margin: 0; padding: 0; box-sizing: border-box; }
body {
font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, sans-serif;
background: #f5f6fa; color: #2d3436; min-height: 100vh; padding: 2rem;
}
.container { max-width: 720px; margin: 0 auto; }
h1 { font-size: 1.5rem; margin-bottom: 0.5rem; }
.subtitle { color: #636e72; font-size: 0.9rem; margin-bottom: 1.5rem; }
.upload-area {
border: 2px dashed #dfe6e9; border-radius: 12px; padding: 2.5rem 1.5rem;
text-align: center; cursor: pointer; transition: all .2s; background: #fff;
}
.upload-area:hover { border-color: #0984e3; background: #f8f9fa; }
.upload-area.dragover { border-color: #0984e3; background: #e3f2fd; }
.upload-area i { font-size: 2rem; color: #b2bec3; margin-bottom: 0.5rem; display: block; }
.upload-area p { color: #636e72; font-size: 0.9rem; }
.format-selector {
margin: 1.5rem 0; display: flex; align-items: center; gap: 1rem;
}
.format-selector label { font-weight: 600; color: #2d3436; }
select {
padding: 0.5rem 1rem; border: 1px solid #dfe6e9; border-radius: 8px;
font-size: 0.9rem; background: #fff; cursor: pointer;
}
.btn-convert {
display: block; width: 100%; padding: 0.8rem; margin-top: 1rem;
background: #0984e3; color: #fff; border: none; border-radius: 8px;
font-size: 1rem; font-weight: 600; cursor: pointer; transition: .2s;
}
.btn-convert:hover:not(:disabled) { background: #074c8a; }
.btn-convert:disabled { background: #b2bec3; cursor: not-allowed; }
.file-list { margin-top: 1.5rem; }
.file-item {
display: flex; align-items: center; justify-content: space-between;
padding: 0.8rem 1rem; border-radius: 8px; margin-bottom: 0.5rem;
background: #fff; border: 1px solid #dfe6e9;
}
.file-info { display: flex; align-items: center; gap: 0.8rem; }
.file-name { font-size: 0.9rem; max-width: 320px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap; }
.file-status { font-size: 0.8rem; padding: 0.2rem 0.6rem; border-radius: 12px; }
.status-pending { background: #f1f2f6; color: #636e72; }
.status-upload { background: #fff3cd; color: #856404; }
.status-convert { background: #fff3cd; color: #856404; }
.status-done { background: #d4edda; color: #155724; }
.status-error { background: #f8d7da; color: #721c24; }
.btn-download { padding: 0.3rem 0.8rem; background: #00b894; color: #fff; border: none; border-radius: 6px; font-size: 0.8rem; cursor: pointer; }
.btn-download:hover { background: #0984e3; }
.supported { margin-top: 1rem; font-size: 0.8rem; color: #636e72; }
.supported span { color: #0984e3; font-weight: 600; }
/* 弹窗模态框 - 毛玻璃居中效果 */
.modal-overlay {
position: fixed; top: 0; left: 0; right: 0; bottom: 0;
background: rgba(0, 0, 0, 0.45);
display: flex; align-items: center; justify-content: center;
z-index: 10000;
}
.modal-content {
background: rgba(255, 255, 255, 0.25);
border-radius: 16px; padding: 1.8rem 1.5rem;
text-align: center; width: 340px; max-width: 90%;
backdrop-filter: blur(12px);
-webkit-backdrop-filter: blur(12px);
border: 1px solid rgba(255, 255, 255, 0.35);
box-shadow: 0 8px 32px rgba(0, 0, 0, 0.25);
}
.modal-content h3 { color: #2d3436; font-size: 1.1rem; margin-bottom: 0.5rem; }
.modal-content p { color: #636e72; font-size: 0.85rem; margin-bottom: 1.2rem; line-height: 1.4; }
.modal-btn {
padding: 0.5rem 1.5rem; background: #0984e3; color: #fff;
border: none; border-radius: 8px; font-size: 0.9rem; cursor: pointer;
transition: background .2s;
}
.modal-btn:hover { background: #074c8a; }
@media (max-width: 480px) {
body { padding: 1rem; }
h1 { font-size: 1.2rem; }
}
</style>
</head>
<body>
<div class="container">
<h1>vipsong — 音乐加密格式转码</h1>
<p class="subtitle">支持 .ncm (网易云) / .kgm .vpr (酷狗) → mp3 / flac / wav / ogg / m4a</p>
<form id="uploadForm">
<div class="upload-area" id="dropZone">
<i>📁</i>
<p>点击选择文件，或拖拽到这里</p>
<input type="file" id="fileInput" multiple accept=".ncm,.kgm,.kgma,.vpr" style="display:none">
</div>
<div class="format-selector">
<label>输出格式:</label>
<select id="formatSelect">
<option value="mp3">MP3 (默认)</option>
<option value="flac">FLAC (无损)</option>
<option value="wav">WAV (无损)</option>
<option value="ogg">OGG</option>
<option value="m4a">M4A</option>
</select>
</div>
<button type="button" id="btnConvert" class="btn-convert" disabled>开始转码</button>
</form>
<div class="file-list" id="fileList"></div>
<p class="supported">
<span>支持格式:</span> .ncm / .kgm / .kgma / .vpr
&nbsp;|&nbsp; <span>输出格式:</span> mp3 / flac / wav / ogg / m4a
</p>
</div>
<script>
const dropZone = document.getElementById('dropZone');
const fileInput = document.getElementById('fileInput');
const formatSelect = document.getElementById('formatSelect');
const btnConvert = document.getElementById('btnConvert');
const fileList = document.getElementById('fileList');
let files = [];
const SUPPORTED_EXTENSIONS = ['.ncm', '.kgm', '.kgma', '.vpr'];
function showModal(message) {
const existing = document.querySelector('.modal-overlay');
if (existing) existing.remove();
const overlay = document.createElement('div');
overlay.className = 'modal-overlay';
overlay.innerHTML =
'<div class="modal-content">' +
'<h3>⚠️ 文件格式不支持</h3>' +
'<p>' + message + '</p>' +
'<button class="modal-btn" onclick="this.closest(\'.modal-overlay\').remove()">确定</button>' +
'</div>';
document.body.appendChild(overlay);
}
// 点击上传区域选择文件
dropZone.addEventListener('click', () => fileInput.click());
// 拖拽
['dragenter', 'dragover'].forEach(evt => {
dropZone.addEventListener(evt, e => {
e.preventDefault(); e.stopPropagation();
dropZone.classList.add('dragover');
});
});
['dragleave', 'drop'].forEach(evt => {
dropZone.addEventListener(evt, e => {
e.preventDefault(); e.stopPropagation();
dropZone.classList.remove('dragover');
});
});
// 拖拽放下
dropZone.addEventListener('drop', e => {
const dt = e.dataTransfer;
handleFiles(dt.files);
});
// 文件选择
fileInput.addEventListener('change', e => {
handleFiles(e.target.files);
});
function handleFiles(fileListObj) {
for (const f of fileListObj) {
const ext = f.name.slice(f.name.lastIndexOf('.')).toLowerCase();
if (!SUPPORTED_EXTENSIONS.includes(ext)) {
showModal('文件 "' + f.name + '" 格式不受支持<br>仅支持: .ncm / .kgm / .kgma / .vpr 格式的音乐文件');
continue;
}
if (files.some(existing => existing.name === f.name)) continue;
files.push(f);
}
renderFileList();
}
function renderFileList() {
fileList.innerHTML = '';
if (files.length === 0) {
btnConvert.disabled = true;
return;
}
btnConvert.disabled = false;
files.forEach((f, idx) => {
const item = document.createElement('div');
item.className = 'file-item';
item.innerHTML = `
<div class="file-info">
<span class="file-name">${f.name}</span>
<span class="file-status status-pending" id="status-${idx}">等待</span>
</div>
<div id="actions-${idx}"></div>
`;
fileList.appendChild(item);
});
}
function setStatus(idx, text, cls) {
const el = document.getElementById(`status-${idx}`);
if (el) { el.textContent = text; el.className = 'file-status status-' + cls; }
}
async function convertFile(file, idx) {
const formData = new FormData();
formData.append('file', file);
formData.append('format', formatSelect.value);
setStatus(idx, '上传中', 'upload');
try {
const resp = await fetch('/convert', { method: 'POST', body: formData });
const data = await resp.json();
if (!resp.ok) throw new Error(data.detail || '转换失败');
setStatus(idx, '完成', 'done');
const actions = document.getElementById(`actions-${idx}`);
const a = document.createElement('a');
a.href = data.download_url;
a.textContent = '下载';
a.className = 'btn-download';
a.download = data.output_name;
actions.appendChild(a);
} catch (error) {
setStatus(idx, '失败: ' + error.message, 'error');
}
}
// 开始转码
btnConvert.addEventListener('click', async () => {
btnConvert.disabled = true;
btnConvert.textContent = '转码中...';
for (let i = 0; i < files.length; i++) {
await convertFile(files[i], i);
}
btnConvert.disabled = false;
btnConvert.textContent = '开始转码';
});
</script>
</body>
</html>
vipsong.py
+47 -289
View File
@@ -1,5 +1,5 @@
#!/usr/bin/env python3
import os, sys, glob, struct, base64, json, io, re
import os, sys, glob, struct, base64, json
from pathlib import Path
from typing import Optional
from Crypto.Cipher import AES
@@ -9,124 +9,13 @@ 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
return struct.unpack(b"<I", data[pos:pos+4])[0]
def ncm_stream_bytes(key):
sbox = bytearray(range(256))
@@ -143,7 +32,7 @@ def ncm_stream_bytes(key):
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")
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()
@@ -266,26 +155,16 @@ class KgmDecoder:
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):
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))
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
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
@@ -331,24 +210,7 @@ def _add_tags(output_path, tags, cover=None):
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):
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():
@@ -393,18 +255,7 @@ def convert_file(file_path, output_fmt="mp3", delete_flag=False, output_path=Non
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
output_path = _resolve_output(file_path, out_fmt)
try:
_convert_audio(raw, container, out_fmt, output_path)
except Exception as e:
@@ -441,107 +292,18 @@ def expand_files(patterns):
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"]))
print("=== Conversion Stats ===")
print(" Success: " + str(stats["success"]) + " | Failed: " + str(stats["failed"]))
if stats["files"]:
print()
print(" 文件:")
print(" Files:")
for fp, ok in stats["files"]:
status = "成功" if ok else "失败"
status = "OK" if ok else "FAIL"
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:
@@ -553,7 +315,7 @@ def main():
while i < len(argv):
arg = argv[i]
if arg == "-f" and i + 1 < len(argv):
output_fmt = _normalize_format(argv[i + 1], "mp3")
output_fmt = argv[i + 1].lower()
i += 2
elif arg == "-d":
delete_flag = True
@@ -564,83 +326,79 @@ def main():
patterns.append(arg)
i += 1
if not patterns:
print("用法: vipsong.py [选项] <文件...>")
print(" 选项: -f <格式> 输出格式 (mp3/flac/wav/ogg/m4a/original)")
print(" -d 转码后删除原文件")
print(" 示例:")
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)
_process_patterns(patterns, output_fmt, delete_flag)
files = expand_files(patterns)
for f in files:
convert_file(f, 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("=* 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("再见！")
print("Bye!")
break
if not line:
continue
parts = _split_line(line)
parts = line.split()
if not parts:
continue
# 检查命令（支持带/或不带/前缀），避免指令被当作文件路径处理
cmd = parts[0].lstrip("/")
if cmd == "exit":
print("再见！")
cmd = parts[0]
if cmd == "/exit":
print("Bye!")
break
if cmd == "get":
elif cmd == "/get":
print_stats()
continue
# 解析参数和文件模式（从 i=0 开始，parts[0] 可能是文件路径）
patterns = []
output_fmt = "mp3"
delete_flag = False
i = 0
i = 1
while i < len(parts):
if parts[i] == "-f" and i + 1 < len(parts):
output_fmt = _normalize_format(parts[i + 1], "mp3")
output_fmt = parts[i + 1].lower()
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():
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] 或 <filepath>>output.xxx")
print(" Usage: <filepath> [-f format] [-d]")
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)
files = expand_files(patterns)
for f in files:
convert_file(f, output_fmt, delete_flag)
print()
print(" 当前状态: 成功 " + str(stats["success"]) + " | 失败 " + str(stats["failed"]))
print(" Current: Success " + str(stats["success"]) + " | Failed " + str(stats["failed"]))
print()
if __name__ == "__main__":