master_adapt_detect/process_detection.py

#!/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)
        }