Create app.py

app.py

@@ -0,0 +1,207 @@
+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()