app.py

#!/usr/bin/env python3
"""
Madverse Music API
AI Music Detection Service
"""

from fastapi import FastAPI, HTTPException, BackgroundTasks, Header, Depends
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from pydantic import BaseModel, HttpUrl
import torch
import librosa
import tempfile
import os
import requests
from pathlib import Path
import time
from typing import Optional, Annotated, List, Union
import uvicorn
import asyncio
from contextlib import asynccontextmanager
import socket

# Global model variable
model = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    """Application lifespan management"""
    # Startup
    global model
    try:
        from sonics import HFAudioClassifier
        print("🔄 Loading Madverse Music AI model...")
        model = HFAudioClassifier.from_pretrained("awsaf49/sonics-spectttra-alpha-120s")
        model.eval()
        print("✅ Model loaded successfully!")
    except Exception as e:
        print(f"❌ Failed to load model: {e}")
        raise
    
    yield
    
    # Shutdown
    print("🔄 Shutting down...")

# Initialize FastAPI app with lifespan
app = FastAPI(
    title="Madverse Music API",
    description="AI-powered music detection API to identify AI-generated vs human-created music",
    version="1.0.0",
    docs_url="/",
    redoc_url="/docs",
    lifespan=lifespan
)

# API Key Configuration
API_KEY = os.getenv("MADVERSE_API_KEY", "madverse-music-api-key-2024")  # Default key for demo

async def verify_api_key(x_api_key: Annotated[Union[str, None], Header()] = None):
    """Verify API key from header"""
    if x_api_key is None:
        raise HTTPException(
            status_code=401,
            detail="Missing API key. Please provide a valid X-API-Key header."
        )
    if x_api_key != API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key. Please provide a valid X-API-Key header."
        )
    return x_api_key

class MusicAnalysisRequest(BaseModel):
    urls: List[HttpUrl]

def check_api_key_first(request: MusicAnalysisRequest, x_api_key: Annotated[Union[str, None], Header()] = None):
    """Check API key before processing request"""
    if x_api_key is None:
        raise HTTPException(
            status_code=401,
            detail="Missing API key. Please provide a valid X-API-Key header."
        )
    if x_api_key != API_KEY:
        raise HTTPException(
            status_code=401,
            detail="Invalid API key. Please provide a valid X-API-Key header."
        )
    return request
    
class FileAnalysisResult(BaseModel):
    url: str
    success: bool
    classification: Optional[str] = None  # "Real" or "Fake"
    confidence: Optional[float] = None    # 0.0 to 1.0
    probability: Optional[float] = None   # Raw sigmoid probability
    raw_score: Optional[float] = None     # Raw model output
    duration: Optional[float] = None      # Audio duration in seconds
    message: str
    processing_time: Optional[float] = None
    error: Optional[str] = None

class MusicAnalysisResponse(BaseModel):
    success: bool
    total_files: int
    successful_analyses: int
    failed_analyses: int
    results: List[FileAnalysisResult]
    total_processing_time: float
    message: str

class ErrorResponse(BaseModel):
    success: bool
    error: str
    message: str

def cleanup_file(file_path: str):
    """Background task to cleanup temporary files"""
    try:
        if os.path.exists(file_path):
            os.unlink(file_path)
    except:
        pass

def download_audio(url: str, max_size_mb: int = 100) -> str:
    """Download audio file from URL with size validation"""
    try:
        # Check if URL is accessible
        response = requests.head(str(url), timeout=10)
        
        # Check content size
        content_length = response.headers.get('Content-Length')
        if content_length and int(content_length) > max_size_mb * 1024 * 1024:
            raise HTTPException(
                status_code=413, 
                detail=f"File too large. Maximum size: {max_size_mb}MB"
            )
        
        # Download file
        response = requests.get(str(url), timeout=30, stream=True)
        response.raise_for_status()
        
        # Create temporary file
        with tempfile.NamedTemporaryFile(delete=False, suffix='.tmp') as tmp_file:
            downloaded_size = 0
            for chunk in response.iter_content(chunk_size=8192):
                downloaded_size += len(chunk)
                if downloaded_size > max_size_mb * 1024 * 1024:
                    os.unlink(tmp_file.name)
                    raise HTTPException(
                        status_code=413,
                        detail=f"File too large. Maximum size: {max_size_mb}MB"
                    )
                tmp_file.write(chunk)
            
            return tmp_file.name
            
    except requests.exceptions.RequestException as e:
        raise HTTPException(
            status_code=400,
            detail=f"Failed to download audio: {str(e)}"
        )
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Error downloading file: {str(e)}"
        )

def classify_audio(file_path: str) -> dict:
    """Classify audio file using the AI model"""
    try:
        # Load audio (model uses 16kHz sample rate)
        audio, sr = librosa.load(file_path, sr=16000)
        
        # Convert to tensor and add batch dimension
        audio_tensor = torch.FloatTensor(audio).unsqueeze(0)
        
        # Get prediction
        with torch.no_grad():
            output = model(audio_tensor)
            
            # Convert logit to probability using sigmoid
            prob = torch.sigmoid(output).item()
            
            # Classify: prob < 0.5 = Real, prob >= 0.5 = Fake
            if prob < 0.5:
                classification = "Real"
                confidence = (1 - prob) * 2  # Convert to 0-1 scale
            else:
                classification = "Fake"
                confidence = (prob - 0.5) * 2  # Convert to 0-1 scale
        
        return {
            "classification": classification,
            "confidence": min(confidence, 1.0),  # Cap at 1.0
            "probability": prob,
            "raw_score": output.item(),
            "duration": len(audio) / sr
        }
        
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Error analyzing audio: {str(e)}"
        )

async def process_single_url(url: str) -> FileAnalysisResult:
    """Process a single URL and return result"""
    start_time = time.time()
    
    try:
        # Download audio file
        temp_file = download_audio(url)
        
        # Classify audio
        result = classify_audio(temp_file)
        
        # Calculate processing time
        processing_time = time.time() - start_time
        
        # Cleanup file in background
        try:
            os.unlink(temp_file)
        except:
            pass
        
        # Prepare response
        emoji = "🎤" if result["classification"] == "Real" else "🤖"
        message = f'{emoji} Detected as {result["classification"].lower()} music'
        
        return FileAnalysisResult(
            url=str(url),
            success=True,
            classification=result["classification"],
            confidence=result["confidence"],
            probability=result["probability"],
            raw_score=result["raw_score"],
            duration=result["duration"],
            message=message,
            processing_time=processing_time
        )
        
    except Exception as e:
        processing_time = time.time() - start_time
        error_msg = str(e)
        
        return FileAnalysisResult(
            url=str(url),
            success=False,
            message=f"❌ Failed to process: {error_msg}",
            processing_time=processing_time,
            error=error_msg
        )

@app.post("/analyze", response_model=MusicAnalysisResponse)
async def analyze_music(
    request: MusicAnalysisRequest = Depends(check_api_key_first)
):
    """
    Analyze music from URL(s) to detect if it's AI-generated or human-created
    
    - **urls**: Array of direct URLs to audio files (MP3, WAV, FLAC, M4A, OGG)
    - Returns classification results for each file
    - Processes files concurrently for better performance when multiple URLs provided
    """
    start_time = time.time()
    
    if not model:
        raise HTTPException(
            status_code=503,
            detail="Model not loaded. Please try again later."
        )
    
    if len(request.urls) > 50:  # Limit processing
        raise HTTPException(
            status_code=400,
            detail="Too many URLs. Maximum 50 files per request."
        )
    
    if len(request.urls) == 0:
        raise HTTPException(
            status_code=400,
            detail="At least one URL is required."
        )
    
    try:
        # Process all URLs concurrently with limited concurrency
        semaphore = asyncio.Semaphore(5)  # Limit to 5 concurrent downloads
        
        async def process_with_semaphore(url):
            async with semaphore:
                return await process_single_url(str(url))
        
        # Create tasks for all URLs
        tasks = [process_with_semaphore(url) for url in request.urls]
        
        # Wait for all tasks to complete
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # Process results and handle any exceptions
        processed_results = []
        successful_count = 0
        failed_count = 0
        
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                # Handle exception case
                processed_results.append(FileAnalysisResult(
                    url=str(request.urls[i]),
                    success=False,
                    message=f"❌ Processing failed: {str(result)}",
                    error=str(result)
                ))
                failed_count += 1
            else:
                processed_results.append(result)
                if result.success:
                    successful_count += 1
                else:
                    failed_count += 1
        
        # Calculate total processing time
        total_processing_time = time.time() - start_time
        
        # Prepare summary message
        total_files = len(request.urls)
        if total_files == 1:
            # Single file message
            if successful_count == 1:
                message = processed_results[0].message
            else:
                message = processed_results[0].message
        else:
            # Multiple files message
            if successful_count == total_files:
                message = f"✅ Successfully analyzed all {total_files} files"
            elif successful_count > 0:
                message = f"⚠️ Analyzed {successful_count}/{total_files} files successfully"
            else:
                message = f"❌ Failed to analyze any files"
        
        return MusicAnalysisResponse(
            success=successful_count > 0,
            total_files=total_files,
            successful_analyses=successful_count,
            failed_analyses=failed_count,
            results=processed_results,
            total_processing_time=total_processing_time,
            message=message
        )
        
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Internal server error during processing: {str(e)}"
        )

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {
        "status": "healthy",
        "model_loaded": model is not None,
        "service": "Madverse Music API"
    }

@app.get("/info")
async def get_info():
    """Get API information"""
    return {
        "name": "Madverse Music API",
        "version": "1.0.0",
        "description": "AI-powered music detection to identify AI-generated vs human-created music",
        "model": "SpecTTTra-α (120s)",
        "accuracy": {
            "f1_score": 0.97,
            "sensitivity": 0.96,
            "specificity": 0.99
        },
        "supported_formats": ["MP3", "WAV", "FLAC", "M4A", "OGG"],
        "max_file_size": "100MB",
        "max_duration": "120 seconds",
        "authentication": {
            "required": True,
            "type": "API Key",
            "header": "X-API-Key",
            "example": "X-API-Key: your-api-key-here"
        },
        "usage": {
            "curl_example": "curl -X POST 'http://localhost:8000/analyze' -H 'X-API-Key: your-api-key' -H 'Content-Type: application/json' -d '{\"url\":\"https://example.com/song.mp3\"}'"
        }
    }

def find_available_port(start_port: int = 8000, max_attempts: int = 10) -> int:
    """Find an available port starting from start_port"""
    import random
    import time
    
    # Add some randomization to avoid race conditions
    time.sleep(random.uniform(0.1, 0.5))
    
    for port in range(start_port, start_port + max_attempts):
        try:
            # Try to bind to the port with proper error handling
            with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
                s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                s.bind(('0.0.0.0', port))
                s.listen(1)
                print(f"✅ Port {port} is available")
                return port
        except OSError as e:
            print(f"❌ Port {port} is busy: {e}")
            continue
    
    # If no port found, raise an exception
    raise RuntimeError(f"No available port found in range {start_port}-{start_port + max_attempts - 1}")

if __name__ == "__main__":
    try:
        # Check if we're in a Hugging Face environment
        is_hf_space = os.getenv('SPACE_ID') is not None
        hf_port = os.getenv('PORT')  # HF Spaces sets this
        
        if is_hf_space and hf_port:
            # Use HF Spaces assigned port
            port = int(hf_port)
            print(f"🤗 Running in Hugging Face Spaces on port {port}")
        elif is_hf_space:
            print("🤗 Running in Hugging Face Spaces environment")
            # Use standard HF Spaces port
            port = 7860
        else:
            # Find an available port for local development
            port = find_available_port(8000, 10)
            
        print(f"🚀 Starting server on port {port}")
        
        # For HF Spaces, don't use the retry logic as it might cause issues
        if is_hf_space:
            uvicorn.run(app, host="0.0.0.0", port=port)
        else:
            # Add retry logic for local development
            max_retries = 3
            for attempt in range(max_retries):
                try:
                    uvicorn.run(app, host="0.0.0.0", port=port)
                    break  # If successful, break out of retry loop
                except OSError as e:
                    if "Address already in use" in str(e) and attempt < max_retries - 1:
                        print(f"⚠️ Port {port} became busy, trying next port...")
                        port = find_available_port(port + 1, 10)
                        print(f"🔄 Retrying on port {port}")
                    else:
                        raise
                    
    except RuntimeError as e:
        print(f"❌ {e}")
        print("💡 Suggestions:")
        print("   1. Wait a moment and try again (another instance might be shutting down)")
        print("   2. Manually specify a different port:")
        print("      uvicorn app:app --host 0.0.0.0 --port 8001")
        print("   3. Check for running processes: ps aux | grep python")
    except KeyboardInterrupt:
        print("\n🛑 Server stopped by user")
    except Exception as e:
        print(f"❌ Failed to start server: {e}")
        import traceback
        traceback.print_exc()