import mimetypes
from io import BufferedReader, BytesIO
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Union, cast

import httpx
from httpx._types import RequestFiles

import litellm
from litellm.llms.base_llm.videos.transformation import BaseVideoConfig
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 (
    CharacterObject,
    VideoCreateOptionalRequestParams,
    VideoObject,
)
from litellm.types.videos.utils import (
    encode_video_id_with_provider,
    extract_original_character_id,
    extract_original_video_id,
)

if TYPE_CHECKING:
    from litellm.litellm_core_utils.litellm_logging import Logging as _LiteLLMLoggingObj

    from ...base_llm.chat.transformation import BaseLLMException as _BaseLLMException

    LiteLLMLoggingObj = _LiteLLMLoggingObj
    BaseLLMException = _BaseLLMException
else:
    LiteLLMLoggingObj = Any
    BaseLLMException = Any


class OpenAIVideoConfig(BaseVideoConfig):
    """
    Configuration class for OpenAI video generation.
    """

    def __init__(self):
        super().__init__()

    def get_supported_openai_params(self, model: str) -> list:
        """
        Get the list of supported OpenAI parameters for video generation.
        """
        return [
            "model",
            "prompt",
            "input_reference",
            "seconds",
            "size",
            "characters",
            "user",
            "extra_headers",
        ]

    def map_openai_params(
        self,
        video_create_optional_params: VideoCreateOptionalRequestParams,
        model: str,
        drop_params: bool,
    ) -> Dict:
        """No mapping applied since inputs are in OpenAI spec already"""
        return dict(video_create_optional_params)

    def validate_environment(
        self,
        headers: dict,
        model: str,
        api_key: Optional[str] = None,
        litellm_params: Optional[GenericLiteLLMParams] = None,
    ) -> dict:
        # Use api_key from litellm_params if available, otherwise fall back to other sources
        if litellm_params and litellm_params.api_key:
            api_key = api_key or litellm_params.api_key

        api_key = (
            api_key
            or litellm.api_key
            or litellm.openai_key
            or get_secret_str("OPENAI_API_KEY")
        )
        headers.update(
            {
                "Authorization": f"Bearer {api_key}",
            }
        )
        return headers

    def get_complete_url(
        self,
        model: str,
        api_base: Optional[str],
        litellm_params: dict,
    ) -> str:
        """
        Get the complete URL for OpenAI video generation.
        """
        if api_base is None:
            api_base = "https://api.openai.com/v1"

        return f"{api_base.rstrip('/')}/videos"

    def transform_video_create_request(
        self,
        model: str,
        prompt: str,
        api_base: str,
        video_create_optional_request_params: Dict,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> Tuple[Dict, RequestFiles, str]:
        """
        Transform the video creation request for OpenAI API.
        """
        # Remove model and extra_headers from optional params as they're handled separately
        video_create_optional_request_params = {
            k: v
            for k, v in video_create_optional_request_params.items()
            if k not in ["model", "extra_headers", "prompt"]
        }

        # Create the request data
        video_create_request = CreateVideoRequest(
            model=model, prompt=prompt, **video_create_optional_request_params
        )
        request_dict = cast(Dict, video_create_request)
        request_dict = self._decode_character_ids_in_create_video_request(request_dict)

        # Handle input_reference parameter if provided
        _input_reference = video_create_optional_request_params.get("input_reference")
        data_without_files = {
            k: v for k, v in request_dict.items() if k not in ["input_reference"]
        }
        files_list: List[Tuple[str, Any]] = []

        # Handle input_reference parameter
        if _input_reference is not None:
            self._add_image_to_files(
                files_list=files_list,
                image=_input_reference,
                field_name="input_reference",
            )
        return data_without_files, files_list, api_base

    def _decode_character_ids_in_create_video_request(self, request_dict: Dict) -> Dict:
        """
        Decode LiteLLM-managed encoded character ids for provider requests.

        OpenAI expects character ids like `char_...`. If a caller sends
        `character_<base64-encoded-provider-payload>`, convert it back to the
        original provider id before forwarding upstream.
        """
        raw_characters = request_dict.get("characters")
        if not isinstance(raw_characters, list):
            return request_dict

        decoded_characters: List[Any] = []
        for character in raw_characters:
            if not isinstance(character, dict):
                decoded_characters.append(character)
                continue

            character_id = character.get("id")
            if isinstance(character_id, str):
                decoded_character = dict(character)
                decoded_character["id"] = extract_original_character_id(character_id)
                decoded_characters.append(decoded_character)
            else:
                decoded_characters.append(character)

        request_dict["characters"] = decoded_characters
        return request_dict

    def transform_video_create_response(
        self,
        model: str,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        custom_llm_provider: Optional[str] = None,
        request_data: Optional[Dict] = None,
    ) -> VideoObject:
        """Transform the OpenAI video creation response."""
        response_data = raw_response.json()

        video_obj = VideoObject(**response_data)  # type: ignore[arg-type]

        if custom_llm_provider and video_obj.id:
            video_obj.id = encode_video_id_with_provider(
                video_obj.id, custom_llm_provider, model
            )

        usage_data = {}
        if video_obj:
            if hasattr(video_obj, "seconds") and video_obj.seconds:
                try:
                    usage_data["duration_seconds"] = float(video_obj.seconds)
                except (ValueError, TypeError):
                    pass
        video_obj.usage = usage_data

        return video_obj

    def transform_video_content_request(
        self,
        video_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
        variant: Optional[str] = None,
    ) -> Tuple[str, Dict]:
        """
        Transform the video content request for OpenAI API.

        OpenAI API expects the following request:
        - GET /v1/videos/{video_id}/content
        - GET /v1/videos/{video_id}/content?variant=thumbnail
        """
        original_video_id = extract_original_video_id(video_id)

        # Construct the URL for video content download
        url = f"{api_base.rstrip('/')}/{original_video_id}/content"
        if variant is not None:
            url = f"{url}?variant={variant}"

        # No additional data needed for GET content request
        data: Dict[str, Any] = {}

        return url, data

    def transform_video_remix_request(
        self,
        video_id: str,
        prompt: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
        extra_body: Optional[Dict[str, Any]] = None,
    ) -> Tuple[str, Dict]:
        """
        Transform the video remix request for OpenAI API.

        OpenAI API expects the following request:
        - POST /v1/videos/{video_id}/remix
        """
        original_video_id = extract_original_video_id(video_id)

        # Construct the URL for video remix
        url = f"{api_base.rstrip('/')}/{original_video_id}/remix"

        # Prepare the request data
        data = {"prompt": prompt}

        # Add any extra body parameters
        if extra_body:
            data.update(extra_body)

        return url, data

    def transform_video_content_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> bytes:
        """Transform the OpenAI video content download response."""
        return raw_response.content

    def transform_video_remix_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        custom_llm_provider: Optional[str] = None,
    ) -> VideoObject:
        """
        Transform the OpenAI video remix response.
        """
        response_data = raw_response.json()

        # Transform the response data
        video_obj = VideoObject(**response_data)  # type: ignore[arg-type]

        if custom_llm_provider and video_obj.id:
            video_obj.id = encode_video_id_with_provider(
                video_obj.id, custom_llm_provider, None
            )

        # Create usage object with duration information for cost calculation
        # Video remix API doesn't provide usage, so we create one with duration
        usage_data = {}
        if video_obj:
            if hasattr(video_obj, "seconds") and video_obj.seconds:
                try:
                    usage_data["duration_seconds"] = float(video_obj.seconds)
                except (ValueError, TypeError):
                    pass
        # Create the response
        video_obj.usage = usage_data

        return video_obj

    def transform_video_list_request(
        self,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
        after: Optional[str] = None,
        limit: Optional[int] = None,
        order: Optional[str] = None,
        extra_query: Optional[Dict[str, Any]] = None,
    ) -> Tuple[str, Dict]:
        """
        Transform the video list request for OpenAI API.

        OpenAI API expects the following request:
        - GET /v1/videos
        """
        # Use the api_base directly for video list
        url = api_base

        # Prepare query parameters
        params = {}
        if after is not None:
            # Decode the wrapped video ID back to the original provider ID
            params["after"] = extract_original_video_id(after)
        if limit is not None:
            params["limit"] = str(limit)
        if order is not None:
            params["order"] = order

        # Add any extra query parameters
        if extra_query:
            params.update(extra_query)

        return url, params

    def transform_video_list_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        custom_llm_provider: Optional[str] = None,
    ) -> Dict[str, str]:
        response_data = raw_response.json()

        if custom_llm_provider and "data" in response_data:
            for video_obj in response_data.get("data", []):
                if isinstance(video_obj, dict) and "id" in video_obj:
                    video_obj["id"] = encode_video_id_with_provider(
                        video_obj["id"],
                        custom_llm_provider,
                        video_obj.get("model"),
                    )

            # Encode pagination cursor IDs so they remain consistent
            # with the wrapped data[].id format
            data_list = response_data.get("data", [])
            if response_data.get("first_id"):
                first_model = None
                if data_list and isinstance(data_list[0], dict):
                    first_model = data_list[0].get("model")
                response_data["first_id"] = encode_video_id_with_provider(
                    response_data["first_id"],
                    custom_llm_provider,
                    first_model,
                )
            if response_data.get("last_id"):
                last_model = None
                if data_list and isinstance(data_list[-1], dict):
                    last_model = data_list[-1].get("model")
                response_data["last_id"] = encode_video_id_with_provider(
                    response_data["last_id"],
                    custom_llm_provider,
                    last_model,
                )

        return response_data

    def transform_video_delete_request(
        self,
        video_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> Tuple[str, Dict]:
        """
        Transform the video delete request for OpenAI API.

        OpenAI API expects the following request:
        - DELETE /v1/videos/{video_id}
        """
        original_video_id = extract_original_video_id(video_id)

        # Construct the URL for video delete
        url = f"{api_base.rstrip('/')}/{original_video_id}"

        # No data needed for DELETE request
        data: Dict[str, Any] = {}

        return url, data

    def transform_video_delete_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
    ) -> VideoObject:
        """
        Transform the OpenAI video delete response.
        """
        response_data = raw_response.json()

        # Transform the response data
        video_obj = VideoObject(**response_data)  # type: ignore[arg-type]  # type: ignore[arg-type]

        return video_obj

    def transform_video_status_retrieve_request(
        self,
        video_id: str,
        api_base: str,
        litellm_params: GenericLiteLLMParams,
        headers: dict,
    ) -> Tuple[str, Dict]:
        """
        Transform the OpenAI video retrieve request.
        """
        # Extract the original video_id (remove provider encoding if present)
        original_video_id = extract_original_video_id(video_id)

        # For video retrieve, we just need to construct the URL
        url = f"{api_base.rstrip('/')}/{original_video_id}"

        # No additional data needed for GET request
        data: Dict[str, Any] = {}

        return url, data

    def transform_video_status_retrieve_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        custom_llm_provider: Optional[str] = None,
    ) -> VideoObject:
        """
        Transform the OpenAI video retrieve response.
        """
        response_data = raw_response.json()
        # Transform the response data
        video_obj = VideoObject(**response_data)  # type: ignore[arg-type]

        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 get_error_class(
        self, error_message: str, status_code: int, headers: Union[dict, httpx.Headers]
    ) -> BaseLLMException:
        from ...base_llm.chat.transformation import BaseLLMException

        raise BaseLLMException(
            status_code=status_code,
            message=error_message,
            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]],
        image: Any,
        field_name: str,
    ) -> None:
        """Add an image to the files list with appropriate content type"""
        image_content_type = ImageEditRequestUtils.get_image_content_type(image)

        if isinstance(image, BufferedReader):
            files_list.append((field_name, (image.name, image, image_content_type)))
        else:
            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"