okooooooo/grok2api
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

video

Browse Source
This commit is contained in:
yyg-max committed 2025-10-28 13:54:37 +08:00
1 parent 3df652f943
commit 275e7bb4b4
4 files changed
+590
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

app/api/v1/video.py
+176
View File
@@ -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)}")
app/services/grok/video.py
+362
View File
@@ -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()
main.py
+2
View File
@@ -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)
# 挂载静态文件
test_video.py
+50
View File
@@ -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())