Repository: meta-llama/llama-stack
File: llama_stack/providers/utils/memory/vector_store.py (Python)

Code before:
        data = data.encode(encoding)
    return content_from_data_and_mime_type(data, parts["mimetype"], parts.get("encoding", None))


def content_from_data_and_mime_type(data: bytes | str, mime_type: str | None, encoding: str | None = None) -> str:
    if isinstance(data, bytes):
        if not encoding:
            import chardet

            detected = chardet.detect(data)
            encoding = detected["encoding"]

    mime_category = mime_type.split("/")[0] if mime_type else None
    if mime_category == "text":
        # For text-based files (including CSV, MD)
        try:
            return data.decode(encoding)
        except Exception as e:
            # try to decode with utf-8
            if encoding != "utf-8":
                try:
                    log.warning(f"Decoding data as utf-8. encoding: {encoding}")
                    return data.decode("utf-8")
                except Exception:
                    log.error(f"Could not decode data as utf-8. {e}")
                    return ""
            else:
                # UTF-8 was the primary encoding and it failed, return empty string
                log.error(f"Could not decode data as {encoding}. {e}")
                return ""

    elif mime_type == "application/pdf":
        return parse_pdf(data)

    else:
        log.error("Could not extract content from data_url properly.")
        return ""


def concat_interleaved_content(content: list[InterleavedContent]) -> InterleavedContent:
    """concatenate interleaved content into a single list. ensure that 'str's are converted to TextContentItem when in a list"""

    ret = []

    def _process(c):
        if isinstance(c, str):
            ret.append(TextContentItem(text=c))
        elif isinstance(c, list):
            for item in c:
                _process(item)
        else:

Review comment (suggestion):
avoid swallowing exceptions. the error log is not visible to users requesting the decode. uses of the method do not check the return value but they do catch exceptions.

Code after:
        data = data.encode(encoding)
    return content_from_data_and_mime_type(data, parts["mimetype"], parts.get("encoding", None))


def content_from_data_and_mime_type(data: bytes | str, mime_type: str | None, encoding: str | None = None) -> str:
    if isinstance(data, bytes):
        if not encoding:
            import chardet

            detected = chardet.detect(data)
            encoding = detected["encoding"]

    mime_category = mime_type.split("/")[0] if mime_type else None
    if mime_category == "text":
        # For text-based files (including CSV, MD)
        encodings_to_try = [encoding]
        if encoding != "utf-8":
            encodings_to_try.append("utf-8")
        first_exception = None
        for encoding in encodings_to_try:
            try:
                return data.decode(encoding)
            except UnicodeDecodeError as e:
                if first_exception is None:
                    first_exception = e
                log.warning(f"Decoding failed with {encoding}: {e}")
        # raise the origional exception, if we got here there was at least 1 exception
        log.error(f"Could not decode data as any of {encodings_to_try}")
        raise first_exception

    elif mime_type == "application/pdf":
        return parse_pdf(data)

    else:
        log.error("Could not extract content from data_url properly.")
        return ""


def concat_interleaved_content(content: list[InterleavedContent]) -> InterleavedContent:
    """concatenate interleaved content into a single list. ensure that 'str's are converted to TextContentItem when in a list"""

    ret = []

    def _process(c):
        if isinstance(c, str):
            ret.append(TextContentItem(text=c))
        elif isinstance(c, list):
            for item in c:
                _process(item)
        else:
            ret.append(c)