333 lines
14 KiB
Python
Executable File
333 lines
14 KiB
Python
Executable File
import json
|
||
import time
|
||
import requests
|
||
from django.http import JsonResponse
|
||
from .models import VideoGeneration
|
||
from django.views.decorators.csrf import csrf_exempt
|
||
from .base import logger,deduct_points
|
||
def generate_video_id(user_id):
|
||
"""
|
||
根据用户ID和当前时间戳生成唯一的视频ID
|
||
"""
|
||
timestamp = int(time.time())
|
||
video_id = f"{timestamp}_{user_id}"
|
||
print(f"生成的视频ID: {video_id}")
|
||
return video_id
|
||
|
||
def get_dimensions(model, width=None, height=None):
|
||
"""
|
||
根据模型返回相应的宽高分辨率:
|
||
gen3: 最大1280x768
|
||
gen2: 可传入前端指定的宽高，默认最大1920x1080
|
||
"""
|
||
if model == 'gen3':
|
||
return 1280, 768
|
||
elif model == 'gen2':
|
||
# 如果前端提供了 width 和 height，则使用传入的值，否则使用默认值
|
||
if width and height:
|
||
return width, height
|
||
return 1920, 1080
|
||
else:
|
||
raise ValueError(f"不支持的模型: {model}")
|
||
|
||
def get_headers():
|
||
"""
|
||
返回统一的请求头
|
||
"""
|
||
return {
|
||
"accept": "application/json",
|
||
"content-type": "application/json",
|
||
"Authorization": "12e76710fad2047db8c0cc6b25987e2a2" # 替换为你的真实授权密钥
|
||
}
|
||
|
||
@csrf_exempt
|
||
def text_to_video(request):
|
||
"""
|
||
文本生成视频接口
|
||
"""
|
||
if request.method == "POST":
|
||
try:
|
||
if not request.user.is_authenticated:
|
||
return JsonResponse({'code': 401, 'message': '用户未登录'})
|
||
user = request.user
|
||
data = json.loads(request.body)
|
||
|
||
# 获取参数
|
||
text_prompt = data.get('text_prompt')
|
||
style = data.get('style', '') # 选择的风格，默认是 'general'
|
||
model = data.get('model', 'gen3') # 选择模型，默认是 'gen3'
|
||
time_duration = int(data.get('time', 5)) # 视频时长
|
||
motion = int(data.get('motion', 5)) # 视频丰富度参数
|
||
width = data.get('width') # 由前端传入的宽度（仅限gen2）
|
||
height = data.get('height') # 由前端传入的高度（仅限gen2）
|
||
|
||
print(f"生成模型: {model}, 时长: {time_duration}, 文本: {text_prompt}, 宽高: {width}x{height}, 画面丰富度: {motion}")
|
||
|
||
result = deduct_points(user, "text-to-video")
|
||
if result is not True:
|
||
return result # 返回积分不足的响应
|
||
|
||
# 验证模型和时长限制
|
||
if model == 'gen3' and time_duration not in [5, 10]:
|
||
return JsonResponse({'code': 400, 'message': 'gen3 模型只支持 5s 或 10s 的时长'})
|
||
if model == 'gen2' and time_duration != 4:
|
||
return JsonResponse({'code': 400, 'message': 'gen2 模型只支持 4s 的时长'})
|
||
|
||
# 生成视频ID
|
||
video_id = generate_video_id(user.id)
|
||
|
||
# 根据模型获取宽高
|
||
width, height = get_dimensions(model, width, height)
|
||
print(f"生成的视频分辨率: {width}x{height}")
|
||
|
||
# 准备POST请求数据
|
||
payload = {
|
||
"text_prompt": f"{text_prompt}. Style: {style}",
|
||
"model": model,
|
||
"width": width,
|
||
"height": height,
|
||
"motion": motion, # 视频画面丰富度
|
||
"seed": 0,
|
||
"upscale": False,
|
||
"interpolate": False,
|
||
"callback_url": "https://www.typeframes.ai/api/callback/",
|
||
"time": time_duration # 视频时长
|
||
}
|
||
print(f'text_to_video 请求体: {payload}')
|
||
|
||
# 发送请求到API
|
||
response = requests.post('https://api.aivideoapi.com/runway/generate/text', json=payload, headers=get_headers())
|
||
print(f"API响应状态码: {response.status_code}")
|
||
print(f"API响应内容: {response.text}")
|
||
|
||
# 解析响应数据
|
||
if response.status_code == 200:
|
||
res_data = response.json()
|
||
uuid = res_data.get('uuid')
|
||
|
||
# 保存生成记录到数据库
|
||
video_generation = VideoGeneration.objects.create(
|
||
video_id=video_id,
|
||
user=user,
|
||
text=text_prompt,
|
||
status='in_progress',
|
||
pid=uuid,
|
||
media_type=f'{model}_text_to_video',
|
||
slug=f'text-to-video',
|
||
ratio=f"{width}:{height}"
|
||
)
|
||
video_generation.save()
|
||
|
||
# 返回成功响应
|
||
return JsonResponse({
|
||
'code': 200,
|
||
'message': '请求成功，视频正在生成中',
|
||
'data': {
|
||
'video_id': video_generation.video_id,
|
||
'status': video_generation.status,
|
||
'pid': uuid
|
||
}
|
||
})
|
||
else:
|
||
return JsonResponse({'code': response.status_code, 'message': 'API请求失败', 'data': {}})
|
||
|
||
except Exception as e:
|
||
print(f"发生错误: {str(e)}")
|
||
return JsonResponse({'code': 500, 'message': f'内部错误: {str(e)}', 'data': {}})
|
||
|
||
return JsonResponse({'code': 405, 'message': '请求方法错误', 'data': {}})
|
||
|
||
|
||
|
||
@csrf_exempt
|
||
def image_to_video(request):
|
||
"""
|
||
图片生成视频接口，图片宽高自适应
|
||
"""
|
||
if request.method == "POST":
|
||
try:
|
||
if not request.user.is_authenticated:
|
||
return JsonResponse({'code': 401, 'message': '用户未登录'})
|
||
user = request.user
|
||
data = json.loads(request.body)
|
||
|
||
# 获取参数
|
||
text_prompt = data.get('text_prompt')
|
||
image_url = data.get('image_url')
|
||
model = data.get('model', 'gen3') # 选择模型，默认是 'gen3'
|
||
time_duration = int(data.get('time', 5)) # 视频时长
|
||
motion = int(data.get('motion', 5)) # 视频丰富度
|
||
|
||
print(f"生成模型: {model}, 图片地址: {image_url}, 时长: {time_duration}, 画面丰富度: {motion}")
|
||
|
||
result = deduct_points(user, "img-to-video")
|
||
if result is not True:
|
||
return result # 返回积分不足的响应
|
||
|
||
# 验证模型和时长限制
|
||
if model == 'gen3' and time_duration not in [5, 10]:
|
||
return JsonResponse({'code': 400, 'message': 'gen3 模型只支持 5s 或 10s 的时长'})
|
||
if model == 'gen2' and time_duration != 4:
|
||
return JsonResponse({'code': 400, 'message': 'gen2 模型只支持 4s 的时长'})
|
||
|
||
# 生成视频ID
|
||
video_id = generate_video_id(user.id)
|
||
|
||
# 准备POST请求数据，宽高不需要传入，图片自适应
|
||
payload = {
|
||
"text_prompt": text_prompt,
|
||
"model": model,
|
||
"img_prompt": image_url,
|
||
"image_as_end_frame": False, # 图片不作为最后一帧
|
||
"motion": motion, # 视频丰富度
|
||
"seed": 0,
|
||
"upscale": False,
|
||
"interpolate": False,
|
||
"callback_url": "https://www.typeframes.ai/api/callback/",
|
||
"time": time_duration # 视频时长
|
||
}
|
||
print(f'img_to_video 请求体: {payload}')
|
||
|
||
# 请求头
|
||
headers = get_headers()
|
||
|
||
# 发送请求到API
|
||
response = requests.post('https://api.aivideoapi.com/runway/generate/imageDescription', json=payload, headers=headers)
|
||
print(f"API响应状态码: {response.status_code}")
|
||
print(f"API响应内容: {response.text}")
|
||
|
||
# 解析响应数据
|
||
if response.status_code == 200:
|
||
res_data = response.json()
|
||
uuid = res_data.get('uuid')
|
||
|
||
# 保存生成记录到数据库
|
||
video_generation = VideoGeneration.objects.create(
|
||
video_id=video_id,
|
||
user=user,
|
||
text=text_prompt,
|
||
status='in_progress',
|
||
pid=uuid,
|
||
media_type=f'{model}_img_to_video',
|
||
slug=f'img-to-video',
|
||
ratio="auto" # 图片自适应宽高
|
||
)
|
||
video_generation.save()
|
||
|
||
# 返回成功响应
|
||
return JsonResponse({
|
||
'code': 200,
|
||
'message': '请求成功，视频正在生成中',
|
||
'data': {
|
||
'video_id': video_generation.video_id,
|
||
'status': video_generation.status,
|
||
'pid': uuid
|
||
}
|
||
})
|
||
else:
|
||
return JsonResponse({'code': response.status_code, 'message': 'API请求失败', 'data': {}})
|
||
|
||
except Exception as e:
|
||
print(f"发生错误: {str(e)}")
|
||
return JsonResponse({'code': 500, 'message': f'内部错误: {str(e)}', 'data': {}})
|
||
|
||
return JsonResponse({'code': 405, 'message': '请求方法错误', 'data': {}})
|
||
|
||
|
||
|
||
@csrf_exempt
|
||
def extend_video(request):
|
||
"""
|
||
延长视频的函数，发送请求到 /runway/extend
|
||
"""
|
||
if request.method == "POST":
|
||
try:
|
||
if not request.user.is_authenticated:
|
||
return JsonResponse({'code': 401, 'message': '用户未登录'})
|
||
user = request.user
|
||
data = json.loads(request.body)
|
||
|
||
# 获取参数
|
||
pid = data.get('pid')
|
||
motion = int(data.get('motion', 5)) # 视频丰富度
|
||
|
||
# 获取当前视频生成记录
|
||
video_generation = VideoGeneration.objects.get(pid=pid)
|
||
|
||
# 通过 media_type 判断模型是 gen3 还是 gen2
|
||
if 'gen3' in video_generation.media_type:
|
||
model = 'gen3'
|
||
# 验证 gen3 的延长逻辑
|
||
if video_generation.extension_count >= 1 or video_generation.time_duration >= 10:
|
||
return JsonResponse({'code': 400, 'message': 'gen3 最大时长为 10s，无法继续延长'})
|
||
time_extension = 5 # 每次延长 5 秒
|
||
|
||
elif 'gen2' in video_generation.media_type:
|
||
model = 'gen2'
|
||
# 验证 gen2 的延长逻辑
|
||
if video_generation.extension_count >= 2 or video_generation.time_duration >= 12:
|
||
return JsonResponse({'code': 400, 'message': 'gen2 最大时长为 12s，无法继续延长'})
|
||
time_extension = 4 # 每次延长 4 秒
|
||
|
||
else:
|
||
return JsonResponse({'code': 400, 'message': '未知的模型类型'})
|
||
|
||
result = deduct_points(user, video_generation.slug)
|
||
if result is not True:
|
||
return result # 返回积分不足的响应
|
||
|
||
# 生成新的视频ID
|
||
new_video_id = generate_video_id(user.id)
|
||
|
||
# 默认请求体
|
||
payload = {
|
||
"uuid": pid, # 视频的 UUID
|
||
"motion": motion, # 视频丰富度
|
||
"seed": 0, # 默认 seed 为 0
|
||
"upscale": False, # 默认开启 upscale
|
||
"interpolate": False, # 默认开启 interpolate
|
||
"callback_url": "https://www.typeframes.ai/api/callback/"
|
||
}
|
||
print(f"extend_video 请求体: {payload}")
|
||
|
||
headers = get_headers()
|
||
|
||
# 发送 POST 请求
|
||
response = requests.post('https://api.aivideoapi.com/runway/extend', json=payload, headers=headers)
|
||
print(f"API响应状态码: {response.status_code}")
|
||
print(f"API响应内容: {response.text}")
|
||
|
||
# 解析响应内容
|
||
if response.status_code == 200:
|
||
res_data = response.json()
|
||
new_pid = res_data.get('uuid')
|
||
|
||
# 创建新的延长视频记录，复制上一条视频的参数
|
||
new_video_generation = VideoGeneration.objects.create(
|
||
video_id=new_video_id,
|
||
user=user,
|
||
text=video_generation.text,
|
||
voice_name=video_generation.voice_name,
|
||
style=video_generation.style,
|
||
rate=video_generation.rate,
|
||
media_type=video_generation.media_type,
|
||
ratio=video_generation.ratio,
|
||
status='in_progress',
|
||
pid=new_pid,
|
||
time_duration=video_generation.time_duration + time_extension, # 增加对应模型的时长
|
||
extension_count=video_generation.extension_count + 1, # 增加延长次数
|
||
slug=video_generation.slug # slug 使用中划线
|
||
)
|
||
print(f"延长视频成功: {res_data}")
|
||
|
||
return JsonResponse({'code': 200, 'message': '视频延长请求成功', 'data': {'new_pid': new_pid}})
|
||
|
||
else:
|
||
return JsonResponse({'code': response.status_code, 'message': 'API请求失败', 'data': {}})
|
||
|
||
except Exception as e:
|
||
print(f"发生错误: {str(e)}")
|
||
return JsonResponse({'code': 500, 'message': f'内部错误: {str(e)}', 'data': {}})
|
||
|
||
return JsonResponse({'code': 405, 'message': '请求方法错误', 'data': {}})
|