136 lines
5.5 KiB
Python
136 lines
5.5 KiB
Python
import base64
|
|
import json
|
|
import os
|
|
import tempfile
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
import loguru
|
|
import modal
|
|
import requests
|
|
from fastapi import APIRouter, UploadFile, Form, File, Depends, HTTPException
|
|
from fastapi.responses import FileResponse
|
|
|
|
from BowongModalFunctions.config.settings import minimax_settings
|
|
from BowongModalFunctions.middleware.authorization import verify_token
|
|
from BowongModalFunctions.models.responses.models import ModalTaskResponse
|
|
from BowongModalFunctions.models.settings.cluster import WorkerConfig
|
|
|
|
router = APIRouter(prefix='/hl_router', tags=['海螺API'])
|
|
config = WorkerConfig()
|
|
|
|
|
|
def image_generation(image_file: UploadFile, prompt: str, aspect_ratio):
|
|
data = base64.b64encode(image_file.file.read()).decode('utf-8')
|
|
|
|
payload = json.dumps({
|
|
"model": "image-01",
|
|
"prompt": prompt,
|
|
"subject_reference": [
|
|
{
|
|
"type": "character",
|
|
"image_file": f"data:image/jpeg;base64,{data}"
|
|
}
|
|
],
|
|
"aspect_ratio": aspect_ratio,
|
|
"n": 1
|
|
})
|
|
headers = {
|
|
'Authorization': f'Bearer {minimax_settings.api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
|
|
response = requests.request("POST", "https://api.minimaxi.com/v1/image_generation", headers=headers, data=payload,
|
|
timeout=150)
|
|
response.raise_for_status()
|
|
return response.json()
|
|
|
|
|
|
def music_generation(lyrics=None, refer_voice=None, refer_instrumental=None):
|
|
headers = {
|
|
'Authorization': f'Bearer {minimax_settings.api_key}',
|
|
'Content-Type': 'application/json'
|
|
}
|
|
payload = {
|
|
"model": "music-01",
|
|
"audio_setting": {
|
|
"sample_rate": 44100,
|
|
"bitrate": 256000,
|
|
"format": "mp3"
|
|
}
|
|
}
|
|
if lyrics:
|
|
payload["lyrics"] = lyrics
|
|
if refer_voice:
|
|
payload["refer_voice"] = refer_voice
|
|
if refer_instrumental:
|
|
payload["refer_instrumental"] = refer_instrumental
|
|
payload = json.dumps(payload)
|
|
loguru.logger.info("music generation payload: {}".format(payload))
|
|
response = requests.request("POST", "https://api.minimaxi.com/v1/music_generation", headers=headers, data=payload,
|
|
timeout=150)
|
|
response.raise_for_status()
|
|
resp = response.json()
|
|
if resp["base_resp"]["status_code"] == 0:
|
|
return response.json()
|
|
else:
|
|
raise Exception(f"status_code={resp['base_resp']['status_code']}")
|
|
|
|
|
|
@router.post('/async/upload/music', summary="上传人声/伴奏/音乐", dependencies=[Depends(verify_token)])
|
|
async def upload_music_hl(music: UploadFile = File(description="参考音频--10秒以上10分钟以内, 与目标对应"),
|
|
purpose: str = Form(default="voice",
|
|
description="目标分类 voice(人声)/instrumental(伴奏)/song(人声+伴奏)")):
|
|
file_path = os.path.join(config.S3_mount_dir,
|
|
"upload/{}.{}".format(str(uuid.uuid4()), music.filename.split(".")[-1]))
|
|
with open(file_path, "wb") as file:
|
|
file.write(music.file.read())
|
|
file.close()
|
|
|
|
fn = modal.Function.from_name(config.modal_app_name, "music_upload",
|
|
environment_name=config.modal_environment)
|
|
fn_call = fn.spawn(file_path, music.content_type, purpose)
|
|
return ModalTaskResponse(success=True, taskId=fn_call.object_id)
|
|
|
|
|
|
@router.post('/sync/generate/image', summary="生成图片", dependencies=[Depends(verify_token)])
|
|
async def generate_image_hl(prompt: str = Form(..., description="图片生成提示词"),
|
|
image_file: UploadFile = File(description="样貌参考图片"),
|
|
aspect_ratio=Form(default="9:16", description="图片宽高比")):
|
|
try:
|
|
return image_generation(image_file, prompt, aspect_ratio)
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=str(e))
|
|
|
|
|
|
@router.post('/sync/generate/music', summary="生成音乐",
|
|
description="refer_voice、refer_instrumental至少填写1个, 传refer_instrumental可不传lyrics",
|
|
dependencies=[Depends(verify_token)])
|
|
async def generate_music_hl(lyrics: Optional[str] = Form(default=None, description="歌词"),
|
|
refer_voice: Optional[str] = Form(default=None, description="参考音色ID, 需要先上传"),
|
|
refer_instrumental: Optional[str] = Form(default=None,
|
|
description="参考伴奏ID, 需要先上传")):
|
|
try:
|
|
if not refer_voice and not refer_instrumental:
|
|
raise Exception("refer_voice、refer_instrumental至少填写1个!")
|
|
if refer_voice and not lyrics:
|
|
raise Exception("refer_voice需传lyrics")
|
|
result = music_generation(lyrics, refer_voice, refer_instrumental)
|
|
hex_data = result["data"]["audio"]
|
|
# 将HEX编码的字符串转换为二进制数据
|
|
audio_data = bytes.fromhex(hex_data)
|
|
|
|
audio_file = f"dev/audios/{uuid.uuid4()}.mp3"
|
|
with open(os.path.join(config.S3_mount_dir, audio_file), "wb") as file:
|
|
# 写入音频数据
|
|
file.write(audio_data)
|
|
file.close()
|
|
|
|
# 返回文件响应
|
|
return {"audio_url": f"https://cdn.roasmax.cn/" + audio_file}
|
|
|
|
except ValueError:
|
|
raise HTTPException(status_code=400, detail="无效的HEX编码")
|
|
except Exception as e:
|
|
raise HTTPException(status_code=500, detail=f"生成音频时出错: {str(e)}")
|