[Bug Fix] process transparent image

fastdeploy/multimodal/image.py

@@ -22,6 +22,7 @@
 from PIL import Image
 
 from .base import MediaIO
+from .utils import process_transparency
 
 
 class ImageMediaIO(MediaIO[Image.Image]):
@@ -59,6 +60,7 @@ def load_bytes(self, data: bytes) -> Image.Image:
         """
         image = Image.open(BytesIO(data))
         image.load()
+        image = process_transparency(image)
         return image.convert(self.image_mode)
 
     def load_base64(self, media_type: str, data: str) -> Image.Image:
@@ -93,6 +95,7 @@ def load_file(self, filepath: str) -> Image.Image:
         """
         image = Image.open(filepath)
         image.load()
+        image = process_transparency(image)
         return image.convert(self.image_mode)
 
     def load_file_request(self, request: Any) -> Image.Image:
@@ -112,6 +115,7 @@ def load_file_request(self, request: Any) -> Image.Image:
         """
         image = Image.open(requests.get(request, stream=True).raw)
         image.load()
+        image = process_transparency(image)
         return image.convert(self.image_mode)
 
     def encode_base64(

fastdeploy/multimodal/utils.py

@@ -14,141 +14,27 @@
 # limitations under the License.
 """
 
-import base64
-import io
-import ipaddress
-import mimetypes
-import os
-import socket
-import subprocess
-import tempfile
-from urllib.parse import urlparse
-
-import cairosvg
-import pyheif
-import requests
-from pdf2image import convert_from_path
 from PIL import Image, ImageOps
 
 from fastdeploy.utils import data_processor_logger
 
 
-def process_image_data(image_data, mime_type, url):
-    """处理不同类型的图像数据并返回 PIL 图像对象"""
-
-    if mime_type in ["image/heif", "image/heic"] or url.lower().endswith(".heif") or url.lower().endswith(".heic"):
-        heif_file = pyheif.read(image_data)
-        pil_image = Image.frombytes(
-            heif_file.mode,
-            heif_file.size,
-            heif_file.data,
-            "raw",
-            heif_file.mode,
-            heif_file.stride,
-        )
-    elif mime_type == "application/pdf" or url.lower().endswith(".pdf"):
-        with tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as temp_pdf:
-            temp_pdf.write(image_data.getvalue())
-            temp_pdf_path = temp_pdf.name
-        images = convert_from_path(temp_pdf_path)
-        pil_image = images[0]
-        os.remove(temp_pdf_path)
-    elif mime_type == "image/svg+xml" or url.lower().endswith(".svg"):
-        png_data = cairosvg.svg2png(bytestring=image_data.getvalue())
-        pil_image = Image.open(io.BytesIO(png_data))
-    elif mime_type in [
-        "application/postscript",
-        "application/illustrator",
-    ] or url.lower().endswith(".ai"):
-        with (
-            tempfile.NamedTemporaryFile(delete=False, suffix=".ai") as ai_temp,
-            tempfile.NamedTemporaryFile(delete=False, suffix=".pdf") as pdf_temp,
-        ):
-            ai_temp_path = ai_temp.name
-            pdf_temp_path = pdf_temp.name
-            ai_temp.write(image_data.getvalue())
-            ai_temp.close()
-            subprocess.run(
-                ["inkscape", ai_temp_path, "--export-pdf=" + pdf_temp_path],
-                check=True,
-            )
-            images = convert_from_path(pdf_temp_path)
-            pil_image = images[0]
-            os.remove(ai_temp_path)
-            os.remove(pdf_temp_path)
-
-    elif mime_type == "image/gif" or url.lower().endswith(".gif"):
-        pil_image = Image.open(image_data)
-    else:
-        pil_image = Image.open(image_data)
-
-    return pil_image
-
-
-def http_to_pil_image(url):
-    """http_to_pil_image"""
-
-    response = requests.get(url)
-    if response.status_code != 200:
-        raise Exception("Failed to download the image from URL.")
-    image_data = io.BytesIO(response.content)
-
-    mime_type = response.headers.get("Content-Type")
-    if mime_type is None:
-        mime_type, _ = mimetypes.guess_type(url)
-
-    data_processor_logger.info(f"Detected MIME type: {mime_type}")  # 调试信息
-    pil_image = process_image_data(image_data, mime_type, url)
-
-    return pil_image
-
-
-def base64_to_pil_image(base64_string):
-    """base64_to_pil_image"""
-    image_bytes = base64.b64decode(base64_string)
-    buffer = io.BytesIO(image_bytes)
-    pil_image = Image.open(buffer)
-    return pil_image
-
-
-def is_public_url(url):
-    """判断是否公网url"""
-    try:
-        # 解析URL
-        parsed_url = urlparse(url)
-        hostname = parsed_url.hostname
-        if hostname is None:
-            return False
-        # 尝试将域名解析为IP地址
-        ip_address = socket.gethostbyname(hostname)
-        # 转换为IP地址对象
-        ip_obj = ipaddress.ip_address(ip_address)
-        # 判断是否为私有IP或保留IP地址
-        if ip_obj.is_private or ip_obj.is_loopback or ip_obj.is_link_local or ip_obj.is_reserved:
-            return False
-        else:
-            return True
-    except Exception as e:
-        print(f"Error checking URL: {e}")
-        return False
-
-
 def process_transparency(image):
     """process transparency."""
 
     def _is_transparent(image):
-        # 检查图片是否有alpha通道
+        # Check if image has alpha channel
         if image.mode in ("RGBA", "LA") or (image.mode == "P" and "transparency" in image.info):
-            # 获取alpha通道
+            # Get alpha channel
             alpha = image.convert("RGBA").split()[-1]
-            # 如果alpha通道中存在0，说明图片有透明部分
+            # If alpha channel contains 0, image has transparent part
             if alpha.getextrema()[0] < 255:
                 return True
         return False
 
     def _convert_transparent_paste(image):
         width, height = image.size
-        new_image = Image.new("RGB", (width, height), (255, 255, 255))  # 生成一张白色底图
+        new_image = Image.new("RGB", (width, height), (255, 255, 255))  # Generate an image with white background
         new_image.paste(image, (0, 0), image)
         return new_image