lambda_handler.py

import json
import os
import subprocess
import tempfile
import urllib.request
import logging
from typing import Dict, Any

# Configure logging
logger = logging.getLogger()
logger.setLevel(logging.INFO)

def download_audio(url: str, duration: int, output_path: str) -> bool:
    """
    Download audio from a stream URL using ffmpeg.
    
    Args:
        url: The stream URL to download from
        duration: Duration in seconds to record
        output_path: Path to save the audio file
    
    Returns:
        bool: True if successful, False otherwise
    """
    try:
        # Use ffmpeg to download the stream for the specified duration
        cmd = [
            'ffmpeg',
            '-i', url,
            '-t', str(duration),  # Duration limit
            '-c:a', 'mp3',        # Audio codec
            '-y',                  # Overwrite output file
            output_path
        ]
        
        logger.info(f"Downloading audio from {url} for {duration} seconds")
        result = subprocess.run(cmd, capture_output=True, text=True, timeout=duration + 30)
        
        if result.returncode != 0:
            logger.error(f"ffmpeg failed: {result.stderr}")
            return False
            
        logger.info(f"Successfully downloaded audio to {output_path}")
        return True
        
    except subprocess.TimeoutExpired:
        logger.error("Download timed out")
        return False
    except Exception as e:
        logger.error(f"Error downloading audio: {str(e)}")
        return False

def run_chromaprint(audio_file: str) -> dict:
    try:
        fpcalc_path = '/opt/layer/bin/fpcalc'
        if not os.path.exists(fpcalc_path):
            fpcalc_path = 'fpcalc'
        cmd = [fpcalc_path, '-json', audio_file]
        result = subprocess.run(cmd, capture_output=True, text=True)
        if result.returncode != 0:
            logger.error(f"fpcalc failed: {result.stderr}")
            return {"error": "Failed to generate fingerprint"}
        logger.info(f"fpcalc raw output: {result.stdout}")
        # Parse the JSON output directly!
        try:
            fingerprint_data = json.loads(result.stdout)
        except Exception as e:
            logger.error(f"Failed to parse fpcalc JSON: {e}")
            return {"error": "Failed to parse fpcalc JSON"}
        logger.info(f"Final fingerprint data: {fingerprint_data}")
        return fingerprint_data
    except Exception as e:
        logger.error(f"Error running chromaprint: {str(e)}")
        return {"error": f"Failed to run chromaprint: {str(e)}"}

def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
    """
    AWS Lambda handler function.
    
    Expected event format:
    {
        "stream_url": "https://example.com/stream",
        "duration": 30
    }
    
    Returns:
        Dict containing the fingerprint and metadata
    """
    try:
        # Parse input
        body = event.get('body', '{}')
        if isinstance(body, str):
            body = json.loads(body)
        
        stream_url = body.get('stream_url')
        duration = body.get('duration', 30)  # Default 30 seconds
        
        if not stream_url:
            return {
                'statusCode': 400,
                'body': json.dumps({
                    'error': 'stream_url is required'
                })
            }
        
        # Validate duration
        if not isinstance(duration, int) or duration <= 0 or duration > 300:
            return {
                'statusCode': 400,
                'body': json.dumps({
                    'error': 'duration must be a positive integer between 1 and 300'
                })
            }
        
        logger.info(f"Processing request: URL={stream_url}, Duration={duration}")
        
        # Create temporary directory for audio file
        with tempfile.TemporaryDirectory() as temp_dir:
            audio_file = os.path.join(temp_dir, 'audio.mp3')
            
            # Download audio
            if not download_audio(stream_url, duration, audio_file):
                return {
                    'statusCode': 500,
                    'body': json.dumps({
                        'error': 'Failed to download audio stream'
                    })
                }
            
            # Check if file was created and has content
            if not os.path.exists(audio_file) or os.path.getsize(audio_file) == 0:
                return {
                    'statusCode': 500,
                    'body': json.dumps({
                        'error': 'Downloaded audio file is empty or missing'
                    })
                }
            
            # Run chromaprint
            fingerprint_result = run_chromaprint(audio_file)
            
            if 'error' in fingerprint_result:
                return {
                    'statusCode': 500,
                    'body': json.dumps(fingerprint_result)
                }
            
            # Add metadata
            result = {
                'streamUrl': stream_url,
                'duration': fingerprint_result['duration'],
                'fingerprint': fingerprint_result['fingerprint']
            } 
            
            return {
                'statusCode': 200,
                'body': json.dumps(result),
                'headers': {
                    'Content-Type': 'application/json'
                }
            }
            
    except json.JSONDecodeError:
        return {
            'statusCode': 400,
            'body': json.dumps({
                'error': 'Invalid JSON in request body'
            })
        }
    except Exception as e:
        logger.error(f"Unexpected error: {str(e)}")
        return {
            'statusCode': 500,
            'body': json.dumps({
                'error': f'Internal server error: {str(e)}'
            })
        }