from flask import Flask, jsonify, request from flask_cors import CORS import random from datetime import datetime, timedelta import json import logging from firebase_admin import firestore from pytz import timezone IST = timezone('Asia/Kolkata') now_ist = datetime.now(IST) # Firebase Admin imports try: import firebase_admin from firebase_admin import credentials, messaging FIREBASE_AVAILABLE = True except ImportError: FIREBASE_AVAILABLE = False logging.warning("Firebase Admin SDK not installed. Install with: pip install firebase-admin") app = Flask(__name__) CORS(app) # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) # Firebase Admin initialization firebase_app = None if FIREBASE_AVAILABLE: try: # Initialize Firebase Admin SDK # Place your firebase-service-account.json file in the same directory cred = credentials.Certificate('sg.json') firebase_app = firebase_admin.initialize_app(cred) logger.info("Firebase Admin SDK initialized successfully") except Exception as e: logger.error(f"Failed to initialize Firebase Admin SDK: {e}") firebase_app = None SEVERITY_COLOR_MAP = { "Minor": "Green", "Moderate": "Yellow", "Strong": "Orange", "Major": "Red", "Great": "Red", "Low": "Green", "High": "Orange", "Very High": "Red", "Extreme": "Red", "Critical": "Red", "Severe": "Orange", "Very Severe": "Red", "Super Cyclonic": "Red", "Warning": "Orange", "Major Warning": "Red", } # Disaster data with varying field structures DISASTERS = { "earthquake": { "name": "Earthquake Alert", "emoji": "๐Ÿข", "helpline": "1070", "emergency_contacts": ["100", "102", "108"], "severity_levels": ["Low", "Moderate", "High", "Critical"], "instructions": [ "Drop, Cover, and Hold On immediately", "Stay away from windows and heavy objects", "Move to open areas if outdoors", "Check for injuries and hazards after shaking stops" ], "locations": ["Delhi", "Mumbai", "Guwahati", "Shimla", "Gangtok"], "has_magnitude": True, "has_depth": True, "has_tsunami_risk": True, "has_aftershock_warning": True }, "flood": { "name": "Flood Warning", "emoji": "๐ŸŒŠ", "helpline": "1077", "emergency_contacts": ["100", "101", "102"], "severity_levels":["Low", "Moderate", "High", "Critical"], "instructions": [ "Move to higher ground immediately", "Avoid walking in moving water", "Stay away from electrical equipment", "Listen to emergency broadcasts continuously" ], "locations": ["Kerala", "Assam", "Bihar", "West Bengal", "Maharashtra"], "has_water_level": True, "has_dam_status": True, "has_boat_rescue": True }, "cyclone": { "name": "Cyclone Alert", "emoji": "๐ŸŒ€", "helpline": "1078", "emergency_contacts": ["100", "101", "108"], "severity_levels": ["Low", "Moderate", "High", "Critical"], "instructions": [ "Stay indoors and away from windows", "Stock up on essential supplies", "Secure loose outdoor items", "Monitor weather updates constantly" ], "locations": ["Odisha", "Andhra Pradesh", "Tamil Nadu", "West Bengal", "Gujarat"], "has_wind_speed": True, "has_storm_surge": True, "has_landfall_time": True, "has_eye_wall": True }, "wildfire": { "name": "Wildfire Emergency", "emoji": "๐Ÿ”ฅ", "helpline": "101", "emergency_contacts": ["101"], "severity_levels": ["Low", "Moderate", "High", "Critical"], "instructions": [ "Evacuate immediately if ordered", "Close all windows and doors", "Turn on lights to aid visibility", "Stay low to avoid smoke inhalation" ], "locations": ["Uttarakhand", "Himachal Pradesh", "Karnataka", "Maharashtra", "Rajasthan"], "has_fire_spread_rate": True, "has_smoke_direction": True, "has_containment": True }, "heatwave": { "name": "Heatwave Alert", "emoji": "๐ŸŒก๏ธ", "helpline": "104", "emergency_contacts": ["102", "108"], "severity_levels": ["Low", "Moderate", "High", "Critical"], "instructions": [ "Stay indoors during peak hours (11 AM - 4 PM)", "Drink plenty of water regularly", "Wear loose, light-colored clothing", "Avoid alcohol and caffeine" ], "locations": ["Rajasthan", "Haryana", "Delhi", "Uttar Pradesh", "Madhya Pradesh"], "has_heat_index": True, "has_uv_index": True, "has_cooling_centers": True }, "tsunami": { "name": "Tsunami Warning", "emoji": "๐ŸŒŠ", "helpline": "1070", "emergency_contacts": ["100", "102", "108"], "severity_levels": ["Low", "Moderate", "High", "Critical"], "instructions": [ "Move to higher ground immediately", "Stay away from the coast", "Listen to official warnings only", "Do not return until all-clear is given" ], "locations": ["Tamil Nadu", "Kerala", "Andhra Pradesh", "Odisha", "West Bengal"], "has_wave_height": True, "has_arrival_time": True, "has_coastal_impact": True, "has_sea_level": True } } def generate_disaster_data(): """Generate random disaster data - same logic as /anything endpoint""" disaster_type = random.choice(list(DISASTERS.keys())) disaster_info = DISASTERS[disaster_type] severity = random.choice(disaster_info["severity_levels"]) location = random.choice(disaster_info["locations"]) affected_people = random.randint(100, 10000) # Base disaster alert data disaster_data = { "alert_id": f"ALERT_{random.randint(1000, 9999)}", "disaster_type": disaster_type, "disaster_name": disaster_info["name"], "severity": severity, "location": location, "description": f"{severity} {disaster_info['name']} reported in {location}. Take immediate action.", "emergency_contacts": disaster_info["emergency_contacts"], "safety_instructions": disaster_info["instructions"], "issued_at": datetime.now().isoformat(), "affected_people": affected_people } # Add helpline if available if "helpline" in disaster_info: disaster_data["primary_helpline"] = disaster_info["helpline"] # Add disaster-specific fields if disaster_type == "earthquake": disaster_data.update({ "magnitude": round(random.uniform(3.0, 8.5), 1), "depth_km": random.randint(5, 200), "tsunami_risk": random.choice(["Low", "Medium", "High"]), "aftershock_warning": random.choice([True, False]), "epicenter_distance": f"{random.randint(5, 500)} km from {location}" }) elif disaster_type == "flood": disaster_data.update({ "water_level": f"{random.randint(2, 15)} meters above normal", "dam_status": random.choice(["Normal", "Alert", "Danger"]), "boat_rescue_available": random.choice([True, False]), "affected_villages": random.randint(5, 50) }) elif disaster_type == "cyclone": disaster_data.update({ "wind_speed_kmh": random.randint(60, 250), "storm_surge_height": f"{random.randint(1, 8)} meters", "expected_landfall": (datetime.now() + timedelta(hours=random.randint(6, 48))).isoformat(), "eye_wall_diameter": f"{random.randint(20, 100)} km", "pressure_mb": random.randint(900, 1000) }) elif disaster_type == "wildfire": disaster_data.update({ "fire_spread_rate": f"{random.randint(1, 15)} km/hour", "smoke_direction": random.choice(["North", "South", "East", "West", "Northeast", "Southwest"]), "containment_percentage": random.randint(0, 85), "area_burned": f"{random.randint(10, 5000)} hectares" }) elif disaster_type == "heatwave": disaster_data.update({ "heat_index": random.randint(40, 55), "uv_index": random.randint(8, 15), "cooling_centers_open": random.randint(3, 20), "peak_temperature_time": "2:00 PM - 4:00 PM", "heat_related_cases": random.randint(10, 200) }) elif disaster_type == "tsunami": disaster_data.update({ "estimated_wave_height": f"{random.randint(2, 20)} meters", "estimated_arrival_time": (datetime.now() + timedelta(minutes=random.randint(30, 300))).isoformat(), "coastal_impact_zones": random.sample(["Beach areas", "Ports", "Fishing villages", "Tourist areas"], random.randint(2, 4)), "sea_level_rise": f"{random.randint(1, 10)} meters above normal" }) # Add optional fields randomly if random.choice([True, False]): disaster_data["urgency"] = random.choice(["Low", "Medium", "High", "Critical"]) if random.choice([True, False]): disaster_data["expires_at"] = (datetime.now() + timedelta(days=7)).isoformat() if random.choice([True, False]): disaster_data["evacuation_status"] = random.choice(["Not Required", "Recommended", "Mandatory"]) if random.choice([True, False]): disaster_data["relief_centers"] = random.randint(2, 10) if random.choice([True, False]): disaster_data["rescue_teams_deployed"] = random.randint(1, 8) # Add weather conditions for relevant disasters if disaster_type in ["cyclone", "flood", "heatwave"] and random.choice([True, False]): disaster_data["weather_conditions"] = { "temperature": random.randint(25, 45), "humidity": random.randint(40, 90), "wind_speed": random.randint(10, 100) } # Add additional helplines sometimes if random.choice([True, False]): disaster_data["additional_helplines"] = { "ndrf": "9711077372", "women_helpline": "1091", "disaster_management": "1070" } # Always add meta fields disaster_data.update({ "last_updated": datetime.now().isoformat(), "source": "National Disaster Management Authority", "alert_color": SEVERITY_COLOR_MAP.get(severity, "Yellow") }) return disaster_data def send_firebase_notification(disaster_data): """Send Firebase notification to 'all' topic with disaster data""" if not firebase_app: raise Exception("Firebase Admin SDK not initialized") disaster_info = DISASTERS.get(disaster_data["disaster_type"], {}) emoji = disaster_info.get("emoji", "๐Ÿšจ") # Create notification title and body title = f"{emoji} {disaster_data['disaster_name']}" body = f"{disaster_data['severity']} alert in {disaster_data['location']}. Tap for details." # Prepare data payload - convert all values to strings for FCM data_payload = {} for key, value in disaster_data.items(): if isinstance(value, (dict, list)): data_payload[key] = json.dumps(value) else: data_payload[key] = str(value) # Add notification metadata data_payload.update({ "notification_type": "disaster_alert", "click_action": "FLUTTER_NOTIFICATION_CLICK", "sound": "default" }) # Create FCM message message = messaging.Message( notification=messaging.Notification( title=title, body=body ), data=data_payload, topic="all", # Send to all subscribed users android=messaging.AndroidConfig( priority='high', notification=messaging.AndroidNotification( icon='@drawable/ic_notification', color='#FF0000', sound='default', channel_id='disaster_alerts', priority='high', default_sound=True, default_vibrate_timings=True ) ), apns=messaging.APNSConfig( payload=messaging.APNSPayload( aps=messaging.Aps( alert=messaging.ApsAlert( title=title, body=body ), badge=1, sound='default', content_available=True ) ) ) ) # Send the message response = messaging.send(message) return response @app.route('/') def home(): firebase_status = "โœ… Connected" if firebase_app else "โŒ Not Connected" return f"""

๐Ÿšจ Disaster Management API

Real-time disaster alerts with Firebase notifications!

Firebase Status: {firebase_status}

Available Endpoints:

Setup Instructions:

  1. Place your firebase-service-account.json file in the app directory
  2. Install Firebase Admin SDK: pip install firebase-admin
  3. Your Flutter app should subscribe to 'all' topic

Perfect for testing your Flutter disaster management app!

""" @app.route('/anything') def get_random_disaster(): """๐ŸŽฒ Returns random disaster data - Original endpoint""" disaster_data = generate_disaster_data() return jsonify(disaster_data) @app.route('/notify', methods=['GET', 'POST']) def send_notification(): """๐Ÿ“ฑ Generate random disaster data, send notification, and store it in Firestore""" if not FIREBASE_AVAILABLE: return jsonify({ "error": "Firebase Admin SDK not installed", "message": "Install with: pip install firebase-admin", "status": "failed" }), 500 if not firebase_app: return jsonify({ "error": "Firebase not initialized", "message": "Check firebase-service-account.json file", "status": "failed" }), 500 try: # Generate disaster data disaster_data = generate_disaster_data() # Force expiry to 7 days if not set # Always override expiry to 7 days from now disaster_data["expires_at"] = (datetime.now() + timedelta(days=7)).isoformat() logger.info(f"Generated disaster data: {disaster_data['alert_id']} - {disaster_data['disaster_name']}") # Send Firebase notification response = send_firebase_notification(disaster_data) logger.info(f"Firebase notification sent: {response}") # Prepare notification metadata notification_details = { "title": f"{DISASTERS[disaster_data['disaster_type']]['emoji']} {disaster_data['disaster_name']}", "body": f"{disaster_data['severity']} alert in {disaster_data['location']}. Tap for details.", "timestamp": datetime.now().isoformat(), "firebase_response": response, "topic": "all" } # Store in Firestore db = firestore.client() alert_id = disaster_data["alert_id"] now = datetime.now(IST) db.collection("alerts").document(alert_id).set({ "alert_id": alert_id, "data": disaster_data, "notification": notification_details, "status": "sent", "created_at": datetime.now(IST), "last_updated": datetime.now(IST), "expires_at": datetime.fromisoformat(disaster_data["expires_at"]) }) return jsonify({ "status": "success", "message": "Disaster notification sent to all users", "firebase_response": response, "disaster_data": disaster_data, "notification_details": notification_details }) except Exception as e: logger.error(f"Error sending/storing notification: {str(e)}") return jsonify({ "error": "Failed to send notification", "message": str(e), "status": "failed" }), 500 @app.route('/test-firebase') def test_firebase(): """๐Ÿ”ง Test Firebase connection""" if not FIREBASE_AVAILABLE: return jsonify({ "firebase_available": False, "error": "Firebase Admin SDK not installed" }) if not firebase_app: return jsonify({ "firebase_available": True, "firebase_initialized": False, "error": "Firebase not initialized - check service account file" }) try: # Send a test message test_message = messaging.Message( notification=messaging.Notification( title="๐Ÿงช Test Notification", body="Firebase connection is working!" ), data={"test": "true", "timestamp": datetime.now().isoformat()}, topic="all" ) response = messaging.send(test_message) return jsonify({ "firebase_available": True, "firebase_initialized": True, "test_message_sent": True, "response": response, "status": "success" }) except Exception as e: return jsonify({ "firebase_available": True, "firebase_initialized": True, "test_message_sent": False, "error": str(e), "status": "failed" }) if __name__ == '__main__': print("๐Ÿšจ Starting Disaster Management API...") print(f"Firebase Available: {FIREBASE_AVAILABLE}") print(f"Firebase Initialized: {firebase_app is not None}") app.run(host='0.0.0.0', port=7860, debug=True)