Delete api.py

api.py

@@ -1,385 +0,0 @@
-#!/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
-import uvicorn
-import asyncio
-
-# Initialize FastAPI app
-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"
-)
-
-# API Key Configuration
-API_KEY = os.getenv("MADVERSE_API_KEY", "madverse-music-api-key-2024")  # Default key for demo
-
-# Global model variable
-model = None
-
-async def verify_api_key(x_api_key: Annotated[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[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
-
-@app.on_event("startup")
-async def load_model():
-    """Load the AI model on 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
-
-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\"}'"
-        }
-    }
-
-if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8000)