File size: 2,072 Bytes
30c1037
 
5c821e5
 
30c1037
5c821e5
30c1037
5c821e5
4cce576
5c821e5
 
 
 
 
 
 
 
 
30c1037
 
 
5c821e5
 
 
 
30c1037
5c821e5
30c1037
 
 
 
 
5c821e5
 
30c1037
4cce576
 
30c1037
4cce576
5c821e5
30c1037
5c821e5
30c1037
5c821e5
 
 
30c1037
 
5c821e5
 
 
30c1037
5c821e5
 
4cce576
30c1037
 
 
5c821e5
30c1037
 
5c821e5
30c1037
4cce576
30c1037
4cce576
 
30c1037
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
from pathlib import Path

import joblib
import numpy as np
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel

app = FastAPI(title="Bot Detection API")

app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

BASE_DIR = Path(__file__).resolve().parent
MODEL_PATH = BASE_DIR / "bot_detection_model.pkl"


class InputData(BaseModel):
    features: list[float]


try:
    model = joblib.load(MODEL_PATH)
    EXPECTED_FEATURES = int(getattr(model, "n_features_in_", 14))
except Exception as ex:
    raise RuntimeError(f"Model failed to load: {ex}") from ex


@app.get("/")
def home() -> dict[str, str | int]:
    return {
        "message": "Bot detection model running",
        "expected_features": EXPECTED_FEATURES,
    }


@app.get("/health")
def health() -> dict[str, str]:
    return {"status": "ok"}


@app.post("/predict")
def predict(data: InputData) -> dict[str, int | float | str]:
    if len(data.features) != EXPECTED_FEATURES:
        raise HTTPException(
            status_code=400,
            detail=f"Expected {EXPECTED_FEATURES} features",
        )

    try:
        x = np.asarray(data.features, dtype=np.float64).reshape(1, -1)
        if not np.isfinite(x).all():
            raise HTTPException(status_code=400, detail="Features contain NaN or Inf")

        pred = int(model.predict(x)[0])
        result = "bot_detected" if pred == 1 else "normal_traffic"

        response: dict[str, int | float | str] = {
            "prediction": result,
            "raw_prediction": pred,
        }

        if hasattr(model, "predict_proba"):
            proba = model.predict_proba(x)[0]
            bot_idx = list(model.classes_).index(1)
            response["bot_probability"] = float(proba[bot_idx])

        return response
    except HTTPException:
        raise
    except Exception as ex:
        raise HTTPException(status_code=500, detail=f"Prediction failed: {ex}") from ex