def threaded_download_and_preprocess_content(allocated_docs: List[dict],
                                             media_repo: dict,
                                             tensor_fields: List[str],
                                             device: str = None,
                                             media_field_types_mapping: Optional[Dict[str, FieldType]] = None,
                                             media_download_headers: Optional[Dict] = None,
                                             metric_obj: Optional[RequestMetrics] = None,
                                             preprocessors: Optional[Preprocessors] = None,
                                             marqo_index_type: Optional[IndexType] = None,
                                             marqo_index_model: Optional[Model] = None,
                                             audio_preprocessing: Optional[AudioPreProcessing] = None,
                                             video_preprocessing: Optional[VideoPreProcessing] = None,
                                             force_download: bool = False) -> None:
    """A thread calls this function to download images for its allocated documents

    This should be called only if treat URLs as images is True.

    Args:
        allocated_docs: docs with images to be downloaded by this thread,
        media_repo: dictionary that will be mutated by this thread. It will add media
            as values and the URLs as keys
        tensor_fields: A tuple of tensor_fields. Images will be downloaded for these fields only.
        media_download_headers: A dict of headers for image download. Can be used
            to authenticate image downloads
        force_download: If True, skip the _is_image check and download the fields as images.
    Side Effects:
        Adds members to the image_repo dict. Each key is a string which is identified as a URL.
        Each value is either a PIL image, or UnidentifiedImageError, if there were any errors encountered retrieving
        the image.
        For example:
        {
            'https://google.com/my_dog.png': UnidentifiedImageError, # error because such an image doesn't exist
            'https://raw.githubusercontent.com/marqo-ai/marqo-api-tests/mainline/assets/ai_hippo_realistic.png': <PIL image>
        }
    Returns:
        None

    """
    # Determine index type
    is_structured_index = marqo_index_type == IndexType.Structured
    is_unstructured_index = marqo_index_type in [IndexType.Unstructured, IndexType.SemiStructured]
    # Generate pseudo-unique ID for thread metrics.
    _id = f'image_download.{threading.get_ident()}'
    TIMEOUT_SECONDS = 3
    if metric_obj is None:  # Occurs predominately in testing.
        metric_obj = RequestMetricsStore.for_request()
        RequestMetricsStore.set_in_request(metrics=metric_obj)

    # For backward compatibility, we should accept both supportedModalities and supported_modalities
    if marqo_index_model.properties.get("supported_modalities") and marqo_index_model.properties.get(
            "supportedModalities"):
        raise InvalidModelPropertiesError(
            "Model properties must have either 'supported_modalities' or 'supportedModalities', not both"
        )
    else:
        supported_modalities = marqo_index_model.properties.get('supported_modalities') or \
                               marqo_index_model.properties.get('supportedModalities') or []

    with metric_obj.time(f"{_id}.thread_time"):
        for doc in allocated_docs:
            for field in list(doc):
                if field not in tensor_fields:
                    continue
                if isinstance(doc[field], str) or force_download:
                    try:
                        inferred_modality = infer_modality(doc[field], media_download_headers)
                    except MediaDownloadError as e:
                        if is_structured_index and media_field_types_mapping[field] == FieldType.ImagePointer:
                            # Continue processing for structured indexes with image fields
                            inferred_modality = Modality.IMAGE
                        else:
                            media_repo[doc[field]] = MediaDownloadError(
                                f"Error inferring modality of media file {doc[field]}: {e}")
                            continue

                    if (inferred_modality == Modality.IMAGE and is_unstructured_index) or (
                            is_structured_index and media_field_types_mapping[
                        field] == FieldType.ImagePointer):  # Don't use infer modality in structured image pointers

                        if marqo_index_model.properties.get('type') in [ModelType.LanguageBind] \
                                and supported_modalities is not None \
                                and Modality.IMAGE not in supported_modalities:
                            media_repo[doc[field]] = UnsupportedModalityError(
                                f"Model {marqo_index_model.name} does not support {inferred_modality}")
                            continue

                        # Existing logic
                        if doc[field] in media_repo:
                            continue

                        try:
                            media_repo[doc[field]] = clip_utils.load_image_from_path(doc[field], media_download_headers,
                                                                                     timeout_ms=int(
                                                                                         TIMEOUT_SECONDS * 1000),
                                                                                     metrics_obj=metric_obj)
                        except PIL.UnidentifiedImageError as e:
                            media_repo[doc[field]] = e
                            metric_obj.increment_counter(f"{doc.get(field, '')}.UnidentifiedImageError")
                            continue
                        # preprocess image to tensor
                        if preprocessors is not None and preprocessors.image is not None:
                            if not device or not isinstance(device, str):
                                raise ValueError("Device must be provided for preprocessing images")
                            try:
                                preprocessed_results = preprocessors.image(media_repo[doc[field]])
                                if isinstance(preprocessed_results, torch.Tensor):
                                    media_repo[doc[field]] = preprocessed_results.to(device)
                                elif isinstance(preprocessed_results, dict):
                                    media_repo[doc[field]] = {k: v.to(device) for k, v in preprocessed_results.items()}
                                # media_repo[doc[field]] = preprocessors.image(media_repo[doc[field]]).to(device)
                            except OSError as e:
                                if "image file is truncated" in str(e):
                                    media_repo[doc[field]] = e
                                    metric_obj.increment_counter(f"{doc.get(field, '')}.OSError")
                                    continue
                                else:
                                    raise e

                    elif (inferred_modality in [Modality.VIDEO, Modality.AUDIO] and is_unstructured_index) or (
                            is_structured_index and media_field_types_mapping[field] in [FieldType.AudioPointer,
                                                                                         FieldType.VideoPointer] and inferred_modality in [
                                Modality.AUDIO, Modality.VIDEO]):
                        if marqo_index_model.properties.get('type') not in [ModelType.LanguageBind]:
                            media_repo[doc[field]] = UnsupportedModalityError(
                                f"Model {marqo_index_model.name} does not support {inferred_modality}")
                            continue

                        if inferred_modality not in supported_modalities:
                            media_repo[doc[field]] = UnsupportedModalityError(
                                f"Model {marqo_index_model.name} does not support {inferred_modality}")
                            continue

                        if is_structured_index:
                            if inferred_modality is Modality.VIDEO and media_field_types_mapping[
                                field] is FieldType.AudioPointer:
                                media_repo[doc[field]] = MediaMismatchError(
                                    f"Invalid audio file. Error processing media file {doc}, detected as video, but field type is not VideoPointer")
                                continue

                            if inferred_modality is Modality.AUDIO and media_field_types_mapping[
                                field] is FieldType.VideoPointer:
                                media_repo[doc[field]] = MediaMismatchError(
                                    f"Invalid video file. Error processing media file {doc}, detected as audio, but field type is not AudioPointer")
                                continue

                        try:
                            processed_chunks = download_and_chunk_media(
                                url=doc[field], device=device, modality=inferred_modality,
                                preprocessors=preprocessors, audio_preprocessing=audio_preprocessing,
                                video_preprocessing=video_preprocessing, media_download_headers=media_download_headers
                            )
                            media_repo[doc[field]] = processed_chunks
                        except (ffmpeg.Error, S2InferenceError) as e:
                            logger.error(f"Error processing {inferred_modality} file: {str(e)}")
                            media_repo[doc[field]] = S2InferenceError(
                                f"Error processing {inferred_modality} file: {str(e)}")

                    elif inferred_modality is Modality.TEXT and is_structured_index and media_field_types_mapping[
                        field] in [FieldType.AudioPointer, FieldType.VideoPointer, FieldType.ImagePointer]:
                        media_repo[doc[field]] = S2InferenceError(
                            f"Error processing media file {doc}, detected as text, expected a {media_field_types_mapping[field]} pointer")
                    else:
                        pass
                else:
                    raise InternalError(f"Invalid field type for {field} to be added in media repo. Must be a string "
                                        f"but {type(field)}.")
A thread calls this function to download images for its allocated documents

    This should be called only if treat URLs as images is True.

    Args:
        allocated_docs: docs with images to be downloaded by this thread,
        media_repo: dictionary that will be mutated by this thread. It will add media
            as values and the URLs as keys
        tensor_fields: A tuple of tensor_fields. Images will be downloaded for these fields only.
        media_download_headers: A dict of headers for image download. Can be used
            to authenticate image downloads
        force_download: If True, skip the _is_image check and download the fields as images.
    Side Effects:
        Adds members to the image_repo dict. Each key is a string which is identified as a URL.
        Each value is either a PIL image, or UnidentifiedImageError, if there were any errors encountered retrieving
        the image.
        For example:
        {
            'https://google.com/my_dog.png': UnidentifiedImageError, # error because such an image doesn't exist
            'https://raw.githubusercontent.com/marqo-ai/marqo-api-tests/mainline/assets/ai_hippo_realistic.png': <PIL image>
        }
    Returns:
        None

    
src/marqo/tensor_search/add_docs.py