"""
Demo resource for the OpenEventDatabase.
"""
import falcon
import requests
import json
import os
from collections import defaultdict
from oedb.utils.logging import logger
from oedb.utils.db import load_env_from_file
class DemoResource:
"""
Resource for the demo endpoint.
Handles the /demo endpoint and related demo pages.
"""
def on_get_edit(self, req, resp, id=None):
"""
Handle GET requests to the /demo/edit endpoint.
Returns an HTML page with a form for editing an existing event.
Args:
req: The request object.
resp: The response object.
id: The event ID to edit.
"""
logger.info(f"Processing GET request to /demo/edit for event ID: {id}")
if id is None:
resp.status = falcon.HTTP_400
resp.text = "Event ID is required"
return
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Fetch the event data from the API
response = requests.get(f'http://api.openevent/event/{id}')
if response.status_code != 200:
resp.status = falcon.HTTP_404
resp.text = f"Event with ID {id} not found"
return
event_data = response.json()
# Create HTML response with form
html = f"""
Edit Event - OpenEventDatabase
"""
# Set the response body and status
resp.text = html.replace('{event_data}', json.dumps(event_data))
resp.status = falcon.HTTP_200
logger.success(f"Successfully processed GET request to /demo/edit for event ID: {id}")
except Exception as e:
logger.error(f"Error processing GET request to /demo/edit: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get(self, req, resp):
"""
Handle GET requests to the /demo endpoint.
Returns an HTML page with a MapLibre map showing current events.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Create HTML response with MapLibre map
html = """
OpenEventDatabase Demo
OpenEventDatabase Demo
This map shows current events from the OpenEventDatabase.
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo")
except Exception as e:
logger.error(f"Error processing GET request to /demo: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_by_what(self, req, resp):
"""
Handle GET requests to the /demo/by-what endpoint.
Returns an HTML page with links to events organized by their "what" type.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/by-what")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Fetch events from the API
try:
response = requests.get('/event?limit=1000')
if response.status_code == 200 and response.text:
events_data = response.json()
else:
logger.error(f"Error fetching events: Status code {response.status_code}, Response: {response.text}")
events_data = {"features": []}
except json.JSONDecodeError as e:
logger.error(f"Error parsing JSON response: {e}")
events_data = {"features": []}
except Exception as e:
logger.error(f"Error fetching events: {e}")
events_data = {"features": []}
# Group events by "what" type
events_by_what = defaultdict(list)
if events_data.get('features'):
for feature in events_data['features']:
properties = feature.get('properties', {})
what = properties.get('what', 'Unknown')
events_by_what[what].append({
'id': properties.get('id'),
'label': properties.get('label', 'Unnamed Event'),
'coordinates': feature.get('geometry', {}).get('coordinates', [0, 0])
})
# Create HTML response
html = """
Events by Type - OpenEventDatabase
"
html += """
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_search(self, req, resp):
"""
Handle GET requests to the /demo/search endpoint.
Returns an HTML page with a form for searching events and displaying results.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/search")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Create HTML response with search form
html = """
Search Events - OpenEventDatabase
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/search")
except Exception as e:
logger.error(f"Error processing GET request to /demo/search: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_map_by_what(self, req, resp):
"""
Handle GET requests to the /demo/map-by-what endpoint.
Returns an HTML page with a MapLibre map showing events filtered by "what" type.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/map-by-what")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Create HTML response with MapLibre map and filtering controls
html = """
Map by Event Type - OpenEventDatabase
This map shows events from the OpenEventDatabase filtered by their type.
Use the filter panel on the right to show/hide different event types.
Loading events...
Filter by Event Type
Loading event types...
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/map-by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/map-by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
events_by_what = defaultdict(list)
if events_data.get('features'):
for feature in events_data['features']:
properties = feature.get('properties', {})
what = properties.get('what', 'Unknown')
events_by_what[what].append({
'id': properties.get('id'),
'label': properties.get('label', 'Unnamed Event'),
'coordinates': feature.get('geometry', {}).get('coordinates', [0, 0])
})
# Create HTML response
html = """
Events by Type - OpenEventDatabase
"
html += """
"""
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/by-what")
except Exception as e:
logger.error(f"Error processing GET request to /demo/by-what: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
def on_get_traffic(self, req, resp):
"""
Handle GET requests to the /demo/traffic endpoint.
Returns an HTML page with a form for reporting traffic jams.
Args:
req: The request object.
resp: The response object.
"""
logger.info("Processing GET request to /demo/traffic")
try:
# Set content type to HTML
resp.content_type = 'text/html'
# Load environment variables from .env file
load_env_from_file()
# Get OAuth2 configuration parameters
client_id = os.getenv("CLIENT_ID", "")
client_secret = os.getenv("CLIENT_SECRET", "")
client_authorizations = os.getenv("CLIENT_AUTORIZATIONS", "read_prefs")
client_redirect = os.getenv("CLIENT_REDIRECT", "")
# Check if we have an authorization code in the query parameters
auth_code = req.params.get('code', None)
auth_state = req.params.get('state', None)
# Variables to track authentication state
is_authenticated = False
osm_username = ""
osm_user_id = ""
# If we have an authorization code, exchange it for an access token
if auth_code:
logger.info(f"Received authorization code: {auth_code}")
try:
# Exchange authorization code for access token
token_url = "https://www.openstreetmap.org/oauth2/token"
token_data = {
"grant_type": "authorization_code",
"code": auth_code,
"redirect_uri": client_redirect,
"client_id": client_id,
"client_secret": client_secret
}
token_response = requests.post(token_url, data=token_data)
token_response.raise_for_status()
token_info = token_response.json()
access_token = token_info.get("access_token")
if access_token:
# Use access token to get user information
user_url = "https://api.openstreetmap.org/api/0.6/user/details.json"
headers = {"Authorization": f"Bearer {access_token}"}
user_response = requests.get(user_url, headers=headers)
user_response.raise_for_status()
user_info = user_response.json()
# Extract user information
user = user_info.get("user", {})
osm_username = user.get("display_name", "")
osm_user_id = user.get("id", "")
if osm_username:
is_authenticated = True
logger.info(f"User authenticated: {osm_username} (ID: {osm_user_id})")
else:
logger.error("Failed to get OSM username from user details")
else:
logger.error("Failed to get access token from token response")
except Exception as e:
logger.error(f"Error during OAuth2 token exchange: {e}")
# Create HTML response with form
# Start with the common HTML header
html_header = f"""
Report Traffic Jam - OpenEventDatabase
"""
# Concatenate the HTML parts to form the complete template
html = html_header + auth_section + html_footer
# Set the response body and status
resp.text = html
resp.status = falcon.HTTP_200
logger.success("Successfully processed GET request to /demo/traffic")
except Exception as e:
logger.error(f"Error processing GET request to /demo/traffic: {e}")
resp.status = falcon.HTTP_500
resp.text = f"Error: {str(e)}"
# Create a global instance of DemoResource
demo = DemoResource()