# main.py
from fastapi import FastAPI, HTTPException, Response, Depends, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, EmailStr
import base64
from PIL import Image
import io
import traceback
from CrimeSceneAnalyzer import CrimeSceneAnalyzer # Import your analyzer class
from pymongo import MongoClient
from dotenv import load_dotenv
import os
from passlib.context import CryptContext
from datetime import datetime, timedelta
from typing import Optional
import jwt
from email_validator import validate_email, EmailNotValidError
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig
import httpx
import random
import uuid
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError
load_dotenv()
from fastapi.staticfiles import StaticFiles
app = FastAPI()
os.makedirs("public", exist_ok=True)
app.mount("/static", StaticFiles(directory="public"), name="static")
# Configure CORS to allow your React Native app to access the API
app.add_middleware(
CORSMiddleware,
allow_origins=["*"], # In production, replace with specific origins
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
expose_headers=["*"],
max_age=3600,
)
# MongoDB Connection
MONGO_URI = os.getenv("MONGO_URI")
try:
client = MongoClient(MONGO_URI, serverSelectionTimeoutMS=5000)
db = client.mobile # Database name
# Test connection
client.admin.command('ping')
print("✓ MongoDB connected successfully.")
except Exception as e:
print(f"⚠️ MongoDB connection failed: {e}")
client = None
db = None
# Password Hashing
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
# JWT Settings
SECRET_KEY = os.getenv("SECRET_KEY")
ALGORITHM = os.getenv("ALGORITHM")
ACCESS_TOKEN_EXPIRE_MINUTES = 20 # Changed from 30 to 20 minutes
# Email Configuration
conf = ConnectionConfig(
MAIL_USERNAME=os.getenv("MAIL_USERNAME"),
MAIL_PASSWORD=os.getenv("MAIL_PASSWORD"),
MAIL_FROM=os.getenv("MAIL_FROM"),
MAIL_PORT=int(os.getenv("MAIL_PORT", 587)),
MAIL_SERVER=os.getenv("MAIL_SERVER"),
MAIL_STARTTLS=os.getenv("MAIL_STARTTLS").lower() == 'true',
MAIL_SSL_TLS=os.getenv("MAIL_SSL_TLS").lower() == 'true',
USE_CREDENTIALS=os.getenv("USE_CREDENTIALS").lower() == 'true',
VALIDATE_CERTS=os.getenv("VALIDATE_CERTS").lower() == 'true',
)
analyzer = CrimeSceneAnalyzer()
class ImageRequest(BaseModel):
image: str # Base64 encoded image string
class UserCreate(BaseModel):
username: str
email: EmailStr
password: str
phoneNumber: Optional[str] = None
class UserLogin(BaseModel):
email: EmailStr
password: str
class ForgotPasswordRequest(BaseModel):
email: EmailStr
class VerifyOTPRequest(BaseModel):
email: EmailStr
otp: str
class Token(BaseModel):
access_token: str
token_type: str
class TokenData(BaseModel):
email: Optional[str] = None
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
async def send_otp_email(email: str, otp: str):
"""Send OTP via Brevo HTTP API (HF Spaces blocks outbound SMTP)."""
html_content = f"""
Dear User,
Your OTP for verification is: {otp}
This OTP is valid for 10 minutes.
If you did not request this, please ignore this email.
"""
brevo_api_key = os.getenv("BREVO_API_KEY")
if not brevo_api_key:
# Fallback: log OTP to console so deployment isn't blocked
print(f"[OTP for {email}] {otp} (BREVO_API_KEY not set)")
return
sender_email = os.getenv("BREVO_SENDER_EMAIL", os.getenv("MAIL_FROM", "noreply@example.com"))
sender_name = os.getenv("BREVO_SENDER_NAME", "SceneX")
payload = {
"sender": {"name": sender_name, "email": sender_email},
"to": [{"email": email}],
"subject": "OTP Verification for SceneXMobile",
"htmlContent": html_content,
}
headers = {
"api-key": brevo_api_key,
"Content-Type": "application/json",
"Accept": "application/json",
}
async with httpx.AsyncClient(timeout=15) as client_http:
resp = await client_http.post(
"https://api.brevo.com/v3/smtp/email", json=payload, headers=headers
)
if resp.status_code >= 400:
raise Exception(f"Brevo API error {resp.status_code}: {resp.text}")
@app.post("/signup", response_model=Token)
async def signup(user: UserCreate):
try:
if db is None:
raise HTTPException(status_code=503, detail="Database not connected. Please check MongoDB URI.")
try:
validate_email(user.email)
except EmailNotValidError:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid email format")
if db.users.find_one({"email": user.email}):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Email already registered")
if db.users.find_one({"username": user.username}):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Username already taken")
hashed_password = get_password_hash(user.password)
otp = str(random.randint(100000, 999999))
otp_expiry = datetime.utcnow() + timedelta(minutes=10)
user_data = {
"username": user.username,
"email": user.email,
"password": hashed_password,
"is_verified": False,
"otp": otp,
"otp_expiry": otp_expiry,
"phone": user.phoneNumber if user.phoneNumber else str(uuid.uuid4())
}
try:
db.users.insert_one(user_data)
except Exception as e:
print(f"Error inserting user: {e}")
raise HTTPException(status_code=500, detail=f"Failed to insert user: {str(e)}")
try:
await send_otp_email(user.email, otp)
print(f"OTP email sent to {user.email}")
except Exception as e:
print(f"Error sending OTP email: {e}")
traceback.print_exc()
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
except HTTPException:
raise
except Exception as e:
print(f"UNEXPECTED ERROR in signup: {e}")
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Internal error: {str(e)}")
@app.post("/login")
async def login(user: UserLogin):
db_user = db.users.find_one({"email": user.email})
if not db_user or not verify_password(user.password, db_user["password"]):
raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Incorrect email or password")
if not db_user.get("is_verified"):
raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="Account not verified. Please verify your email with OTP.")
# Generate OTP for 2FA
otp = str(random.randint(100000, 999999))
otp_expiry = datetime.utcnow() + timedelta(minutes=10)
db.users.update_one({"email": user.email}, {"$set": {"otp": otp, "otp_expiry": otp_expiry}})
try:
await send_otp_email(user.email, otp)
except Exception as e:
print(f"Error sending OTP email: {e}")
raise HTTPException(status_code=500, detail=f"Failed to send OTP email: {str(e)}")
return {"message": "OTP sent to your email for verification."}
@app.post("/forgot-password")
async def forgot_password(request: ForgotPasswordRequest):
db_user = db.users.find_one({"email": request.email})
if not db_user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User with this email not found")
otp = str(random.randint(100000, 999999))
otp_expiry = datetime.utcnow() + timedelta(minutes=10)
db.users.update_one({"email": request.email}, {"$set": {"otp": otp, "otp_expiry": otp_expiry}})
await send_otp_email(request.email, otp)
return {"message": "OTP sent to your email for password reset."}
@app.post("/verify-otp")
async def verify_otp(request: VerifyOTPRequest):
db_user = db.users.find_one({"email": request.email})
if not db_user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
if db_user.get("otp") != request.otp:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid OTP")
if datetime.utcnow() > db_user.get("otp_expiry", datetime.utcnow()):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="OTP expired")
db.users.update_one({"email": request.email}, {"$set": {"is_verified": True, "otp": None, "otp_expiry": None}})
return {"message": "Email verified successfully!"}
@app.post("/login/verify-otp", response_model=Token)
async def login_verify_otp(request: VerifyOTPRequest):
db_user = db.users.find_one({"email": request.email})
if not db_user:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
if db_user.get("otp") != request.otp:
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Invalid OTP")
if datetime.utcnow() > db_user.get("otp_expiry", datetime.utcnow()):
raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="OTP expired")
# Clear OTP after successful verification
db.users.update_one({"email": request.email}, {"$set": {"otp": None, "otp_expiry": None}})
# Generate access token
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": request.email},
expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
@app.post("/analyze")
async def analyze_image_endpoint(request: ImageRequest):
"""
Analyzes a crime scene image and returns structured JSON results.
The full PDF report can be downloaded via the /download_report endpoint.
"""
try:
# Decode base64 image
image_bytes = base64.b64decode(request.image)
image_pil = Image.open(io.BytesIO(image_bytes)).convert("RGB")
# Call the analyzer to get both analysis data and PDF bytes
# We need to temporarily modify `analyze_scene` in your class to return more.
# For this example, let's assume `analyze_scene` is modified to return
# a dictionary containing 'analysis_data' and 'pdf_bytes'.
# --- REFACTORING YOUR CrimeSceneAnalyzer.analyze_scene METHOD ---
# Modify CrimeSceneAnalyzer.py's analyze_scene to return a structured dict:
# { "analysis_data": {...}, "pdf_bytes": }
analysis_output = analyzer.analyze_scene(image_pil) # Assume this now returns the dict
if "analysis_data" not in analysis_output:
raise HTTPException(status_code=500, detail="Analysis data not found in response.")
return analysis_output["analysis_data"]
except Exception as e:
print(f"Error during analysis: {e}")
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Analysis failed: {str(e)}")
@app.post("/download_report")
async def download_report_endpoint(request: ImageRequest):
"""
Generates and returns the full crime scene report as a PDF.
"""
try:
# Decode base64 image
image_bytes = base64.b64decode(request.image)
image_pil = Image.open(io.BytesIO(image_bytes)).convert("RGB")
# Perform analysis to get the PDF bytes
analysis_output = analyzer.analyze_scene(image_pil) # Assume this now returns the dict
pdf_bytes = analysis_output.get("pdf_bytes")
if not pdf_bytes:
raise HTTPException(status_code=500, detail="Failed to generate PDF report. No PDF bytes returned.")
return Response(content=bytes(pdf_bytes), media_type="application/pdf", headers={
"Content-Disposition": "attachment; filename=crime_scene_report.pdf"
})
except Exception as e:
print(f"Error generating PDF: {e}")
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"PDF generation failed: {str(e)}")
@app.post("/classify")
async def classify_image_endpoint(request: ImageRequest):
"""
Classifies a crime scene image and returns the predicted crime type.
"""
try:
# Decode base64 image
image_bytes = base64.b64decode(request.image)
image_pil = Image.open(io.BytesIO(image_bytes)).convert("RGB")
# Classify the scene using CrimeSceneAnalyzer
crime_type = analyzer.classify_scene(image_pil)
return {"crime_type": crime_type}
except Exception as e:
print(f"Error during classification: {e}")
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Classification failed: {str(e)}")
from video_classifier import classify_video_from_path
class VideoRequest(BaseModel):
video: str # Base64 encoded video string
@app.post("/video_classify")
async def video_classify_endpoint(request: VideoRequest):
"""
Classifies a video and returns the predicted crime type.
"""
try:
# Decode base64 video
print(f"Received video base64 string length: {len(request.video)}")
video_bytes = base64.b64decode(request.video)
print(f"Decoded video bytes length: {len(video_bytes)}")
# Save to a temporary file using tempfile for robustness
import tempfile
with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as temp_file:
temp_file.write(video_bytes)
temp_video_path = temp_file.name
print(f"Temporary video saved to: {temp_video_path}")
try:
# Classify the video using video_classifier
crime_type = classify_video_from_path(temp_video_path)
finally:
# Clean up temporary file
os.remove(temp_video_path)
print(f"Temporary video file removed: {temp_video_path}")
return {"result": crime_type}
except Exception as e:
print(f"Error during video classification: {e}")
traceback.print_exc()
raise HTTPException(status_code=500, detail=f"Video classification failed: {str(e)}")
# JWT Authentication
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login/verify-otp")
async def get_current_user(token: str = Depends(oauth2_scheme)):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
# Ensure token is properly formatted
if not token or not token.strip():
raise credentials_exception
# Clean the token
token = token.strip()
if not token.startswith('Bearer '):
token = f'Bearer {token}'
# Extract the actual token from the Bearer scheme
token = token.split(' ')[1] if token.startswith('Bearer ') else token
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
except jwt.ExpiredSignatureError:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Token has expired",
headers={"WWW-Authenticate": "Bearer"},
)
except Exception as e:
print(f"JWT decode error: {str(e)}")
raise credentials_exception
email: str = payload.get("sub")
if email is None:
raise credentials_exception
token_data = TokenData(email=email)
except Exception as e:
print(f"Unexpected error in get_current_user: {str(e)}")
raise credentials_exception
user = db.users.find_one({"email": token_data.email})
if user is None:
raise credentials_exception
return {"sub": token_data.email}
# History-related models
class HistoryItem(BaseModel):
id: str
date: str
type: str
title: str
image: str
primaryResult: str
confidence: float
description: str
userId: str
@app.get("/history/{user_id}")
async def get_user_history(user_id: str, current_user: dict = Depends(get_current_user)):
"""
Retrieves the analysis history for a specific user.
"""
try:
# Ensure the requesting user can only access their own history
if current_user["sub"] != user_id:
raise HTTPException(status_code=403, detail="Not authorized to access this user's history")
history_items = list(db.history.find({"userId": user_id}))
# Convert ObjectId to string for JSON serialization
for item in history_items:
item["_id"] = str(item["_id"])
return history_items
except Exception as e:
print(f"Error retrieving history: {e}")
raise HTTPException(status_code=500, detail=f"Failed to retrieve history: {str(e)}")
@app.post("/history")
async def save_history_item(history_item: HistoryItem, current_user: dict = Depends(get_current_user)):
"""
Saves a new history item for a user.
"""
try:
# Ensure the user can only save to their own history
if current_user["sub"] != history_item.userId:
raise HTTPException(status_code=403, detail="Not authorized to save history for this user")
history_dict = history_item.dict()
result = db.history.insert_one(history_dict)
# Return the created item with its ID
history_dict["_id"] = str(result.inserted_id)
return history_dict
except Exception as e:
print(f"Error saving history item: {e}")
raise HTTPException(status_code=500, detail=f"Failed to save history item: {str(e)}")
@app.delete("/history/{item_id}")
async def delete_history_item(item_id: str, current_user: dict = Depends(get_current_user)):
"""
Deletes a specific history item.
"""
try:
# First, get the history item to check ownership
history_item = db.history.find_one({"id": item_id})
if not history_item:
raise HTTPException(status_code=404, detail="History item not found")
# Ensure the user can only delete their own history items
if history_item["userId"] != current_user["sub"]:
raise HTTPException(status_code=403, detail="Not authorized to delete this history item")
result = db.history.delete_one({"id": item_id})
if result.deleted_count == 0:
raise HTTPException(status_code=404, detail="History item not found")
return {"message": "History item deleted successfully"}
except Exception as e:
print(f"Error deleting history item: {e}")
raise HTTPException(status_code=500, detail=f"Failed to delete history item: {str(e)}")