ypshidifu
8dd6aebd8b
- app.py: FastAPI 后端，POST /convert 接收文件上传并调用 vipsong.convert_file 转码 GET /download/<filename> 提供下载，GET /health 健康检查 文件名安全过滤保留扩展名，支持多格式输出(mp3/flac/wav/ogg/m4a) - templates/index.html: 纯HTML/CSS/JS单页应用，拖拽上传、格式选择、状态显示、下载链接 - .gitignore: 添加 static/uploads/、static/outputs/、音频输出文件过滤 - requirements.txt: 添加 fastapi、uvicorn、python-multipart 验证 MCP 可用性: - Baota MCP SystemInfo/SiteList/ServiceStatus 全部正常返回 - nginx+mysql+php-fpm-74/82/85 运行中，5个网站正常 - Web API 端对端测试: NCM→MP3/FLAC/OGG 均输出有效音频，错误处理返回 HTTP 400
89 lines
2.8 KiB
Python
89 lines
2.8 KiB
Python
#!/usr/bin/env python3
|
|
"""vipsong Web — 音乐加密格式转码工具 (Web 版)。"""
|
|
import os
|
|
import shutil
|
|
from pathlib import Path
|
|
from fastapi import FastAPI, File, UploadFile, Form, HTTPException, BackgroundTasks
|
|
from fastapi.responses import HTMLResponse, FileResponse, JSONResponse
|
|
from fastapi.staticfiles import StaticFiles
|
|
import vipsong
|
|
|
|
app = FastAPI(title="vipsong Web", description="音乐加密格式转码工具")
|
|
|
|
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(
|
|
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"}
|