Spaces:
Sleeping
Sleeping
| # 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""" | |
| <html> | |
| <body> | |
| <p>Dear User,</p> | |
| <p>Your OTP for verification is: <strong>{otp}</strong></p> | |
| <p>This OTP is valid for 10 minutes.</p> | |
| <p>If you did not request this, please ignore this email.</p> | |
| </body> | |
| </html> | |
| """ | |
| 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}") | |
| 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)}") | |
| 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."} | |
| 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."} | |
| 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!"} | |
| 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"} | |
| 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": <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)}") | |
| 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)}") | |
| 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 | |
| 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 | |
| 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)}") | |
| 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)}") | |
| 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)}") |