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

Code before:

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)
        return data.decode(encoding)

    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)

    for c in content:
        _process(c)

    return ret


async def content_from_doc(doc: RAGDocument) -> str:
    if isinstance(doc.content, URL):
        if doc.content.uri.startswith("data:"):
            return content_from_data(doc.content.uri)
        async with httpx.AsyncClient() as client:
            r = await client.get(doc.content.uri)
        if doc.mime_type == "application/pdf":
            return parse_pdf(r.content)
        return r.text

Review comment (refactor):
Can we simplify like so?

```python
for enc in {encoding, "utf-8"}:
    try:
        return data.decode(enc)
    except UnicodeDecodeError:
        continue
log.error(f"Could not decode data with {encoding} or utf-8")
raise UnicodeDecodeError(f"Could not decode data with {encoding} or utf-8")
```

Code after:

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)

    for c in content:
        _process(c)