From 78de3c747907cfef12820aa7f4d7b1a5fb427751 Mon Sep 17 00:00:00 2001 From: Chenyme <118253778+chenyme@users.noreply.github.com> Date: Tue, 28 Oct 2025 14:41:28 +0800 Subject: [PATCH] chore: integrate into existing architecture --- app/api/v1/video.py | 176 ---------------- app/services/grok/client.py | 71 +++++-- app/services/grok/create.py | 138 +++++++++++++ app/services/grok/processer.py | 48 +++-- app/services/grok/statsig.py | 2 +- app/services/grok/upload.py | 5 +- app/services/grok/video.py | 362 --------------------------------- main.py | 4 +- test_video.py | 50 ----- 9 files changed, 226 insertions(+), 630 deletions(-) delete mode 100644 app/api/v1/video.py create mode 100644 app/services/grok/create.py delete mode 100644 app/services/grok/video.py delete mode 100644 test_video.py diff --git a/app/api/v1/video.py b/app/api/v1/video.py deleted file mode 100644 index b1de1a8..0000000 --- a/app/api/v1/video.py +++ /dev/null @@ -1,176 +0,0 @@ -"""视频生成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/client.py b/app/services/grok/client.py index 4f7fbb3..05b3dea 100644 --- a/app/services/grok/client.py +++ b/app/services/grok/client.py @@ -13,6 +13,7 @@ from app.services.grok.processer import GrokResponseProcessor from app.services.grok.statsig import get_dynamic_headers from app.services.grok.token import token_manager from app.services.grok.upload import ImageUploadManager +from app.services.grok.create import PostCreateManager from app.core.exception import GrokApiException # 常量定义 @@ -44,7 +45,6 @@ class GrokClient: if len(image_urls) > 1: logger.warning(f"[Client] 视频模型只允许一张图片，当前有{len(image_urls)}张，只使用第一张") image_urls = image_urls[:1] - content = f"{content} --mode=custom" logger.debug(f"[Client] 视频模型文本处理: {content}") # 重试逻辑 @@ -61,11 +61,26 @@ class GrokClient: auth_token = token_manager.get_token(model) # 上传图片 - imgs = await GrokClient._upload_imgs(image_urls, auth_token) + imgs, uris = await GrokClient._upload_imgs(image_urls, auth_token) + + # 视频模型需要额外的create操作 + post_id = None + if is_video and imgs and uris: + logger.debug(f"[Client] 检测到视频模型，执行post create操作") + try: + create_result = await PostCreateManager.create(imgs[0], uris[0], auth_token) + if create_result and create_result.get("success"): + post_id = create_result.get("post_id") + logger.debug(f"[Client] Post创建成功: {post_id}") + else: + logger.warning(f"[Client] Post创建失败，继续使用原有流程") + except Exception as e: + logger.warning(f"[Client] Post创建异常: {e}，继续使用原有流程") # 构建并发送请求 - payload = GrokClient._build_payload(content, model_name, model_mode, imgs, is_video) - return await GrokClient._send_request(payload, auth_token, model, stream) + payload = GrokClient._build_payload(content, model_name, model_mode, imgs, uris, is_video, post_id) + logger.debug(f"[Client] 请求载荷配置: {payload}") + return await GrokClient._send_request(payload, auth_token, model, stream, post_id) except GrokApiException as e: last_err = e @@ -112,23 +127,25 @@ class GrokClient: return "".join(content_parts), image_urls @staticmethod - async def _upload_imgs(image_urls: List[str], auth_token: str) -> List[str]: + async def _upload_imgs(image_urls: List[str], auth_token: str) -> Tuple[List[str], List[str]]: """上传图片并返回附件ID列表""" image_attachments = [] + image_uris = [] # 并发上传所有图片 tasks = [ImageUploadManager.upload(url, auth_token) for url in image_urls] results = await asyncio.gather(*tasks, return_exceptions=True) - for url, result in zip(image_urls, results): - if isinstance(result, Exception): - logger.warning(f"[Client] 图片上传失败: {url}, 错误: {result}") - elif result: - image_attachments.append(result) + for url, (file_id, file_uri) in zip(image_urls, results): + if isinstance(file_id, Exception): + logger.warning(f"[Client] 图片上传失败: {url}, 错误: {file_id}") + elif file_id: + image_attachments.append(file_id) + image_uris.append(file_uri) - return image_attachments + return image_attachments, image_uris @staticmethod - def _build_payload(content: str, model_name: str, model_mode: str, image_attachments: List[str], is_video_model: bool = False) -> Dict[str, Any]: + def _build_payload(content: str, model_name: str, model_mode: str, image_attachments: List[str], image_uris: List[str], is_video_model: bool = False, post_id: str = None) -> Dict[str, Any]: """构建Grok API请求载荷""" payload = { "temporary": setting.grok_config.get("temporary", True), @@ -157,14 +174,32 @@ class GrokClient: } # 视频模型特殊配置 - if is_video_model: - payload["toolOverrides"] = {"videoGen": True} + if is_video_model and image_uris: + image_url = image_uris[0] + logger.debug(f"[Client] 视频模型图片URL: {image_url}") + + # 根据是否有post_id选择不同的URL格式 + if post_id: + logger.debug(f"[Client] 使用PostID构建URL: {post_id}") + image_message = f"https://grok.com/imagine/{post_id} {content} --mode=custom" + else: + logger.debug(f"[Client] 使用文件URI构建URL: {image_url}") + image_message = f"https://assets.grok.com/post/{image_url} {content} --mode=custom" + + payload = { + "temporary": True, + "modelName": "grok-3", + "message": image_message, + "fileAttachments": image_attachments, + "toolOverrides": {"videoGen": True} + } + logger.debug(f"[Client] 视频模型载荷配置: {payload}") logger.debug("[Client] 视频模型载荷配置: toolOverrides.videoGen = True") return payload @staticmethod - async def _send_request(payload: dict, auth_token: str, model: str, stream: bool): + async def _send_request(payload: dict, auth_token: str, model: str, stream: bool, post_id: str = None): """发送HTTP请求到Grok API""" # 验证认证令牌 if not auth_token: @@ -173,6 +208,12 @@ class GrokClient: try: # 构建请求头 headers = GrokClient._build_headers(auth_token) + if model == "grok-imagine-0.9": + # 优先使用传入的post_id，否则使用fileAttachments中的第一个 + referer_id = post_id if post_id else payload.get("fileAttachments", [""])[0] + if referer_id: + headers["Referer"] = f"https://grok.com/imagine/{referer_id}" + logger.debug(f"[Client] 设置Referer: {headers['Referer']}") # 使用服务代理 proxy_url = setting.get_service_proxy() diff --git a/app/services/grok/create.py b/app/services/grok/create.py new file mode 100644 index 0000000..1aeddfa --- /dev/null +++ b/app/services/grok/create.py @@ -0,0 +1,138 @@ +"""Post创建管理器""" + +import json +from typing import Dict, Any, Optional +from curl_cffi.requests import AsyncSession + +from app.services.grok.statsig import get_dynamic_headers +from app.core.exception import GrokApiException +from app.core.config import setting +from app.core.logger import logger + +# 常量定义 +CREATE_ENDPOINT = "https://grok.com/rest/media/post/create" +REQUEST_TIMEOUT = 30 +IMPERSONATE_BROWSER = "chrome133a" + + +class PostCreateManager: + """ + Grok Post创建管理器 + + 提供图片post创建功能，用于视频生成前的准备工作 + """ + + @staticmethod + async def create(file_id: str, file_uri: str, auth_token: str) -> Optional[Dict[str, Any]]: + """ + 创建post记录 + + Args: + file_id: 上传后的文件ID + file_uri: 上传后的文件URI + auth_token: 认证令牌 + + Returns: + 创建的post信息，包含post_id等 + """ + try: + # 验证参数 + if not file_id or not file_uri: + raise GrokApiException("文件ID或URI缺失", "INVALID_PARAMS") + + if not auth_token: + raise GrokApiException("认证令牌缺失", "NO_AUTH_TOKEN") + + # 构建创建数据 + # 构建完整的 media_url + media_url = f"https://assets.grok.com/{file_uri}" + + create_data = { + "media_url": media_url, + "media_type": "MEDIA_POST_TYPE_IMAGE" + } + + logger.debug(f"[PostCreate] 创建post请求: fileId={file_id}, uri={file_uri}, media_url={media_url}") + + # 获取认证令牌和cookie + 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", "") + if proxy_url: + logger.debug(f"[PostCreate] 使用代理: {proxy_url.split('@')[-1] if '@' in proxy_url else proxy_url}") + + proxies = {"http": proxy_url, "https": proxy_url} if proxy_url else None + + # 发送异步请求 + async with AsyncSession() as session: + response = await session.post( + CREATE_ENDPOINT, + headers={ + **get_dynamic_headers("/rest/media/post/create"), + "Cookie": cookie, + }, + json=create_data, + impersonate=IMPERSONATE_BROWSER, + timeout=REQUEST_TIMEOUT, + proxies=proxies, + ) + + # 检查响应 + if response.status_code == 200: + result = response.json() + post_id = result.get("postId", "") + logger.debug(f"[PostCreate] Post创建成功，PostID: {post_id}") + return { + "post_id": post_id, + "file_id": file_id, + "file_uri": file_uri, + "success": True, + "data": result + } + else: + error_msg = f"状态码: {response.status_code}" + try: + error_data = response.json() + error_msg = f"{error_msg}, 详情: {error_data}" + except: + error_msg = f"{error_msg}, 详情: {response.text[:200]}" + + logger.error(f"[PostCreate] Post创建失败: {error_msg}") + raise GrokApiException(f"Post创建失败: {error_msg}", "CREATE_ERROR") + + except GrokApiException: + raise + except Exception as e: + logger.error(f"[PostCreate] Post创建异常: {e}") + raise GrokApiException(f"Post创建异常: {e}", "CREATE_ERROR") from e + + @staticmethod + async def batch_create(file_list: list, auth_token: str) -> list: + """ + 批量创建post记录 + + Args: + file_list: 文件列表，每个元素包含 (file_id, file_uri) + auth_token: 认证令牌 + + Returns: + 创建结果列表 + """ + results = [] + for file_id, file_uri in file_list: + try: + result = await PostCreateManager.create(file_id, file_uri, auth_token) + results.append(result) + except Exception as e: + logger.warning(f"[PostCreate] 批量创建失败: fileId={file_id}, 错误={e}") + results.append({ + "file_id": file_id, + "file_uri": file_uri, + "success": False, + "error": str(e) + }) + + return results + diff --git a/app/services/grok/processer.py b/app/services/grok/processer.py index bcb82a5..1c29564 100644 --- a/app/services/grok/processer.py +++ b/app/services/grok/processer.py @@ -269,7 +269,7 @@ class GrokResponseProcessor: yield "data: [DONE]\n\n" return - logger.debug(f"[Processor] 接收到数据块: {len(chunk)} bytes") + logger.debug(f"[Processor] 接收到数据块: {chunk} \n\n") if not chunk: continue @@ -286,7 +286,7 @@ class GrokResponseProcessor: # 提取响应数据 grok_resp = data.get("result", {}).get("response", {}) - logger.debug(f"[Processor] 解析响应数据: {len(grok_resp)} 字段") + logger.debug(f"[Processor] 解析响应数据: {grok_resp} \n\n") if not grok_resp: continue @@ -298,7 +298,9 @@ class GrokResponseProcessor: # 提取视频数据 if video_resp := grok_resp.get("streamingVideoGenerationResponse"): progress = video_resp.get("progress", 0) + v_url = video_resp.get("videoUrl") + # 处理进度更新（仅当进度增加时） if progress > last_video_progress: last_video_progress = progress @@ -309,31 +311,35 @@ class GrokResponseProcessor: elif progress < 100: content = f"视频已生成{progress}%\n" else: - # 进度100%时关闭 标签并立即处理视频 + # 进度100%时关闭 标签 content = f"视频已生成{progress}%\n" - - # 立即下载并缓存视频 - if v_url := video_resp.get("videoUrl"): - logger.debug(f"[Processor] 视频生成完成: {v_url}") - full_video_url = f"https://assets.grok.com/{v_url}" - - try: - cache_path = await video_cache_service.download_video(f"/{v_url}", auth_token) - if cache_path: - video_path = v_url.replace('/', '-') - base_url = setting.global_config.get("base_url", "") - local_video_url = f"{base_url}/images/{video_path}" if base_url else f"/images/{video_path}" - content += f'\n' - else: - content += f'\n' - except Exception as e: - logger.warning(f"[Processor] 缓存视频失败: {e}") - content += f'\n' yield make_chunk(content) timeout_manager.mark_chunk_received() chunk_index += 1 + # 处理视频URL（单独判断，不依赖进度） + if v_url: + logger.debug(f"[Processor] 视频生成完成: {v_url}") + full_video_url = f"https://assets.grok.com/{v_url}" + + try: + cache_path = await video_cache_service.download_video(f"/{v_url}", auth_token) + if cache_path: + video_path = v_url.replace('/', '-') + base_url = setting.global_config.get("base_url", "") + local_video_url = f"{base_url}/images/{video_path}" if base_url else f"/images/{video_path}" + video_content = f'\n' + else: + video_content = f'\n' + except Exception as e: + logger.warning(f"[Processor] 缓存视频失败: {e}") + video_content = f'\n' + + yield make_chunk(video_content) + timeout_manager.mark_chunk_received() + chunk_index += 1 + continue # 检查生成模式 diff --git a/app/services/grok/statsig.py b/app/services/grok/statsig.py index 2c30601..abeb212 100644 --- a/app/services/grok/statsig.py +++ b/app/services/grok/statsig.py @@ -36,7 +36,7 @@ def get_dynamic_headers(pathname: str = "/rest/app-chat/conversations/new") -> D "Sec-Fetch-Dest": "empty", "Sec-Fetch-Mode": "cors", "Sec-Fetch-Site": "same-origin", - "Baggage": "sentry-public_key=b311e0f2690c81f25e2c4cf6d4f7ce1c", + "Baggage": "sentry-environment=production,sentry-public_key=b311e0f2690c81f25e2c4cf6d4f7ce1c", "x-statsig-id": statsig_id, "x-xai-request-id": str(uuid.uuid4()) } diff --git a/app/services/grok/upload.py b/app/services/grok/upload.py index c7ee7de..973a508 100644 --- a/app/services/grok/upload.py +++ b/app/services/grok/upload.py @@ -86,10 +86,11 @@ class ImageUploadManager: if response.status_code == 200: result = response.json() file_id = result.get("fileMetadataId", "") + file_uri = result.get("fileUri", "") logger.debug(f"[Upload] 图片上传成功，文件ID: {file_id}") - return file_id + return file_id, file_uri - return "" + return "", "" except Exception as e: logger.warning(f"[Upload] 上传图片失败: {e}") diff --git a/app/services/grok/video.py b/app/services/grok/video.py deleted file mode 100644 index f774caa..0000000 --- a/app/services/grok/video.py +++ /dev/null @@ -1,362 +0,0 @@ -"""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 6014e85..2b9c463 100644 --- a/main.py +++ b/main.py @@ -11,7 +11,6 @@ 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中完成） @@ -85,7 +84,6 @@ 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) # 挂载静态文件 @@ -113,4 +111,4 @@ app.mount("", mcp_app) if __name__ == "__main__": import uvicorn - uvicorn.run(app, host="0.0.0.0", port=8001) + uvicorn.run(app, host="0.0.0.0", port=8001) \ No newline at end of file diff --git a/test_video.py b/test_video.py deleted file mode 100644 index ac4f502..0000000 --- a/test_video.py +++ /dev/null @@ -1,50 +0,0 @@ -#!/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())