Upload 7 files

Dockerfile

@@ -1,5 +1,12 @@
-FROM wlhtea/suno2openai:latest
+FROM python:3.10-slim-buster
 
-ENV BASE_URL=https://studio-api.suno.ai
-ENV SESSION_ID=cookie
-ENV SQL_DK=3306
+WORKDIR /app
+
+COPY requirements.txt ./
+RUN --mount=type=cache,target=/root/.cache/pip \
+    pip install -r requirements.txt --no-cache-dir
+
+COPY . .
+
+EXPOSE 8000
+CMD [ "uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000" ]

cookie.py

@@ -0,0 +1,94 @@
+# -*- coding:utf-8 -*-
+
+import os
+import time
+from http.cookies import SimpleCookie
+from threading import Thread
+
+import requests
+
+from utils import COMMON_HEADERS
+
+
+class SunoCookie:
+    def __init__(self):
+        self.cookie = SimpleCookie()
+        self.session_id = None
+        self.token = None
+        self.is_disabled = False
+
+    def load_cookie(self, cookie_str):
+        self.cookie.load(cookie_str)
+
+    def get_cookie(self):
+        return ";".join([f"{i}={self.cookie.get(i).value}" for i in self.cookie.keys()])
+
+    def set_session_id(self, session_id):
+        self.session_id = session_id
+
+    def get_session_id(self):
+        return self.session_id
+
+    def get_token(self):
+        return self.token
+
+    def set_token(self, token: str):
+        self.token = token
+
+    def disable(self):
+        self.is_disabled = True
+
+    def enable(self):
+        self.is_disabled = False
+
+def load_env_cookies():
+    suno_auths = {}
+    for key, value in os.environ.items():
+        if key.startswith("SESSION_ID"):
+            try:
+                i = int(key[10:])  # 提取 SESSION_ID 后的数字
+                cookie = os.getenv(f"COOKIE{i}")
+                if cookie:
+                    suno_auth = SunoCookie()
+                    suno_auth.set_session_id(value)
+                    suno_auth.load_cookie(cookie)
+                    suno_auths[i] = suno_auth
+            except ValueError:
+                continue
+    return suno_auths
+
+def update_token(suno_cookie: SunoCookie):
+    headers = {"cookie": suno_cookie.get_cookie()}
+    headers.update(COMMON_HEADERS)
+    session_id = suno_cookie.get_session_id()
+
+    resp = requests.post(
+        url=f"https://clerk.suno.com/v1/client/sessions/{session_id}/tokens?_clerk_js_version=5.22.3",
+        headers=headers,
+    )
+
+    resp_headers = dict(resp.headers)
+    set_cookie = resp_headers.get("Set-Cookie")
+    suno_cookie.load_cookie(set_cookie)
+    token = resp.json().get("jwt")
+    suno_cookie.set_token(token)
+    # print(set_cookie)
+    # print(f"*** token -> {token} ***")
+
+
+def keep_alive(suno_cookie: SunoCookie):
+    while True:
+        try:
+            update_token(suno_cookie)
+        except Exception as e:
+            print(e)
+        finally:
+            time.sleep(5)
+
+def start_keep_alive(suno_auths):
+    for suno_auth in suno_auths.values():
+        t = Thread(target=keep_alive, args=(suno_auth,))
+        t.start()
+
+suno_auths = load_env_cookies()
+start_keep_alive(suno_auths)

deps.py

@@ -0,0 +1,47 @@
+# -*- coding:utf-8 -*-
+
+from cookie import suno_auths
+from utils import get_credits
+import logging
+import asyncio
+
+async def get_token():
+    disabled_accounts = set()
+    
+    while True:
+        for i, suno_auth in sorted(suno_auths.items()):
+            if suno_auth.is_disabled or i in disabled_accounts:
+                continue
+
+            token = suno_auth.get_token()
+
+            try:
+                credits_info = await get_credits(token)
+                credits = credits_info.get('credits_left', 0)
+                logging.info(f"当前账号 {suno_auth.get_session_id()} 积分: {credits}")
+                
+                if credits > 5:
+                    return token
+            except Exception as e:
+                logging.error(f"账号 {suno_auth.get_session_id()} 获取积分失败: {e}")
+                suno_auth.disable()
+                disabled_accounts.add(i)
+        
+        if len(disabled_accounts) == len(suno_auths):
+            logging.warning("所有账号都已被禁用,等待 1 小时后重试...")
+            await asyncio.sleep(3600)  # 等待1小时 (3600秒)
+            # 重置所有账号的禁用状态
+            for suno_auth in suno_auths.values():
+                suno_auth.enable()
+            disabled_accounts.clear()
+        else:
+            logging.warning("所有可用账号积分已用尽,等待 1 小时后重试...")
+            await asyncio.sleep(3600)  # 等待1小时 (3600秒)
+
+# 在程序启动时重置所有账号的禁用状态
+def reset_account_status():
+    for suno_auth in suno_auths.values():
+        suno_auth.enable()
+
+# 确保在主程序中调用此函数
+reset_account_status()

main.py

@@ -0,0 +1,125 @@
+# -*- coding:utf-8 -*-
+
+import json
+import logging
+from fastapi import Depends, FastAPI, HTTPException, Request, status
+from fastapi.middleware.cors import CORSMiddleware
+
+import schemas
+from deps import get_token, reset_account_status
+from utils import generate_lyrics, generate_music, get_feed, get_lyrics, get_credits
+from cookie import suno_auths
+
+app = FastAPI()
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+# 配置日志
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
+
+@app.on_event("startup")
+async def startup_event():
+    # 在应用启动时重置所有账号状态
+    reset_account_status()
+    logging.info("所有账号状态已重置")
+
+@app.get("/")
+async def get_root():
+    return schemas.Response()
+
+@app.post("/generate")
+async def generate(
+    data: schemas.CustomModeGenerateParam, token: str = Depends(get_token)
+):
+    try:
+        resp = await generate_music(data.dict(), token)
+        return resp
+    except Exception as e:
+        logging.error(f"生成音乐失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.post("/generate/description-mode")
+async def generate_with_song_description(
+    data: schemas.DescriptionModeGenerateParam, token: str = Depends(get_token)
+):
+    try:
+        resp = await generate_music(data.dict(), token)
+        return resp
+    except Exception as e:
+        logging.error(f"根据描述生成音乐失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.get("/feed/{aid}")
+async def fetch_feed(aid: str, token: str = Depends(get_token)):
+    try:
+        resp = await get_feed(aid, token)
+        return resp
+    except Exception as e:
+        logging.error(f"获取feed失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.post("/generate/lyrics/")
+async def generate_lyrics_post(request: Request, token: str = Depends(get_token)):
+    req = await request.json()
+    prompt = req.get("prompt")
+    if prompt is None:
+        raise HTTPException(
+            detail="prompt is required", status_code=status.HTTP_400_BAD_REQUEST
+        )
+
+    try:
+        resp = await generate_lyrics(prompt, token)
+        return resp
+    except Exception as e:
+        logging.error(f"生成歌词失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.get("/lyrics/{lid}")
+async def fetch_lyrics(lid: str, token: str = Depends(get_token)):
+    try:
+        resp = await get_lyrics(lid, token)
+        return resp
+    except Exception as e:
+        logging.error(f"获取歌词失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.get("/get_credits")
+async def fetch_credits(token: str = Depends(get_token)):
+    try:
+        resp = await get_credits(token)
+        return resp
+    except Exception as e:
+        logging.error(f"获取积分失败: {str(e)}")
+        raise HTTPException(
+            detail=str(e), status_code=status.HTTP_500_INTERNAL_SERVER_ERROR
+        )
+
+@app.get("/account_status")
+async def get_account_status():
+    status = {}
+    for i, suno_auth in suno_auths.items():
+        status[i] = {
+            "session_id": suno_auth.get_session_id(),
+            "is_disabled": suno_auth.is_disabled
+        }
+    return status
+
+if __name__ == "__main__":
+    import uvicorn
+    uvicorn.run(app, host="0.0.0.0", port=8000)

requirements.txt

@@ -0,0 +1,6 @@
+aiohttp
+python-dotenv
+fastapi
+uvicorn
+pydantic
+requests

schemas.py

@@ -0,0 +1,53 @@
+# -*- coding:utf-8 -*-
+
+from datetime import datetime
+from typing import Any, List, Optional, Union
+
+from pydantic import BaseModel, Field
+
+
+class Response(BaseModel):
+    code: Optional[int] = 0
+    msg: Optional[str] = "success"
+    data: Optional[Any] = None
+
+
+class CustomModeGenerateParam(BaseModel):
+    """Generate with Custom Mode"""
+
+    prompt: str = Field(..., description="lyrics")
+    mv: str = Field(
+        ...,
+        description="model version, default: chirp-v3-0",
+        examples=["chirp-v3-0"],
+    )
+    title: str = Field(..., description="song title")
+    tags: str = Field(..., description="style of music")
+    continue_at: Optional[int] = Field(
+        default=None,
+        description="continue a new clip from a previous song, format number",
+        examples=[120],
+    )
+    continue_clip_id: Optional[str] = None
+
+    negative_tags: str = ""
+
+
+class DescriptionModeGenerateParam(BaseModel):
+    """Generate with Song Description"""
+
+    gpt_description_prompt: str
+    make_instrumental: bool = False
+    mv: str = Field(
+        default='chirp-v3-0',
+        description="model version, default: chirp-v3-0",
+        examples=["chirp-v3-0"],
+    )
+    
+    prompt: str = Field(
+        default="",
+        description="Placeholder, keep it as an empty string, do not modify it",
+    )
+
+    generation_type: str = "TEXT"
+    user_uploaded_images_b64: str = ""

utils.py

@@ -0,0 +1,75 @@
+import json
+import os
+import time
+
+import aiohttp
+from dotenv import load_dotenv
+
+load_dotenv()
+
+BASE_URL = "https://studio-api.suno.ai"
+
+COMMON_HEADERS = {
+    "Content-Type": "text/plain;charset=UTF-8",
+    "User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/123.0.0.0 Safari/537.36",
+    "Referer": "https://suno.com",
+    "Origin": "https://suno.com",
+}
+
+
+async def fetch(url, headers=None, data=None, method="POST"):
+    if headers is None:
+        headers = {}
+    headers.update(COMMON_HEADERS)
+    if data is not None:
+        data = json.dumps(data)
+
+    print(data, method, headers, url)
+
+    async with aiohttp.ClientSession() as session:
+        try:
+            async with session.request(
+                method=method, url=url, data=data, headers=headers
+            ) as resp:
+                return await resp.json()
+        except Exception as e:
+            return f"An error occurred: {e}"
+
+
+async def get_feed(ids, token):
+    headers = {"Authorization": f"Bearer {token}"}
+    api_url = f"{BASE_URL}/api/feed/?ids={ids}"
+    response = await fetch(api_url, headers, method="GET")
+    return response
+
+
+async def generate_music(data, token):
+    headers = {"Authorization": f"Bearer {token}"}
+    api_url = f"{BASE_URL}/api/generate/v2/"
+    response = await fetch(api_url, headers, data)
+    return response
+
+
+async def generate_lyrics(prompt, token):
+    headers = {"Authorization": f"Bearer {token}"}
+    api_url = f"{BASE_URL}/api/generate/lyrics/"
+    data = {"prompt": prompt}
+    return await fetch(api_url, headers, data)
+
+
+async def get_lyrics(lid, token):
+    headers = {"Authorization": f"Bearer {token}"}
+    api_url = f"{BASE_URL}/api/generate/lyrics/{lid}"
+    return await fetch(api_url, headers, method="GET")
+
+
+async def get_credits(token):
+    headers = {"Authorization": f"Bearer {token}"}
+    api_url = f"{BASE_URL}/api/billing/info/"
+    respose = await fetch(api_url, headers, method="GET")
+    return {
+        "credits_left": respose['total_credits_left'],
+        #"period": respose['period'],
+        "monthly_limit": respose['monthly_limit'],
+        "monthly_usage": respose['monthly_usage']
+    }