refac

backend/open_webui/env.py

@@ -674,6 +674,15 @@ ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION = (
     os.environ.get('ENABLE_CHAT_RESPONSE_BASE64_IMAGE_URL_CONVERSION', 'False').lower() == 'true'
 )
 
+# When enabled, uses a hardcoded extension-to-MIME dictionary as a last-resort
+# fallback when both mimetypes.guess_type() and file.meta.content_type fail to
+# determine the content type. This can help on minimal container images (e.g.
+# wolfi-base) that lack /etc/mime.types AND have legacy files without stored
+# content_type metadata.
+ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK = (
+    os.environ.get('ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK', 'False').lower() == 'true'
+)
+
 CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE = os.environ.get('CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE', '1')
 
 if CHAT_RESPONSE_STREAM_DELTA_CHUNK_SIZE == '':

backend/open_webui/utils/files.py

@@ -26,12 +26,29 @@ import base64
 import io
 import re
 
-from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL
+from open_webui.env import AIOHTTP_CLIENT_SESSION_SSL, ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK
 from open_webui.utils.session_pool import get_session
 
 BASE64_IMAGE_URL_PREFIX = re.compile(r'data:image/\w+;base64,', re.IGNORECASE)
 MARKDOWN_IMAGE_URL_PATTERN = re.compile(r'!\[(.*?)\]\((.+?)\)', re.IGNORECASE)
 
+# Extension-based MIME fallback, only used when ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK is True.
+_IMAGE_MIME_FALLBACK = {
+    ".webp": "image/webp",
+    ".png": "image/png",
+    ".jpg": "image/jpeg",
+    ".jpeg": "image/jpeg",
+    ".gif": "image/gif",
+    ".svg": "image/svg+xml",
+    ".bmp": "image/bmp",
+    ".tiff": "image/tiff",
+    ".tif": "image/tiff",
+    ".ico": "image/x-icon",
+    ".heic": "image/heic",
+    ".heif": "image/heif",
+    ".avif": "image/avif",
+}
+
 
 async def get_image_base64_from_url(url: str) -> Optional[str]:
     try:
@@ -58,7 +75,14 @@ async def get_image_base64_from_url(url: str) -> Optional[str]:
             if file_path.is_file():
                 with open(file_path, 'rb') as image_file:
                     encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
-                    content_type, _ = mimetypes.guess_type(file_path.name)
+                    content_type = (
+                        mimetypes.guess_type(file_path.name)[0]
+                        or (file.meta or {}).get('content_type')
+                    )
+                    if not content_type and ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK:
+                        content_type = _IMAGE_MIME_FALLBACK.get(file_path.suffix.lower())
+                    if not content_type:
+                        return None
                     return f'data:{content_type};base64,{encoded_string}'
             else:
                 return None
@@ -178,11 +202,16 @@ async def get_image_base64_from_file_id(id: str) -> Optional[str]:
 
         # Check if the file already exists in the cache
         if file_path.is_file():
-            import base64
-
             with open(file_path, 'rb') as image_file:
                 encoded_string = base64.b64encode(image_file.read()).decode('utf-8')
-                content_type, _ = mimetypes.guess_type(file_path.name)
+                content_type = (
+                    mimetypes.guess_type(file_path.name)[0]
+                    or (file.meta or {}).get('content_type')
+                )
+                if not content_type and ENABLE_IMAGE_CONTENT_TYPE_EXTENSION_FALLBACK:
+                    content_type = _IMAGE_MIME_FALLBACK.get(file_path.suffix.lower())
+                if not content_type:
+                    return None
                 return f'data:{content_type};base64,{encoded_string}'
         else:
             return None