ok

app copy 2.py

@@ -0,0 +1,104 @@
+import uuid
+import requests
+from PIL import Image
+import numpy as np
+import gradio as gr
+from encoder import FashionCLIPEncoder
+
+# Constants
+REQUESTS_HEADERS = {
+    'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36'
+}
+BATCH_SIZE = 30  # Define batch size for processing
+
+# Initialize encoder
+encoder = FashionCLIPEncoder()
+
+# Helper function to download images
+def download_image_as_pil(url: str, timeout: int = 10) -> Image.Image:
+    try:
+        response = requests.get(url, stream=True, headers=REQUESTS_HEADERS, timeout=timeout)
+        if response.status_code == 200:
+            return Image.open(response.raw).convert("RGB")  # Ensure consistent format
+        return None
+    except Exception as e:
+        print(f"Error downloading image: {e}")
+        return None
+
+# Embedding function for a batch of images
+def batch_process_images(image_urls: str):
+    # Split the input string by commas and strip whitespace
+    urls = [url.strip() for url in image_urls.split(",") if url.strip()]
+    
+    if not urls:
+        return {"error": "No valid image URLs provided."}
+
+    results = []
+    batch_urls, batch_images = [], []
+
+    for url in urls:
+        try:
+            # Download image
+            image = download_image_as_pil(url)
+            if not image:
+                results.append({"image_url": url, "error": "Failed to download image"})
+                continue
+
+            batch_urls.append(url)
+            batch_images.append(image)
+
+            # Process batch when reaching batch size
+            if len(batch_images) == BATCH_SIZE:
+                process_batch(batch_urls, batch_images, results)
+                batch_urls, batch_images = [], []
+
+        except Exception as e:
+            results.append({"image_url": url, "error": str(e)})
+
+    # Process remaining images in the last batch
+    if batch_images:
+        process_batch(batch_urls, batch_images, results)
+
+    return results
+
+
+# Helper function to process a batch
+def process_batch(batch_urls, batch_images, results):
+    try:
+        # Generate embeddings
+        embeddings = encoder.encode_images(batch_images)
+        
+        for url, embedding in zip(batch_urls, embeddings):
+            # Normalize embedding
+            embedding_normalized = embedding / np.linalg.norm(embedding)
+            
+            # Append results
+            results.append({
+                "image_url": url,
+                "embedding_preview": embedding_normalized[:5].tolist(),  # First 5 values for preview
+                "success": True
+            })
+    except Exception as e:
+        for url in batch_urls:
+            results.append({"image_url": url, "error": str(e)})
+
+
+# Gradio Interface
+iface = gr.Interface(
+    fn=batch_process_images,
+    inputs=gr.Textbox(
+        lines=5,
+        placeholder="Enter image URLs separated by commas",
+        label="Batch Image URLs",
+    ),
+    outputs=gr.JSON(label="Embedding Results"),
+    title="Batch Fashion CLIP Embedding API",
+    description="Enter multiple image URLs (separated by commas) to generate embeddings for the batch. Each embedding preview includes the first 5 values.",
+    examples=[
+        ["https://cdn.shopify.com/s/files/1/0522/2239/4534/files/CT21355-22_1024x1024.webp, https://cdn.shopify.com/s/files/1/0522/2239/4534/files/00907857-C6B0-4D2A-8AEA-688BDE1E67D7_1024x1024.jpg"]
+    ],
+)
+
+# Launch Gradio App
+if __name__ == "__main__":
+    iface.launch()

app.py

@@ -1,21 +1,16 @@
-# app.py
 import os
-from fastapi import FastAPI
-from pydantic import BaseModel
-from typing import List
 import requests
 from PIL import Image, UnidentifiedImageError
 import numpy as np
+import gradio as gr
 from encoder import FashionCLIPEncoder
 from pinecone import Pinecone
 from dotenv import load_dotenv
+import json
 
 # Load environment variables
 load_dotenv()
 
-# Initialize FastAPI app
-app = FastAPI()
-
 # Constants
 PINECONE_API_KEY = os.getenv("PINECONE_API_KEY")
 PINECONE_INDEX_NAME = os.getenv("PINECONE_INDEX_NAME")
@@ -30,14 +25,6 @@ pc = Pinecone(api_key=PINECONE_API_KEY)
 index = pc.Index(PINECONE_INDEX_NAME)
 encoder = FashionCLIPEncoder()
 
-class ProductData(BaseModel):
-    product_id: str
-    url: str
-
-class ProcessRequest(BaseModel):
-    products: List[ProductData]
-    upload_to_pinecone: bool = True
-
 def download_image_as_pil(url: str, timeout: int = 10) -> Image.Image:
     try:
         response = requests.get(url, stream=True, headers=REQUESTS_HEADERS, timeout=timeout)
@@ -51,7 +38,6 @@ def download_image_as_pil(url: str, timeout: int = 10) -> Image.Image:
         print(f"Error downloading image: {e}")
         return None
 
-
 def process_batch(batch_products, batch_images, results):
     try:
         # Generate embeddings
@@ -66,58 +52,77 @@ def process_batch(batch_products, batch_images, results):
                 "product_id": product["product_id"],
                 "image_url": product["url"],
                 "embedding": embedding_normalized.tolist(),
-                "embedding_preview": embedding_normalized[:5].tolist(),  # First 5 values for preview
+                "embedding_preview": embedding_normalized[:5].tolist(),
                 "success": True
             })
     except Exception as e:
         for product in batch_products:
             results.append({
-                "product_id": product["product_id"], 
-                "image_url": product["url"], 
+                "product_id": product["product_id"],
+                "image_url": product["url"],
                 "error": str(e)
             })
 
-def batch_process_images(products):
-    if not products:
-        return {"error": "No products provided."}
+def batch_process_images(json_input: str):
+    try:
+        # Parse JSON input
+        data = json.loads(json_input)
+        products = data.get("products", [])
+        upload_to_pinecone = data.get("upload_to_pinecone", False)
+        
+        if not products:
+            return {"error": "No products provided in JSON input."}
 
-    results = []
-    batch_products, batch_images = [], []
+        results = []
+        batch_products, batch_images = [], []
 
-    for product in products:
-        try:
-            # Download image
-            image = download_image_as_pil(product["url"])
-            if not image:
+        for product in products:
+            try:
+                # Download image
+                image = download_image_as_pil(product["url"])
+                if not image:
+                    results.append({
+                        "product_id": product["product_id"],
+                        "image_url": product["url"],
+                        "error": "Failed to download image"
+                    })
+                    continue
+
+                batch_products.append(product)
+                batch_images.append(image)
+
+                # Process batch when reaching batch size
+                if len(batch_images) == BATCH_SIZE:
+                    process_batch(batch_products, batch_images, results)
+                    batch_products, batch_images = [], []
+
+            except Exception as e:
                 results.append({
                     "product_id": product["product_id"],
                     "image_url": product["url"],
-                    "error": "Failed to download image"
+                    "error": str(e)
                 })
-                continue
-
-            batch_products.append(product)
-            batch_images.append(image)
-
-            # Process batch when reaching batch size
-            if len(batch_images) == BATCH_SIZE:
-                process_batch(batch_products, batch_images, results)
-                batch_products, batch_images = [], []
-
-        except Exception as e:
-            results.append({
-                "product_id": product["product_id"],
-                "image_url": product["url"],
-                "error": str(e)
-            })
 
-    # Process remaining images in the last batch
-    if batch_images:
-        process_batch(batch_products, batch_images, results)
+        # Process remaining images in the last batch
+        if batch_images:
+            process_batch(batch_products, batch_images, results)
 
-    return results
+        # Upload to Pinecone if requested
+        if upload_to_pinecone:
+            upload_result = upload_vector_to_pinecone(results)
+            return {
+                "processing_results": results,
+                "pinecone_upload": upload_result
+            }
+        
+        return {"processing_results": results}
+    
+    except json.JSONDecodeError:
+        return {"error": "Invalid JSON format"}
+    except Exception as e:
+        return {"error": f"Unexpecteddd error: {str(e)}"}
 
-def upload_to_pinecone(processed_results):
+def upload_vector_to_pinecone(processed_results):
     """Upload embeddings to Pinecone"""
     vectors_to_upsert = []
     for result in processed_results:
@@ -136,35 +141,56 @@ def upload_to_pinecone(processed_results):
     
     return {"uploaded_count": len(vectors_to_upsert)}
 
-@app.post("/process")
-async def process_images(request: ProcessRequest):
-    """
-    Process product images and optionally upload their embeddings to Pinecone
-    
-    Parameters:
-    - products: List of products with product_id and url
-    - upload_to_pinecone: Boolean flag to determine if embeddings should be uploaded to Pinecone
-    """
-    # Convert products to list of dicts
-    products_data = [{"product_id": p.product_id, "url": p.url} for p in request.products]
-    
-    # Process images
-    results = batch_process_images(products_data)
-    
-    # Upload to Pinecone if requested
-    if request.upload_to_pinecone:
-        upload_result = upload_to_pinecone(results)
-        return {
-            "processing_results": results,
-            "pinecone_upload": upload_result
+# Example JSON input
+EXAMPLE_INPUT = {
+    "products": [
+        {
+            "product_id": "1",
+            "url": "https://cdn.shopify.com/s/files/1/0522/2239/4534/files/CT21355-22_1024x1024.webp"
         }
-    
-    return {"processing_results": results}
+    ],
+    "upload_to_pinecone": False
+}
 
-@app.get("/health")
-async def health_check():
-    return {"status": "healthy"}
+# Gradio Interface
+iface = gr.Interface(
+    fn=batch_process_images,
+    inputs=gr.Code(
+        label="Input JSON",
+        language="json",
+        value=json.dumps(EXAMPLE_INPUT, indent=4)  # Changed from default to value
+    ),
+    outputs=gr.JSON(label="Processing Results"),
+    title="Fashion CLIP Embedding Generator",
+    description="Provide JSON input with product IDs, URLs, and Pinecone upload preference to generate embeddings.",
+    article="""
+    ### Input JSON Format:
+    ```json
+    {
+        "products": [
+            {
+                "product_id": "string",
+                "url": "string"
+            }
+        ],
+        "upload_to_pinecone": boolean
+    }
+    ```
+    
+    ### Features:
+    - Batch processing of multiple images
+    - Custom product ID support
+    - Embedding generation using Fashion CLIP
+    - Optional Pinecone database integration
+    - Error handling and detailed results
+    
+    Make sure to set up your environment variables in a .env file:
+    - PINECONE_API_KEY
+    - PINECONE_INDEX_NAME
+    - PINECONE_NAMESPACE
+    """
+)
 
-# if __name__ == "__main__":
-#     import uvicorn
-#     uvicorn.run(app, host="0.0.0.0", port=8000)
+# Launch Gradio App
+if __name__ == "__main__":
+    iface.launch()