diff --git a/litellm/llms/openai/videos/transformation.py b/litellm/llms/openai/videos/transformation.py index e224097fb0..0501a67fc1 100644 --- a/litellm/llms/openai/videos/transformation.py +++ b/litellm/llms/openai/videos/transformation.py @@ -1,4 +1,5 @@ -from io import BufferedReader +import mimetypes +from io import BufferedReader, BytesIO from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, cast import httpx @@ -10,7 +11,11 @@ from litellm.llms.openai.image_edit.transformation import ImageEditRequestUtils from litellm.secret_managers.main import get_secret_str from litellm.types.llms.openai import CreateVideoRequest from litellm.types.router import GenericLiteLLMParams -from litellm.types.videos.main import VideoCreateOptionalRequestParams, VideoObject +from litellm.types.videos.main import ( + CharacterObject, + VideoCreateOptionalRequestParams, + VideoObject, +) from litellm.types.videos.utils import ( encode_video_id_with_provider, extract_original_video_id, @@ -430,6 +435,106 @@ class OpenAIVideoConfig(BaseVideoConfig): headers=headers, ) + def transform_video_create_character_request( + self, + name: str, + video: Any, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> Tuple[str, list]: + url = f"{api_base.rstrip('/')}/characters" + files_list: List[Tuple[str, Any]] = [("name", (None, name))] + self._add_video_to_files(files_list, video, "video") + return url, files_list + + def transform_video_create_character_response( + self, + raw_response: httpx.Response, + logging_obj: Any, + ) -> CharacterObject: + return CharacterObject(**raw_response.json()) + + def transform_video_get_character_request( + self, + character_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + ) -> Tuple[str, Dict]: + url = f"{api_base.rstrip('/')}/characters/{character_id}" + return url, {} + + def transform_video_get_character_response( + self, + raw_response: httpx.Response, + logging_obj: Any, + ) -> CharacterObject: + return CharacterObject(**raw_response.json()) + + def transform_video_edit_request( + self, + prompt: str, + video_id: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Tuple[str, Dict]: + original_video_id = extract_original_video_id(video_id) + url = f"{api_base.rstrip('/')}/edits" + data: Dict[str, Any] = {"prompt": prompt, "video": {"id": original_video_id}} + if extra_body: + data.update(extra_body) + return url, data + + def transform_video_edit_response( + self, + raw_response: httpx.Response, + logging_obj: Any, + custom_llm_provider: Optional[str] = None, + ) -> VideoObject: + video_obj = VideoObject(**raw_response.json()) + if custom_llm_provider and video_obj.id: + video_obj.id = encode_video_id_with_provider( + video_obj.id, custom_llm_provider, None + ) + return video_obj + + def transform_video_extension_request( + self, + prompt: str, + video_id: str, + seconds: str, + api_base: str, + litellm_params: GenericLiteLLMParams, + headers: dict, + extra_body: Optional[Dict[str, Any]] = None, + ) -> Tuple[str, Dict]: + original_video_id = extract_original_video_id(video_id) + url = f"{api_base.rstrip('/')}/extensions" + data: Dict[str, Any] = { + "prompt": prompt, + "seconds": seconds, + "video": {"id": original_video_id}, + } + if extra_body: + data.update(extra_body) + return url, data + + def transform_video_extension_response( + self, + raw_response: httpx.Response, + logging_obj: Any, + custom_llm_provider: Optional[str] = None, + ) -> VideoObject: + video_obj = VideoObject(**raw_response.json()) + if custom_llm_provider and video_obj.id: + video_obj.id = encode_video_id_with_provider( + video_obj.id, custom_llm_provider, None + ) + return video_obj + def _add_image_to_files( self, files_list: List[Tuple[str, Any]], @@ -445,3 +550,49 @@ class OpenAIVideoConfig(BaseVideoConfig): files_list.append( (field_name, ("input_reference.png", image, image_content_type)) ) + + def _add_video_to_files( + self, + files_list: List[Tuple[str, Any]], + video: Any, + field_name: str, + ) -> None: + """ + Add a video to files with proper video MIME type detection. + + This path is used by POST /videos/characters and must send video/mp4, + not image/* content types. + """ + filename = getattr(video, "name", None) or "input_video.mp4" + content_type = self._get_video_content_type(video=video, filename=filename) + files_list.append((field_name, (filename, video, content_type))) + + def _get_video_content_type(self, video: Any, filename: str) -> str: + guessed_content_type, _ = mimetypes.guess_type(filename) + if guessed_content_type and guessed_content_type.startswith("video/"): + return guessed_content_type + + # Fast-path detection for common MP4 signatures when filename is missing/incorrect. + try: + header_bytes = b"" + if isinstance(video, BytesIO): + current_pos = video.tell() + video.seek(0) + header_bytes = video.read(64) + video.seek(current_pos) + elif isinstance(video, BufferedReader): + current_pos = video.tell() + video.seek(0) + header_bytes = video.read(64) + video.seek(current_pos) + elif isinstance(video, bytes): + header_bytes = video[:64] + + # MP4 typically includes ftyp in first box. + if b"ftyp" in header_bytes: + return "video/mp4" + except Exception: + pass + + # OpenAI create-character currently supports mp4. + return "video/mp4" diff --git a/litellm/types/videos/main.py b/litellm/types/videos/main.py index b6357f3273..24454890fe 100644 --- a/litellm/types/videos/main.py +++ b/litellm/types/videos/main.py @@ -3,7 +3,7 @@ from typing import Any, Dict, List, Literal, Optional from pydantic import BaseModel from typing_extensions import TypedDict -from litellm.types.utils import FileTypes +FileTypes = Any class VideoObject(BaseModel): @@ -104,3 +104,51 @@ class DecodedVideoId(TypedDict, total=False): custom_llm_provider: Optional[str] model_id: Optional[str] video_id: str + + +class DecodedCharacterId(TypedDict, total=False): + """Structure representing a decoded character ID""" + + custom_llm_provider: Optional[str] + model_id: Optional[str] + character_id: str + + +class CharacterObject(BaseModel): + """Represents a character created from a video.""" + + id: str + object: Literal["character"] = "character" + created_at: int + name: str + _hidden_params: Dict[str, Any] = {} + + def __contains__(self, key): + return hasattr(self, key) + + def get(self, key, default=None): + return getattr(self, key, default) + + def __getitem__(self, key): + return getattr(self, key) + + def json(self, **kwargs): # type: ignore + try: + return self.model_dump(**kwargs) + except Exception: + return self.dict() + + +class VideoEditRequestParams(TypedDict, total=False): + """TypedDict for video edit request parameters.""" + + prompt: str + video: Dict[str, str] # {"id": "video_123"} + + +class VideoExtensionRequestParams(TypedDict, total=False): + """TypedDict for video extension request parameters.""" + + prompt: str + seconds: str + video: Dict[str, str] # {"id": "video_123"}