孩子们这里也有硬编码

Author: Tz-WIND

astrbot/core/provider/entities.py

@@ -169,22 +169,17 @@ def _print_friendly_context(self):
 
     async def assemble_context(self) -> dict:
         """将请求(prompt 和 image_urls)包装成 OpenAI 的消息格式。"""
-        # 构建内容块列表
         content_blocks = []
 
-        # 1. 用户原始发言（OpenAI 建议：用户发言在前）
         if self.prompt and self.prompt.strip():
             content_blocks.append({"type": "text", "text": self.prompt})
         elif self.image_urls:
-            # 如果没有文本但有图片，添加占位文本
             content_blocks.append({"type": "text", "text": "[图片]"})
 
-        # 2. 额外的内容块（系统提醒、指令等）
         if self.extra_user_content_parts:
             for part in self.extra_user_content_parts:
                 content_blocks.append(part.model_dump())
 
-        # 3. 图片内容
         if self.image_urls:
             for image_url in self.image_urls:
                 if image_url.startswith("http"):
@@ -202,7 +197,6 @@ async def assemble_context(self) -> dict:
                     {"type": "image_url", "image_url": {"url": image_data}},
                 )
 
-        # 只有当只有一个来自 prompt 的文本块且没有额外内容块时，才降级为简单格式以保持向后兼容
         if (
             len(content_blocks) == 1
             and content_blocks[0]["type"] == "text"
@@ -211,17 +205,56 @@ async def assemble_context(self) -> dict:
         ):
             return {"role": "user", "content": content_blocks[0]["text"]}
 
-        # 否则返回多模态格式
         return {"role": "user", "content": content_blocks}
 
+    @staticmethod
+    def _detect_mime_type(header_bytes: bytes) -> str:
+        """根据文件头魔术字节检测图片的 MIME 类型，无法识别时回退到 image/jpeg"""
+        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"
+        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"
+        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"
+        if b'<svg' in header_bytes[:256].lower():
+            return "image/svg+xml"
+        if len(header_bytes) >= 12 and header_bytes[4:12] == b'ftypavif':
+            return "image/avif"
+        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"
+        return "image/jpeg"
+
     async def _encode_image_bs64(self, image_url: str) -> str:
-        """将图片转换为 base64"""
+        """将图片转换为 base64，自动检测实际 MIME 类型"""
         if image_url.startswith("base64://"):
-            return image_url.replace("base64://", "data:image/jpeg;base64,")
+            raw_b64 = image_url[len("base64://"):]
+            try:
+                sample = raw_b64[:32]
+                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:
+            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 "data:image/jpeg;base64," + image_bs64
-        return ""
+            return f"data:{mime_type};base64,{image_bs64}"
 
 
 @dataclass
@@ -362,7 +395,7 @@ def completion_text(self, value) -> None:
                 comp
                 for comp in self.result_chain.chain
                 if not isinstance(comp, Comp.Plain)
-            ]  # 清空 Plain 组件
+            ]
             self.result_chain.chain.insert(0, Comp.Plain(value))
         else:
             self._completion_text = value
@@ -397,7 +430,6 @@ def to_openai_to_calls_model(self) -> list[ToolCall]:
                         name=self.tools_call_name[idx],
                         arguments=json.dumps(tool_call_arg),
                     ),
-                    # the extra_content will not serialize if it's None when calling ToolCall.model_dump()
                     extra_content=self.tools_call_extra_content.get(
                         self.tools_call_ids[idx]
                     ),