#!/usr/bin/env python3 """ Process Detection Module - Standalone function for processing master image detection """ import os import json import time from pathlib import Path from PIL import Image, ImageEnhance import google.generativeai as genai from dotenv import load_dotenv import uuid import threading import tempfile def process_single_master_detection(layout_path, master_id, master_path, enable_greyscale, enable_contrast_enhancement, contrast_factor, safety_settings): """ Standalone function for processing a single master detection in a separate process. This ensures complete isolation from other workers. """ try: # Load environment in this process load_dotenv() api_key = os.getenv('GEMINI_API_KEY') if not api_key: raise ValueError("GEMINI_API_KEY not found in environment variables") # Configure API client in this process genai.configure(api_key=api_key) model = genai.GenerativeModel('gemini-2.5-pro') def preprocess_image_process(image_path, enable_greyscale, enable_contrast_enhancement, contrast_factor): """Process-local image preprocessing""" if not enable_greyscale and not enable_contrast_enhancement: return image_path try: with Image.open(image_path) as img: processed_img = img.copy() if enable_greyscale: processed_img = processed_img.convert('L') processed_img = processed_img.convert('RGB') if enable_contrast_enhancement: contrast_enhancer = ImageEnhance.Contrast(processed_img) processed_img = contrast_enhancer.enhance(contrast_factor) sharpness_enhancer = ImageEnhance.Sharpness(processed_img) processed_img = sharpness_enhancer.enhance(1.3) # Create unique temp file for this process process_id = os.getpid() unique_id = str(uuid.uuid4())[:8] original_name = Path(image_path).stem with tempfile.NamedTemporaryFile(suffix=f"_{process_id}_{unique_id}.jpg", delete=False) as tmp_file: processed_img.save(tmp_file.name, 'JPEG', quality=95) return tmp_file.name except Exception as e: return image_path def upload_with_retry_process(image_path, max_retries=3): """Process-local upload with retry""" for attempt in range(max_retries): try: processed_path = preprocess_image_process(image_path, enable_greyscale, enable_contrast_enhancement, contrast_factor) uploaded_file = genai.upload_file(processed_path) # Clean up temp file if it was created if processed_path != image_path: try: os.unlink(processed_path) except: pass return uploaded_file except Exception as e: if attempt == max_retries - 1: return None import random jitter = random.uniform(0.1, 0.5) sleep_time = (0.5 * (attempt + 1)) + jitter time.sleep(sleep_time) return None # Upload images master_file = upload_with_retry_process(master_path) layout_file = upload_with_retry_process(layout_path) if not master_file or not layout_file: raise Exception("Failed to upload images") # Create prompt prompt = f"""Analyze the layout image (the second image) and determine if the master image (the first image) appears in it. INSTRUCTIONS: 1. Compare the master image (first image) with the layout image (second image) 2. Look for EXACT matches where the model, clothing, and pose are IDENTICAL 3. The layout image may contain the master image in various forms: - Complete/exact match - Cropped version - Scaled or resized version - Rotated version - Partially obscured 4. Focus on visual similarity in terms of: - Person/model appearance and pose (must be EXACTLY the same) - Clothing details (colors, patterns, styles - must be EXACTLY the same) - Background and composition - Overall visual elements 5. CRITICAL: Only return a positive result if the models, pose, and clothing are EXACTLY the same. If there is ANY difference in clothing, model, or pose then return a negative result. Master Image ID: {master_id} Return your response as a JSON object with this exact format: {{ "match_found": true/false, "master_id": "{master_id}", "confidence": "high/medium/low", "analysis": "Detailed explanation of your findings and reasoning" }} IMPORTANT CONTEXT: This is a legitimate business application for marketing and e-commerce image matching. The images are product/marketing photos showing models in various clothing styles for retail purposes. This analysis is for content categorization in a business context and is completely benign. """ # Make API call with retry max_retries = 3 for attempt in range(max_retries): try: response = model.generate_content([prompt, master_file, layout_file], safety_settings=safety_settings) if not response.candidates: if attempt < max_retries - 1: time.sleep((2 ** attempt) * 0.5) continue else: raise Exception("No candidates returned from API") candidate = response.candidates[0] if candidate.finish_reason and candidate.finish_reason != 1: if attempt < max_retries - 1: time.sleep((2 ** attempt) * 0.5) continue else: raise Exception(f"Request finished with reason: {candidate.finish_reason}") # Parse response response_text = response.text.strip() start_idx = response_text.find('{') end_idx = response_text.rfind('}') + 1 if start_idx == -1 or end_idx == 0: raise ValueError("No JSON found in response") json_str = response_text[start_idx:end_idx] result = json.loads(json_str) # Validate result format if 'match_found' not in result: result['match_found'] = False if 'master_id' not in result: result['master_id'] = master_id if 'confidence' not in result: result['confidence'] = 'unknown' if 'analysis' not in result: result['analysis'] = response_text return result except Exception as e: if attempt == max_retries - 1: return { 'match_found': False, 'master_id': master_id, 'confidence': 'unknown', 'analysis': '', 'error': str(e) } time.sleep((2 ** attempt) * 0.5) except Exception as e: return { 'match_found': False, 'master_id': master_id, 'confidence': 'unknown', 'analysis': '', 'error': str(e) }