chore: integrate into existing architecture

app/api/v1/video.py

@@ -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)}")

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()

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
+

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%时关闭 <think> 标签并立即处理视频
+                                # 进度100%时关闭 <think> 标签
                                 content = f"视频已生成{progress}%</think>\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'<video src="{local_video_url}" controls="controls"></video>\n'
-                                        else:
-                                            content += f'<video src="{full_video_url}" controls="controls"></video>\n'
-                                    except Exception as e:
-                                        logger.warning(f"[Processor] 缓存视频失败: {e}")
-                                        content += f'<video src="{full_video_url}" controls="controls"></video>\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'<video src="{local_video_url}" controls="controls"></video>\n'
+                                else:
+                                    video_content = f'<video src="{full_video_url}" controls="controls"></video>\n'
+                            except Exception as e:
+                                logger.warning(f"[Processor] 缓存视频失败: {e}")
+                                video_content = f'<video src="{full_video_url}" controls="controls"></video>\n'
+                            
+                            yield make_chunk(video_content)
+                            timeout_manager.mark_chunk_received()
+                            chunk_index += 1
+                        
                         continue
 
                     # 检查生成模式

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())
     }

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}")

app/services/grok/video.py

@@ -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()

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)

test_video.py

@@ -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())