app.py

# import time
# import pickle
# import numpy as np
# from fastapi import FastAPI, Request, status, HTTPException, Depends
# from fastapi.security.api_key import APIKeyHeader
# from pydantic import BaseModel
# from typing import List
# import os
# import joblib

# app = FastAPI(
#     title="ML Music Classifier API",
#     description="Predict 'liked' or not from audio features using 8 ML models.",
#     version="1.0.0"
# )

# DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
# api_key_header = APIKeyHeader(name="X-Token", auto_error=False)

# @app.middleware("http")
# async def add_process_time_header(request: Request, call_next):
#     start_time = time.perf_counter()
#     response = await call_next(request)
#     process_time = time.perf_counter() - start_time
#     response.headers["X-Process-Time"] = str(process_time)
#     return response

# def verify_token(x_token: str = Depends(api_key_header)):
#     if x_token != DEFAULT_TOKEN:
#         raise HTTPException(
#             status_code=status.HTTP_401_UNAUTHORIZED,
#             detail="Invalid or missing token. Use correct 'X-Token' in headers."
#         )

# class SongFeatures(BaseModel):
#     danceability: float
#     energy: float
#     key: int
#     loudness: float
#     mode: int
#     speechiness: float
#     acousticness: float
#     instrumentalness: float
#     liveness: float
#     valence: float
#     tempo: float
#     duration_ms: int
#     time_signature: int

# MODEL_DIR = "models"
# model_names = [
#     "ANN_model", "knn_model", "logistic_regression_model",
#     "neural_model", "Naive_Bayes_model",
#     "random_forest_model", "svm_model", "XGBoost_model"
# ]
# models = {}


# for name in model_names:
#     path = os.path.join(MODEL_DIR, f"{name}.pkl")
#     try:
#         models[name] = joblib.load(path)
#     except Exception as e:
#         print(f"Error loading {name}: {e}")


# @app.post("/predict/{model_name}", dependencies=[Depends(verify_token)])
# def predict(model_name: str, features: SongFeatures):
#     if model_name not in models:
#         raise HTTPException(status_code=404, detail="Model not found")

#     model = models[model_name]
#     input_array = np.array([[getattr(features, field) for field in features.model_fields]])
    
#     try:
#         prediction = model.predict(input_array)
#         return {
#             "model": model_name,
#             "input": features,
#             "prediction": int(prediction[0]),
#             "prediction_label": "liked" if int(prediction[0]) == 1 else "not_liked"
#         }
#     except Exception as e:
#         raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}")





# import time
# import pickle
# import numpy as np
# from fastapi import FastAPI, Request, status, HTTPException, Depends
# from fastapi.security.api_key import APIKeyHeader
# from pydantic import BaseModel
# from typing import List
# import os
# import joblib

# app = FastAPI(
#     title="ML Music Classifier API",
#     description="Predict 'liked' or not from audio features using 8 ML models.",
#     version="1.0.0"
# )

# DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
# api_key_header = APIKeyHeader(name="X-Token", auto_error=False)

# @app.middleware("http")
# async def add_process_time_header(request: Request, call_next):
#     start_time = time.perf_counter()
#     response = await call_next(request)
#     process_time = time.perf_counter() - start_time
#     response.headers["X-Process-Time"] = str(process_time)
#     return response

# def verify_token(x_token: str = Depends(api_key_header)):
#     if x_token != DEFAULT_TOKEN:
#         raise HTTPException(
#             status_code=status.HTTP_401_UNAUTHORIZED,
#             detail="Invalid or missing token. Use correct 'X-Token' in headers."
#         )

# class SongFeatures(BaseModel):
#     danceability: float
#     energy: float
#     key: int
#     loudness: float
#     mode: int
#     speechiness: float
#     acousticness: float
#     instrumentalness: float
#     liveness: float
#     valence: float
#     tempo: float
#     duration_ms: int
#     time_signature: int

# MODEL_DIR = "models"
# model_names = [
#     "ANN_model", "knn_model", "logistic_regression_model",
#     "neural_model", "Naive_Bayes_model",
#     "random_forest_model", "svm_model", "XGBoost_model"
# ]
# models = {}

# for name in model_names:
#     path = os.path.join(MODEL_DIR, f"{name}.pkl")
#     try:
#         models[name] = joblib.load(path)
#         print(f"✅ Successfully loaded {name}")
#     except Exception as e:
#         print(f"❌ Error loading {name}: {e}")

# @app.get("/")
# def root():
#     return {
#         "message": "ML Music Classifier API is running!",
#         "loaded_models": len(models),
#         "available_models": list(models.keys()),
#         "endpoints": {
#             "predict": "/predict/{model_name}",
#             "docs": "/docs",
#             "health": "/health"
#         }
#     }

# @app.get("/health")
# def health_check():
#     return {
#         "status": "healthy",
#         "loaded_models": len(models),
#         "available_models": list(models.keys())
#     }

# @app.post("/predict/{model_name}", dependencies=[Depends(verify_token)])
# def predict(model_name: str, features: SongFeatures):
#     if model_name not in models:
#         raise HTTPException(status_code=404, detail="Model not found")

#     model = models[model_name]
#     input_array = np.array([[getattr(features, field) for field in features.model_fields]])
    
#     try:
#         prediction = model.predict(input_array)
#         return {
#             "model": model_name,
#             "input": features.dict(),
#             "prediction": int(prediction[0]),
#             "prediction_label": "liked" if int(prediction[0]) == 1 else "not_liked"
#         }
#     except Exception as e:
#         raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}")

# if __name__ == "__main__":
#     import uvicorn
#     uvicorn.run(app, host="0.0.0.0", port=7860)




import time
import pickle
import numpy as np
from fastapi import FastAPI, Request, status, HTTPException, Depends
from fastapi.security.api_key import APIKeyHeader
from fastapi.templating import Jinja2Templates
from fastapi.responses import HTMLResponse
from pydantic import BaseModel
from typing import List
import os
import joblib

app = FastAPI(
    title="ML Music Classifier API",
    description="Predict 'liked' or not from audio features using 8 ML models.",
    version="1.0.0"
)

templates = Jinja2Templates(directory="templates")

DEFAULT_TOKEN = "a9b7c7e8b0e44157a99c9a8c5f6a172e10b77e2b44693506a32e5a6a0cd749d0"
api_key_header = APIKeyHeader(name="X-Token", auto_error=False)

@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    start_time = time.perf_counter()
    response = await call_next(request)
    process_time = time.perf_counter() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

def verify_token(x_token: str = Depends(api_key_header)):
    if x_token != DEFAULT_TOKEN:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid or missing token. Use correct 'X-Token' in headers."
        )

class SongFeatures(BaseModel):
    danceability: float
    energy: float
    key: int
    loudness: float
    mode: int
    speechiness: float
    acousticness: float
    instrumentalness: float
    liveness: float
    valence: float
    tempo: float
    duration_ms: int
    time_signature: int

MODEL_DIR = "models"
model_names = [
    "ANN_model", "knn_model", "logistic_regression_model",
    "neural_model", "Naive_Bayes_model",
    "random_forest_model", "svm_model", "XGBoost_model"
]
models = {}

for name in model_names:
    path = os.path.join(MODEL_DIR, f"{name}.pkl")
    try:
        models[name] = joblib.load(path)
        print(f"✅ Successfully loaded {name}")
    except Exception as e:
        print(f"❌ Error loading {name}: {e}")

@app.get("/", response_class=HTMLResponse)
def home(request: Request):
    return templates.TemplateResponse("index.html", {
        "request": request,
        "models": list(models.keys())
    })
    
@app.get("/health")
def health_check():
    return {
        "status": "healthy",
        "loaded_models": len(models),
        "available_models": list(models.keys())
    }

@app.post("/predict/{model_name}", dependencies=[Depends(verify_token)])
def predict(model_name: str, features: SongFeatures):
    if model_name not in models:
        raise HTTPException(status_code=404, detail="Model not found")

    model = models[model_name]
    input_array = np.array([[getattr(features, field) for field in features.model_fields]])
    
    try:
        prediction = model.predict(input_array)
        return {
            "model": model_name,
            "input": features.dict(),
            "prediction": int(prediction[0]),
            "prediction_label": "liked" if int(prediction[0]) == 1 else "Not liked"
        }
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Prediction error: {str(e)}")

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)