改为使用工具模块

Author: Tz-WIND

astrbot/core/provider/entities.py

@@ -1,6 +1,5 @@
 from __future__ import annotations
 
-import base64
 import enum
 import json
 from dataclasses import dataclass, field
@@ -21,6 +20,7 @@
 from astrbot.core.agent.tool import ToolSet
 from astrbot.core.db.po import Conversation
 from astrbot.core.message.message_event_result import MessageChain
+from astrbot.core.utils.image_utils import encode_image_to_base64_url
 from astrbot.core.utils.io import download_image_by_url
 
 
@@ -189,12 +189,13 @@ async def assemble_context(self) -> dict:
             for image_url in self.image_urls:
                 if image_url.startswith("http"):
                     image_path = await download_image_by_url(image_url)
-                    image_data = await self._encode_image_bs64(image_path)
+                    # 统一通过公共工具函数编码，自动检测 MIME 类型
+                    image_data = await encode_image_to_base64_url(image_path)
                 elif image_url.startswith("file:///"):
                     image_path = image_url.replace("file:///", "")
-                    image_data = await self._encode_image_bs64(image_path)
+                    image_data = await encode_image_to_base64_url(image_path)
                 else:
-                    image_data = await self._encode_image_bs64(image_url)
+                    image_data = await encode_image_to_base64_url(image_url)
                 if not image_data:
                     logger.warning(f"图片 {image_url} 得到的结果为空，将忽略。")
                     continue
@@ -214,87 +215,6 @@ async def assemble_context(self) -> dict:
         # 否则返回多模态格式
         return {"role": "user", "content": content_blocks}
 
-    @staticmethod
-    def _detect_mime_type(header_bytes: bytes) -> str:
-        """根据文件头魔术字节（magic bytes）检测图片的实际 MIME 类型。
-
-        依次匹配常见图片格式的文件头特征，均不匹配时回退到 image/jpeg
-        以保持向后兼容。支持的格式：JPEG、PNG、GIF、WebP、BMP、TIFF、
-        ICO、SVG、AVIF、HEIF/HEIC。
-
-        Args:
-            header_bytes: 文件头原始字节，建议至少传入 16 字节。
-
-        Returns:
-            对应的 MIME 类型字符串，例如 "image/png"。
-        """
-        if len(header_bytes) >= 3 and header_bytes[:3] == b'\xff\xd8\xff':
-            return "image/jpeg"
-        if len(header_bytes) >= 8 and header_bytes[:8] == b'\x89PNG\r\n\x1a\n':
-            return "image/png"
-        if len(header_bytes) >= 4 and header_bytes[:4] == b'GIF8':
-            return "image/gif"
-        # WebP: RIFF????WEBP
-        if len(header_bytes) >= 12 and header_bytes[:4] == b'RIFF' and header_bytes[8:12] == b'WEBP':
-            return "image/webp"
-        if len(header_bytes) >= 2 and header_bytes[:2] == b'BM':
-            return "image/bmp"
-        # TIFF: 小端 (II) 或大端 (MM)
-        if len(header_bytes) >= 4 and header_bytes[:4] in (b'II\x2a\x00', b'MM\x00\x2a'):
-            return "image/tiff"
-        if len(header_bytes) >= 4 and header_bytes[:4] == b'\x00\x00\x01\x00':
-            return "image/x-icon"
-        # SVG 为文本格式，检测头部是否含有 <svg 标签
-        if b'<svg' in header_bytes[:256].lower():
-            return "image/svg+xml"
-        # AVIF: ftyp box brand = avif
-        if len(header_bytes) >= 12 and header_bytes[4:12] == b'ftypavif':
-            return "image/avif"
-        # HEIF/HEIC: ftyp box，brand 为 heic/heix/hevc/hevx/mif1
-        if len(header_bytes) >= 12 and header_bytes[4:8] == b'ftyp':
-            brand = header_bytes[8:12]
-            if brand in (b'heic', b'heix', b'hevc', b'hevx', b'mif1'):
-                return "image/heif"
-        # 无法识别，回退到 image/jpeg
-        return "image/jpeg"
-
-    async def _encode_image_bs64(self, image_url: str) -> str:
-        """将图片转换为 base64 Data URL，自动检测实际 MIME 类型。
-
-        原实现硬编码 image/jpeg，会导致 PNG 等格式在严格校验的接口上报错。
-        现通过读取文件头魔术字节来推断正确的 MIME 类型。
-
-        Args:
-            image_url: 本地文件路径，或以 "base64://" 开头的 base64 字符串。
-
-        Returns:
-            形如 "data:image/png;base64,..." 的 Data URL 字符串。
-        """
-        if image_url.startswith("base64://"):
-            raw_b64 = image_url[len("base64://"):]
-            # 从 base64 数据中解码少量字节以检测实际格式，
-            # 取前 32 个 base64 字符可解码出约 24 字节的原始数据，足以覆盖所有魔术字节
-            try:
-                sample = raw_b64[:32]
-                # 确保 base64 填充正确，避免解码报错
-                missing_padding = len(sample) % 4
-                if missing_padding:
-                    sample += '=' * (4 - missing_padding)
-                header_bytes = base64.b64decode(sample)
-                mime_type = self._detect_mime_type(header_bytes)
-            except Exception:
-                # 解码失败时安全回退
-                mime_type = "image/jpeg"
-            return f"data:{mime_type};base64,{raw_b64}"
-
-        with open(image_url, "rb") as f:
-            # 先读取文件头用于格式检测，再 seek 回起点读取完整内容
-            header_bytes = f.read(16)
-            mime_type = self._detect_mime_type(header_bytes)
-            f.seek(0)
-            image_bs64 = base64.b64encode(f.read()).decode("utf-8")
-            return f"data:{mime_type};base64,{image_bs64}"
-
 
 @dataclass
 class TokenUsage: