mule-image/backend/app/services/metadata.py

"""
Metadata extraction service using ExifTool
"""
import json
import logging
import asyncio
from datetime import datetime
from typing import Dict, Optional
import subprocess
from pathlib import Path

from celery import shared_task
from sqlalchemy import select

from app.database import AsyncSessionLocal
from app.models import Photo

logger = logging.getLogger(__name__)

def parse_exif_datetime(date_str: str) -> Optional[datetime]:
    """Parse EXIF datetime string to Python datetime"""
    if not date_str:
        return None
    
    # Common EXIF datetime formats
    formats = [
        "%Y:%m:%d %H:%M:%S",
        "%Y-%m-%d %H:%M:%S",
        "%Y:%m:%d %H:%M:%S.%f",
        "%Y-%m-%dT%H:%M:%S",
        "%Y-%m-%dT%H:%M:%S.%f",
        "%Y-%m-%dT%H:%M:%S%z"
    ]
    
    for fmt in formats:
        try:
            return datetime.strptime(date_str, fmt)
        except ValueError:
            continue
    
    return None

def extract_key_metadata(exif_data: Dict) -> Dict:
    """Extract key metadata fields for FTS indexing"""
    key_fields = []
    
    # Camera information
    if 'Make' in exif_data:
        key_fields.append(exif_data['Make'])
    if 'Model' in exif_data:
        key_fields.append(exif_data['Model'])
    if 'LensModel' in exif_data:
        key_fields.append(exif_data['LensModel'])
    
    # Location information
    if 'GPSLatitude' in exif_data and 'GPSLongitude' in exif_data:
        key_fields.append(f"GPS: {exif_data['GPSLatitude']}, {exif_data['GPSLongitude']}")
    
    # IPTC/XMP keywords
    if 'Keywords' in exif_data:
        if isinstance(exif_data['Keywords'], list):
            key_fields.extend(exif_data['Keywords'])
        else:
            key_fields.append(exif_data['Keywords'])
    
    # Copyright and creator
    if 'Copyright' in exif_data:
        key_fields.append(exif_data['Copyright'])
    if 'Creator' in exif_data:
        key_fields.append(exif_data['Creator'])
    if 'Artist' in exif_data:
        key_fields.append(exif_data['Artist'])
    
    return {
        'exif_text': ' '.join(key_fields),
        'camera_make': exif_data.get('Make'),
        'camera_model': exif_data.get('Model'),
        'lens_model': exif_data.get('LensModel'),
        'gps_latitude': exif_data.get('GPSLatitude'),
        'gps_longitude': exif_data.get('GPSLongitude'),
    }

@shared_task(name='extract_metadata')
def extract_metadata(photo_id: str):
    """Extract metadata from a photo using ExifTool"""
    return asyncio.run(_extract_metadata_async(photo_id))

async def _extract_metadata_async(photo_id: str):
    """Async implementation of metadata extraction"""
    async with AsyncSessionLocal() as session:
        try:
            # Get photo from database
            result = await session.execute(
                select(Photo).where(Photo.id == photo_id)
            )
            photo = result.scalar_one_or_none()
            
            if not photo:
                logger.error(f"Photo not found: {photo_id}")
                return {'status': 'error', 'message': 'Photo not found'}
            
            # Check if file exists
            if not Path(photo.filepath).exists():
                logger.error(f"File not found: {photo.filepath}")
                return {'status': 'error', 'message': 'File not found'}
            
            # Run ExifTool to extract metadata
            cmd = [
                'exiftool',
                '-j',  # JSON output
                '-G',  # Group names
                '-s',  # Short output format
                '-All',  # All metadata
                photo.filepath
            ]
            
            try:
                result = subprocess.run(
                    cmd,
                    capture_output=True,
                    text=True,
                    timeout=30
                )
                
                if result.returncode != 0:
                    logger.error(f"ExifTool error: {result.stderr}")
                    return {'status': 'error', 'message': result.stderr}
                
                # Parse JSON output
                metadata = json.loads(result.stdout)
                if metadata and len(metadata) > 0:
                    exif_data = metadata[0]
                    
                    # Store full metadata as JSON
                    photo.exif_json = json.dumps(exif_data)
                    
                    # Extract taken_at date
                    date_fields = [
                        'EXIF:DateTimeOriginal',
                        'EXIF:CreateDate',
                        'QuickTime:MediaCreateDate',
                        'EXIF:ModifyDate'
                    ]
                    
                    for field in date_fields:
                        if field in exif_data:
                            taken_at = parse_exif_datetime(exif_data[field])
                            if taken_at:
                                photo.taken_at = taken_at
                                photo.taken_at_source = 'exif'
                                break
                    
                    # Extract dimensions if not already set
                    if not photo.width:
                        photo.width = exif_data.get('EXIF:ImageWidth') or exif_data.get('File:ImageWidth')
                    if not photo.height:
                        photo.height = exif_data.get('EXIF:ImageHeight') or exif_data.get('File:ImageHeight')
                    
                    # Extract and store key metadata for search
                    key_metadata = extract_key_metadata(exif_data)
                    
                    # Update FTS table (would be done via trigger in production)
                    # For now, we'll store it in a comment
                    
                    await session.commit()
                    
                    logger.info(f"Metadata extracted for photo {photo_id}")
                    return {
                        'status': 'success',
                        'photo_id': photo_id,
                        'taken_at': photo.taken_at.isoformat() if photo.taken_at else None
                    }
                
            except subprocess.TimeoutExpired:
                logger.error(f"ExifTool timeout for {photo.filepath}")
                return {'status': 'error', 'message': 'ExifTool timeout'}
            except json.JSONDecodeError as e:
                logger.error(f"Failed to parse ExifTool output: {e}")
                return {'status': 'error', 'message': 'Invalid ExifTool output'}
            
        except Exception as e:
            logger.error(f"Error extracting metadata for {photo_id}: {e}")
            return {'status': 'error', 'message': str(e)}