"""VenueFlow — Flask application factory.""" import logging from flask import Flask, redirect, url_for, render_template from config import get_config # Configure logging logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", ) logger = logging.getLogger(__name__) def create_app(config_class=None): """Application factory.""" app = Flask(__name__) # Load configuration if config_class is None: config_class = get_config() app.config.from_object(config_class) # ─── Security Headers ──────────────────────────────────────────────── @app.after_request def set_security_headers(response): response.headers["X-Content-Type-Options"] = "nosniff" response.headers["X-Frame-Options"] = "DENY" response.headers["X-XSS-Protection"] = "1; mode=block" response.headers["Referrer-Policy"] = "strict-origin-when-cross-origin" if not app.debug: response.headers["Strict-Transport-Security"] = ( "max-age=31536000; includeSubDomains" ) return response # ─── Initialize Services ───────────────────────────────────────────── from services.firebase_service import init_firebase from services.gemini_service import init_gemini from services.translation_service import init_translation init_firebase(app) init_gemini(app) init_translation(app) # ─── Initialize Demo Data ──────────────────────────────────────────── from services.simulator import ( create_demo_venue, create_demo_queue_stations, create_demo_event, Simulator, ) from services.crowd_service import CrowdService from services.queue_service import QueueService from services.maps_service import MapsService from services.notification_service import NotificationService venue = create_demo_venue() event = create_demo_event() crowd_service = CrowdService() queue_service = QueueService() maps_service = MapsService(venue) notification_service = NotificationService() # Register queue stations queue_stations = create_demo_queue_stations() queue_service.register_stations(queue_stations) # Create simulator simulator = Simulator(venue, queue_service, crowd_service, notification_service, event) # Wire SSE broadcasting from blueprints.sse import broadcast_update def on_sim_update(): broadcast_update({ "venue": venue.to_dict(), "event": event.to_dict(), "sim_status": simulator.get_status(), }) simulator.set_update_callback(on_sim_update) # Store services in app config for blueprint access app.config["VENUE_OBJ"] = venue app.config["EVENT_OBJ"] = event app.config["CROWD_SERVICE"] = crowd_service app.config["QUEUE_SERVICE"] = queue_service app.config["MAPS_SERVICE"] = maps_service app.config["NOTIFICATION_SERVICE"] = notification_service app.config["SIMULATOR_OBJ"] = simulator # Run initial tick to populate data simulator.tick() # ─── Register Blueprints ───────────────────────────────────────────── from blueprints.auth import auth_bp from blueprints.attendee import attendee_bp from blueprints.operator import operator_bp from blueprints.api import api_bp from blueprints.sse import sse_bp app.register_blueprint(auth_bp) app.register_blueprint(attendee_bp) app.register_blueprint(operator_bp) app.register_blueprint(api_bp) app.register_blueprint(sse_bp) # ─── Root Route ────────────────────────────────────────────────────── @app.route("/") def index(): return redirect(url_for("auth.login")) # ─── Error Handlers ────────────────────────────────────────────────── @app.errorhandler(404) def not_found(e): return render_template("errors/404.html"), 404 @app.errorhandler(500) def server_error(e): return render_template("errors/500.html"), 500 logger.info("🏟️ VenueFlow app created successfully") return app # ─── Entry Point ───────────────────────────────────────────────────────────── if __name__ == "__main__": app = create_app() app.run(host="0.0.0.0", port=5000, debug=True, threaded=True)