feat: Add support for image-to-image and image-to-video generation; enhance upload functionality and create test scripts for validation

Author: mir-ashiq

.gitignore

@@ -110,4 +110,5 @@ QUICK_FIX_GUIDE.md
 # Don't include cloudflare-worker files
 cloudflare-worker
 # network.json
-# network-*.json
+# network-*.json
+test_api.html

src/metaai_api/api_server.py

@@ -162,6 +162,7 @@ class ImageRequest(BaseModel):
     media_ids: Optional[list] = None
     attachment_metadata: Optional[dict] = None  # {'file_size': int, 'mime_type': str}
     orientation: Optional[str] = None  # 'VERTICAL', 'LANDSCAPE' (not HORIZONTAL), or 'SQUARE'
+    num_images: int = Field(1, ge=1, le=4)  # Number of images to generate (1-4)
 
 
 class VideoRequest(BaseModel):
@@ -314,13 +315,16 @@ async def image(body: ImageRequest) -> Dict[str, Any]:
         )
     ai = _meta_ai_instance
     try:
+        # Determine number of images: use 4 for image-to-image, 1 for text-to-image
+        num_images = 4 if body.media_ids else body.num_images
+        
         # Use the new generation API with timeout protection
         result = await asyncio.wait_for(
             run_in_threadpool(
                 ai.generate_image_new,
                 prompt=body.prompt,
                 orientation=body.orientation or "VERTICAL",
-                num_images=1,
+                num_images=num_images,
                 media_ids=body.media_ids,
                 attachment_metadata=body.attachment_metadata
             ),

src/metaai_api/generation.py

@@ -138,6 +138,52 @@ def _build_base_variables(
         if media_ids:
             attachments_v2 = [str(mid) for mid in media_ids]
 
+        # Determine operation type based on media_ids presence
+        is_image_to_image = operation == "TEXT_TO_IMAGE" and media_ids and len(media_ids) > 0
+        is_image_to_video = operation == "TEXT_TO_VIDEO" and media_ids and len(media_ids) > 0
+        
+        # Build imagineOperationRequest based on operation type
+        if is_image_to_image:
+            # Use imageToImageParams for image-to-image generation
+            num_media = kwargs.get('num_images', 4)
+            imagine_request = {
+                "operation": "IMAGE_TO_IMAGE",
+                "imageToImageParams": {
+                    "sourceMediaEntId": str(media_ids[0]),
+                    "instruction": prompt,
+                    "imageSource": "USER_UPLOADED",
+                    "imageUploadType": "GENAI_UPLOADED_FILE",
+                    "mediaType": "UPLOADED_IMAGE",
+                    "numMedia": num_media
+                }
+            }
+        elif is_image_to_video:
+            # Use imageToVideoParams for image-to-video generation
+            imagine_request = {
+                "operation": "IMAGE_TO_VIDEO",
+                "imageToVideoParams": {
+                    "sourceMediaEntId": str(media_ids[0]),
+                    "prompt": prompt,
+                    "numMedia": 1
+                }
+            }
+        else:
+            # Use textToImageParams or textToVideoParams for text-based generation
+            if operation == "TEXT_TO_VIDEO":
+                imagine_request = {
+                    "operation": operation,
+                    "textToVideoParams": {
+                        "prompt": prompt
+                    }
+                }
+            else:
+                imagine_request = {
+                    "operation": operation,
+                    "textToImageParams": {
+                        "prompt": prompt
+                    }
+                }
+
         variables = {
             "conversationId": conversation_id,
             "content": content,
@@ -153,12 +199,7 @@ def _build_base_variables(
             "mentions": None,
             "clippyIp": None,
             "isNewConversation": kwargs.get('is_new_conversation', True),
-            "imagineOperationRequest": {
-                "operation": operation,
-                "textToImageParams": {
-                    "prompt": prompt
-                }
-            },
+            "imagineOperationRequest": imagine_request,
             "qplJoinId": None,
             "clientTimezone": kwargs.get('timezone', "UTC"),
             "developerOverridesForMessage": None,
@@ -208,9 +249,19 @@ def generate_image(
         )
 
         # Add image-specific parameters
-        variables["imagineOperationRequest"]["textToImageParams"]["orientation"] = self._normalize_orientation(orientation)
-        if num_images > 1:
-            self.logger.warning("num_images > 1 is not supported by this endpoint; generating a single image")
+        # Check if we're doing image-to-image generation
+        media_ids = kwargs.get('media_ids')
+        is_image_to_image = media_ids and len(media_ids) > 0
+        
+        if is_image_to_image:
+            # For image-to-image, orientation is not in imageToImageParams
+            # numMedia is already set in _build_base_variables
+            self.logger.info(f"Using IMAGE_TO_IMAGE operation with source media: {media_ids[0]}")
+        else:
+            # For text-to-image, add orientation to textToImageParams
+            variables["imagineOperationRequest"]["textToImageParams"]["orientation"] = self._normalize_orientation(orientation)
+            if num_images > 1:
+                self.logger.warning("num_images > 1 is not supported by this endpoint; generating a single image")
 
         payload = {
             "doc_id": self.IMAGE_DOC_ID,