""" Blob conversion utilities for Gradio image components. Handles conversion of blob data to proper image file formats. """ import hashlib import os from typing import Dict, Any from logging_config import get_logger # Module logger logger = get_logger(__name__) class BlobConverter: """Handles conversion of blob data to proper image file formats.""" # File format signatures FORMAT_SIGNATURES = { b"\x89PNG\r\n\x1a\n": (".png", "image/png"), b"\xff\xd8\xff": (".jpg", "image/jpeg"), b"GIF87a": (".gif", "image/gif"), b"GIF89a": (".gif", "image/gif"), } @classmethod def is_blob_data(cls, image_data: Dict[str, Any]) -> bool: """ Check if the image data represents a blob that needs conversion. Args: image_data: Dictionary containing image metadata Returns: True if the data is a blob that needs conversion """ return ( isinstance(image_data, dict) and "path" in image_data and "blob" in image_data["path"] and image_data.get("size") is None and image_data.get("orig_name") is None and image_data.get("mime_type") is None ) @classmethod def detect_format(cls, content: bytes) -> tuple[str, str]: """ Detect image format from file content. Args: content: Binary content of the file Returns: Tuple of (extension, mime_type) """ for signature, (ext, mime_type) in cls.FORMAT_SIGNATURES.items(): if content.startswith(signature): return ext, mime_type # Default to PNG if format cannot be determined return ".png", "image/png" @classmethod def generate_filename(cls, content: bytes, extension: str) -> str: """ Generate a unique filename for the converted blob. Args: content: Binary content of the file extension: File extension to use Returns: Unique filename """ content_hash = hashlib.md5(content).hexdigest()[:8] return f"flagged_image_{content_hash}{extension}" @classmethod def convert_blob(cls, image_data: Dict[str, Any]) -> Dict[str, Any]: """ Convert blob data to proper image file format. Args: image_data: Original image data dictionary Returns: Updated image data with proper file information """ if not cls.is_blob_data(image_data): return image_data logger.info("Converting blob data: %s", image_data) blob_path = image_data["path"] logger.debug("Blob path: %s", blob_path) # Read blob content with open(blob_path, "rb") as f: content = f.read() file_size = len(content) logger.debug("File size: %d bytes", file_size) # Detect format extension, mime_type = cls.detect_format(content) logger.debug("Detected format: %s, MIME type: %s", extension, mime_type) # Generate filename and path filename = cls.generate_filename(content, extension) temp_dir = os.path.dirname(blob_path) new_path = os.path.join(temp_dir, filename) logger.debug("Generated filename: %s", filename) logger.debug("New path: %s", new_path) # Write converted file with open(new_path, "wb") as f: f.write(content) logger.info("Successfully converted blob to: %s", new_path) # Return updated image data converted_data = { "path": new_path, "url": image_data["url"].replace("blob", filename), "size": file_size, "orig_name": filename, "mime_type": mime_type, "is_stream": False, "meta": image_data.get("meta", {}), } logger.debug("Converted data: %s", converted_data) return converted_data def decode_blob_data(image_data: Dict[str, Any]) -> Dict[str, Any]: """ Convenience function to decode blob data from Gradio image component. Args: image_data: Image data dictionary from Gradio Returns: Converted image data or original data if not a blob """ logger.debug("Processing image data: %s", image_data) result = BlobConverter.convert_blob(image_data) if result is image_data: logger.debug("Not a blob, skipping conversion") else: logger.info("Blob conversion completed") return result def is_blob_data(image_data: Dict[str, Any]) -> bool: """ Check if the image data represents a blob that needs conversion. Args: image_data: Dictionary containing image metadata Returns: True if the data is a blob that needs conversion """ return BlobConverter.is_blob_data(image_data)