feat:Adapt outgoing images to the formats supported by the provider, with support for animation strategies and local conversion caching.

[piexian]

各家模型提供商对图片格式支持不一（例如 xAI 官方仅支持 JPEG/PNG，BMP/HEIC 几乎到处被拒），动图 GIF 要么直接报错要么浪费 token。本 PR 在发送前将外发图片自动适配为目标提供商官方支持的格式，并为动图提供可配置的处理策略。
关联issue #9295

Modifications / 改动点

• astrbot/core/provider/provider.py：新增 Provider.supported_image_formats 类属性，以及resolve_allowed_image_formats()（优先级：实例级 image_formats 配置 > 类声明 > 保守回退 jpeg/png）和 get_animated_image_strategy()（first_frame / multi_frame，最大帧数钳制在 1–16）。

• astrbot/core/utils/media_utils.py：新增 resolve_image_ref_to_images() —— 不支持的静图经 Pillow 转换（透明图 → PNG，不透明图 → JPEG）；动图按策略取首帧或均匀抽 N 帧。转换结果按内容寻址缓存到 AstrBot 临时目录（由现有 TempDirCleaner 统一管理），写入全程原子化（临时文件 + rename；帧集通过目录级原子 rename 发布），崩溃或并发不会产生残缺缓存。

• 提供商适配器：为 OpenAI（png/jpeg/webp/gif）、Anthropic（jpeg/png/gif/webp）、Gemini（png/jpeg/webp/heic/heif）、xAI 和智谱（jpeg/png）声明官方格式集；聚合网关（OpenRouter、AIHubMix、SSYCloud）显式置回 None，避免误继承 OpenAI 的格式集而走保守默认值。Anthropic 历史上下文中不支持/无法识别格式的图片现在会跳过并告警，不再被误标为 image/jpeg。

• astrbot/core/config/default.py：新增提供商配置项 image_formats（复选列表，* 表示不限制）、animated_image_strategy（first_frame / multi_frame）、animated_image_max_frames（默认 4）。image_formats 条目全部无效时会输出告警日志再回退默认值。

• 测试：新增 tests/unit/test_provider_image_formats.py（配置优先级、聚合器继承、策略钳制、端到端 provider 组装），并在 tests/test_media_utils.py / tests/test_openai_source.py 补充格式转换、透明通道、均匀抽帧、缓存命中/失效、残留 staging 目录、不可解码透传等用例。

• This is NOT a breaking change. / 这不是一个破坏性变更。

Screenshots or Test Results / 运行截图或测试结果

pytest tests/unit/test_provider_image_formats.py tests/test_media_utils.py tests/test_openai_source.py
140 passed in 6.21s

---

Checklist / 检查清单

• 😊 If there are new features added in the PR, I have discussed it with the authors through issues/emails, etc.
  / 如果 PR 中有新加入的功能，已经通过 Issue / 邮件等方式和作者讨论过。

• 👀 My changes have been well-tested, and "Verification Steps" and "Screenshots" have been provided above.
  / 我的更改经过了良好的测试，并已在上方提供了“验证步骤”和“运行截图”。

• 🤓 I have ensured that no new dependencies are introduced, OR if new dependencies are introduced, they have been added to the appropriate locations in requirements.txt and pyproject.toml.
  / 我确保没有引入新依赖库，或者引入了新依赖库的同时将其添加到 requirements.txt 和 pyproject.toml 文件相应位置。

• 😮 My changes do not introduce malicious code.
  / 我的更改没有引入恶意代码。

Summary by Sourcery

Adapt image handling across providers to respect each API’s supported formats, with configurable animated image strategies and shared conversion caching.

New Features:

• Introduce provider-level configuration for allowed image formats and animated image handling, including strategy selection and maximum frame count.
• Add support for resolving image references into one or multiple provider-ready images, with automatic format adaptation and animated frame extraction.

Enhancements:

• Define official image format sets for major vision providers and ensure aggregator gateways use conservative defaults instead of inheriting vendor-specific formats.
• Improve media resolution to convert incompatible still images, preserve transparency where possible, and cache converted outputs under the AstrBot temp directory.
• Update Anthropic, OpenAI, Gemini, xAI, Zhipu, and generic provider context assembly to use the new image adaptation pipeline and avoid mislabeling or rejecting unsupported formats.

Tests:

• Add unit tests covering provider image format resolution, aggregation behavior, animated strategy clamping, and end-to-end context assembly with different providers.
• Extend media utility and OpenAI source tests to verify image format conversion, alpha-channel handling, multi-frame extraction, cache hits and recovery from partial caches, and strict vs non-strict decoding behavior.

[sourcery-ai]

Hey - I've found 3 issues, and left some high level feedback:

• In resolve_image_ref_to_images, the logger.info on every animated-image extraction may be quite noisy in production; consider downgrading this to debug or adding rate limiting if you expect frequent image usage.
• The animated frame clamping is done both in Provider.get_animated_image_strategy and again inside resolve_image_ref_to_images; you could simplify by trusting the provider-level clamping and avoiding the second clamp to reduce duplicated logic.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `resolve_image_ref_to_images`, the `logger.info` on every animated-image extraction may be quite noisy in production; consider downgrading this to `debug` or adding rate limiting if you expect frequent image usage.
- The animated frame clamping is done both in `Provider.get_animated_image_strategy` and again inside `resolve_image_ref_to_images`; you could simplify by trusting the provider-level clamping and avoiding the second clamp to reduce duplicated logic.

## Individual Comments

### Comment 1
<location path="astrbot/core/utils/media_utils.py" line_range="1151" />
<code_context>
+    )
+
+
+async def resolve_image_ref_to_images(
+    image_ref: MediaRefStr,
+    *,
</code_context>
<issue_to_address>
**issue (complexity):** Consider merging the frame-saving helpers, simplifying the frame index calculation, and optionally introducing an image profile object to make the image-resolution pipeline easier to follow.

- The `resolve_image_ref_to_images` pipeline is clear but dense, and some of the helper decomposition adds indirection without much gain. You can reduce cognitive load without losing any features by consolidating a couple of helpers and simplifying the frame-selection logic.

### 1. Merge `_save_current_image_frame` and `_save_image_frame_atomic`

You currently have two tightly-coupled helpers: one for “how to save a frame” and one for “save atomically”. You can merge them into a single helper that encapsulates both the Pillow conversion and the atomic write, and use it everywhere.

```python
def _save_image_frame(
    image: PILImage.Image,
    target_mime_type: str,
    output_path: Path,
) -> None:
    """Convert & save the current frame atomically to avoid partial cache writes."""
    working: PILImage.Image | None = None
    fd, tmp_name = tempfile.mkstemp(dir=output_path.parent, suffix=".tmp")
    os.close(fd)
    tmp_path = Path(tmp_name)
    try:
        frame = image
        if target_mime_type == "image/jpeg" and image.mode != "RGB":
            working = image.convert("RGB")
            frame = working
        elif (
            target_mime_type == "image/png"
            and image.mode == "P"
            and "transparency" in image.info
        ):
            working = image.convert("RGBA")
            frame = working

        save_kwargs: dict[str, int] = {}
        if target_mime_type == "image/jpeg":
            save_kwargs = {
                "quality": IMAGE_COMPRESS_DEFAULT_QUALITY,
                "subsampling": 0,
            }

        frame.save(tmp_path, _MIME_PIL_FORMAT[target_mime_type], **save_kwargs)
        os.replace(tmp_path, output_path)
    finally:
        if working is not None:
            working.close()
        tmp_path.unlink(missing_ok=True)
```

Call sites become simpler:

```python
# _convert_image_bytes_sync
with PILImage.open(io.BytesIO(source_bytes)) as image:
    if frame_index is not None:
        image.seek(frame_index)
    _save_image_frame(image, target_mime_type, output_path)

# _extract_animation_frames_sync
with PILImage.open(io.BytesIO(source_bytes)) as image:
    total_frames = getattr(image, "n_frames", 1)
    for out_index, frame_index in enumerate(
        _even_frame_indices(total_frames, max_frames)
    ):
        frame_path = staging_dir / f"f{out_index}{suffix}"
        image.seek(frame_index)
        _save_image_frame(image, target_mime_type, frame_path)
```

This keeps all existing behavior (including atomic writes) but removes a layer of indirection.

### 2. Simplify `_even_frame_indices`

The current implementation uses a set + `round` + `sorted`, which is correct but harder to reason about quickly. You can keep the “evenly spaced” behavior with a simpler, more linear formulation:

```python
def _even_frame_indices(total_frames: int, max_frames: int) -> list[int]:
    """Pick up to `max_frames` frame indices evenly spaced over the animation."""
    count = min(max_frames, total_frames)
    if count <= 1:
        return [0]

    # Step across [0, total_frames - 1] with (count - 1) intervals.
    step = (total_frames - 1) / (count - 1)
    indices: list[int] = []
    last = -1
    for i in range(count):
        idx = int(round(i * step))
        if idx != last:
            indices.append(idx)
            last = idx
    return indices
```

This preserves the intent (“even spread, no duplicates, up to max_frames”) with a straightforward loop that is easier to follow at a glance.

### 3. Factor image inspection into a small domain object (optional but helps `resolve_image_ref_to_images`)

You can reduce branching inside `resolve_image_ref_to_images` by introducing a tiny `ImageProfile` and a helper that encapsulates the inspection and classification. That keeps the function’s high-level flow more obvious:

```python
@dataclass(frozen=True)
class ImageProfile:
    has_alpha: bool
    frame_count: int
    detected_mime: str | None

def _profile_image(
    image_bytes: bytes,
    detected_mime: str | None,
) -> ImageProfile:
    has_alpha, frame_count = _inspect_image(image_bytes)
    return ImageProfile(
        has_alpha=has_alpha,
        frame_count=frame_count,
        detected_mime=detected_mime,
    )
```

Then in `resolve_image_ref_to_images`:

```python
image_bytes = media_data.to_bytes()
unrestricted = allowed_mime_types is None or "*" in allowed_mime_types

try:
    profile = await asyncio.to_thread(
        _profile_image,
        image_bytes,
        media_data.mime_type,
    )
except Exception as exc:
    # existing “Pillow cannot decode” fallback logic here...

if profile.frame_count > 1 and (
    not unrestricted or animated_strategy == ANIMATED_STRATEGY_MULTI_FRAME
):
    # animated handling using `profile.has_alpha`, etc.
else:
    # still-image handling using `profile.has_alpha`, `profile.detected_mime`
```

This doesn’t change behavior, but it makes the coarse-grained steps (“profile image”, “handle animated”, “handle still”) more explicit and reduces the mental overhead of tracing the current conditional branches.
</issue_to_address>

### Comment 2
<location path="astrbot/core/provider/provider.py" line_range="96" />
<code_context>
         super().__init__(provider_config)
         self.provider_settings = provider_settings

+    def resolve_allowed_image_formats(self) -> frozenset[str] | None:
+        """Resolve the image MIME types allowed for this provider instance.
+
</code_context>
<issue_to_address>
**issue (complexity):** Consider extracting the image format and animated strategy parsing logic into dedicated helper functions to keep the base Provider class focused on high-level behavior.

The new logic is fairly dense and mixes concerns in the base class. You can reduce complexity by pushing the configuration parsing into small helpers and keeping the provider thin, while preserving behavior.

### 1. Extract image format resolution into a helper

Move the normalization/mapping/validation into a standalone function (e.g. in `media_utils` or a new `provider_config_utils`), and have the method delegate to it:

```python
# media_utils.py (or a new helper module)
from astrbot import logger
from astrbot.core.utils.media_utils import IMAGE_SHORT_MIME_TYPES

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

def resolve_allowed_image_formats(
    provider_config: dict,
    supported_image_formats: frozenset[str] | None,
) -> frozenset[str] | None:
    configured = provider_config.get("image_formats")
    if configured:
        normalized = {
            str(value).strip().lower()
            for value in configured
            if str(value).strip()
        }
        if "*" in normalized:
            return None

        mapped = {
            IMAGE_SHORT_MIME_TYPES.get(value, value)
            for value in normalized
            if value.startswith("image/") or value in IMAGE_SHORT_MIME_TYPES
        }
        if mapped:
            return frozenset(mapped)

        logger.warning(
            "Provider %s: image_formats %s contains no valid entries; "
            "falling back to the default format set.",
            provider_config.get("id"),
            sorted(normalized),
        )

    if supported_image_formats is not None:
        return supported_image_formats
    return DEFAULT_FALLBACK_IMAGE_FORMATS
```

Then in the `Provider` class:

```python
from astrbot.core.utils.media_utils import resolve_allowed_image_formats

class Provider(AbstractProvider):
    supported_image_formats: ClassVar[frozenset[str] | None] = None

    def resolve_allowed_image_formats(self) -> frozenset[str] | None:
        return resolve_allowed_image_formats(
            self.provider_config,
            self.supported_image_formats,
        )
```

This keeps the behavior identical but removes config parsing from the core abstraction and makes the logic easier to unit test in isolation.

### 2. Extract animated image strategy parsing

Do the same for `get_animated_image_strategy`:

```python
# media_utils.py (or helper module)
from astrbot.core.utils.media_utils import (
    ANIMATED_DEFAULT_MAX_FRAMES,
    ANIMATED_MAX_FRAMES_LIMIT,
    ANIMATED_STRATEGY_FIRST_FRAME,
    ANIMATED_STRATEGY_MULTI_FRAME,
)

def resolve_animated_image_strategy(provider_config: dict) -> tuple[str, int]:
    strategy = str(
        provider_config.get("animated_image_strategy")
        or ANIMATED_STRATEGY_FIRST_FRAME
    )
    if strategy not in (
        ANIMATED_STRATEGY_FIRST_FRAME,
        ANIMATED_STRATEGY_MULTI_FRAME,
    ):
        strategy = ANIMATED_STRATEGY_FIRST_FRAME

    raw_max_frames = provider_config.get("animated_image_max_frames")
    try:
        max_frames = (
            ANIMATED_DEFAULT_MAX_FRAMES if raw_max_frames is None else int(raw_max_frames)
        )
    except (TypeError, ValueError):
        max_frames = ANIMATED_DEFAULT_MAX_FRAMES

    max_frames = min(max(max_frames, 1), ANIMATED_MAX_FRAMES_LIMIT)
    return strategy, max_frames
```

And in `Provider`:

```python
from astrbot.core.utils.media_utils import resolve_animated_image_strategy

class Provider(AbstractProvider):
    # ...

    def get_animated_image_strategy(self) -> tuple[str, int]:
        return resolve_animated_image_strategy(self.provider_config)
```

This preserves the existing semantics (including clamping and defaulting) but makes the base provider’s public surface more declarative and shifts the configuration-heavy logic into dedicated helpers.
</issue_to_address>

### Comment 3
<location path="astrbot/core/provider/sources/openai_source.py" line_range="269" />
<code_context>
             },
         }

-    async def _transform_content_part(self, part: dict) -> dict:
+    async def _transform_content_part(self, part: dict) -> dict | list[dict]:
         if not isinstance(part, dict):
</code_context>
<issue_to_address>
**issue (complexity):** Consider normalizing image and content resolution helpers around a single 'list of parts' contract to simplify return types and caller logic while preserving multi-image behavior.

You can keep the new multi‑image functionality while simplifying the flow and types by normalizing on a “list of parts” contract and trimming wrappers.

### 1. Make `_transform_content_part` always return a list

This removes the dual `dict | list[dict]` shape and simplifies callers:

```python
# Before
async def _transform_content_part(self, part: dict) -> dict | list[dict]:
    ...

# After
async def _transform_content_part(self, part: dict) -> list[dict]:
    if not isinstance(part, dict):
        return [part]

    if part.get("type") == "image_url":
        url, image_detail = self._extract_image_part_info(part)
        if not url:
            return [part]

        try:
            resolved_parts = await self._resolve_image_parts(url, image_detail=image_detail)
        except Exception as exc:
            logger.warning(
                "图片 %s 预处理失败，将保留原始内容。错误: %s",
                url,
                exc,
            )
            return [part]

        return resolved_parts or [part]

    if part.get("type") == "audio_url":
        audio_ref = self._extract_audio_part_info(part)
        if not audio_ref:
            return [part]

        resolved_part = await self._resolve_audio_part(audio_ref)
        return [resolved_part] if resolved_part else [part]

    return [part]
```

Then `*_materialize_*` can unconditionally extend:

```python
async def _materialize_message_image_parts(self, message: dict) -> dict:
    content = message.get("content")
    if not isinstance(content, list):
        return {**message}

    new_content: list[dict] = []
    for part in content:
        new_content.extend(await self._transform_content_part(part))

    return {**message, "content": new_content}
```

This keeps multi‑frame expansion intact while simplifying all callers.

### 2. Use a single canonical “resolve to parts” helper

You already have `_image_ref_to_images` and `_resolve_image_parts`. You can make `_resolve_image_parts` the canonical “list of JSON parts” helper and keep `_image_ref_to_data_url` as a thin adapter that just returns the first image:

```python
async def _image_ref_to_images(
    self,
    image_ref: str,
    *,
    mode: Literal["safe", "strict"] = "safe",
) -> list[ResolvedMediaData]:
    strategy, max_frames = self.get_animated_image_strategy()
    return await resolve_image_ref_to_images(
        image_ref,
        allowed_mime_types=self.resolve_allowed_image_formats(),
        animated_strategy=strategy,
        animated_max_frames=max_frames,
        strict=mode == "strict",
    )


async def _resolve_image_parts(
    self,
    image_ref: str,
    *,
    image_detail: str | None = None,
    mode: Literal["safe", "strict"] = "safe",
) -> list[dict]:
    images = await self._image_ref_to_images(image_ref, mode=mode)
    if not images:
        logger.warning("图片预处理结果为空，将忽略。")
        return []

    parts: list[dict] = []
    for image_data in images:
        image_payload: dict = {"url": image_data.to_data_url()}
        if image_detail:
            image_payload["detail"] = image_detail
        parts.append({"type": "image_url", "image_url": image_payload})
    return parts


async def _image_ref_to_data_url(
    self,
    image_ref: str,
    *,
    mode: Literal["safe", "strict"] = "safe",
) -> str | None:
    images = await self._image_ref_to_images(image_ref, mode=mode)
    return images[0].to_data_url() if images else None
```

Then all call sites should use `_resolve_image_parts` (with `extend`) when they want OpenAI JSON parts, and `_image_ref_to_data_url` only for single‑image needs:

```python
# extra_user_content_parts
elif isinstance(part, ImageURLPart):
    image_parts = await self._resolve_image_parts(part.image_url.url)
    content_blocks.extend(image_parts)

# image_urls
for image_url in image_urls:
    image_parts = await self._resolve_image_parts(image_url)
    content_blocks.extend(image_parts)
```

This keeps the new animated/frame‑splitting behavior and provider‑specific formats, but:

- callers always deal with `list[dict]`
- there is a clear canonical entry point for “image ref -> OpenAI image_url parts”
- the single‑image helper is a simple adapter instead of a separate flow.
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: f8911ebab8

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.

[chatgpt-codex-connector]

You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard.