import base64
import json
import os
import time
from collections.abc import Iterable, Mapping, MutableMapping, Sequence
from contextlib import suppress
from functools import cache
from itertools import chain
from types import MappingProxyType
from typing import Any, Final, TypeAlias, TypedDict
from urllib.parse import unquote

import httpx
from httpx import Headers, Response
from openai.types.file_deleted import FileDeleted
from pydantic import BaseModel, ConfigDict, TypeAdapter
from typing_extensions import ReadOnly

from litellm._logging import verbose_logger
from litellm._uuid import uuid
from litellm.constants import BEDROCK_INVOKE_PROVIDERS_LITERAL
from litellm.files.utils import FilesAPIUtils
from litellm.litellm_core_utils.aws_partition import get_aws_dns_suffix
from litellm.litellm_core_utils.cloud_storage_security import (
    BEDROCK_MANAGED_S3_BATCH_PREFIX,
    BEDROCK_MANAGED_S3_PREFIXES,
    BEDROCK_MANAGED_S3_UPLOAD_PREFIX,
    build_managed_cloud_object_name,
    encode_s3_object_key_for_url,
    sanitize_cloud_object_component,
    should_allow_legacy_cloud_file_ids,
    split_configured_cloud_bucket_name,
    validate_managed_cloud_file_id,
)
from litellm.litellm_core_utils.prompt_templates.common_utils import (
    extract_file_data,
    text_completion_prompt_to_messages,
)
from litellm.llms.base_llm.chat.transformation import BaseLLMException
from litellm.llms.base_llm.files.transformation import (
    BaseFilesConfig,
    LiteLLMLoggingObj,
)
from litellm.types.llms.bedrock import BedrockBatchRecordKind
from litellm.types.llms.openai import (
    AllMessageValues,
    CreateFileRequest,
    FileContentRequest,
    FileTypes,
    HttpxBinaryResponseContent,
    OpenAICreateFileRequestOptionalParams,
    OpenAIFileObject,
    PathLike,
    ResponseInputParam,
    ResponsesAPIOptionalRequestParams,
)
from litellm.types.utils import ExtractedFileData, LlmProviders, SpecialEnums
from litellm.utils import get_llm_provider

from ..base_aws_llm import BaseAWSLLM
from ..common_utils import BedrockError, merge_bedrock_aws_request_params, resolve_s3_encryption_key_id

# litellm_params key used to hand the SigV4-signed GET headers from
# `transform_file_content_request` to `validate_environment` (the only hook
# the shared file-content HTTP handler exposes for setting request headers).
# Same pattern as the `upload_url` handoff in `transform_create_file_request`.
S3_SIGNED_GET_HEADERS_PARAM: Final = "_s3_signed_get_headers"

# litellm_params key carrying the size of the body uploaded to S3, handed from
# `transform_create_file_request` to `transform_create_file_response`.
UPLOAD_CONTENT_LENGTH_PARAM: Final = "_s3_upload_content_length"


def _frozen_mapping(items: Iterable[tuple[str, object]]) -> Mapping[str, object]:
    return MappingProxyType(dict(items))


def _strip_llm_routing_prefix(model: str) -> str:
    try:
        stripped_model, _, _, _ = get_llm_provider(model=model, custom_llm_provider=None)
    except Exception as e:
        verbose_logger.exception(
            "litellm.llms.bedrock.files.transformation.py::_strip_llm_routing_prefix() - Error inferring custom_llm_provider - %s",
            e,
        )
        return model
    return stripped_model


_EmbeddingBatchInput: TypeAlias = (
    str | int | float | Sequence[str] | Sequence[int] | Sequence[Sequence[int]] | Mapping[str, object]
)


class _OpenAIBatchRecordBody(TypedDict, total=False):
    model: ReadOnly[str]
    prompt: ReadOnly[str | Sequence[str] | Sequence[int] | Sequence[Sequence[int]]]
    input: ReadOnly[_EmbeddingBatchInput]
    metadata: ReadOnly[Mapping[str, object]]


class _OpenAIBatchRecord(TypedDict, total=False):
    custom_id: ReadOnly[str]
    url: ReadOnly[str]
    body: ReadOnly[_OpenAIBatchRecordBody]


class _BedrockBatchRecord(TypedDict):
    recordId: ReadOnly[str]
    modelInput: ReadOnly[Mapping[str, object]]


class _S3UploadResponse(TypedDict, total=False):
    Key: ReadOnly[str]
    Bucket: ReadOnly[str]
    ContentLength: ReadOnly[int]


# JSONL batch records are untyped json, so the `/v1/responses` fields are
# validated into their concrete Responses API types before being handed to the
# Responses-to-Chat bridge. Both adapters drop keys the Responses API doesn't
# define, which is what the bridge would ignore anyway. Built on first use
# rather than at import: `ResponseInputParam` is a deep union and only batch
# files carrying `/v1/responses` records need it.
@cache
def _responses_input_adapter() -> TypeAdapter[str | ResponseInputParam]:
    return TypeAdapter(str | ResponseInputParam)


@cache
def _responses_request_adapter() -> TypeAdapter[ResponsesAPIOptionalRequestParams]:
    return TypeAdapter(ResponsesAPIOptionalRequestParams)


class _BedrockS3RequestParams(BaseModel):
    """Typed view of the credential/region params the S3 GetObject path reads."""

    model_config = ConfigDict(extra="ignore")

    aws_access_key_id: str | None = None
    aws_secret_access_key: str | None = None
    aws_session_token: str | None = None
    aws_region_name: str | None = None
    aws_session_name: str | None = None
    aws_profile_name: str | None = None
    aws_role_name: str | None = None
    aws_web_identity_token: str | None = None
    aws_sts_endpoint: str | None = None
    aws_external_id: str | None = None
    s3_region_name: str | None = None
    s3_endpoint_url: str | None = None


class _TrustedS3ModelCredentials(BaseModel):
    """The S3 buckets the server trusts file ids against, from the deployment snapshot."""

    model_config = ConfigDict(extra="ignore")

    s3_bucket_name: str | None = None
    s3_output_bucket_name: str | None = None


def extract_s3_uri_from_file_id(file_id: str) -> str:
    """
    Resolve a Bedrock file id to its S3 URI.

    Accepts either a base64-encoded LiteLLM unified file id (whose decoded
    form carries `llm_output_file_id,s3://...`) or a direct `s3://` URI.
    """
    try:
        padded: Final = file_id + "=" * (-len(file_id) % 4)
        decoded: Final = base64.urlsafe_b64decode(padded).decode()

        if decoded.startswith(SpecialEnums.LITELM_MANAGED_FILE_ID_PREFIX.value):
            if "llm_output_file_id," in decoded:
                return decoded.split("llm_output_file_id,")[1].split(";")[0]
    except Exception:
        pass

    if file_id.startswith("s3://"):
        return file_id

    raise ValueError("file_id must be a managed LiteLLM S3 file id")


_S3_BUCKET_REQUIRED_ERROR: Final = "S3 bucket_name is required. Set 's3_bucket_name' in proxy config or AWS_S3_BUCKET_NAME for Bedrock file content retrieval."


def _trusted_s3_model_credentials(litellm_params: Mapping[str, object]) -> _TrustedS3ModelCredentials:
    trusted_model_credentials: Final = litellm_params.get("_litellm_internal_model_credentials")
    if not isinstance(trusted_model_credentials, MappingProxyType):
        return _TrustedS3ModelCredentials()
    snapshot: Final[dict[str, object]] = {}
    snapshot.update(trusted_model_credentials)  # any-ok: untyped snapshot
    return _TrustedS3ModelCredentials.model_validate(snapshot)


def get_configured_s3_bucket_name(litellm_params: Mapping[str, object]) -> str:
    """
    Resolve the server-configured S3 bucket for Bedrock file operations.

    Only trusts the immutable server-side credential snapshot or the
    environment; never a request-supplied param, since the bucket is what
    `validate_managed_cloud_file_id` checks file ids against.
    """
    bucket_name: Final = _trusted_s3_model_credentials(litellm_params).s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME")
    if not bucket_name:
        raise ValueError(_S3_BUCKET_REQUIRED_ERROR)
    return bucket_name


def get_configured_s3_bucket_names(litellm_params: Mapping[str, object]) -> tuple[str, ...]:
    """
    Resolve the server-configured S3 buckets a Bedrock file id may live in.

    Bedrock batch outputs land in ``s3_output_bucket_name`` when it differs from
    the input bucket, so retrieval validates against both. Same trust rules as
    ``get_configured_s3_bucket_name``: only the immutable credential snapshot or
    the environment, never a request param.
    """
    trusted: Final = _trusted_s3_model_credentials(litellm_params)
    input_bucket: Final = trusted.s3_bucket_name or os.getenv("AWS_S3_BUCKET_NAME")
    output_bucket: Final = trusted.s3_output_bucket_name or os.getenv("AWS_S3_OUTPUT_BUCKET_NAME")
    buckets: Final = tuple(dict.fromkeys(bucket for bucket in (input_bucket, output_bucket) if bucket))
    if not buckets:
        raise ValueError(_S3_BUCKET_REQUIRED_ERROR)
    return buckets


def _validate_file_id_against_configured_buckets(
    s3_uri: str,
    configured_bucket_names: tuple[str, ...],
    allow_legacy_cloud_file_ids: bool,
) -> tuple[str, str]:
    def validate_against(configured_bucket_name: str) -> tuple[str, str]:
        return validate_managed_cloud_file_id(
            file_id=s3_uri,
            scheme="s3://",
            configured_bucket_name=configured_bucket_name,
            allowed_object_prefixes=BEDROCK_MANAGED_S3_PREFIXES,
            allow_legacy_cloud_file_ids=allow_legacy_cloud_file_ids,
        )

    for candidate_bucket_name in configured_bucket_names[:-1]:
        with suppress(ValueError):
            return validate_against(candidate_bucket_name)
    return validate_against(configured_bucket_names[-1])


def _uploaded_object_size(litellm_params: Mapping[str, object], raw_response: Response) -> int:
    """
    S3 answers PutObject with an empty body, so the stored object size comes from the
    signed request recorded by `transform_create_file_request`, not the response headers.
    """
    uploaded_size: Final = litellm_params.get(UPLOAD_CONTENT_LENGTH_PARAM)
    if isinstance(uploaded_size, int):
        return uploaded_size
    response_content_length: Final = raw_response.headers.get("Content-Length", "0")
    return int(response_content_length) if response_content_length.isdigit() else 0


class BedrockFilesConfig(BaseAWSLLM, BaseFilesConfig):
    """
    Config for Bedrock Files - handles S3 uploads for Bedrock batch processing
    """

    def __init__(self):
        self.jsonl_transformation = BedrockJsonlFilesTransformation()
        super().__init__()

    @property
    def custom_llm_provider(self) -> LlmProviders:
        return LlmProviders.BEDROCK

    @property
    def file_upload_http_method(self) -> str:
        """
        Bedrock files are uploaded to S3, which requires PUT requests
        """
        return "PUT"

    def validate_environment(
        self,
        headers: MutableMapping[str, object],
        model: str,
        messages: list[AllMessageValues],
        optional_params: dict,
        litellm_params: MutableMapping[str, object],
        api_key: str | None = None,
        api_base: str | None = None,
    ) -> dict:
        result: Final[dict[str, object]] = {}
        result.update(headers)
        signed_headers: Final = litellm_params.pop(S3_SIGNED_GET_HEADERS_PARAM, None)
        if isinstance(signed_headers, Mapping):
            result.update(signed_headers)  # any-ok: untyped handoff headers
        # otherwise no extra headers - AWS credentials are handled by BaseAWSLLM
        return result

    def _get_content_from_openai_file(self, openai_file_content: FileTypes) -> str:
        """
        Helper to extract content from various OpenAI file types and return as string.

        Handles:
        - Direct content (str, bytes, IO[bytes])
        - Tuple formats: (filename, content, [content_type], [headers])
        - PathLike objects
        """
        content: str | bytes = b""
        # Extract file content from tuple if necessary
        if isinstance(openai_file_content, tuple):
            # Take the second element which is always the file content
            file_content = openai_file_content[1]
        else:
            file_content = openai_file_content

        # Handle different file content types
        if isinstance(file_content, str):
            # String content can be used directly
            content = file_content
        elif isinstance(file_content, bytes):
            # Bytes content can be decoded
            content = file_content
        elif isinstance(file_content, PathLike):  # PathLike
            with open(str(file_content), "rb") as f:
                content = f.read()
        elif hasattr(file_content, "read"):  # IO[bytes]
            # File-like objects need to be read
            content = file_content.read()

        # Ensure content is string
        if isinstance(content, bytes):
            content = content.decode("utf-8")

        return content

    def _get_s3_object_name_from_batch_jsonl(
        self,
        openai_jsonl_content: Sequence[_OpenAIBatchRecord],
    ) -> str:
        """
        Gets a unique S3 object name for the Bedrock batch processing job

        named as: litellm-bedrock-files/{model}/{uuid}
        """
        _model = openai_jsonl_content[0].get("body", {}).get("model", "")
        # Remove bedrock/ prefix if present
        _model = _model.removeprefix("bedrock/")

        safe_model: Final = sanitize_cloud_object_component(_model.replace(":", "-"), fallback="model")

        object_name: Final = f"{BEDROCK_MANAGED_S3_BATCH_PREFIX}{safe_model}-{uuid.uuid4()}.jsonl"
        return object_name

    def get_object_name(self, extracted_file_data: ExtractedFileData, purpose: str) -> str:
        """
        Get the object name for the request
        """
        extracted_file_data_content: Final = extracted_file_data.get("content")

        if extracted_file_data_content is None:
            raise ValueError("file content is required")

        if purpose == "batch":
            ## 1. If jsonl, check if there's a model name
            file_content: Final = self._get_content_from_openai_file(extracted_file_data_content)

            # Split into lines and parse each line as JSON
            openai_jsonl_content: Final = [json.loads(line) for line in file_content.splitlines() if line.strip()]
            if len(openai_jsonl_content) > 0:
                return self._get_s3_object_name_from_batch_jsonl(openai_jsonl_content)

        ## 2. If not jsonl, store under a server-generated managed object name
        filename: Final = extracted_file_data.get("filename")
        return build_managed_cloud_object_name(
            prefix=BEDROCK_MANAGED_S3_UPLOAD_PREFIX,
            filename=filename,
            fallback_filename="file",
        )

    def get_complete_file_url(
        self,
        api_base: str | None,
        api_key: str | None,
        model: str,
        optional_params: dict,
        litellm_params: dict,
        data: CreateFileRequest,
    ) -> str:
        """
        Get the complete S3 URL for the file upload request
        """
        request_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
        bucket_name = litellm_params.get("s3_bucket_name") or os.getenv("AWS_S3_BUCKET_NAME")
        if not bucket_name:
            raise ValueError(
                "S3 bucket_name is required. Set 's3_bucket_name' in litellm_params or AWS_S3_BUCKET_NAME env var"
            )
        bucket_name, object_prefix = split_configured_cloud_bucket_name(bucket_name)

        s3_region_name: Final = litellm_params.get("s3_region_name") or optional_params.get("s3_region_name")
        aws_region_name: Final = s3_region_name or self._get_aws_region_name(request_params, model)

        file_data: Final = data.get("file")
        purpose: Final = data.get("purpose")
        if file_data is None:
            raise ValueError("file is required")
        if purpose is None:
            raise ValueError("purpose is required")
        extracted_file_data: Final = extract_file_data(file_data)
        object_name = self.get_object_name(extracted_file_data, purpose)
        if object_prefix:
            object_name = f"{object_prefix}/{object_name}"
        encoded_object_name: Final = encode_s3_object_key_for_url(object_name)

        # S3 endpoint URL format
        s3_endpoint_url: Final = (
            request_params.get("s3_endpoint_url")
            or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}"
        ).rstrip("/")

        return f"{s3_endpoint_url}/{bucket_name}/{encoded_object_name}"

    def get_supported_openai_params(self, model: str) -> list[OpenAICreateFileRequestOptionalParams]:
        return []

    def map_openai_params(
        self,
        non_default_params: dict,
        optional_params: dict,
        model: str,
        drop_params: bool,
    ) -> dict:
        return optional_params

    # Providers whose InvokeModel body uses the Converse API format
    # (messages + inferenceConfig + image blocks). Nova is the primary
    # example; add others here as they adopt the same schema.
    CONVERSE_INVOKE_PROVIDERS = ("nova",)

    # OpenAI batch URLs that select which request shape a JSONL line carries.
    # Per the OpenAI Batch API spec every record carries a `url`, so we use it
    # as the authoritative routing signal instead of inferring from the
    # presence of `input` vs `prompt` vs `messages`.
    OPENAI_EMBEDDINGS_URL = "/v1/embeddings"
    OPENAI_TEXT_COMPLETIONS_URL = "/v1/completions"
    OPENAI_RESPONSES_URL = "/v1/responses"

    @staticmethod
    def _classify_batch_record(openai_jsonl_record: _OpenAIBatchRecord) -> BedrockBatchRecordKind:
        """
        Decide which OpenAI endpoint shape an OpenAI batch JSONL line carries.

        Precedence (strict - any recognized `url` short-circuits):
          1. A `url` matching a supported endpoint wins. Authoritative per the
             OpenAI Batch API spec, which requires it on every record.
          2. Any other non-empty `url` -> chat. We trust the caller's explicit
             signal rather than re-deriving it from the body, and an
             unexpectedly-shaped body fails cleanly inside the chat
             transformer instead of being silently misrouted.
          3. `url` missing/empty -> fall back to body shape. `messages` wins
             over the other keys so a malformed record carrying several of
             them keeps its conversation instead of having it dropped, and a
             bare `input` stays an embedding for backwards compatibility
             (that ambiguity with `/v1/responses` is only resolvable from
             `url`).
        """
        match openai_jsonl_record.get("url"):
            case BedrockFilesConfig.OPENAI_EMBEDDINGS_URL:
                return BedrockBatchRecordKind.EMBEDDING
            case BedrockFilesConfig.OPENAI_TEXT_COMPLETIONS_URL:
                return BedrockBatchRecordKind.TEXT_COMPLETION
            case BedrockFilesConfig.OPENAI_RESPONSES_URL:
                return BedrockBatchRecordKind.RESPONSES
            case None | "":
                pass
            case _:
                return BedrockBatchRecordKind.CHAT

        body: Final = openai_jsonl_record.get("body")
        if not isinstance(body, Mapping):
            return BedrockBatchRecordKind.CHAT
        if "messages" in body:
            return BedrockBatchRecordKind.CHAT
        if "prompt" in body:
            return BedrockBatchRecordKind.TEXT_COMPLETION
        if "input" in body:
            return BedrockBatchRecordKind.EMBEDDING
        return BedrockBatchRecordKind.CHAT

    # Identifier for the Bedrock Titan v2 InvokeModel body schema as stored
    # in `model_prices_and_context_window.json`. Centralized so future
    # embedding-schema variants can add their own value
    # (e.g. `cohere_v3`, `titan_g1`, `titan_multimodal`) without touching
    # the detection logic.
    _TITAN_V2_INVOCATION_SCHEMA = "titan_v2"

    # Substring marker used as a fallback when the registry can't resolve
    # the model id - notably cross-region inference profile prefixes
    # (`us.amazon.titan-embed-text-v2:0`) and Bedrock ARN forms, which
    # `get_model_info` doesn't normalize today.
    _TITAN_V2_EMBED_MODEL_MARKER = "titan-embed-text-v2"

    # Nested field name under `provider_specific_entry` that identifies the
    # Bedrock InvokeModel body schema for batch inference.
    # `provider_specific_entry` is the registry's escape hatch for fields
    # `get_model_info` doesn't promote to top-level - exactly what we need
    # here. Documented in the `sample_spec` entry of
    # `model_prices_and_context_window.json` and surfaced by
    # `get_model_info` (see `ModelInfo.provider_specific_entry`).
    _BEDROCK_INVOCATION_SCHEMA_FIELD = "bedrock_invocation_schema"

    @staticmethod
    def _is_titan_v2_embed_model(model: str) -> bool:
        """
        True iff `model` refers to Amazon Titan Text Embeddings V2.

        Resolution order:
          1. `model_prices_and_context_window.json` via `get_model_info`.
             The Titan v2 registry entry carries an explicit
             `provider_specific_entry.bedrock_invocation_schema` discriminator
             (`"titan_v2"`). When the registry resolves the id we trust that
             field as the source of truth - no hardcoded model-id comparison
             needed.
          2. Substring fallback (`titan-embed-text-v2` followed by `:`, `/`,
             or end-of-string) for ids the registry can't normalize. This
             catches cross-region inference profile prefixes
             (`us.amazon.titan-embed-text-v2:0`) and Bedrock ARN forms; the
             marker boundary check rejects lookalikes like
             `titan-embed-text-v20` or `titan-embed-text-v2-experimental`.

        Tolerant of common id shapes:
          - "amazon.titan-embed-text-v2:0"
          - "bedrock/amazon.titan-embed-text-v2:0"
          - "us.amazon.titan-embed-text-v2:0" (cross-region inference profile)
          - ARN forms ending in ".../amazon.titan-embed-text-v2:0"
        """
        # Registry-driven path: when get_model_info resolves the id we trust
        # the registry's discriminator. A resolved id with a different (or
        # absent) schema value here is intentionally not given a substring
        # second-chance - the registry is authoritative for ids it knows.
        registry_schema: Final = BedrockFilesConfig._lookup_provider_specific_field(
            model, BedrockFilesConfig._BEDROCK_INVOCATION_SCHEMA_FIELD
        )
        if registry_schema is not None:
            return registry_schema == BedrockFilesConfig._TITAN_V2_INVOCATION_SCHEMA

        # Registry silence -> substring fallback for unmapped ids only.
        normalized = model.lower()
        normalized = normalized.removeprefix("bedrock/")
        marker: Final = BedrockFilesConfig._TITAN_V2_EMBED_MODEL_MARKER
        idx: Final = normalized.find(marker)
        if idx < 0:
            return False
        end: Final = idx + len(marker)
        return end == len(normalized) or normalized[end] in (":", "/")

    @staticmethod
    def _lookup_provider_specific_field(model_id: str, field: str) -> str | None:
        """
        Read a nested string field from the registry entry's
        `provider_specific_entry` dict via `litellm.get_model_info`.

        Returns the field's string value when:
          - the registry resolves `model_id`,
          - the entry exposes `provider_specific_entry` as a dict, and
          - that dict has `field` mapped to a non-empty string.
        Otherwise returns `None`.

        Isolating this means feature detectors (Titan v2 today, future
        Cohere Embed / Nova Multimodal branches) share one defensive
        try/except shape instead of duplicating it. The `None` return
        covers every realistic failure mode: `get_model_info` raises
        (cross-region profile prefixes, Bedrock ARN forms, unreleased
        models), returns a non-dict, has no `provider_specific_entry`, or
        the requested field is missing / non-string / empty.
        """
        try:
            from litellm import get_model_info

            info: Final = get_model_info(model_id)
        except Exception:
            return None
        if not isinstance(info, dict):
            return None
        provider_specific: Final = info.get("provider_specific_entry")
        if not isinstance(provider_specific, dict):
            return None
        value: Final = provider_specific.get(field)
        return value if isinstance(value, str) and value else None

    @staticmethod
    def _coerce_embedding_input_to_string(raw_input: _EmbeddingBatchInput | None, model: str = "") -> str:
        """
        Normalize an OpenAI /v1/embeddings `input` field into the single
        string that Bedrock Titan v2 InvokeModel expects in `inputText`.

        Accepts: a string, or a single-element list containing one string.
        Rejects (with actionable messages):
          - None / missing -> ValueError
          - Multi-element string lists -> ValueError, prompts caller to
            emit one JSONL line per input
          - Pre-tokenized inputs (List[int], List[List[int]]) -> NotImplementedError
          - Any other type -> ValueError

        Extracted so the validation can be exercised in isolation and so
        future embedding-provider branches (Titan G1, Cohere) can reuse it
        without duplicating the type-shaping logic.
        """
        if raw_input is None:
            raise ValueError(f"Embedding batch record is missing required `input` field: model={model}")

        # Bedrock InvokeModel for Titan v2 takes exactly one string `inputText`
        # per call. Pre-tokenized inputs and multi-element string lists are
        # explicitly unsupported so callers emit one JSONL line per embedding
        # instead of relying on us to silently fan out or concatenate.
        if isinstance(raw_input, list):
            if len(raw_input) == 1:
                candidate = raw_input[0]
            else:
                raise ValueError(
                    "Bedrock batch embedding requires one input per JSONL "
                    "record. Got a list with "
                    f"{len(raw_input)} items for model={model}; emit one "
                    "JSONL line per input string instead."
                )
        else:
            candidate = raw_input

        # Catches pre-tokenized inputs (List[int] from OpenAI spec, or a
        # single int slipping past the list-unwrap above).
        # NOTE: bool is a subclass of int but treating True/False as a token
        # is meaningless either way, so the broad check is fine.
        if isinstance(candidate, (list, int)):
            raise NotImplementedError(
                "Bedrock Titan v2 batch embedding does not support "
                "pre-tokenized integer inputs. Pass `input` as a string "
                f"(model={model})."
            )
        if not isinstance(candidate, str):
            raise ValueError(
                "Bedrock batch embedding `input` must be a string (or a "
                "single-element list of strings). Got type "
                f"{type(candidate).__name__} for model={model}."
            )
        return candidate

    def _map_openai_embedding_to_bedrock_params(
        self,
        openai_request_body: _OpenAIBatchRecordBody,
        model: str,
    ) -> dict[str, object]:
        """
        Transform an OpenAI /v1/embeddings request body into the
        Bedrock InvokeModel `modelInput` for embedding models that AWS
        supports via batch inference (CreateModelInvocationJob).

        Currently routes Amazon Titan Text Embeddings V2 only; other
        embedding providers (Titan G1, Titan Multimodal, Cohere Embed,
        Nova Multimodal Embeddings) raise NotImplementedError until they
        get a dedicated branch. Splitting them keeps PR scope tight and
        lets each model's request schema be exercised by its own tests.

        AWS docs (Titan v2 InvokeModel body):
        https://docs.aws.amazon.com/bedrock/latest/userguide/model-parameters-titan-embed-text.html
        """
        from litellm.llms.bedrock.embed.amazon_titan_v2_transformation import (
            AmazonTitanV2Config,
        )

        if not self._is_titan_v2_embed_model(model):
            # Refuse early instead of silently shaping the body for the wrong
            # provider. The synchronous /v1/embeddings path supports more
            # models, but each has a different InvokeModel schema; mapping
            # them here without dedicated tests would risk corrupt batches.
            raise NotImplementedError(
                "Bedrock batch embedding currently supports only Amazon "
                "Titan Text Embeddings V2 (model id contains "
                f"'titan-embed-text-v2'). Got model={model!r}. Track other "
                "embedding models in https://github.com/BerriAI/litellm/issues."
            )

        input_text: Final = self._coerce_embedding_input_to_string(openai_request_body.get("input"), model=model)

        # Map OpenAI-style params (dimensions, encoding_format) onto the
        # Titan v2 schema (dimensions, embeddingTypes) via the embed config
        # so this stays in sync with the synchronous /v1/embeddings path.
        non_default_params: Final = {k: v for k, v in openai_request_body.items() if k not in ("model", "input")}
        titan_config: Final = AmazonTitanV2Config()
        inference_params: Final = titan_config.map_openai_params(
            non_default_params=non_default_params,
            optional_params={},
        )
        return dict(titan_config._transform_request(input=input_text, inference_params=inference_params))

    @staticmethod
    def _transform_text_completion_body_to_chat_body(
        openai_request_body: _OpenAIBatchRecordBody,
    ) -> Mapping[str, object]:
        """
        Rewrite an OpenAI `/v1/completions` batch body as a Chat Completions body.

        Bedrock batch `modelInput` is the model's InvokeModel/Converse body, and
        no Bedrock batch model takes a bare `prompt`, so the wrapping that
        `litellm.text_completion` does in real time has to happen here too.
        """
        prompt: Final = openai_request_body.get("prompt")
        if prompt is None:
            raise ValueError(
                "Batch record for /v1/completions is missing required `prompt` field: "
                f"model={openai_request_body.get('model', '')}"
            )
        return _frozen_mapping(
            chain(
                ((key, value) for key, value in openai_request_body.items() if key != "prompt"),
                (("messages", text_completion_prompt_to_messages(prompt)),),
            )
        )

    @staticmethod
    def _transform_responses_body_to_chat_body(openai_request_body: _OpenAIBatchRecordBody) -> Mapping[str, object]:
        """
        Rewrite an OpenAI `/v1/responses` batch body as a Chat Completions body.

        Delegates to the same Responses-to-Chat bridge the real-time path uses
        for providers without a native Responses API (which is every Bedrock
        model), so `input`, `instructions`, `max_output_tokens` and the tool
        params translate identically in batch and real time. The bridge always
        emits a `tools` key; an empty one is dropped rather than shipped as an
        empty array inside `modelInput`.
        """
        from litellm.responses.litellm_completion_transformation.transformation import (
            LiteLLMCompletionResponsesConfig,
        )

        responses_input: Final = openai_request_body.get("input")
        if responses_input is None:
            raise ValueError(
                "Batch record for /v1/responses is missing required `input` field: "
                f"model={openai_request_body.get('model', '')}"
            )
        chat_body: Final[Mapping[str, object]] = (
            LiteLLMCompletionResponsesConfig.transform_responses_api_request_to_chat_completion_request(
                model=openai_request_body.get("model", ""),
                input=_responses_input_adapter().validate_python(responses_input),
                responses_api_request=_responses_request_adapter().validate_python(
                    _frozen_mapping(
                        (key, value) for key, value in openai_request_body.items() if key not in ("model", "input")
                    )
                ),
                metadata=openai_request_body.get("metadata"),
            )
        )
        return _frozen_mapping((key, value) for key, value in chat_body.items() if key != "tools" or value)

    @staticmethod
    def _transform_batch_body_to_chat_body(
        openai_request_body: _OpenAIBatchRecordBody,
        record_kind: BedrockBatchRecordKind,
    ) -> Mapping[str, object]:
        """
        Normalize a non-embedding batch body to the Chat Completions shape the
        per-provider Bedrock transformations expect.
        """
        match record_kind:
            case BedrockBatchRecordKind.TEXT_COMPLETION:
                return BedrockFilesConfig._transform_text_completion_body_to_chat_body(openai_request_body)
            case BedrockBatchRecordKind.RESPONSES:
                return BedrockFilesConfig._transform_responses_body_to_chat_body(openai_request_body)
            case BedrockBatchRecordKind.CHAT:
                return openai_request_body
            case BedrockBatchRecordKind.EMBEDDING:
                raise ValueError("Embedding batch records do not have a chat-completion equivalent")

    def _map_openai_to_bedrock_params(
        self,
        openai_request_body: Mapping[str, Any],
        model: str,
        provider: str | None = None,
    ) -> dict[str, object]:
        """
        Transform OpenAI request body to Bedrock-compatible modelInput
        parameters using existing transformation logic.

        Routes to the correct per-provider transformation so that the
        resulting dict matches the InvokeModel body that Bedrock expects
        for batch inference.
        """
        from litellm.types.utils import LlmProviders

        messages: Final = openai_request_body.get("messages", [])
        optional_params: Final = {k: v for k, v in openai_request_body.items() if k not in ["model", "messages"]}

        # --- Anthropic: use existing AmazonAnthropicClaudeConfig ---
        if provider == LlmProviders.ANTHROPIC:
            from litellm.llms.bedrock.chat.invoke_transformations.anthropic_claude3_transformation import (
                AmazonAnthropicClaudeConfig,
            )

            config: Final = AmazonAnthropicClaudeConfig()
            mapped_params = config.map_openai_params(
                non_default_params={},
                optional_params=optional_params,
                model=model,
                drop_params=False,
            )
            return config.transform_request(
                model=model,
                messages=messages,
                optional_params=mapped_params,
                litellm_params={},
                headers={},
            )

        # --- Converse API providers (e.g. Nova): use AmazonConverseConfig
        #     to correctly convert image_url blocks to Bedrock image format
        #     and wrap inference params inside inferenceConfig. ---
        if provider in self.CONVERSE_INVOKE_PROVIDERS:
            from litellm.llms.bedrock.chat.converse_transformation import (
                AmazonConverseConfig,
            )

            converse_config: Final = AmazonConverseConfig()
            mapped_params = converse_config.map_openai_params(
                non_default_params=optional_params,
                optional_params={},
                model=model,
                drop_params=False,
            )
            return converse_config.transform_request(
                model=model,
                messages=messages,
                optional_params=mapped_params,
                litellm_params={},
                headers={},
            )

        # --- All other providers: passthrough (OpenAI-compatible models
        #     like openai.gpt-oss-*, qwen, deepseek, etc.) ---
        return {
            "messages": messages,
            **optional_params,
        }

    def _resolve_batch_record_model_and_provider(
        self,
        record_model: str,
        target_model: str,
    ) -> tuple[str, BEDROCK_INVOKE_PROVIDERS_LITERAL | None]:
        record_provider: Final = self.get_bedrock_invoke_provider(_strip_llm_routing_prefix(record_model))
        if record_provider is not None or not target_model:
            return record_model, record_provider
        target_provider: Final = self.get_bedrock_invoke_provider(_strip_llm_routing_prefix(target_model))
        if target_provider is None:
            return record_model, record_provider
        return target_model, target_provider

    def _transform_openai_jsonl_content_to_bedrock_jsonl_content(
        self, openai_jsonl_content: Sequence[_OpenAIBatchRecord], target_model: str = ""
    ) -> list[_BedrockBatchRecord]:
        """
        Transforms OpenAI JSONL content to Bedrock batch format

        Bedrock batch format: { "recordId": "alphanumeric string", "modelInput": {JSON body} }
        Example:
        {
            "recordId": "CALL0000001",
            "modelInput": {
                "anthropic_version": "bedrock-2023-05-31",
                "max_tokens": 1024,
                "messages": [
                    {
                        "role": "user",
                        "content": [{"type": "text", "text": "Hello"}]
                    }
                ]
            }
        }
        """

        import litellm

        bedrock_jsonl_content: Final = []
        for idx, _openai_jsonl_content in enumerate(openai_jsonl_content):
            # Extract the request body from OpenAI format
            openai_body = _openai_jsonl_content.get("body", {})
            record_model = openai_body.get("model", "")
            resolved_model = litellm.model_alias_map.get(record_model, record_model)
            model_for_transform, provider = self._resolve_batch_record_model_and_provider(
                record_model=resolved_model, target_model=target_model
            )

            # Route to the embedding transformer when the OpenAI batch line
            # targets /v1/embeddings; every other endpoint shape is normalized
            # to chat completions first. We branch here (rather than inside
            # `_map_openai_to_bedrock_params`) so the chat helper keeps its
            # narrow contract and the embedding helper can evolve independently.
            record_kind = self._classify_batch_record(_openai_jsonl_content)
            if record_kind is BedrockBatchRecordKind.EMBEDDING:
                model_input = self._map_openai_embedding_to_bedrock_params(
                    openai_request_body=openai_body, model=model_for_transform
                )
            else:
                model_input = self._map_openai_to_bedrock_params(
                    openai_request_body=self._transform_batch_body_to_chat_body(openai_body, record_kind),
                    model=model_for_transform,
                    provider=provider,
                )

            # Create Bedrock batch record
            record_id = _openai_jsonl_content.get("custom_id", f"CALL{str(idx).zfill(7)}")
            bedrock_record = {"recordId": record_id, "modelInput": model_input}

            bedrock_jsonl_content.append(bedrock_record)
        return bedrock_jsonl_content

    def transform_create_file_request(
        self,
        model: str,
        create_file_data: CreateFileRequest,
        optional_params: dict,
        litellm_params: dict,
    ) -> bytes | str | dict:
        """
        Transform file request and return a pre-signed request for S3.
        This keeps the HTTP handler clean by doing all the signing here.
        """
        file_data: Final = create_file_data.get("file")
        if file_data is None:
            raise ValueError("file is required")
        extracted_file_data: Final = extract_file_data(file_data)
        extracted_file_data_content: Final = extracted_file_data.get("content")

        if extracted_file_data_content is None:
            raise ValueError("file content is required")

        # Get and transform the file content
        if FilesAPIUtils.is_batch_jsonl_file(
            create_file_data=create_file_data,
            extracted_file_data=extracted_file_data,
        ):
            ## Transform JSONL content to Bedrock format
            original_file_content: Final = self._get_content_from_openai_file(extracted_file_data_content)
            openai_jsonl_content = [json.loads(line) for line in original_file_content.splitlines() if line.strip()]
            litellm_params_model: Final = litellm_params.get("model")
            target_model: Final = model or (litellm_params_model if isinstance(litellm_params_model, str) else "")
            bedrock_jsonl_content = self._transform_openai_jsonl_content_to_bedrock_jsonl_content(
                openai_jsonl_content, target_model=target_model
            )
            file_content = "\n".join(json.dumps(item) for item in bedrock_jsonl_content)
        elif isinstance(extracted_file_data_content, bytes):
            file_content = extracted_file_data_content.decode("utf-8")
        elif isinstance(extracted_file_data_content, str):
            file_content = extracted_file_data_content
        else:
            raise ValueError("Unsupported file content type")

        # Get the S3 URL for upload
        api_base: Final = self.get_complete_file_url(
            api_base=None,
            api_key=None,
            model=model,
            optional_params=optional_params,
            litellm_params=litellm_params,
            data=create_file_data,
        )

        # s3_region_name always wins for S3 operations (same priority as in
        # get_complete_file_url above). Overwrite aws_region_name unconditionally,
        # after the deployment-credential merge, so the SigV4 region matches the
        # URL region, avoiding SignatureDoesNotMatch.
        merged_params: Final = merge_bedrock_aws_request_params(litellm_params, optional_params)
        s3_region_name: Final = litellm_params.get("s3_region_name") or optional_params.get("s3_region_name")
        request_params: Final = (
            {**merged_params, "aws_region_name": s3_region_name} if s3_region_name else merged_params
        )

        # Sign the request and return a pre-signed request object
        signed_headers, signed_body = self._sign_s3_request(
            content=file_content,
            api_base=api_base,
            optional_params=request_params,
            s3_encryption_key_id=resolve_s3_encryption_key_id(
                litellm_params=litellm_params,
                optional_params=request_params,
            ),
        )

        litellm_params["upload_url"] = api_base
        upload_content_length: Final = len(file_content.encode("utf-8"))
        litellm_params[UPLOAD_CONTENT_LENGTH_PARAM] = upload_content_length  # rebind-ok: same handoff as upload_url

        # Return a dict that tells the HTTP handler exactly what to do
        return {
            "method": "PUT",
            "url": api_base,
            "headers": signed_headers,
            "data": signed_body or file_content,
        }

    def _sign_s3_request(
        self,
        content: str,
        api_base: str,
        optional_params: dict,
        s3_encryption_key_id: str | None = None,
    ) -> tuple[dict, str]:
        """
        Sign S3 PUT request using the same proven logic as S3Logger.
        Reuses the exact pattern from litellm/integrations/s3_v2.py
        """
        try:
            import hashlib

            import requests
            from botocore.auth import S3SigV4Auth
            from botocore.awsrequest import AWSRequest
        except ImportError:
            raise ImportError("Missing boto3 to call bedrock. Run 'pip install boto3'.")

        # Get AWS credentials using existing methods
        aws_region_name: Final = self._get_aws_region_name(optional_params=optional_params, model="")
        credentials: Final = self.get_credentials(
            aws_access_key_id=optional_params.get("aws_access_key_id"),
            aws_secret_access_key=optional_params.get("aws_secret_access_key"),
            aws_session_token=optional_params.get("aws_session_token"),
            aws_region_name=aws_region_name,
            aws_session_name=optional_params.get("aws_session_name"),
            aws_profile_name=optional_params.get("aws_profile_name"),
            aws_role_name=optional_params.get("aws_role_name"),
            aws_web_identity_token=optional_params.get("aws_web_identity_token"),
            aws_sts_endpoint=optional_params.get("aws_sts_endpoint"),
            aws_external_id=optional_params.get("aws_external_id"),
        )

        # Calculate SHA256 hash of the content (REQUIRED for S3)
        content_hash: Final = hashlib.sha256(content.encode("utf-8")).hexdigest()

        # Prepare headers with required S3 headers (same as s3_v2.py)
        sse_headers: Final = (
            MappingProxyType(
                {
                    "x-amz-server-side-encryption": "aws:kms",
                    "x-amz-server-side-encryption-aws-kms-key-id": s3_encryption_key_id,
                }
            )
            if s3_encryption_key_id
            else MappingProxyType({})
        )
        request_headers: Final = MappingProxyType(
            {
                "Content-Type": "application/json",  # JSONL files are JSON content
                "x-amz-content-sha256": content_hash,  # REQUIRED by S3
                "Content-Language": "en",
                "Cache-Control": "private, immutable, max-age=31536000, s-maxage=0",
                **sse_headers,
            }
        )

        # Use requests.Request to prepare the request (same pattern as s3_v2.py)
        req: Final = requests.Request("PUT", api_base, data=content, headers=request_headers)
        prepped: Final = req.prepare()

        # Sign the request with S3 service
        aws_request: Final = AWSRequest(
            method=prepped.method,
            url=prepped.url,
            data=prepped.body,
            headers=prepped.headers,
        )

        # Get region name for non-LLM API calls (same as s3_v2.py)
        signing_region: Final = self.get_aws_region_name_for_non_llm_api_calls(aws_region_name=aws_region_name)

        S3SigV4Auth(credentials, "s3", signing_region).add_auth(aws_request)

        # Return signed headers and body
        signed_body = aws_request.body
        if isinstance(signed_body, bytes):
            signed_body = signed_body.decode("utf-8")
        elif signed_body is None:
            signed_body = content  # Fallback to original content

        return dict(aws_request.headers), signed_body

    def _convert_https_url_to_s3_uri(self, https_url: str) -> tuple[str, str]:
        """
        Convert HTTPS S3 URL to s3:// URI format.

        Args:
            https_url: HTTPS S3 URL (e.g., "https://s3.us-west-2.amazonaws.com/bucket/key")

        Returns:
            Tuple of (s3_uri, filename)

        Example:
            Input: "https://s3.us-west-2.amazonaws.com/litellm-proxy/file.jsonl"
            Output: ("s3://litellm-proxy/file.jsonl", "file.jsonl")
        """
        import re

        # Match HTTPS S3 URL patterns
        # Pattern 1: https://s3.region.amazonaws.com/bucket/key
        # Pattern 2: https://bucket.s3.region.amazonaws.com/key

        pattern1: Final = r"https://s3\.([^.]+)\.amazonaws\.com/([^/]+)/(.+)"
        pattern2: Final = r"https://([^.]+)\.s3\.([^.]+)\.amazonaws\.com/(.+)"

        match1: Final = re.match(pattern1, https_url)
        match2: Final = re.match(pattern2, https_url)

        if match1:
            # Pattern: https://s3.region.amazonaws.com/bucket/key
            region, bucket, key = match1.groups()
            key = unquote(key)
            s3_uri = f"s3://{bucket}/{key}"
        elif match2:
            # Pattern: https://bucket.s3.region.amazonaws.com/key
            bucket, region, key = match2.groups()
            key = unquote(key)
            s3_uri = f"s3://{bucket}/{key}"
        else:
            # Fallback: try to extract bucket and key from URL path
            from urllib.parse import urlparse

            parsed: Final = urlparse(https_url)
            path_parts: Final = parsed.path.lstrip("/").split("/", 1)
            if len(path_parts) >= 2:
                bucket, key = path_parts[0], path_parts[1]
                key = unquote(key)
                s3_uri = f"s3://{bucket}/{key}"
            else:
                raise ValueError(f"Unable to parse S3 URL: {https_url}")

        # Extract filename from key
        filename: Final = key.split("/")[-1] if "/" in key else key

        return s3_uri, filename

    def transform_create_file_response(
        self,
        model: str | None,
        raw_response: Response,
        logging_obj: LiteLLMLoggingObj,
        litellm_params: dict,
    ) -> OpenAIFileObject:
        """
        Transform S3 File upload response into OpenAI-style FileObject
        """
        # Use the actual upload URL that was used for the S3 upload
        upload_url: Final = litellm_params.get("upload_url")
        file_id: str = ""
        filename: str = ""
        if upload_url:
            # Convert HTTPS S3 URL to s3:// URI format
            file_id, filename = self._convert_https_url_to_s3_uri(upload_url)

        return OpenAIFileObject(
            purpose="batch",  # Default purpose for Bedrock files
            id=file_id,
            filename=filename,
            created_at=int(time.time()),  # Current timestamp
            status="uploaded",
            bytes=_uploaded_object_size(litellm_params=litellm_params, raw_response=raw_response),
            object="file",
        )

    def get_error_class(self, error_message: str, status_code: int, headers: dict | Headers) -> BaseLLMException:
        return BedrockError(status_code=status_code, message=error_message, headers=headers)

    def transform_retrieve_file_request(
        self,
        file_id: str,
        optional_params: dict,
        litellm_params: dict,
    ) -> tuple[str, dict]:
        raise NotImplementedError("BedrockFilesConfig does not support file retrieval")

    def transform_retrieve_file_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        litellm_params: dict,
    ) -> OpenAIFileObject:
        raise NotImplementedError("BedrockFilesConfig does not support file retrieval")

    def transform_delete_file_request(
        self,
        file_id: str,
        optional_params: dict,
        litellm_params: dict,
    ) -> tuple[str, dict]:
        raise NotImplementedError("BedrockFilesConfig does not support file deletion")

    def transform_delete_file_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        litellm_params: dict,
    ) -> FileDeleted:
        raise NotImplementedError("BedrockFilesConfig does not support file deletion")

    def transform_list_files_request(
        self,
        purpose: str | None,
        optional_params: dict,
        litellm_params: dict,
    ) -> tuple[str, dict]:
        raise NotImplementedError("BedrockFilesConfig does not support file listing")

    def transform_list_files_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        litellm_params: dict,
    ) -> list[OpenAIFileObject]:
        raise NotImplementedError("BedrockFilesConfig does not support file listing")

    def transform_file_content_request(
        self,
        file_content_request: FileContentRequest,
        optional_params: Mapping[str, object],
        litellm_params: MutableMapping[str, object],
    ) -> tuple[str, dict[str, str]]:
        """
        Build a SigV4-signed S3 GetObject request for a Bedrock batch file.

        Bedrock batch file ids are `s3://bucket/key` URIs (or unified ids
        that decode to one); the bucket and key are validated against the
        server-configured bucket before any request is signed.
        """
        file_id: Final = file_content_request.get("file_id")
        if not file_id:
            raise ValueError("file_id is required for Bedrock file content retrieval")

        s3_uri: Final = extract_s3_uri_from_file_id(file_id)
        bucket_name, object_key = _validate_file_id_against_configured_buckets(
            s3_uri=s3_uri,
            configured_bucket_names=get_configured_s3_bucket_names(litellm_params),
            allow_legacy_cloud_file_ids=should_allow_legacy_cloud_file_ids(litellm_params),
        )

        # The shared file-content handler passes optional_params={}, so AWS
        # credentials/region arrive via litellm_params here (unlike the upload
        # path). s3_region_name wins over aws_region_name, same priority as
        # get_complete_file_url above.
        merged_params: Final[dict[str, object]] = {}
        merged_params.update(litellm_params)
        merged_params.update(optional_params)
        request_params: Final = _BedrockS3RequestParams.model_validate(merged_params)

        region_preference: Final = request_params.s3_region_name or request_params.aws_region_name
        region_params: Final[dict[str, str | None]] = {"aws_region_name": region_preference}
        aws_region_name: Final = self._get_aws_region_name(optional_params=region_params, model="")

        s3_endpoint_url = (
            request_params.s3_endpoint_url or f"https://s3.{aws_region_name}.{get_aws_dns_suffix(aws_region_name)}"
        ).rstrip("/")
        url: Final = f"{s3_endpoint_url}/{bucket_name}/{encode_s3_object_key_for_url(object_key)}"

        litellm_params[S3_SIGNED_GET_HEADERS_PARAM] = self._sign_s3_get_request(
            api_base=url,
            aws_region_name=aws_region_name,
            request_params=request_params,
        )
        return url, {}

    def _sign_s3_get_request(
        self,
        api_base: str,
        aws_region_name: str,
        request_params: _BedrockS3RequestParams,
    ) -> dict[str, str]:
        """
        SigV4-sign an S3 GetObject request, mirroring `_sign_s3_request` (PUT).
        """
        try:
            import hashlib

            from botocore.auth import S3SigV4Auth
            from botocore.awsrequest import AWSRequest
        except ImportError:
            raise ImportError("Missing boto3 to call bedrock. Run 'pip install boto3'.")

        credentials: Final = self.get_credentials(  # any-ok: boto3 Credentials is untyped
            aws_access_key_id=request_params.aws_access_key_id,
            aws_secret_access_key=request_params.aws_secret_access_key,
            aws_session_token=request_params.aws_session_token,
            aws_region_name=aws_region_name,
            aws_session_name=request_params.aws_session_name,
            aws_profile_name=request_params.aws_profile_name,
            aws_role_name=request_params.aws_role_name,
            aws_web_identity_token=request_params.aws_web_identity_token,
            aws_sts_endpoint=request_params.aws_sts_endpoint,
            aws_external_id=request_params.aws_external_id,
        )

        empty_body_hash: Final = hashlib.sha256(b"").hexdigest()
        aws_request: Final = AWSRequest(  # any-ok: botocore AWSRequest is untyped
            method="GET",
            url=api_base,
            headers={"x-amz-content-sha256": empty_body_hash},
        )
        auth: Final = S3SigV4Auth(credentials, "s3", aws_region_name)  # any-ok: botocore untyped
        auth.add_auth(aws_request)  # any-ok: botocore request mutation is untyped
        return dict(aws_request.headers)  # any-ok: botocore headers are untyped

    def transform_file_content_response(
        self,
        raw_response: httpx.Response,
        logging_obj: LiteLLMLoggingObj,
        litellm_params: dict,
    ) -> HttpxBinaryResponseContent:
        if raw_response.status_code >= 400:
            raise BedrockError(
                status_code=raw_response.status_code,
                message=raw_response.text,
                headers=raw_response.headers,
            )
        return HttpxBinaryResponseContent(response=raw_response)


class BedrockJsonlFilesTransformation:
    """
    Transforms OpenAI /v1/files/* requests to Bedrock S3 file uploads for batch processing
    """

    def transform_openai_file_content_to_bedrock_file_content(
        self, openai_file_content: FileTypes | None = None
    ) -> tuple[str, str]:
        """
        Transforms OpenAI FileContentRequest to Bedrock S3 file format
        """

        if openai_file_content is None:
            raise ValueError("contents of file are None")
        # Read the content of the file
        file_content: Final = self._get_content_from_openai_file(openai_file_content)

        # Split into lines and parse each line as JSON
        openai_jsonl_content: Final = [json.loads(line) for line in file_content.splitlines() if line.strip()]
        bedrock_jsonl_content = self._transform_openai_jsonl_content_to_bedrock_jsonl_content(openai_jsonl_content)
        bedrock_jsonl_string: Final = "\n".join(json.dumps(item) for item in bedrock_jsonl_content)
        object_name: Final = self._get_s3_object_name(openai_jsonl_content=openai_jsonl_content)
        return bedrock_jsonl_string, object_name

    def _transform_openai_jsonl_content_to_bedrock_jsonl_content(
        self, openai_jsonl_content: Sequence[_OpenAIBatchRecord]
    ):
        """
        Delegate to the main BedrockFilesConfig transformation method
        """
        config: Final = BedrockFilesConfig()
        return config._transform_openai_jsonl_content_to_bedrock_jsonl_content(openai_jsonl_content)

    def _get_s3_object_name(
        self,
        openai_jsonl_content: Sequence[_OpenAIBatchRecord],
    ) -> str:
        """
        Gets a unique S3 object name for the Bedrock batch processing job

        named as: litellm-bedrock-files-{model}-{uuid}
        """
        _model = openai_jsonl_content[0].get("body", {}).get("model", "")
        # Remove bedrock/ prefix if present
        _model = _model.removeprefix("bedrock/")
        safe_model: Final = sanitize_cloud_object_component(_model.replace(":", "-"), fallback="model")
        object_name: Final = f"{BEDROCK_MANAGED_S3_BATCH_PREFIX}{safe_model}-{uuid.uuid4()}.jsonl"
        return object_name

    def _get_content_from_openai_file(self, openai_file_content: FileTypes) -> str:
        """
        Helper to extract content from various OpenAI file types and return as string.

        Handles:
        - Direct content (str, bytes, IO[bytes])
        - Tuple formats: (filename, content, [content_type], [headers])
        - PathLike objects
        """
        content: str | bytes = b""
        # Extract file content from tuple if necessary
        if isinstance(openai_file_content, tuple):
            # Take the second element which is always the file content
            file_content = openai_file_content[1]
        else:
            file_content = openai_file_content

        # Handle different file content types
        if isinstance(file_content, str):
            # String content can be used directly
            content = file_content
        elif isinstance(file_content, bytes):
            # Bytes content can be decoded
            content = file_content
        elif isinstance(file_content, PathLike):  # PathLike
            with open(str(file_content), "rb") as f:
                content = f.read()
        elif hasattr(file_content, "read"):  # IO[bytes]
            # File-like objects need to be read
            content = file_content.read()

        # Ensure content is string
        if isinstance(content, bytes):
            content = content.decode("utf-8")

        return content

    def transform_s3_bucket_response_to_openai_file_object(
        self, create_file_data: CreateFileRequest, s3_upload_response: _S3UploadResponse
    ) -> OpenAIFileObject:
        """
        Transforms S3 Bucket upload file response to OpenAI FileObject
        """
        # S3 response typically contains ETag, key, etc.
        object_key: Final = s3_upload_response.get("Key", "")
        bucket_name: Final = s3_upload_response.get("Bucket", "")

        # Extract filename from object key
        filename: Final = object_key.split("/")[-1] if "/" in object_key else object_key

        return OpenAIFileObject(
            purpose=create_file_data.get("purpose", "batch"),
            id=f"s3://{bucket_name}/{object_key}",
            filename=filename,
            created_at=int(time.time()),  # Current timestamp
            status="uploaded",
            bytes=s3_upload_response.get("ContentLength", 0),
            object="file",
        )
