Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion astrbot/core/provider/sources/openai_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from astrbot.core.provider.entities import LLMResponse, TokenUsage, ToolCallsResult
from astrbot.core.utils.media_utils import (
describe_media_ref,
normalize_image_for_provider,
resolve_media_ref_to_base64_data,
)
from astrbot.core.utils.network_utils import (
Expand Down Expand Up @@ -187,7 +188,8 @@ async def _image_ref_to_data_url(
media_type="image",
strict=mode == "strict",
)
return image_data.to_data_url() if image_data else None
normalized = normalize_image_for_provider(image_data)

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

issue (bug_risk): An unsupported or incorrectly labelled image whose resolved bytes are not a valid Pillow image causes PILImage.open to raise, and _image_ref_to_data_url does not catch that exception. This breaks the safe image path by propagating the decoding error instead of returning None and letting _resolve_image_part ignore the invalid attachment.

Triggers: When a safe-mode image reference resolves successfully but contains malformed bytes or a format unsupported by Pillow.

Suggested fix: Catch Pillow decoding/conversion errors in the safe path, or make normalize_image_for_provider return None for invalid image bytes.

return normalized.to_data_url() if normalized else None

async def _resolve_image_part(
self,
Expand Down
73 changes: 73 additions & 0 deletions astrbot/core/utils/media_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -146,6 +146,79 @@ def to_data_url(self) -> str:
return f"data:{self.mime_type};base64,{self.base64_data}"


IMAGE_PROVIDER_SUPPORTED_MIME_TYPES = frozenset(
{
"image/gif",
"image/jpeg",
"image/png",
}
)


def normalize_image_for_provider(
image_data: ResolvedMediaData | None,
supported_mimes: set[str] | frozenset[str] | None = None,
) -> ResolvedMediaData | None:
"""Normalize image bytes to a MIME type accepted by a vision provider.

Args:
image_data: Resolved image bytes and metadata.
supported_mimes: MIME types accepted by the provider. Defaults to JPEG,
PNG, and GIF.

Returns:
Validated image data with corrected MIME metadata, or converted image data
when the source format is not supported.
"""
if image_data is None:
return None

supported = supported_mimes or IMAGE_PROVIDER_SUPPORTED_MIME_TYPES

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

nitpick (bug_risk): An explicitly supplied empty supported_mimes set is treated as if no provider-specific set was supplied, so the function silently falls back to JPEG, PNG, and GIF instead of honoring the empty accepted-format set. The function can therefore return or create a format that the caller explicitly declared unsupported.

Triggers: When a caller passes supported_mimes=set() to indicate that no formats are accepted.

Suggested fix: Use supported_mimes if supported_mimes is not None else IMAGE_PROVIDER_SUPPORTED_MIME_TYPES.

Suggested change
supported = supported_mimes or IMAGE_PROVIDER_SUPPORTED_MIME_TYPES
supported = supported_mimes if supported_mimes is not None else IMAGE_PROVIDER_SUPPORTED_MIME_TYPES

if image_data.mime_type in supported:
return image_data

raw = image_data.to_bytes()
with PILImage.open(io.BytesIO(raw)) as image:
actual_mime = {
"GIF": "image/gif",
"JPEG": "image/jpeg",
"PNG": "image/png",
"WEBP": "image/webp",
}.get(str(image.format or "").upper())
if actual_mime in supported:
return ResolvedMediaData(
base64_data=image_data.base64_data,
mime_type=actual_mime,
format=image_data.format,
)

has_alpha = image.mode in {"RGBA", "LA", "PA"} or "transparency" in image.info
if has_alpha and "image/png" in supported:
output_format = "PNG"
output_mime = "image/png"
converted = image.convert("RGBA")
elif "image/jpeg" in supported:
output_format = "JPEG"
output_mime = "image/jpeg"
converted = image.convert("RGB")
elif "image/png" in supported:
output_format = "PNG"
output_mime = "image/png"
converted = image.convert("RGB")
else:
return None

try:
output = io.BytesIO()
converted.save(output, format=output_format)
return ResolvedMediaData(
base64_data=base64.b64encode(output.getvalue()).decode("utf-8"),
mime_type=output_mime,
)
finally:
converted.close()


@dataclass(slots=True)
class _LocalMediaFile:
path: Path
Expand Down
35 changes: 35 additions & 0 deletions tests/test_media_utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -228,6 +228,41 @@ def test_detect_image_mime_type_sniffs_common_headers():
)


def test_normalize_image_for_provider_converts_webp_bytes():
from PIL import Image as PILImage

image_buffer = BytesIO()
PILImage.new("RGB", (2, 2), (255, 0, 0)).save(image_buffer, format="WEBP")
image_data = media_utils.ResolvedMediaData(
base64_data=base64.b64encode(image_buffer.getvalue()).decode("ascii"),
mime_type="image/webp",
)

normalized = media_utils.normalize_image_for_provider(image_data)

assert normalized is not None
assert normalized.mime_type == "image/jpeg"
with PILImage.open(BytesIO(normalized.to_bytes())) as image:
assert image.format == "JPEG"


def test_normalize_image_for_provider_preserves_png_bytes():
from PIL import Image as PILImage

image_buffer = BytesIO()
PILImage.new("RGB", (2, 2), (255, 0, 0)).save(image_buffer, format="PNG")
image_data = media_utils.ResolvedMediaData(
base64_data=base64.b64encode(image_buffer.getvalue()).decode("ascii"),
mime_type="image/png",
)

normalized = media_utils.normalize_image_for_provider(image_data)

assert normalized is not None
assert normalized.mime_type == "image/png"
assert normalized.base64_data == image_data.base64_data


def test_detect_image_mime_type_returns_default_for_unknown_input():
"""Unknown or empty headers fall back to the provided default."""
assert (
Expand Down
24 changes: 24 additions & 0 deletions tests/test_openai_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -1164,6 +1164,30 @@ async def test_resolve_image_part_preserves_base64_png_mime_type():
await provider.terminate()


@pytest.mark.asyncio
async def test_resolve_image_part_normalizes_webp_to_jpeg():
provider = _make_provider()
try:
image_buffer = BytesIO()
PILImage.new("RGB", (2, 2), (255, 0, 0)).save(
image_buffer,
format="WEBP",
)
image_base64 = base64.b64encode(image_buffer.getvalue()).decode("ascii")

image_part = await provider._resolve_image_part(f"base64://{image_base64}")

assert image_part is not None
image_url = image_part["image_url"]["url"]
assert image_url.startswith("data:image/jpeg;base64,")
with PILImage.open(
BytesIO(base64.b64decode(image_url.split(",", 1)[1]))
) as image:
assert image.format == "JPEG"
finally:
await provider.terminate()


@pytest.mark.asyncio
async def test_prepare_chat_payload_materializes_context_localhost_file_uri_image_urls(
tmp_path,
Expand Down