import asyncio from datetime import datetime, timedelta from typing import List import httpx from fastapi import BackgroundTasks, Depends, FastAPI, HTTPException, status from fastapi.middleware.cors import CORSMiddleware from sqlalchemy import func, select, update from sqlalchemy.ext.asyncio import AsyncSession from services.emergency_service.models import EmergencyAlert, EmergencyResponse from services.emergency_service.schemas import ( EmergencyAlertCreate, EmergencyAlertResponse, EmergencyResponseCreate, EmergencyResponseResponse, EmergencyStats, ) from services.user_service.models import User from shared.auth import get_current_user_from_token from shared.config import settings from shared.database import get_db app = FastAPI(title="Emergency Service", version="1.0.0") # CORS middleware app.add_middleware( CORSMiddleware, allow_origins=settings.CORS_ORIGINS, allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) async def get_current_user( user_data: dict = Depends(get_current_user_from_token), db: AsyncSession = Depends(get_db), ): """Get current user from token via auth dependency.""" # Get full user object from database result = await db.execute(select(User).filter(User.id == user_data["user_id"])) user = result.scalars().first() if user is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="User not found" ) return user @app.get("/health") async def health_check(): """Health check endpoint""" return {"status": "healthy", "service": "emergency_service"} async def get_nearby_users( latitude: float, longitude: float, radius_km: float = 1.0 ) -> List[dict]: """Get users within radius using Location Service""" async with httpx.AsyncClient() as client: try: response = await client.get( f"http://localhost:8003/api/v1/nearby-users", params={ "latitude": latitude, "longitude": longitude, "radius_km": radius_km, }, timeout=5.0, ) if response.status_code == 200: return response.json() return [] except Exception: return [] async def send_emergency_notifications(alert_id: int, nearby_users: List[dict]): """Send push notifications to nearby users""" async with httpx.AsyncClient() as client: try: await client.post( "http://localhost:8005/api/v1/send-emergency-notifications", json={ "alert_id": alert_id, "user_ids": [user["user_id"] for user in nearby_users], }, timeout=10.0, ) except Exception as e: print(f"Failed to send notifications: {e}") @app.post("/api/v1/alert", response_model=EmergencyAlertResponse) async def create_emergency_alert( alert_data: EmergencyAlertCreate, background_tasks: BackgroundTasks, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Create new emergency alert""" # Create alert db_alert = EmergencyAlert( user_id=current_user.id, latitude=alert_data.latitude, longitude=alert_data.longitude, address=alert_data.address, alert_type=alert_data.alert_type, message=alert_data.message, ) db.add(db_alert) await db.commit() await db.refresh(db_alert) # Process alert in background background_tasks.add_task( process_emergency_alert_in_background, int(db_alert.id), # Convert to int explicitly alert_data.latitude, alert_data.longitude ) return EmergencyAlertResponse.model_validate(db_alert) async def process_emergency_alert_in_background(alert_id: int, latitude: float, longitude: float): """Process emergency alert - notify nearby users""" try: # Get nearby users nearby_users = await get_nearby_users(latitude, longitude) if nearby_users: # Create new database session for background task from shared.database import AsyncSessionLocal async with AsyncSessionLocal() as db: try: # Update alert with notification count await db.execute( update(EmergencyAlert) .where(EmergencyAlert.id == alert_id) .values(notified_users_count=len(nearby_users)) ) await db.commit() # Send notifications await send_emergency_notifications(alert_id, nearby_users) except Exception as e: print(f"Error processing emergency alert: {e}") await db.rollback() except Exception as e: print(f"Error in process_emergency_alert_in_background: {e}") @app.post("/api/v1/alert/{alert_id}/respond", response_model=EmergencyResponseResponse) async def respond_to_alert( alert_id: int, response_data: EmergencyResponseCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Respond to emergency alert""" # Check if alert exists result = await db.execute(select(EmergencyAlert).filter(EmergencyAlert.id == alert_id)) alert = result.scalars().first() if not alert: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Alert not found" ) # Check if already responded existing_response = await db.execute( select(EmergencyResponse).filter( EmergencyResponse.alert_id == alert_id, EmergencyResponse.user_id == current_user.id ) ) if existing_response.scalars().first(): raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="You have already responded to this alert" ) # Create response db_response = EmergencyResponse( alert_id=alert_id, user_id=current_user.id, response_type=response_data.response_type, message=response_data.message, eta_minutes=response_data.eta_minutes, ) db.add(db_response) # Update responded users count await db.execute( update(EmergencyAlert) .where(EmergencyAlert.id == alert_id) .values(responded_users_count=EmergencyAlert.responded_users_count + 1) ) await db.commit() await db.refresh(db_response) return EmergencyResponseResponse.model_validate(db_response) @app.put("/api/v1/alert/{alert_id}/resolve") async def resolve_alert( alert_id: int, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Resolve emergency alert""" result = await db.execute(select(EmergencyAlert).filter(EmergencyAlert.id == alert_id)) alert = result.scalars().first() if not alert: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail="Alert not found" ) # Only alert creator can resolve if alert.user_id != current_user.id: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Only alert creator can resolve the alert" ) # Update alert await db.execute( update(EmergencyAlert) .where(EmergencyAlert.id == alert_id) .values( is_resolved=True, resolved_at=datetime.utcnow(), resolved_by=current_user.id ) ) await db.commit() return {"message": "Alert resolved successfully"} @app.get("/api/v1/alerts/my", response_model=List[EmergencyAlertResponse]) async def get_my_alerts( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): """Get current user's emergency alerts""" result = await db.execute( select(EmergencyAlert) .filter(EmergencyAlert.user_id == current_user.id) .order_by(EmergencyAlert.created_at.desc()) ) alerts = result.scalars().all() return [EmergencyAlertResponse.model_validate(alert) for alert in alerts] @app.get("/api/v1/alerts/active", response_model=List[EmergencyAlertResponse]) async def get_active_alerts( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): """Get active emergency alerts""" result = await db.execute( select(EmergencyAlert) .filter(EmergencyAlert.is_resolved == False) .order_by(EmergencyAlert.created_at.desc()) .limit(50) ) alerts = result.scalars().all() return [EmergencyAlertResponse.model_validate(alert) for alert in alerts] @app.get("/api/v1/emergency/reports", response_model=List[EmergencyAlertResponse]) async def get_emergency_reports( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): """Get emergency reports/alerts for admin users or general statistics""" # For now, return all active alerts (in production, add admin check) result = await db.execute( select(EmergencyAlert) .filter(EmergencyAlert.is_resolved == False) .order_by(EmergencyAlert.created_at.desc()) .limit(50) ) alerts = result.scalars().all() return [EmergencyAlertResponse.model_validate(alert) for alert in alerts] @app.get("/api/v1/stats", response_model=EmergencyStats) async def get_emergency_stats( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db) ): """Get emergency statistics""" # Get total alerts count total_alerts_result = await db.execute(select(func.count(EmergencyAlert.id))) total_alerts = total_alerts_result.scalar() or 0 # Get active alerts count active_alerts_result = await db.execute( select(func.count(EmergencyAlert.id)).filter(EmergencyAlert.is_resolved == False) ) active_alerts = active_alerts_result.scalar() or 0 # Calculate resolved alerts resolved_alerts = total_alerts - active_alerts # Get total responders count total_responders_result = await db.execute(select(func.count(EmergencyResponse.id))) total_responders = total_responders_result.scalar() or 0 return EmergencyStats( total_alerts=total_alerts, active_alerts=active_alerts, resolved_alerts=resolved_alerts, avg_response_time_minutes=None, # TODO: Calculate this total_responders=total_responders, ) if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8002)