140 lines
4.5 KiB
Python
140 lines
4.5 KiB
Python
#!/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"}
|