fix(video): enforce character endpoint video MIME handling

Use typed character response models and video multipart helpers so /videos/characters forwards uploaded MP4 files with video/* content type.

Made-with: Cursor
This commit is contained in:
Sameer Kankute
2026-03-16 16:12:07 +05:30
parent 61519d6c65
commit 4a7ef7b1d2
2 changed files with 202 additions and 3 deletions
+153 -2
View File
@@ -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"
+49 -1
View File
@@ -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"}