skin_analysis.py

import os
import json
import time
import hashlib
from typing import Optional, Dict, Any
from datetime import datetime
from google import genai
from google.genai.types import Part

# ==================== CONFIGURATION ====================
API_KEY = os.getenv("GEMINI_API_KEY")

# Use Gemini 2.0 Flash for cost efficiency (or 1.5 Pro for higher quality)
MODEL_COMBINED = "models/gemini-2.0-flash-exp"  # or "models/gemini-1.5-pro"

# Cache for avoiding duplicate analyses
_analysis_cache = {}

# Usage tracking
_usage_log = []


# ==================== HELPER FUNCTIONS ====================

def load_client():
    """Initialize Gemini client"""
    return genai.Client(api_key=API_KEY)


def get_image_hash(image_path: str) -> str:
    """Generate hash of image for caching"""
    with open(image_path, "rb") as f:
        return hashlib.md5(f.read()).hexdigest()


def log_api_usage(tokens_used: int, cost: float, success: bool = True):
    """Track API usage for monitoring"""
    _usage_log.append({
        "timestamp": datetime.now().isoformat(),
        "tokens": tokens_used,
        "cost": cost,
        "success": success
    })
    
    # Optional: Write to file
    with open("api_usage.log", "a") as f:
        f.write(f"{datetime.now()},{tokens_used},{cost},{success}\n")


def retry_with_backoff(func, max_retries: int = 3, initial_delay: float = 2.0):
    """
    Retry API calls with exponential backoff
    """
    delay = initial_delay
    
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            error_msg = str(e).lower()
            
            # Check if it's a retryable error
            is_retryable = any(keyword in error_msg for keyword in [
                "500", "503", "502", "504",
                "timeout", "overload", "unavailable",
                "internal error", "service unavailable"
            ])
            
            if is_retryable and attempt < max_retries - 1:
                wait_time = delay * (2 ** attempt)
                print(f"⚠️  Attempt {attempt + 1}/{max_retries} failed: {e}")
                print(f"   Retrying in {wait_time:.1f}s...")
                time.sleep(wait_time)
            elif attempt == max_retries - 1:
                print(f"❌ All {max_retries} attempts failed: {e}")
                raise
            else:
                # Non-retryable error, raise immediately
                raise
    
    return None


# ==================== MAIN ANALYSIS FUNCTION ====================

def analyze_skin_complete(
    image_path: str,
    use_cache: bool = True,
    max_retries: int = 3
) -> Optional[Dict[str, Any]]:
    """
    Complete skin analysis with ONE API call including pores and wrinkles.
    
    Args:
        image_path: Path to the facial image
        use_cache: Whether to use cached results if available
        max_retries: Number of retry attempts on failure
    
    Returns:
        Dictionary containing all analysis results, or None on failure
    """
    
    # Check cache first
    cache_key = f"complete_v2_{get_image_hash(image_path)}"
    if use_cache and cache_key in _analysis_cache:
        print("✓ Using cached analysis results")
        return _analysis_cache[cache_key]
    
    def _call():
        client = load_client()
        
        # Load image
        with open(image_path, "rb") as f:
            image_bytes = f.read()
        
        image_part = Part.from_bytes(data=image_bytes, mime_type="image/jpeg")
        
        # Combined comprehensive prompt with pores and wrinkles
        prompt = """
You are an advanced AI skin analysis system. Analyze the face in this image comprehensively.

Return STRICT JSON with ALL these fields (use exact field names):

{
  "hydration": {
    "texture": float (0.0-1.0, smoothness level),
    "radiance": float (0.0-1.0, natural glow),
    "flakiness": float (0.0-1.0, visible dry flakes - higher is worse),
    "oil_balance": float (0.0-1.0, healthy surface moisture),
    "fine_lines": float (0.0-1.0, dryness lines - higher is worse)
  },
  "pigmentation": {
    "dark_spots": float (0.0-1.0, severity of dark spots),
    "hyperpigmentation": float (0.0-1.0, overall hyperpigmentation),
    "under_eye_pigmentation": float (0.0-1.0, dark circles),
    "redness": float (0.0-1.0, skin redness),
    "melanin_unevenness": float (0.0-1.0, uneven melanin distribution),
    "uv_damage": float (0.0-1.0, visible UV damage),
    "overall_evenness": float (0.0-1.0, overall skin tone evenness)
  },
  "acne": {
    "active_acne": float (0.0-1.0, active breakouts),
    "comedones": float (0.0-1.0, blackheads/whiteheads),
    "cystic_acne": float (0.0-1.0, deep cystic acne),
    "inflammation": float (0.0-1.0, inflammatory response),
    "oiliness": float (0.0-1.0, excess sebum production),
    "scarring": float (0.0-1.0, acne scarring),
    "congestion": float (0.0-1.0, pore congestion)
  },
  "pores": {
    "visibility": float (0.0-1.0, how visible/prominent pores are),
    "size": float (0.0-1.0, average pore size - larger is worse),
    "enlarged_pores": float (0.0-1.0, percentage of enlarged pores),
    "clogged_pores": float (0.0-1.0, degree of pore clogging),
    "texture_roughness": float (0.0-1.0, roughness due to pores),
    "t_zone_prominence": float (0.0-1.0, pore visibility in T-zone),
    "cheek_prominence": float (0.0-1.0, pore visibility on cheeks)
  },
  "wrinkles": {
    "forehead_lines": float (0.0-1.0, horizontal forehead wrinkles),
    "frown_lines": float (0.0-1.0, glabellar lines between eyebrows),
    "crows_feet": float (0.0-1.0, eye corner wrinkles),
    "nasolabial_folds": float (0.0-1.0, nose-to-mouth lines),
    "marionette_lines": float (0.0-1.0, mouth-to-chin lines),
    "under_eye_wrinkles": float (0.0-1.0, fine lines under eyes),
    "lip_lines": float (0.0-1.0, perioral wrinkles around mouth),
    "neck_lines": float (0.0-1.0, horizontal neck wrinkles if visible),
    "overall_severity": float (0.0-1.0, overall wrinkle severity),
    "depth": float (0.0-1.0, average depth of wrinkles),
    "dynamic_wrinkles": float (0.0-1.0, expression-related wrinkles),
    "static_wrinkles": float (0.0-1.0, wrinkles at rest)
  },
  "age_analysis": {
    "fitzpatrick_type": integer (1-6, skin type based on melanin),
    "eye_age": integer (estimated age of eye area),
    "skin_age": integer (estimated overall skin age)
  }
}

DETAILED ANALYSIS GUIDELINES:

PORES:
- Assess pore visibility across different facial zones
- Consider pore size relative to skin type
- Note if pores appear stretched, enlarged, or clogged
- T-zone (forehead, nose, chin) typically has more prominent pores
- Cheeks may show different pore characteristics

WRINKLES:
- Distinguish between dynamic (expression) and static (at rest) wrinkles
- Forehead lines: horizontal lines across forehead
- Frown lines: vertical lines between eyebrows (11 lines)
- Crow's feet: radiating lines from outer eye corners
- Nasolabial folds: lines from nose to mouth corners
- Marionette lines: lines from mouth corners downward
- Assess depth (superficial vs deep wrinkles)
- Consider fine lines vs established wrinkles

CRITICAL RULES:
- Return ONLY raw JSON, no markdown formatting
- No explanations, no text outside JSON
- All float values must be between 0.0 and 1.0
- All integer values must be positive integers
- Base analysis ONLY on visible features in the image
- Do NOT guess or infer anything not visible
- Ensure all fields are present in the response
- If a feature is not visible or applicable, use 0.0
"""
        
        # Make API call
        start_time = time.time()
        response = client.models.generate_content(
            model=MODEL_COMBINED,
            contents=[prompt, image_part],
            config={
                "temperature": 0,
                "top_p": 1,
                "top_k": 1
            }
        )
        elapsed = time.time() - start_time
        
        # Parse response
        clean_text = response.text.strip()
        clean_text = clean_text.replace("```json", "").replace("```", "").strip()
        
        result = json.loads(clean_text)
        
        # Estimate tokens (rough estimate)
        estimated_tokens = len(prompt) / 4 + len(clean_text) / 4 + 1000  # +1000 for image
        cost = (estimated_tokens / 1_000_000) * 0.075  # Gemini 2.0 Flash pricing
        
        # Log usage
        log_api_usage(int(estimated_tokens), cost, success=True)
        
        print(f"✓ Analysis completed in {elapsed:.2f}s (est. cost: ${cost:.6f})")
        
        return result
    
    try:
        # Execute with retry logic
        result = retry_with_backoff(_call, max_retries=max_retries)
        
        if result and use_cache:
            _analysis_cache[cache_key] = result
        
        return result
        
    except Exception as e:
        print(f"❌ Failed to analyze skin after {max_retries} attempts: {e}")
        log_api_usage(0, 0, success=False)
        return None


# ==================== SCORE CALCULATION FUNCTIONS ====================

def compute_hydration_score(hydration_factors: Dict[str, float]) -> Optional[float]:
    """
    Calculate hydration score (0-100) from factors
    """
    if not hydration_factors:
        return None
    
    try:
        score = (
            hydration_factors["radiance"] * 30 +
            (1 - hydration_factors["flakiness"]) * 25 +
            (1 - hydration_factors["fine_lines"]) * 20 +
            hydration_factors["oil_balance"] * 15 +
            hydration_factors["texture"] * 10
        )
        return round(score, 1)
    except (KeyError, TypeError):
        return None


def compute_pigmentation_score(pigmentation_factors: Dict[str, float]) -> Optional[float]:
    """
    Calculate pigmentation score (0-100) from factors
    Higher score = more pigmentation issues
    """
    if not pigmentation_factors:
        return None
    
    try:
        score = (
            pigmentation_factors["hyperpigmentation"] * 30 +
            pigmentation_factors["dark_spots"] * 25 +
            pigmentation_factors["melanin_unevenness"] * 20 +
            pigmentation_factors["under_eye_pigmentation"] * 10 +
            pigmentation_factors["uv_damage"] * 10 +
            pigmentation_factors["redness"] * 5
        )
        return round(score, 1)
    except (KeyError, TypeError):
        return None


def compute_acne_score(acne_factors: Dict[str, float]) -> Optional[float]:
    """
    Calculate acne score (0-100) from factors
    Higher score = more acne issues
    """
    if not acne_factors:
        return None
    
    try:
        score = (
            acne_factors["active_acne"] * 40 +
            acne_factors["comedones"] * 20 +
            acne_factors["inflammation"] * 15 +
            acne_factors["cystic_acne"] * 15 +
            acne_factors["scarring"] * 10
        )
        return round(score, 1)
    except (KeyError, TypeError):
        return None


def compute_pores_score(pores_factors: Dict[str, float]) -> Optional[float]:
    """
    Calculate pores score (0-100) from factors
    Higher score = more visible/problematic pores
    """
    if not pores_factors:
        return None
    
    try:
        score = (
            pores_factors["visibility"] * 25 +
            pores_factors["size"] * 25 +
            pores_factors["enlarged_pores"] * 20 +
            pores_factors["clogged_pores"] * 15 +
            pores_factors["texture_roughness"] * 15
        )
        return round(score, 1)
    except (KeyError, TypeError):
        return None


def compute_wrinkles_score(wrinkles_factors: Dict[str, float]) -> Optional[float]:
    """
    Calculate wrinkles score (0-100) from factors
    Higher score = more severe wrinkling
    """
    if not wrinkles_factors:
        return None
    
    try:
        score = (
            wrinkles_factors["overall_severity"] * 30 +
            wrinkles_factors["depth"] * 20 +
            wrinkles_factors["forehead_lines"] * 10 +
            wrinkles_factors["crows_feet"] * 10 +
            wrinkles_factors["nasolabial_folds"] * 10 +
            wrinkles_factors["frown_lines"] * 8 +
            wrinkles_factors["static_wrinkles"] * 7 +
            wrinkles_factors["under_eye_wrinkles"] * 5
        )
        return round(score, 1)
    except (KeyError, TypeError):
        return None


def get_comprehensive_analysis(image_path: str) -> Optional[Dict[str, Any]]:
    """
    Get complete skin analysis with computed scores including pores and wrinkles.
    
    Returns a dictionary with all raw factors plus computed scores.
    """
    raw_analysis = analyze_skin_complete(image_path)
    
    if not raw_analysis:
        return None
    
    # Add computed scores
    result = {
        "raw_data": raw_analysis,
        "scores": {
            "hydration": compute_hydration_score(raw_analysis.get("hydration")),
            "pigmentation": compute_pigmentation_score(raw_analysis.get("pigmentation")),
            "acne": compute_acne_score(raw_analysis.get("acne")),
            "pores": compute_pores_score(raw_analysis.get("pores")),
            "wrinkles": compute_wrinkles_score(raw_analysis.get("wrinkles"))
        },
        "age_analysis": raw_analysis.get("age_analysis"),
        "metadata": {
            "analyzed_at": datetime.now().isoformat(),
            "model_used": MODEL_COMBINED
        }
    }
    
    return result


# ==================== LEGACY COMPATIBILITY FUNCTIONS ====================

def get_hydration_factors(image_path: str) -> Optional[Dict[str, float]]:
    """Legacy function - extracts hydration from complete analysis"""
    result = analyze_skin_complete(image_path)
    return result.get("hydration") if result else None


def get_pigmentation_factors(image_path: str) -> Optional[Dict[str, float]]:
    """Legacy function - extracts pigmentation from complete analysis"""
    result = analyze_skin_complete(image_path)
    return result.get("pigmentation") if result else None


def get_acne_factors(image_path: str) -> Optional[Dict[str, float]]:
    """Legacy function - extracts acne from complete analysis"""
    result = analyze_skin_complete(image_path)
    return result.get("acne") if result else None


def get_pores_factors(image_path: str) -> Optional[Dict[str, float]]:
    """Get pores analysis from complete analysis"""
    result = analyze_skin_complete(image_path)
    return result.get("pores") if result else None


def get_wrinkles_factors(image_path: str) -> Optional[Dict[str, float]]:
    """Get wrinkles analysis from complete analysis"""
    result = analyze_skin_complete(image_path)
    return result.get("wrinkles") if result else None


def get_fitzpatrick_type(image_path: str) -> Optional[int]:
    """Legacy function - extracts Fitzpatrick type from complete analysis"""
    result = analyze_skin_complete(image_path)
    if result and "age_analysis" in result:
        return result["age_analysis"].get("fitzpatrick_type")
    return None


def get_eye_age(image_path: str) -> Optional[int]:
    """Legacy function - extracts eye age from complete analysis"""
    result = analyze_skin_complete(image_path)
    if result and "age_analysis" in result:
        return result["age_analysis"].get("eye_age")
    return None


def get_skin_age(image_path: str) -> Optional[int]:
    """Legacy function - extracts skin age from complete analysis"""
    result = analyze_skin_complete(image_path)
    if result and "age_analysis" in result:
        return result["age_analysis"].get("skin_age")
    return None


# ==================== DETAILED ANALYSIS FUNCTIONS ====================

def get_wrinkle_breakdown(image_path: str) -> Optional[Dict[str, Any]]:
    """
    Get detailed breakdown of wrinkles by type and severity
    """
    wrinkles = get_wrinkles_factors(image_path)
    if not wrinkles:
        return None
    
    # Categorize wrinkles by location
    upper_face = {
        "forehead_lines": wrinkles.get("forehead_lines", 0),
        "frown_lines": wrinkles.get("frown_lines", 0),
        "average": (wrinkles.get("forehead_lines", 0) + wrinkles.get("frown_lines", 0)) / 2
    }
    
    eye_area = {
        "crows_feet": wrinkles.get("crows_feet", 0),
        "under_eye_wrinkles": wrinkles.get("under_eye_wrinkles", 0),
        "average": (wrinkles.get("crows_feet", 0) + wrinkles.get("under_eye_wrinkles", 0)) / 2
    }
    
    lower_face = {
        "nasolabial_folds": wrinkles.get("nasolabial_folds", 0),
        "marionette_lines": wrinkles.get("marionette_lines", 0),
        "lip_lines": wrinkles.get("lip_lines", 0),
        "average": (wrinkles.get("nasolabial_folds", 0) + 
                   wrinkles.get("marionette_lines", 0) + 
                   wrinkles.get("lip_lines", 0)) / 3
    }
    
    # Categorize by type
    wrinkle_types = {
        "dynamic": wrinkles.get("dynamic_wrinkles", 0),
        "static": wrinkles.get("static_wrinkles", 0),
        "predominant_type": "static" if wrinkles.get("static_wrinkles", 0) > wrinkles.get("dynamic_wrinkles", 0) else "dynamic"
    }
    
    return {
        "by_location": {
            "upper_face": upper_face,
            "eye_area": eye_area,
            "lower_face": lower_face
        },
        "by_type": wrinkle_types,
        "severity": {
            "depth": wrinkles.get("depth", 0),
            "overall": wrinkles.get("overall_severity", 0)
        },
        "most_affected_area": max(
            [("upper_face", upper_face["average"]),
             ("eye_area", eye_area["average"]),
             ("lower_face", lower_face["average"])],
            key=lambda x: x[1]
        )[0]
    }


def get_pores_breakdown(image_path: str) -> Optional[Dict[str, Any]]:
    """
    Get detailed breakdown of pores by zone and characteristics
    """
    pores = get_pores_factors(image_path)
    if not pores:
        return None
    
    # Analyze by facial zone
    zones = {
        "t_zone": {
            "prominence": pores.get("t_zone_prominence", 0),
            "severity": "severe" if pores.get("t_zone_prominence", 0) > 0.7 
                       else "moderate" if pores.get("t_zone_prominence", 0) > 0.4 
                       else "mild"
        },
        "cheeks": {
            "prominence": pores.get("cheek_prominence", 0),
            "severity": "severe" if pores.get("cheek_prominence", 0) > 0.7 
                       else "moderate" if pores.get("cheek_prominence", 0) > 0.4 
                       else "mild"
        }
    }
    
    # Overall pore characteristics
    characteristics = {
        "size": pores.get("size", 0),
        "visibility": pores.get("visibility", 0),
        "enlarged_percentage": pores.get("enlarged_pores", 0) * 100,
        "clogging_level": pores.get("clogged_pores", 0)
    }
    
    # Recommendations based on severity
    severity_score = (pores.get("visibility", 0) + pores.get("size", 0)) / 2
    
    return {
        "by_zone": zones,
        "characteristics": characteristics,
        "texture_impact": pores.get("texture_roughness", 0),
        "overall_severity": "severe" if severity_score > 0.7 
                          else "moderate" if severity_score > 0.4 
                          else "mild",
        "most_affected_zone": "t_zone" if zones["t_zone"]["prominence"] > zones["cheeks"]["prominence"] else "cheeks"
    }


# ==================== USAGE STATISTICS ====================

def get_usage_stats() -> Dict[str, Any]:
    """Get API usage statistics"""
    if not _usage_log:
        return {"message": "No usage data available"}
    
    total_calls = len(_usage_log)
    successful_calls = sum(1 for log in _usage_log if log["success"])
    total_cost = sum(log["cost"] for log in _usage_log)
    total_tokens = sum(log["tokens"] for log in _usage_log)
    
    return {
        "total_calls": total_calls,
        "successful_calls": successful_calls,
        "failed_calls": total_calls - successful_calls,
        "success_rate": f"{(successful_calls/total_calls*100):.1f}%",
        "total_cost": f"${total_cost:.6f}",
        "total_tokens": total_tokens,
        "avg_cost_per_call": f"${(total_cost/total_calls):.6f}" if total_calls > 0 else "$0"
    }


def clear_cache():
    """Clear the analysis cache"""
    global _analysis_cache
    _analysis_cache = {}
    print("✓ Cache cleared")


# ==================== EXAMPLE USAGE ====================

if __name__ == "__main__":
    # Example 1: Get complete analysis with pores and wrinkles
    print("=" * 70)
    print("COMPLETE SKIN ANALYSIS (INCLUDING PORES & WRINKLES)")
    print("=" * 70)
    
    image_path = "path/to/face_image.jpg"
    
    # Single API call for everything
    analysis = get_comprehensive_analysis(image_path)
    
    if analysis:
        print("\n✓ Analysis successful!")
        print(f"\n{'='*70}")
        print("SCORES (0-100, higher = more concern)")
        print(f"{'='*70}")
        print(f"Hydration Score:      {analysis['scores']['hydration']:.1f}/100")
        print(f"Pigmentation Score:   {analysis['scores']['pigmentation']:.1f}/100")
        print(f"Acne Score:           {analysis['scores']['acne']:.1f}/100")
        print(f"Pores Score:          {analysis['scores']['pores']:.1f}/100")
        print(f"Wrinkles Score:       {analysis['scores']['wrinkles']:.1f}/100")
        
        print(f"\n{'='*70}")
        print("AGE ANALYSIS")
        print(f"{'='*70}")
        print(f"Fitzpatrick Type:     {analysis['age_analysis']['fitzpatrick_type']}")
        print(f"Eye Age:              {analysis['age_analysis']['eye_age']} years")
        print(f"Skin Age:             {analysis['age_analysis']['skin_age']} years")
        
        # Save to file
        with open("analysis_result.json", "w") as f:
            json.dump(analysis, f, indent=2)
        print("\n✓ Results saved to analysis_result.json")
    else:
        print("\n❌ Analysis failed")
    
    # Example 2: Get detailed pores breakdown
    print("\n" + "=" * 70)
    print("DETAILED PORES ANALYSIS")
    print("=" * 70)
    
    pores_detail = get_pores_breakdown(image_path)
    if pores_detail:
        print(f"\nMost Affected Zone:   {pores_detail['most_affected_zone'].upper()}")
        print(f"Overall Severity:     {pores_detail['overall_severity'].upper()}")
        print(f"\nT-Zone Prominence:    {pores_detail['by_zone']['t_zone']['prominence']:.2f}")
        print(f"Cheek Prominence:     {pores_detail['by_zone']['cheeks']['prominence']:.2f}")
        print(f"Enlarged Pores:       {pores_detail['characteristics']['enlarged_percentage']:.1f}%")
        print(f"Pore Size:            {pores_detail['characteristics']['size']:.2f}")
        print(f"Visibility:           {pores_detail['characteristics']['visibility']:.2f}")
    
    # Example 3: Get detailed wrinkles breakdown
    print("\n" + "=" * 70)
    print("DETAILED WRINKLES ANALYSIS")
    print("=" * 70)
    
    wrinkles_detail = get_wrinkle_breakdown(image_path)
    if wrinkles_detail:
        print(f"\nMost Affected Area:   {wrinkles_detail['most_affected_area'].upper()}")
        print(f"Predominant Type:     {wrinkles_detail['by_type']['predominant_type'].upper()}")
        print(f"Overall Severity:     {wrinkles_detail['severity']['overall']:.2f}")
        print(f"Average Depth:        {wrinkles_detail['severity']['depth']:.2f}")
        
        print(f"\nUpper Face (avg):     {wrinkles_detail['by_location']['upper_face']['average']:.2f}")
        print(f"  - Forehead:         {wrinkles_detail['by_location']['upper_face']['forehead_lines']:.2f}")
        print(f"  - Frown Lines:      {wrinkles_detail['by_location']['upper_face']['frown_lines']:.2f}")
        
        print(f"\nEye Area (avg):       {wrinkles_detail['by_location']['eye_area']['average']:.2f}")
        print(f"  - Crow's Feet:      {wrinkles_detail['by_location']['eye_area']['crows_feet']:.2f}")
        print(f"  - Under Eye:        {wrinkles_detail['by_location']['eye_area']['under_eye_wrinkles']:.2f}")
        
        print(f"\nLower Face (avg):     {wrinkles_detail['by_location']['lower_face']['average']:.2f}")
        print(f"  - Nasolabial:       {wrinkles_detail['by_location']['lower_face']['nasolabial_folds']:.2f}")
        print(f"  - Marionette:       {wrinkles_detail['by_location']['lower_face']['marionette_lines']:.2f}")
    
    # Example 4: Using legacy functions (still works, uses same cached data)
    print("\n" + "=" * 70)
    print("USING LEGACY FUNCTIONS")
    print("=" * 70)
    
    hydration = get_hydration_factors(image_path)
    print(f"\nHydration factors: {hydration}")
    
    pigmentation = get_pigmentation_factors(image_path)
    print(f"Pigmentation factors: {pigmentation}")
    
    pores = get_pores_factors(image_path)
    print(f"Pores factors: {pores}")
    
    wrinkles = get_wrinkles_factors(image_path)
    print(f"Wrinkles factors: {wrinkles}")
    
    # Example 5: Check usage statistics
    print("\n" + "=" * 70)
    print("API USAGE STATISTICS")
    print("=" * 70)
    stats = get_usage_stats()
    print(json.dumps(stats, indent=2))