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 import http.client import random import urllib import hashlib from langdetect import detect from langdetect.lang_detect_exception import LangDetectException 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 appid = '20240601002067404' # 填写你的appid secretKey = '6pRZ9HCSqGuMqzLO55hB' # 填写你的密钥 def translate(text, from_lang='auto', to_lang='en'): myurl = '/api/trans/vip/translate' salt = random.randint(32768, 65536) sign = appid + text + str(salt) + secretKey sign = hashlib.md5(sign.encode()).hexdigest() myurl = (myurl + '?appid=' + appid + '&q=' + urllib.parse.quote(text) + '&from=' + from_lang + '&to=' + to_lang + '&salt=' + str(salt) + '&sign=' + sign) try: httpClient = http.client.HTTPConnection('api.fanyi.baidu.com') httpClient.request('GET', myurl) # response是HTTPResponse对象 response = httpClient.getresponse() result_all = response.read().decode("utf-8") result = json.loads(result_all) print(f'请求结果{result}') return result['trans_result'][0]['dst'] except Exception as e: return str(e) finally: if httpClient: httpClient.close() 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') print(f'文生视频原内容---：{text_prompt}') # 判断是否需要翻译 try: language = detect(text_prompt) print(f"检测到的语言: {language}") if language == 'en': description = text_prompt # 英文无需翻译 else: description = translate(text_prompt, 'auto', 'en') # 翻译成英文 print(f'文生视频翻译内容---：{description}') except LangDetectException: return JsonResponse({'code': 400, 'message': '无法检测语言，请输入有效文本'}) # 获取参数 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"{description}. Style: {style}", "model": model, "width": width, "height": height, "motion": motion, # 视频画面丰富度 "seed": 0, "upscale": True, "interpolate": True, "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') print(f' 图片生视频原内容---：{text_prompt}') # 判断是否需要翻译 try: language = detect(text_prompt) print(f"检测到的语言: {language}") if language == 'en': description = text_prompt # 英文无需翻译 else: description = translate(text_prompt, 'auto', 'en') # 翻译成英文 print(f'图片生视翻译内容---：{description}') except LangDetectException: return JsonResponse({'code': 400, 'message': '无法检测语言，请输入有效文本'}) 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": description, "model": model, "img_prompt": image_url, "image_as_end_frame": False, # 图片不作为最后一帧 "motion": motion, # 视频丰富度 "seed": 0, "upscale": True, "interpolate": True, "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) print(data) # 获取参数 pid = data.get('task_id') motion = 6 # 视频丰富度 # 获取当前视频生成记录 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": True, # 默认开启 upscale "interpolate": True, # 默认开启 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': {}})