app.py

from audio_separator.separator import Separator
from flask import Flask, request, jsonify, send_file
import requests
import os
import uuid
from pathlib import Path
import threading
import time

separator = Separator()

# 任务队列和状态存储
tasks = {}

def process_audio_separation(task_id, mp3_url):
    try:
        # 下载MP3文件到本地
        response = requests.get(mp3_url)
        mp3_filename = mp3_url.split('/')[-1]
        with open("/tmp/" + mp3_filename, 'wb') as f:
            f.write(response.content)

        # 执行音频分离操作
        mdxnet_stem1, mdxnet_stem2 = mdx_vr_separator([response.content, 'UVR-MDX-NET-Inst_HQ_3.onnx'])

        # 生成分离后的文件名
        vocals_filename = f"{os.path.splitext(mp3_filename)[0]}_vocals.wav"
        no_vocals_filename = f"{os.path.splitext(mp3_filename)[0]}_no_vocals.wav"

        # 保存结果
        with open(vocals_filename, 'wb') as file:
            file.write(mdxnet_stem1)

        with open(no_vocals_filename, 'wb') as file:
            file.write(mdxnet_stem2)

        # 提供文件的永久直链
        vocals_url = f"/download/{os.path.basename(vocals_filename)}"
        no_vocals_url = f"/download/{os.path.basename(no_vocals_filename)}"

        # 更新任务状态
        tasks[task_id] = {
            "status": "completed",
            "vocals_url": vocals_url,
            "no_vocals_url": no_vocals_url
        }
    except Exception as e:
        tasks[task_id] = {
            "status": "error",
            "message": str(e)
        }

app = Flask(__name__)

@app.route('/', methods=['GET'])
def hello():
    return "Hello! This is an api server, and it is running successfully. For usage, please contact the person who hosted this api server."

@app.route('/api/audio_separation', methods=['GET'])
def audio_separation():
    mp3_url = request.args.get('url')
    if not mp3_url:
        return jsonify({"error": "URL parameter is required"}), 400

    task_id = str(uuid.uuid4())
    tasks[task_id] = {"status": "processing"}

    # 异步执行任务
    threading.Thread(target=process_audio_separation, args=(task_id, mp3_url)).start()

    return jsonify({"task_id": task_id}), 202

@app.route('/api/tasks/<task_id>', methods=['GET'])
def get_task_status(task_id):
    task = tasks.get(task_id)
    if task:
        return jsonify(task)
    else:
        return jsonify({"error": "Task not found"}), 404

@app.route('/download/<filename>', methods=['GET'])
def download(filename):
    return send_file("/tmp/" + filename, as_attachment=True)

if __name__ == '__main__':
    app.run(debug=False)