Spaces:
Sleeping
Sleeping
File size: 3,176 Bytes
154408f | 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 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 | from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, HttpUrl
from typing import Optional, Dict, List, Any
import logging
import time
from src.core.pipeline import Pipeline
from src.ml import slm_explainer
# Configure Logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("api")
# Initialize App
app = FastAPI(
title="PhishingInsight API",
description="AI-powered Phishing Detection & URL Analysis API",
version="1.0.0"
)
# Initialize Pipeline (Global Singleton)
logger.info("API: Initializing Global Pipeline...")
pipeline = Pipeline()
# Load SLM in background on startup
@app.on_event("startup")
async def startup_event():
logger.info("API: Pre-loading SLM Model...")
# We do this asynchronously or threading to not block,
# but slm_explainer.load_model() blocking is safer for "ready" state.
# For now, we rely on the main process having loaded it previously
# or it loading on first request if isolated.
try:
slm_explainer.load_model()
logger.info("API: SLM Model Loaded.")
except Exception as e:
logger.error(f"API: Model load warning: {e}")
# --- Data Models ---
class URLRequest(BaseModel):
url: str
class SignalScores(BaseModel):
url_analysis: float
domain_check: float
class AnalysisResponse(BaseModel):
success: bool
url: str
verdict: str # SAFE, WARNING, DANGER
score: float # 0.0 to 1.0 (Risk)
is_phishing: bool
summary: str
explanation: str
advice: str
signals: SignalScores
findings: List[str]
timestamp: str
time_ms: int
error: Optional[str] = None
# --- Endpoints ---
@app.get("/health")
def health_check():
"""Health check endpoint for Docker/Kubernetes."""
return {"status": "healthy", "service": "PhishingInsight-API"}
@app.get("/version")
def version_info():
"""Return API and Model versions."""
return {
"api_version": "1.0.0",
"model_type": "Hybrid (LightGBM + TinyLlama)"
}
@app.post("/analyze", response_model=AnalysisResponse)
def analyze_url(request: URLRequest):
"""
Analyze a URL for phishing threats.
"""
logger.info(f"API Request: Analyze {request.url}")
result = pipeline.analyze(request.url)
if not result.get('success'):
logger.warning(f"API Analysis Failed: {result.get('error')}")
# Even if logical failure (invalid URL), we return a structured response with error
return {
"success": False,
"url": request.url,
"verdict": "ERROR",
"score": 0.0,
"is_phishing": False,
"summary": "Analysis Failed",
"explanation": result.get('error', "Unknown error"),
"advice": "Check the URL and try again.",
"signals": {"url_analysis": 0, "domain_check": 0},
"findings": [],
"timestamp": "",
"time_ms": 0,
"error": result.get('error')
}
return result
|