import gradio as gr import requests import os import time import zipfile import shutil from concurrent.futures import ThreadPoolExecutor, as_completed # ========================================== # ⚙️ 核心配置 # ========================================== SUNO_API_KEY = "2335a91def6e8c7b851c286ffd80f0df" UPLOAD_URL = "https://kieai.redpandaai.co/api/file-stream-upload" COVER_URL = "https://api.kie.ai/api/v1/generate/upload-cover" QUERY_URL = "https://api.kie.ai/api/v1/generate/record-info" DUMMY_CALLBACK = "https://example.com/callback" # 最大并发数 MAX_WORKERS = 5 def log(msg): return f"[{time.strftime('%H:%M:%S')}] {msg}" # ========================================== # 🛠️ 基础功能 # ========================================== def upload_file(file_obj): try: file_path = file_obj.name file_name = os.path.basename(file_path) headers = {"Authorization": f"Bearer {SUNO_API_KEY}"} files = {'file': (file_name, open(file_path, 'rb'))} data = {'uploadPath': 'suno-inst-copy'} res = requests.post(UPLOAD_URL, headers=headers, files=files, data=data) res_json = res.json() url = res_json.get('data', {}).get('downloadUrl') or res_json.get('data', {}).get('fileUrl') if url: return {"status": "success", "name": file_name, "url": url} return {"status": "failed", "name": file_name, "msg": "No URL returned"} except Exception as e: return {"status": "error", "name": os.path.basename(file_obj.name), "msg": str(e)} # ========================================== # 🎹 核心：复刻任务提交 (接收相似度参数) # ========================================== def submit_copy_task(upload_data, similarity_val): if upload_data['status'] != 'success': return None # 万能高保真 Prompt universal_prompt = "Instrumental, High Fidelity, Cinematic, Clear Production, Masterpiece, Original Vibe" payload = { "uploadUrl": upload_data['url'], "callBackUrl": DUMMY_CALLBACK, "model": "V5", "instrumental": True, # 必须：纯音乐 "prompt": universal_prompt, # 通用高保真提示词 "lyrics": "", # 无歌词 "customMode": True, "title": f"{upload_data['name']} (Sim_{int(similarity_val*100)}%)", # ⚡️ 核心修改点：动态相似度 ⚡️ "styleWeight": 0.95, # 风格权重保持高位，确保流派不偏 "audioWeight": similarity_val # 音频权重由滑块控制 } try: res = requests.post(COVER_URL, headers={"Authorization": f"Bearer {SUNO_API_KEY}", "Content-Type": "application/json"}, json=payload) data = res.json() if data.get('code') == 200: return {"id": data['data']['taskId'], "name": upload_data['name']} return {"failed": True, "name": upload_data['name'], "msg": data.get('msg')} except Exception as e: return {"failed": True, "name": upload_data['name'], "msg": str(e)} # ========================================== # 🔄 流程控制 # ========================================== def run_one_click_copy(files, similarity): logs = [log(f"🚀 启动复刻流程 | 目标相似度: {similarity}")] if not files: return "\n".join(logs), None # 1. 上传 logs.append(log(f"⬆️ 正在上传 {len(files)} 个音频...")) with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: uploads = list(pool.map(upload_file, files)) # 2. 提交 logs.append(log(f"⚙️ 提交任务...")) tasks = [] with ThreadPoolExecutor(max_workers=MAX_WORKERS) as pool: # 将 similarity 参数传递给 submit 函数 futures = [pool.submit(submit_copy_task, up, similarity) for up in uploads] for f in as_completed(futures): res = f.result() if res and not res.get('failed'): tasks.append(res) logs.append(log(f" ✅ 任务建立: {res['name']}")) else: logs.append(log(f" ❌ 建立失败: {res.get('name')} - {res.get('msg')}")) if not tasks: logs.append(log("❌ 无有效任务，流程结束")) return "\n".join(logs), None # 3. 监控 logs.append(log(f"⏳ 等待处理 (共 {len(tasks)} 个)...")) yield "\n".join(logs), None completed_urls = [] headers = {"Authorization": f"Bearer {SUNO_API_KEY}"} start_time = time.time() while len(completed_urls) < len(tasks): if time.time() - start_time > 1200: break pending = False for task in tasks: if any(c['id'] == task['id'] for c in completed_urls): continue if task.get('failed_final'): continue try: r = requests.get(QUERY_URL, headers=headers, params={"taskId": task['id']}).json() status = r.get('data', {}).get('status') if status == 'SUCCESS': data = r['data']['response'].get('sunoData', []) if data: info = {"id": task['id'], "url": data[0]['audioUrl'], "name": task['name']} completed_urls.append(info) logs.append(log(f" 🎉 完成: {task['name']}")) yield "\n".join(logs), None elif status in ['GENERATE_AUDIO_FAILED', 'SENSITIVE_WORD_ERROR']: task['failed_final'] = True logs.append(log(f" ❌ 平台生成失败: {task['name']}")) yield "\n".join(logs), None else: pending = True except: pending = True if pending: time.sleep(5) else: break # 4. 打包 logs.append(log("📦 打包文件中...")) temp_dir = f"Inst_Copy_{int(time.time())}" os.makedirs(temp_dir, exist_ok=True) cnt = 0 for item in completed_urls: try: with requests.get(item['url'], stream=True) as r: fname = f"{os.path.splitext(item['name'])[0]}_Remix.mp3" with open(os.path.join(temp_dir, fname), 'wb') as f: f.write(r.content) cnt += 1 except: pass if cnt > 0: zip_path = f"{temp_dir}.zip" with zipfile.ZipFile(zip_path, 'w') as z: for root, _, fs in os.walk(temp_dir): for f in fs: z.write(os.path.join(root, f), f) shutil.rmtree(temp_dir) logs.append(log(f"✅ 完成！成功下载 {cnt} 个文件")) yield "\n".join(logs), zip_path else: shutil.rmtree(temp_dir) yield "\n".join(logs), None # ========================================== # 🖥️ 界面构建 # ========================================== with gr.Blocks(title="Suno 纯音乐复刻专家") as demo: gr.Markdown("## 🎹 Suno 纯音乐复刻专家 (Instrumental Copy)") gr.Markdown("上传音频 -> 调节相似度 -> 一键生成。") with gr.Row(): with gr.Column(scale=1): file_in = gr.File(label="拖入音频文件 (支持批量)", file_count="multiple", height=200) # 🔥 新增：相似度调节滑块 🔥 similarity_in = gr.Slider( minimum=0.1, maximum=0.99, value=0.95, step=0.01, label="🎚️ 复刻相似度 (Audio Weight)", info="⚠️ 调节说明：\n" "【0.90 - 0.99】：高保真复刻，几乎和原曲一样，仅音质/混音微调。\n" "【0.60 - 0.80】：保留旋律框架，但在乐器和氛围上有明显变化（二创）。\n" "【0.30 - 0.50】：仅保留一点原曲的影子，大幅度重写（魔改）。" ) btn = gr.Button("🚀 开始制作", variant="primary", size="lg") with gr.Column(scale=1): log_box = gr.Textbox(label="运行日志", lines=12) out_file = gr.File(label="下载结果 (ZIP)") # 将 similarity_in 加入输入列表 btn.click(run_one_click_copy, inputs=[file_in, similarity_in], outputs=[log_box, out_file]) if __name__ == "__main__": demo.launch()