diff --git a/app/api/v1/video.py b/app/api/v1/video.py new file mode 100644 index 0000000..b1de1a8 --- /dev/null +++ b/app/api/v1/video.py @@ -0,0 +1,176 @@ +"""视频生成API端点""" + +from fastapi import APIRouter, HTTPException, UploadFile, File, Form +from fastapi.responses import JSONResponse +from typing import Optional +import tempfile +import os + +from app.core.logger import logger +from app.services.grok.video import video_generator + +router = APIRouter(prefix="/api/v1/video", tags=["video"]) + + +@router.post("/generate") +async def generate_video_from_image( + image: UploadFile = File(..., description="图片文件"), + mode: str = Form(default="normal", description="视频生成模式"), + model_name: str = Form(default="grok-3", description="模型名称"), + file_name: Optional[str] = Form(default=None, description="自定义文件名") +): + """ + 从图片生成视频 + + Args: + image: 上传的图片文件 + mode: 视频生成模式 (normal, custom等) + model_name: 使用的模型名称 + file_name: 自定义文件名 + + Returns: + JSON响应包含视频生成结果 + """ + try: + # 验证文件类型 + if not image.content_type or not image.content_type.startswith('image/'): + raise HTTPException(status_code=400, detail="只支持图片文件") + + # 创建临时文件 + with tempfile.NamedTemporaryFile(delete=False, suffix=f".{image.filename.split('.')[-1] if '.' in image.filename else 'jpg'}") as temp_file: + content = await image.read() + temp_file.write(content) + temp_file_path = temp_file.name + + try: + # 调用视频生成服务 + result = await video_generator.generate_video_from_image( + image_path=temp_file_path, + file_name=file_name or image.filename, + mode=mode, + model_name=model_name + ) + + if result.get("success"): + logger.info(f"[Video API] 视频生成成功: {result.get('conversation_result', {}).get('data', {}).get('final_video_url', 'N/A')}") + return JSONResponse(content={ + "success": True, + "message": "视频生成成功", + "data": { + "video_url": result.get('conversation_result', {}).get('data', {}).get('final_video_url'), + "video_id": result.get('conversation_result', {}).get('data', {}).get('video_id'), + "progress": result.get('conversation_result', {}).get('data', {}).get('progress'), + "upload_info": result.get('upload_result', {}).get('data'), + "post_info": result.get('post_result', {}).get('data') + } + }) + else: + logger.error(f"[Video API] 视频生成失败: {result.get('error', '未知错误')}") + return JSONResponse( + status_code=500, + content={ + "success": False, + "message": "视频生成失败", + "error": result.get('error', '未知错误'), + "details": result + } + ) + + finally: + # 清理临时文件 + if os.path.exists(temp_file_path): + os.unlink(temp_file_path) + + except Exception as e: + logger.error(f"[Video API] 视频生成异常: {str(e)}") + raise HTTPException(status_code=500, detail=f"视频生成异常: {str(e)}") + + +@router.post("/generate-from-path") +async def generate_video_from_path( + image_path: str = Form(..., description="图片文件路径"), + mode: str = Form(default="normal", description="视频生成模式"), + model_name: str = Form(default="grok-3", description="模型名称"), + file_name: Optional[str] = Form(default=None, description="自定义文件名") +): + """ + 从本地图片路径生成视频 + + Args: + image_path: 本地图片文件路径 + mode: 视频生成模式 + model_name: 使用的模型名称 + file_name: 自定义文件名 + + Returns: + JSON响应包含视频生成结果 + """ + try: + # 验证文件是否存在 + if not os.path.exists(image_path): + raise HTTPException(status_code=400, detail="图片文件不存在") + + # 调用视频生成服务 + result = await video_generator.generate_video_from_image( + image_path=image_path, + file_name=file_name, + mode=mode, + model_name=model_name + ) + + if result.get("success"): + logger.info(f"[Video API] 视频生成成功: {result.get('conversation_result', {}).get('data', {}).get('final_video_url', 'N/A')}") + return JSONResponse(content={ + "success": True, + "message": "视频生成成功", + "data": { + "video_url": result.get('conversation_result', {}).get('data', {}).get('final_video_url'), + "video_id": result.get('conversation_result', {}).get('data', {}).get('video_id'), + "progress": result.get('conversation_result', {}).get('data', {}).get('progress'), + "upload_info": result.get('upload_result', {}).get('data'), + "post_info": result.get('post_result', {}).get('data') + } + }) + else: + logger.error(f"[Video API] 视频生成失败: {result.get('error', '未知错误')}") + return JSONResponse( + status_code=500, + content={ + "success": False, + "message": "视频生成失败", + "error": result.get('error', '未知错误'), + "details": result + } + ) + + except Exception as e: + logger.error(f"[Video API] 视频生成异常: {str(e)}") + raise HTTPException(status_code=500, detail=f"视频生成异常: {str(e)}") + + +@router.get("/status/{video_id}") +async def get_video_status(video_id: str): + """ + 获取视频生成状态 + + Args: + video_id: 视频ID + + Returns: + JSON响应包含视频状态信息 + """ + try: + # 这里可以实现视频状态查询逻辑 + # 目前返回基本信息 + return JSONResponse(content={ + "success": True, + "message": "状态查询功能待实现", + "data": { + "video_id": video_id, + "status": "unknown" + } + }) + + except Exception as e: + logger.error(f"[Video API] 状态查询异常: {str(e)}") + raise HTTPException(status_code=500, detail=f"状态查询异常: {str(e)}") diff --git a/app/services/grok/video.py b/app/services/grok/video.py new file mode 100644 index 0000000..f774caa --- /dev/null +++ b/app/services/grok/video.py @@ -0,0 +1,362 @@ +"""Grok 视频生成服务模块""" + +import json +import asyncio +import base64 +from typing import Dict, Any, Optional, List +from pathlib import Path + +from curl_cffi.requests import AsyncSession + +from app.core.config import setting +from app.core.logger import logger +from app.core.exception import GrokApiException +from app.services.grok.statsig import get_dynamic_headers +from app.services.grok.token import token_manager +from app.services.grok.upload import ImageUploadManager + +# 常量定义 +CREATE_POST_ENDPOINT = "https://grok.com/rest/media/post/create" +CONVERSATION_ENDPOINT = "https://grok.com/rest/app-chat/conversations/new" +REQUEST_TIMEOUT = 60 +IMPERSONATE_BROWSER = "chrome133a" + + +class GrokVideoGenerator: + """Grok 视频生成器""" + + @staticmethod + async def generate_video_from_image( + image_path: str, + file_name: Optional[str] = None, + mode: str = "normal", + model_name: str = "grok-3" + ) -> Dict[str, Any]: + """ + 完整流程：上传图片并生成视频 + + Args: + image_path: 图片文件路径 + file_name: 自定义文件名（可选） + mode: 视频生成模式 + model_name: 模型名称 + + Returns: + dict: 完整流程结果 + """ + try: + # 第一步：上传图片 + upload_result = await GrokVideoGenerator._upload_image_file(image_path, file_name) + + if not upload_result.get("success"): + return upload_result + + # 获取文件信息 + file_metadata_id = upload_result["data"].get("fileMetadataId") + file_uri = upload_result["data"].get("fileUri") + + if not file_metadata_id or not file_uri: + return {"error": "上传成功但未获取到必要信息"} + + # 第二步：创建媒体帖子 + post_result = await GrokVideoGenerator._create_media_post(file_uri, file_metadata_id) + + if not post_result.get("success"): + return { + "upload_result": upload_result, + "post_result": post_result, + "error": "创建媒体帖子失败" + } + + # 构建图片URL + image_url = f"https://assets.grok.com/{file_uri}" + + # 第三步：创建对话并生成视频 + conversation_result = await GrokVideoGenerator._create_video_conversation( + file_metadata_id, + image_url, + mode, + model_name + ) + + return { + "upload_result": upload_result, + "post_result": post_result, + "conversation_result": conversation_result, + "success": conversation_result.get("success", False) + } + + except Exception as e: + logger.error(f"[Video] 视频生成流程异常: {str(e)}") + return {"error": f"视频生成异常: {str(e)}"} + + @staticmethod + async def _upload_image_file(image_path: str, file_name: Optional[str] = None) -> Dict[str, Any]: + """ + 上传图片文件到Grok Imagine + + Args: + image_path: 图片文件路径 + file_name: 自定义文件名（可选） + + Returns: + dict: 上传结果 + """ + try: + # 读取图片文件 + image_path = Path(image_path) + if not image_path.exists(): + return {"error": "文件不存在"} + + # 获取文件信息 + file_name = file_name or image_path.name + file_mime_type = GrokVideoGenerator._get_mime_type(image_path.suffix) + + # 读取并编码图片 + with open(image_path, 'rb') as f: + image_data = f.read() + content = base64.b64encode(image_data).decode('utf-8') + + # 构建请求数据 + data = { + "content": content, + "fileMimeType": file_mime_type, + "fileName": file_name, + "fileSource": "IMAGINE_SELF_UPLOAD_FILE_SOURCE" + } + + # 获取认证令牌 + auth_token = token_manager.get_token("grok-3") + cf_clearance = setting.grok_config.get("cf_clearance", "") + cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token + + # 获取代理配置 + proxy_url = setting.grok_config.get("proxy_url", "") + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + if proxy_url: + logger.debug(f"[Video] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") + + # 构建上传请求的headers + upload_headers = get_dynamic_headers("/rest/app-chat/upload-file") + upload_headers["referer"] = "https://grok.com/imagine" + upload_headers["Cookie"] = cookie + + # 发送请求 + async with AsyncSession() as session: + response = await session.post( + "https://grok.com/rest/app-chat/upload-file", + headers=upload_headers, + json=data, + impersonate=IMPERSONATE_BROWSER, + timeout=REQUEST_TIMEOUT, + proxies=proxies + ) + + if response.status_code == 200: + return { + "success": True, + "data": response.json(), + "status_code": response.status_code + } + else: + return { + "success": False, + "error": f"上传失败，状态码: {response.status_code}", + "response": response.text + } + + except Exception as e: + logger.error(f"[Video] 上传图片异常: {str(e)}") + return {"error": f"上传异常: {str(e)}"} + + @staticmethod + async def _create_media_post(file_uri: str, file_metadata_id: str, media_type: str = "MEDIA_POST_TYPE_IMAGE") -> Dict[str, Any]: + """ + 创建媒体帖子 + + Args: + file_uri: 上传文件后返回的fileUri + file_metadata_id: 文件元数据ID，用于构建referer + media_type: 媒体类型，默认为图片 + + Returns: + dict: 创建结果 + """ + try: + # 构建请求数据 + data = { + "mediaType": media_type, + "mediaUrl": f"https://assets.grok.com/{file_uri}" + } + + # 获取认证令牌 + auth_token = token_manager.get_token("grok-3") + cf_clearance = setting.grok_config.get("cf_clearance", "") + cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token + + # 获取代理配置 + proxy_url = setting.grok_config.get("proxy_url", "") + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + # 构建媒体帖子请求的headers + post_headers = get_dynamic_headers("/rest/media/post/create") + post_headers["referer"] = f"https://grok.com/imagine/{file_metadata_id}" + post_headers["Cookie"] = cookie + + # 发送请求 + async with AsyncSession() as session: + response = await session.post( + CREATE_POST_ENDPOINT, + headers=post_headers, + json=data, + impersonate=IMPERSONATE_BROWSER, + timeout=REQUEST_TIMEOUT, + proxies=proxies + ) + + if response.status_code == 200: + return { + "success": True, + "data": response.json(), + "status_code": response.status_code + } + else: + return { + "success": False, + "error": f"创建帖子失败，状态码: {response.status_code}", + "response": response.text + } + + except Exception as e: + logger.error(f"[Video] 创建帖子异常: {str(e)}") + return {"error": f"创建帖子异常: {str(e)}"} + + @staticmethod + async def _create_video_conversation( + file_metadata_id: str, + image_url: str, + mode: str = "normal", + model_name: str = "grok-3" + ) -> Dict[str, Any]: + """ + 创建新对话并生成视频 + + Args: + file_metadata_id: 文件元数据ID + image_url: 图片URL + mode: 生成模式，默认为normal + model_name: 模型名称，默认为grok-3 + + Returns: + dict: 对话创建和视频生成结果 + """ + try: + # 构建请求数据 + data = { + "fileAttachments": [file_metadata_id], + "message": f"{image_url} --mode={mode}", + "modelName": model_name, + "temporary": True, + "toolOverrides": {"videoGen": True}, + "videoGen": True + } + + # 获取认证令牌 + auth_token = token_manager.get_token(model_name) + cf_clearance = setting.grok_config.get("cf_clearance", "") + cookie = f"{auth_token};{cf_clearance}" if cf_clearance else auth_token + + # 获取代理配置 + proxy_url = setting.grok_config.get("proxy_url", "") + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + # 构建对话请求的headers + conversation_headers = get_dynamic_headers("/rest/app-chat/conversations/new") + conversation_headers["referer"] = f"https://grok.com/imagine/{file_metadata_id}" + conversation_headers["Cookie"] = cookie + + # 发送请求 + async with AsyncSession() as session: + response = await session.post( + CONVERSATION_ENDPOINT, + headers=conversation_headers, + json=data, + impersonate=IMPERSONATE_BROWSER, + timeout=REQUEST_TIMEOUT, + proxies=proxies + ) + + if response.status_code == 200: + # 处理流式响应 + response_text = response.text + responses = [] + + # 分割多个JSON响应 + json_objects = response_text.strip().split('\n') + for json_str in json_objects: + if json_str.strip(): + try: + json_obj = json.loads(json_str) + responses.append(json_obj) + except json.JSONDecodeError: + continue + + # 查找最终视频URL + video_url = None + video_id = None + progress = 0 + + for resp in responses: + if 'result' in resp and 'response' in resp['result']: + response_data = resp['result']['response'] + + # 检查是否有视频生成响应 + if 'streamingVideoGenerationResponse' in response_data: + video_data = response_data['streamingVideoGenerationResponse'] + video_id = video_data.get('videoId') + progress = video_data.get('progress', 0) + + # 如果进度100%，获取最终视频URL + if progress == 100 and 'videoUrl' in video_data: + video_url = video_data['videoUrl'] + break + + return { + "success": True, + "data": { + "responses": responses, + "video_id": video_id, + "video_url": video_url, + "progress": progress, + "final_video_url": f"https://assets.grok.com/{video_url}" if video_url else None + }, + "status_code": response.status_code + } + else: + return { + "success": False, + "error": f"创建对话失败，状态码: {response.status_code}", + "response": response.text + } + + except Exception as e: + logger.error(f"[Video] 创建对话异常: {str(e)}") + return {"error": f"创建对话异常: {str(e)}"} + + @staticmethod + def _get_mime_type(file_extension: str) -> str: + """根据文件扩展名获取MIME类型""" + mime_types = { + '.jpg': 'image/jpeg', + '.jpeg': 'image/jpeg', + '.png': 'image/png', + '.gif': 'image/gif', + '.webp': 'image/webp', + '.bmp': 'image/bmp' + } + return mime_types.get(file_extension.lower(), 'image/jpeg') + + +# 全局视频生成器实例 +video_generator = GrokVideoGenerator() diff --git a/main.py b/main.py index 9bc39a1..6014e85 100644 --- a/main.py +++ b/main.py @@ -11,6 +11,7 @@ from app.services.grok.token import token_manager from app.api.v1.chat import router as chat_router from app.api.v1.models import router as models_router from app.api.v1.images import router as images_router +from app.api.v1.video import router as video_router from app.api.admin.manage import router as admin_router # 导入MCP服务器（认证配置在server.py中完成） @@ -84,6 +85,7 @@ register_exception_handlers(app) app.include_router(chat_router, prefix="/v1") app.include_router(models_router, prefix="/v1") app.include_router(images_router) +app.include_router(video_router) app.include_router(admin_router) # 挂载静态文件 diff --git a/test_video.py b/test_video.py new file mode 100644 index 0000000..ac4f502 --- /dev/null +++ b/test_video.py @@ -0,0 +1,50 @@ +#!/usr/bin/env python3 +"""测试视频生成功能""" + +import asyncio +import sys +from pathlib import Path + +# 添加项目根目录到Python路径 +project_root = Path(__file__).parent +sys.path.insert(0, str(project_root)) + +from app.services.grok.video import video_generator + + +async def test_video_generation(): + """测试视频生成功能""" + print("开始测试视频生成功能...") + + # 测试图片路径（请替换为实际存在的图片路径） + test_image_path = "C:\\Users\\21449\\Pictures\\Saved Pictures\\R.gif" # 请确保这个路径存在 + + if not Path(test_image_path).exists(): + print(f"测试图片不存在: {test_image_path}") + return + + try: + # 调用视频生成服务 + result = await video_generator.generate_video_from_image( + image_path=test_image_path, + file_name="test_video", + mode="normal", + model_name="grok-3" + ) + + if result.get("success"): + print("✅ 视频生成成功!") + print(f"视频URL: {result.get('conversation_result', {}).get('data', {}).get('final_video_url', 'N/A')}") + print(f"视频ID: {result.get('conversation_result', {}).get('data', {}).get('video_id', 'N/A')}") + print(f"进度: {result.get('conversation_result', {}).get('data', {}).get('progress', 'N/A')}%") + else: + print("❌ 视频生成失败!") + print(f"错误: {result.get('error', '未知错误')}") + print(f"详细信息: {result}") + + except Exception as e: + print(f"❌ 测试异常: {str(e)}") + + +if __name__ == "__main__": + asyncio.run(test_video_generation())