File size: 4,735 Bytes
93c5df6 | 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 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 | """
Restaurant Inspector API.
Analyzes restaurant reviews and provides scores for various aspects.
"""
from datetime import datetime
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field
from transformers import pipeline
# Initialize FastAPI app
app = FastAPI(
title="Restaurant Inspector",
description="AI-powered restaurant review aspect analyzer",
version="1.0.0",
)
# Aspect names mapping
ASPECT_NAMES = ["FOOD", "SERVICE", "HYGIENE", "PARKING", "CLEANLINESS"]
# Load model at startup (once!)
print("🚀 Loading model...")
try:
classifier = pipeline(
"text-classification",
model="dpratapx/restaurant-inspector",
device=-1, # CPU mode
top_k=None, # Return all scores
)
print("✅ Model loaded successfully!")
except Exception as e:
print(f"❌ Error loading model: {e}")
print("⚠️ Make sure you've run 'python train.py' first!")
classifier = None
# Request/Response models
class ReviewRequest(BaseModel):
"""Request model for review analysis."""
text: str = Field(
...,
min_length=10,
max_length=1000,
description="Restaurant review text to analyze",
examples=["Great food but terrible parking and dirty bathrooms"],
)
class AspectScores(BaseModel):
"""Aspect scores model."""
FOOD: float = Field(..., ge=0.0, le=1.0, description="Food quality score (0-1)")
SERVICE: float = Field(..., ge=0.0, le=1.0, description="Service quality score (0-1)")
HYGIENE: float = Field(..., ge=0.0, le=1.0, description="Hygiene score (0-1)")
PARKING: float = Field(..., ge=0.0, le=1.0, description="Parking availability score (0-1)")
CLEANLINESS: float = Field(..., ge=0.0, le=1.0, description="Cleanliness score (0-1)")
class AnalysisResponse(BaseModel):
"""Response model for review analysis."""
review: str = Field(..., description="Original review text")
scores: AspectScores = Field(..., description="Aspect scores")
timestamp: str = Field(..., description="Analysis timestamp (ISO 8601)")
# API Endpoints
@app.get("/")
async def root():
"""Root endpoint with API information."""
return {
"message": "Restaurant Inspector API",
"version": "1.0.0",
"endpoints": {
"POST /analyze": "Analyze a restaurant review",
"GET /health": "Health check",
"GET /docs": "API documentation",
},
}
@app.get("/health")
async def health_check():
"""Health check endpoint."""
model_status = "ready" if classifier is not None else "not_loaded"
return {
"status": "healthy" if model_status == "ready" else "degraded",
"model": model_status,
"timestamp": datetime.utcnow().isoformat(),
}
@app.post("/analyze", response_model=AnalysisResponse)
async def analyze_review(request: ReviewRequest):
"""
Analyze a restaurant review and return aspect scores.
Args:
request: ReviewRequest with review text
Returns:
AnalysisResponse with scores for each aspect
Raises:
HTTPException: If model not loaded or analysis fails
"""
if classifier is None:
raise HTTPException(
status_code=503,
detail="Model not loaded. Please run 'python train.py' first.",
)
try:
# Get predictions
results = classifier(request.text)
# Parse scores - results is a list of lists of dicts
# Format: [[{'label': 'LABEL_0', 'score': 0.9}, ...]]
scores_dict = {}
if isinstance(results[0], list):
# top_k=None returns list of all labels with scores
for item in results[0]:
label_idx = int(item["label"].split("_")[1])
if 0 <= label_idx < len(ASPECT_NAMES):
aspect_name = ASPECT_NAMES[label_idx]
scores_dict[aspect_name] = round(item["score"], 3)
else:
# Fallback for different pipeline output format
for i, aspect_name in enumerate(ASPECT_NAMES):
scores_dict[aspect_name] = 0.5 # Default score
# Ensure all aspects are present
for aspect in ASPECT_NAMES:
if aspect not in scores_dict:
scores_dict[aspect] = 0.5
return AnalysisResponse(
review=request.text,
scores=AspectScores(**scores_dict),
timestamp=datetime.utcnow().isoformat(),
)
except Exception as e:
raise HTTPException(
status_code=500,
detail=f"Analysis failed: {str(e)}",
)
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
|