Update app.py

app.py

@@ -8,50 +8,85 @@ import time
 import requests
 import base64
 from simple_salesforce import Salesforce
+import logging
+
+# Configure logging
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
 
 # Initialize MediaPipe
 mp_pose = mp.solutions.pose
 pose = mp_pose.Pose(static_image_mode=True, model_complexity=2)
 
-# Salesforce Configuration
-SF_USERNAME = "your_salesforce_username"
-SF_PASSWORD = "your_salesforce_password"
-SF_SECURITY_TOKEN = "your_security_token"
-SF_DOMAIN = "yourdomain.my"
-
-# Initialize Salesforce connection
-sf = Salesforce(
-    username=SF_USERNAME,
-    password=SF_PASSWORD,
-    security_token=SF_SECURITY_TOKEN,
-    domain=SF_DOMAIN
-)
+# Salesforce Configuration (use environment variables in production)
+SF_CONFIG = {
+    "username": "your_salesforce_username@example.com",
+    "password": "your_password",
+    "security_token": "your_security_token",
+    "domain": "login"  # For production: "login" or "test"
+}
 
-def fetch_dresses_from_salesforce():
-    """Dynamically fetch dresses from Salesforce"""
-    try:
-        result = sf.query("SELECT Id, Name, Image_URL__c FROM Product2 WHERE IsActive = TRUE AND Family = 'Dress'")
-        dresses = {}
+class SalesforceConnector:
+    def __init__(self):
+        self.sf = None
+        self.connect()
+    
+    def connect(self):
+        try:
+            self.sf = Salesforce(
+                username=SF_CONFIG["username"],
+                password=SF_CONFIG["password"],
+                security_token=SF_CONFIG["security_token"],
+                domain=SF_CONFIG["domain"]
+            )
+            logger.info("Successfully connected to Salesforce")
+        except Exception as e:
+            logger.error(f"Salesforce connection failed: {str(e)}")
+            self.sf = None
+    
+    def is_connected(self):
+        return self.sf is not None
+    
+    def query_dresses(self):
+        if not self.is_connected():
+            self.connect()
+            if not self.is_connected():
+                return {}
         
-        for item in result['records']:
-            if not item.get('Image_URL__c'):
-                continue
-                
-            try:
-                img_response = requests.get(item['Image_URL__c'], timeout=10)
-                img = Image.open(BytesIO(img_response.content)).convert("RGBA")
-                dresses[item['Id']] = {
-                    'name': item['Name'],
-                    'image': img,
-                    'thumbnail': generate_thumbnail(img)
-                }
-            except Exception as e:
-                print(f"Failed to load dress {item['Name']}: {str(e)}")
-                
-        return dresses
-    except Exception as e:
-        print(f"Salesforce query failed: {str(e)}")
-        return {}
+        try:
+            result = self.sf.query(
+                "SELECT Id, Name, Image_URL__c FROM Product2 "
+                "WHERE IsActive = TRUE AND Family = 'Dress' "
+                "LIMIT 20"
+            )
+            return result.get("records", [])
+        except Exception as e:
+            logger.error(f"Salesforce query failed: {str(e)}")
+            return []
+
+sf_connector = SalesforceConnector()
+
+def fetch_dresses():
+    """Fetch dresses with automatic reconnection"""
+    dresses = {}
+    records = sf_connector.query_dresses()
+    
+    for item in records:
+        if not item.get('Image_URL__c'):
+            continue
+            
+        try:
+            img_response = requests.get(item['Image_URL__c'], timeout=10)
+            img = Image.open(BytesIO(img_response.content)).convert("RGBA")
+            dresses[item['Id']] = {
+                'name': item['Name'],
+                'image': img,
+                'thumbnail': generate_thumbnail(img)
+            }
+        except Exception as e:
+            logger.warning(f"Failed to load dress {item.get('Name')}: {str(e)}")
+    
+    return dresses
 
 def generate_thumbnail(img, size=(150, 200)):
     """Generate base64 thumbnail for UI"""
@@ -61,67 +96,44 @@ def generate_thumbnail(img, size=(150, 200)):
     img.save(buffered, format="PNG")
     return f"data:image/png;base64,{base64.b64encode(buffered.getvalue()).decode()}"
 
-def get_body_landmarks(image):
-    """Detect body pose using MediaPipe"""
-    results = pose.process(cv2.cvtColor(image, cv2.COLOR_BGR2RGB))
-    if not results.pose_landmarks:
-        return None
-        
-    h, w = image.shape[:2]
-    landmarks = results.pose_landmarks.landmark
-    return {
-        'left_shoulder': (landmarks[11].x * w, landmarks[11].y * h),
-        'right_shoulder': (landmarks[12].x * w, landmarks[12].y * h),
-        'left_hip': (landmarks[23].x * w, landmarks[23].y * h),
-        'right_hip': (landmarks[24].x * w, landmarks[24].y * h)
-    }
-
-def warp_and_overlay(user_img, dress_img, body_pts):
-    """Thin Plate Spline warping with alpha blending"""
-    # Warp dress
-    h, w = dress_img.shape[:2]
-    src_pts = np.float32([[0,0], [w,0], [w,h], [0,h]])
-    dst_pts = np.float32([body_pts['left_shoulder'], body_pts['right_shoulder'], 
-                         body_pts['right_hip'], body_pts['left_hip']])
-    
-    tps = cv2.createThinPlateSplineShapeTransformer()
-    tps.estimateTransformation(dst_pts, src_pts)
-    warped_dress = tps.warpImage(dress_img)
-    
-    # Blend images
-    alpha = warped_dress[:,:,3] / 255.0
-    for c in range(3):
-        user_img[:,:,c] = user_img[:,:,c] * (1 - alpha) + warped_dress[:,:,c] * alpha
-    
-    return user_img
+# Rest of your existing functions (get_body_landmarks, warp_and_overlay, etc.)
 
 def process_try_on(user_image, dress_id):
-    """Main processing pipeline"""
+    """Main processing with enhanced error handling"""
     if user_image is None:
         return None, "Please upload an image"
         
     start_time = time.time()
-    dresses = fetch_dresses_from_salesforce()
-    
-    if not dresses or dress_id not in dresses:
-        return user_image, "No dresses found in Salesforce"
     
     try:
+        # Convert to numpy array
         user_img = np.array(user_image)
+        
+        # Get body landmarks
         body_pts = get_body_landmarks(user_img)
         if not body_pts:
             return user_image, "Stand facing camera with arms slightly away"
             
+        # Get dresses
+        dresses = fetch_dresses()
+        if not dresses:
+            return user_image, "No dresses available - check Salesforce connection"
+        
+        if dress_id not in dresses:
+            return user_image, "Selected dress not found"
+            
+        # Process try-on
         dress_img = np.array(dresses[dress_id]['image'])
         result = warp_and_overlay(user_img.copy(), dress_img, body_pts)
         
         return Image.fromarray(result), f"Done in {time.time()-start_time:.2f}s"
     except Exception as e:
+        logger.error(f"Try-on failed: {str(e)}")
         return user_image, f"Error: {str(e)}"
 
 # Gradio Interface
-with gr.Blocks(title="Salesforce Virtual Try-On", css=".thumbnail { height: 100px !important }") as demo:
-    gr.Markdown("# 👗 Virtual Try-On (Powered by Salesforce)")
+with gr.Blocks(title="Virtual Try-On", css=".thumbnail { height: 100px !important }") as demo:
+    gr.Markdown("# 👗 Virtual Try-On (Salesforce Connected)")
     
     with gr.Row():
         with gr.Column():
@@ -130,19 +142,37 @@ with gr.Blocks(title="Salesforce Virtual Try-On", css=".thumbnail { height: 100p
                 refresh_btn = gr.Button("🔄 Refresh Dresses")
                 try_btn = gr.Button("👗 Try On Dress", variant="primary")
             dress_dropdown = gr.Dropdown(label="Select Dress", interactive=True)
+            connection_status = gr.Textbox(label="Salesforce Status", 
+                                        value="Connected" if sf_connector.is_connected() else "Disconnected")
             
         with gr.Column():
             output_image = gr.Image(label="Your Virtual Try-On", interactive=False)
             status = gr.Textbox(label="Status")
     
     def update_dress_dropdown():
-        dresses = fetch_dresses_from_salesforce()
+        dresses = fetch_dresses()
         choices = [(f"{d['name']}", id) for id, d in dresses.items()]
-        return gr.Dropdown(choices=choices, value=choices[0][1] if choices else None)
+        status = "Connected" if sf_connector.is_connected() else "Disconnected"
+        return (
+            gr.Dropdown(choices=choices, value=choices[0][1] if choices else None),
+            f"Salesforce: {status} | {len(choices)} dresses loaded"
+        )
+    
+    refresh_btn.click(
+        update_dress_dropdown, 
+        outputs=[dress_dropdown, connection_status]
+    )
+    
+    try_btn.click(
+        process_try_on, 
+        inputs=[input_image, dress_dropdown], 
+        outputs=[output_image, status]
+    )
     
-    refresh_btn.click(update_dress_dropdown, outputs=dress_dropdown)
-    try_btn.click(process_try_on, [input_image, dress_dropdown], [output_image, status])
-    demo.load(update_dress_dropdown, outputs=dress_dropdown)
+    demo.load(
+        update_dress_dropdown, 
+        outputs=[dress_dropdown, connection_status]
+    )
 
 if __name__ == "__main__":
     demo.launch(server_name="0.0.0.0", server_port=7860)