Edward Betts
10716f9874
Each Gantt bar and table row gets a data-conf-key attribute (ISO start date + conference name). A small JS lookup map connects them so hovering either element highlights both simultaneously: - Gantt bar: filter brightness + inset white box-shadow - Table row: yellow tint via background-color Hovering a Gantt bar also scrolls the matching table row into view (scrollIntoView nearest) so future conferences are reachable without manual scrolling. The key field is pre-computed in build_conference_timeline() to keep the template simple. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
1325 lines
43 KiB
Python
Executable file
1325 lines
43 KiB
Python
Executable file
#!/usr/bin/python3
|
||
|
||
"""Web page to show upcoming events."""
|
||
|
||
import decimal
|
||
import hashlib
|
||
import importlib
|
||
import inspect
|
||
import json
|
||
import operator
|
||
import os.path
|
||
import sys
|
||
import time
|
||
import traceback
|
||
import typing
|
||
from collections import defaultdict
|
||
from datetime import date, datetime, timedelta, timezone
|
||
from zoneinfo import ZoneInfo
|
||
|
||
import flask
|
||
import pytz
|
||
import werkzeug
|
||
import werkzeug.debug.tbtools
|
||
import yaml
|
||
from authlib.integrations.flask_client import OAuth
|
||
from werkzeug.middleware.proxy_fix import ProxyFix
|
||
|
||
import agenda.data
|
||
import agenda.error_mail
|
||
import agenda.fx
|
||
import agenda.holidays
|
||
import agenda.meteors
|
||
import agenda.stats
|
||
import agenda.thespacedevs
|
||
import agenda.trip
|
||
import agenda.trip_schengen
|
||
import agenda.uk_school_holiday
|
||
import agenda.utils
|
||
import agenda.weather
|
||
from agenda import ical, calendar, format_list_with_ampersand, travel, uk_tz
|
||
from agenda.event import Event
|
||
from agenda.types import StrDict, Trip
|
||
|
||
app = flask.Flask(__name__)
|
||
app.debug = False
|
||
app.config.from_object("config.default")
|
||
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
|
||
|
||
agenda.error_mail.setup_error_mail(app)
|
||
|
||
oauth = OAuth(app)
|
||
authentik_url = app.config["AUTHENTIK_URL"]
|
||
oauth.register(
|
||
name="authentik",
|
||
client_id=app.config["AUTHENTIK_CLIENT_ID"],
|
||
client_secret=app.config["AUTHENTIK_CLIENT_SECRET"],
|
||
server_metadata_url=f"{authentik_url}/application/o/agenda/.well-known/openid-configuration",
|
||
client_kwargs={"scope": "openid email profile"},
|
||
)
|
||
|
||
|
||
class User:
|
||
"""Simple user object for template compatibility."""
|
||
|
||
def __init__(self, is_authenticated: bool) -> None:
|
||
"""Init."""
|
||
self.is_authenticated = is_authenticated
|
||
|
||
|
||
@app.before_request
|
||
def handle_auth() -> None:
|
||
"""Set global user from session."""
|
||
flask.g.user = User(is_authenticated=bool(flask.session.get("user")))
|
||
|
||
|
||
@app.errorhandler(werkzeug.exceptions.InternalServerError)
|
||
def exception_handler(e: werkzeug.exceptions.InternalServerError) -> tuple[str, int]:
|
||
"""Handle exception."""
|
||
exec_type, exc_value, current_traceback = sys.exc_info()
|
||
assert exc_value
|
||
tb = werkzeug.debug.tbtools.DebugTraceback(exc_value)
|
||
|
||
summary = tb.render_traceback_html(include_title=False)
|
||
exc_lines = "".join(tb._te.format_exception_only())
|
||
|
||
last_frame = list(traceback.walk_tb(current_traceback))[-1][0]
|
||
last_frame_args = inspect.getargs(last_frame.f_code)
|
||
assert tb._te.exc_type
|
||
|
||
return (
|
||
flask.render_template(
|
||
"show_error.html",
|
||
plaintext=tb.render_traceback_text(),
|
||
exception=exc_lines,
|
||
exception_type=tb._te.exc_type.__name__,
|
||
summary=summary,
|
||
last_frame=last_frame,
|
||
last_frame_args=last_frame_args,
|
||
),
|
||
500,
|
||
)
|
||
|
||
|
||
@app.route("/")
|
||
async def index() -> str:
|
||
"""Index page."""
|
||
t0 = time.time()
|
||
now = datetime.now()
|
||
data = await agenda.data.get_data(now, app.config)
|
||
|
||
events = data.pop("events")
|
||
today = now.date()
|
||
|
||
markets_arg = flask.request.args.get("markets")
|
||
if markets_arg == "hide":
|
||
events = [e for e in events if e.name != "market"]
|
||
if markets_arg != "show":
|
||
agenda.data.hide_markets_while_away(events, data["accommodation_events"])
|
||
|
||
return flask.render_template(
|
||
"event_list.html",
|
||
today=today,
|
||
events=events,
|
||
get_country=agenda.get_country,
|
||
current_trip=agenda.trip.get_current_trip(today),
|
||
start_event_list=date.today() - timedelta(days=1),
|
||
end_event_list=date.today() + timedelta(days=365 * 2),
|
||
render_time=(time.time() - t0),
|
||
home_weather=get_home_weather(),
|
||
**data,
|
||
)
|
||
|
||
|
||
@app.route("/calendar")
|
||
async def calendar_page() -> str:
|
||
"""Index page."""
|
||
now = datetime.now()
|
||
data = await agenda.data.get_data(now, app.config)
|
||
|
||
events = data.pop("events")
|
||
|
||
markets_arg = flask.request.args.get("markets")
|
||
if markets_arg == "hide":
|
||
events = [e for e in events if e.name != "market"]
|
||
if markets_arg != "show":
|
||
agenda.data.hide_markets_while_away(events, data["accommodation_events"])
|
||
|
||
# Use the new function to build events and pass both calendars and events
|
||
return flask.render_template(
|
||
"calendar.html",
|
||
today=now.date(),
|
||
events=events,
|
||
toastui_events=calendar.build_toastui_events(events),
|
||
toastui_calendars=calendar.toastui_calendars,
|
||
**data,
|
||
)
|
||
|
||
|
||
@app.route("/recent")
|
||
async def recent() -> str:
|
||
"""Index page."""
|
||
t0 = time.time()
|
||
now = datetime.now()
|
||
data = await agenda.data.get_data(now, app.config)
|
||
|
||
events = data.pop("events")
|
||
|
||
markets_arg = flask.request.args.get("markets")
|
||
if markets_arg == "hide":
|
||
events = [e for e in events if e.name != "market"]
|
||
if markets_arg != "show":
|
||
agenda.data.hide_markets_while_away(events, data["accommodation_events"])
|
||
|
||
return flask.render_template(
|
||
"event_list.html",
|
||
today=now.date(),
|
||
events=events,
|
||
start_event_list=date.today() - timedelta(days=14),
|
||
end_event_list=date.today(),
|
||
render_time=(time.time() - t0),
|
||
**data,
|
||
)
|
||
|
||
|
||
@app.route("/launches")
|
||
def launch_list() -> str:
|
||
"""Web page showing List of space launches."""
|
||
now = datetime.now(timezone.utc)
|
||
data_dir = app.config["DATA_DIR"]
|
||
rocket_dir = os.path.join(data_dir, "thespacedevs")
|
||
launches = agenda.thespacedevs.get_launches(rocket_dir, limit=100)
|
||
assert launches
|
||
active_crewed = agenda.thespacedevs.get_active_crewed_flights(rocket_dir) or []
|
||
active_crewed_slugs = {
|
||
launch["slug"]
|
||
for launch in active_crewed
|
||
if isinstance(launch.get("slug"), str)
|
||
}
|
||
|
||
for launch in launches:
|
||
launch_net = agenda.thespacedevs.parse_api_datetime(launch.get("net"))
|
||
launch["is_future"] = bool(launch_net and launch_net > now)
|
||
launch["is_active_crewed"] = launch.get("slug") in active_crewed_slugs
|
||
|
||
mission_type_filter = flask.request.args.get("type")
|
||
rocket_filter = flask.request.args.get("rocket")
|
||
orbit_filter = flask.request.args.get("orbit")
|
||
|
||
mission_types = {
|
||
launch["mission"]["type"] for launch in launches if launch["mission"]
|
||
}
|
||
|
||
orbits = {
|
||
(launch["orbit"]["name"], launch["orbit"]["abbrev"])
|
||
for launch in launches
|
||
if launch.get("orbit")
|
||
}
|
||
rockets = {launch["rocket"]["full_name"] for launch in launches}
|
||
|
||
launches = [
|
||
launch
|
||
for launch in launches
|
||
if (
|
||
not mission_type_filter
|
||
or (launch["mission"] and launch["mission"]["type"] == mission_type_filter)
|
||
)
|
||
and (not rocket_filter or launch["rocket"]["full_name"] == rocket_filter)
|
||
and (
|
||
not orbit_filter
|
||
or (launch.get("orbit") and launch["orbit"]["abbrev"] == orbit_filter)
|
||
)
|
||
]
|
||
|
||
return flask.render_template(
|
||
"launches.html",
|
||
launches=launches,
|
||
rockets=rockets,
|
||
now=now,
|
||
get_country=agenda.get_country,
|
||
mission_types=mission_types,
|
||
orbits=orbits,
|
||
)
|
||
|
||
|
||
@app.route("/meteors")
|
||
def meteor_list() -> str:
|
||
"""Web page showing meteor shower information."""
|
||
meteors = agenda.meteors.get_meteor_data()
|
||
|
||
return flask.render_template(
|
||
"meteors.html",
|
||
meteors=meteors,
|
||
today=date.today(),
|
||
)
|
||
|
||
|
||
@app.route("/gaps")
|
||
async def gaps_page() -> str:
|
||
"""List of available gaps."""
|
||
now = datetime.now()
|
||
trip_list = agenda.trip.build_trip_list()
|
||
busy_events = agenda.busy.get_busy_events(now.date(), app.config, trip_list)
|
||
gaps = agenda.busy.find_gaps(busy_events)
|
||
return flask.render_template("gaps.html", today=now.date(), gaps=gaps)
|
||
|
||
|
||
@app.route("/weekends")
|
||
async def weekends() -> str:
|
||
"""List of available weekends using an optional date, week, or year parameter."""
|
||
today = datetime.now().date()
|
||
min_year = 2020
|
||
max_year = today.year + 5
|
||
|
||
def validate_year(year: int) -> None:
|
||
"""Validate year parameter range for weekends page."""
|
||
if year < min_year or year > max_year:
|
||
flask.abort(
|
||
400, description=f"Year must be between {min_year} and {max_year}."
|
||
)
|
||
|
||
date_str = flask.request.args.get("date")
|
||
week_str = flask.request.args.get("week")
|
||
year_str = flask.request.args.get("year")
|
||
|
||
if date_str:
|
||
try:
|
||
start = datetime.strptime(date_str, "%Y-%m-%d").date()
|
||
validate_year(start.year)
|
||
except ValueError:
|
||
return flask.abort(400, description="Invalid date format. Use YYYY-MM-DD.")
|
||
elif week_str:
|
||
try:
|
||
week = int(week_str)
|
||
year = int(year_str) if year_str else today.year
|
||
validate_year(year)
|
||
if week < 1 or week > 53:
|
||
return flask.abort(
|
||
400, description="Week number must be between 1 and 53."
|
||
)
|
||
# Calculate the date of the first day of the given week
|
||
jan_1 = date(year, 1, 1)
|
||
week_1_start = jan_1 - timedelta(days=jan_1.weekday())
|
||
start = week_1_start + timedelta(weeks=week - 1)
|
||
except ValueError:
|
||
return flask.abort(
|
||
400, description="Invalid week or year format. Use integers."
|
||
)
|
||
elif year_str:
|
||
try:
|
||
year = int(year_str)
|
||
validate_year(year)
|
||
start = date(year, 1, 1)
|
||
except ValueError:
|
||
return flask.abort(400, description="Invalid year format. Use an integer.")
|
||
else:
|
||
start = date(today.year, 1, 1)
|
||
|
||
current_week_number = today.isocalendar().week
|
||
|
||
trip_list = agenda.trip.build_trip_list()
|
||
busy_events = agenda.busy.get_busy_events(start, app.config, trip_list)
|
||
weekends = agenda.busy.weekends(
|
||
start, busy_events, trip_list, app.config["PERSONAL_DATA"]
|
||
)
|
||
|
||
home_weather = get_home_weather()
|
||
home_weather_by_date = {f["date"]: f for f in home_weather}
|
||
for weekend in weekends:
|
||
saturday = weekend["date"]
|
||
sunday = saturday + timedelta(days=1)
|
||
weekend["saturday_weather"] = home_weather_by_date.get(saturday.isoformat())
|
||
weekend["sunday_weather"] = home_weather_by_date.get(sunday.isoformat())
|
||
|
||
return flask.render_template(
|
||
"weekends.html",
|
||
items=weekends,
|
||
current_week_number=current_week_number,
|
||
today=today,
|
||
)
|
||
|
||
|
||
@app.route("/travel")
|
||
def travel_list() -> str:
|
||
"""Page showing a list of upcoming travel."""
|
||
data_dir = app.config["PERSONAL_DATA"]
|
||
flights = agenda.trip.load_flight_bookings(data_dir)
|
||
trains = [
|
||
item
|
||
for item in travel.parse_yaml("trains", data_dir)
|
||
if isinstance(item["depart"], datetime)
|
||
]
|
||
|
||
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
|
||
|
||
for train in trains:
|
||
for leg in train["legs"]:
|
||
agenda.travel.add_leg_route_distance(leg, route_distances)
|
||
|
||
if all("distance" in leg for leg in train["legs"]):
|
||
train["distance"] = sum(leg["distance"] for leg in train["legs"])
|
||
|
||
return flask.render_template(
|
||
"travel.html",
|
||
flights=flights,
|
||
trains=trains,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
)
|
||
|
||
|
||
def build_conference_list() -> list[StrDict]:
|
||
"""Build conference list."""
|
||
data_dir = app.config["PERSONAL_DATA"]
|
||
filepath = os.path.join(data_dir, "conferences.yaml")
|
||
items: list[StrDict] = yaml.safe_load(open(filepath))
|
||
conference_trip_lookup = {}
|
||
|
||
for trip in agenda.trip.build_trip_list():
|
||
for trip_conf in trip.conferences:
|
||
key = (trip_conf["start"], trip_conf["name"])
|
||
conference_trip_lookup[key] = trip
|
||
|
||
for conf in items:
|
||
conf["start_date"] = agenda.utils.as_date(conf["start"])
|
||
conf["end_date"] = agenda.utils.as_date(conf["end"])
|
||
|
||
price = conf.get("price")
|
||
if price:
|
||
conf["price"] = decimal.Decimal(price)
|
||
|
||
key = (conf["start"], conf["name"])
|
||
if this_trip := conference_trip_lookup.get(key):
|
||
conf["linked_trip"] = this_trip
|
||
|
||
items.sort(key=operator.itemgetter("start_date"))
|
||
return items
|
||
|
||
|
||
def _conference_uid(conf: StrDict) -> str:
|
||
"""Generate deterministic UID for conference events."""
|
||
start = agenda.utils.as_date(conf["start"])
|
||
raw = f"conference|{start.isoformat()}|{conf.get('name','unknown')}"
|
||
digest = hashlib.sha1(raw.encode("utf-8")).hexdigest()
|
||
return f"conference-{digest}@agenda-codex"
|
||
|
||
|
||
def _conference_location(conf: StrDict) -> str | None:
|
||
"""Build conference location string."""
|
||
parts: list[str] = []
|
||
venue = conf.get("venue")
|
||
location = conf.get("location")
|
||
if isinstance(venue, str) and venue.strip():
|
||
parts.append(venue.strip())
|
||
if isinstance(location, str) and location.strip():
|
||
parts.append(location.strip())
|
||
if country_code := conf.get("country"):
|
||
country = agenda.get_country(country_code)
|
||
if country:
|
||
parts.append(country.name)
|
||
return ", ".join(parts) if parts else None
|
||
|
||
|
||
def _conference_description(conf: StrDict) -> str:
|
||
"""Build textual description for conferences."""
|
||
lines: list[str] = []
|
||
if topic := conf.get("topic"):
|
||
lines.append(f"Topic: {topic}")
|
||
if venue := conf.get("venue"):
|
||
lines.append(f"Venue: {venue}")
|
||
if address := conf.get("address"):
|
||
lines.append(f"Address: {address}")
|
||
if url := conf.get("url"):
|
||
lines.append(f"URL: {url}")
|
||
status_bits: list[str] = []
|
||
if conf.get("going"):
|
||
status_bits.append("attending")
|
||
if conf.get("speaking"):
|
||
status_bits.append("speaking")
|
||
if status_bits:
|
||
lines.append(f"Status: {', '.join(status_bits)}")
|
||
return "\n".join(lines) if lines else "Conference"
|
||
|
||
|
||
def build_conference_timeline(
|
||
current: list[StrDict], future: list[StrDict], today: date, days: int = 90
|
||
) -> dict | None:
|
||
"""Build data for a Gantt-style timeline of upcoming conferences."""
|
||
timeline_start = today
|
||
timeline_end = today + timedelta(days=days)
|
||
|
||
visible = [
|
||
c
|
||
for c in (current + future)
|
||
if c["start_date"] <= timeline_end and c["end_date"] >= today
|
||
]
|
||
if not visible:
|
||
return None
|
||
|
||
visible.sort(key=lambda c: c["start_date"])
|
||
|
||
# Greedy interval-coloring: assign each conference a lane (row)
|
||
lane_ends: list[date] = []
|
||
conf_data = []
|
||
for conf in visible:
|
||
lane = next(
|
||
(i for i, end in enumerate(lane_ends) if end < conf["start_date"]),
|
||
len(lane_ends),
|
||
)
|
||
if lane == len(lane_ends):
|
||
lane_ends.append(conf["end_date"])
|
||
else:
|
||
lane_ends[lane] = conf["end_date"]
|
||
|
||
start_off = max((conf["start_date"] - timeline_start).days, 0)
|
||
end_off = min((conf["end_date"] - timeline_start).days + 1, days)
|
||
left_pct = round(start_off / days * 100, 2)
|
||
width_pct = max(round((end_off - start_off) / days * 100, 2), 0.5)
|
||
|
||
conf_data.append(
|
||
{
|
||
"name": conf["name"],
|
||
"url": conf.get("url"),
|
||
"lane": lane,
|
||
"left_pct": left_pct,
|
||
"width_pct": width_pct,
|
||
"key": f"{conf['start_date'].isoformat()}|{conf['name']}",
|
||
"label": (
|
||
f"{conf['name']}"
|
||
f" ({conf['start_date'].strftime('%-d %b')}–"
|
||
f"{conf['end_date'].strftime('%-d %b')})"
|
||
),
|
||
}
|
||
)
|
||
|
||
# Month markers for x-axis labels
|
||
months = []
|
||
d = today.replace(day=1)
|
||
while d <= timeline_end:
|
||
off = max((d - timeline_start).days, 0)
|
||
months.append({"label": d.strftime("%b %Y"), "left_pct": round(off / days * 100, 2)})
|
||
# advance to next month
|
||
d = (d.replace(day=28) + timedelta(days=4)).replace(day=1)
|
||
|
||
return {
|
||
"confs": conf_data,
|
||
"lane_count": len(lane_ends),
|
||
"months": months,
|
||
"days": days,
|
||
}
|
||
|
||
|
||
def build_conference_ical(items: list[StrDict]) -> bytes:
|
||
"""Build iCalendar feed for all conferences."""
|
||
lines = [
|
||
"BEGIN:VCALENDAR",
|
||
"VERSION:2.0",
|
||
"PRODID:-//Agenda Codex//Conferences//EN",
|
||
"CALSCALE:GREGORIAN",
|
||
"METHOD:PUBLISH",
|
||
"X-WR-CALNAME:Conferences",
|
||
]
|
||
generated = datetime.now(tz=timezone.utc)
|
||
|
||
for conf in items:
|
||
start_date = agenda.utils.as_date(conf["start"])
|
||
end_date = agenda.utils.as_date(conf["end"])
|
||
end_exclusive = end_date + timedelta(days=1)
|
||
|
||
lines.append("BEGIN:VEVENT")
|
||
ical.append_property(lines, "UID", _conference_uid(conf))
|
||
ical.append_property(lines, "DTSTAMP", ical.format_datetime_utc(generated))
|
||
ical.append_property(lines, "DTSTART;VALUE=DATE", ical.format_date(start_date))
|
||
ical.append_property(lines, "DTEND;VALUE=DATE", ical.format_date(end_exclusive))
|
||
summary = ical.escape_text(f"conference: {conf['name']}")
|
||
ical.append_property(lines, "SUMMARY", summary)
|
||
description = ical.escape_text(_conference_description(conf))
|
||
ical.append_property(lines, "DESCRIPTION", description)
|
||
if location := _conference_location(conf):
|
||
ical.append_property(lines, "LOCATION", ical.escape_text(location))
|
||
lines.append("END:VEVENT")
|
||
|
||
lines.append("END:VCALENDAR")
|
||
ical_text = "\r\n".join(lines) + "\r\n"
|
||
return ical_text.encode("utf-8")
|
||
|
||
|
||
@app.route("/conference")
|
||
def conference_list() -> str:
|
||
"""Page showing a list of conferences."""
|
||
today = date.today()
|
||
items = build_conference_list()
|
||
|
||
current = [
|
||
conf
|
||
for conf in items
|
||
if conf["start_date"] <= today and conf["end_date"] >= today
|
||
]
|
||
future = [conf for conf in items if conf["start_date"] > today]
|
||
|
||
timeline = build_conference_timeline(current, future, today)
|
||
|
||
return flask.render_template(
|
||
"conference_list.html",
|
||
current=current,
|
||
future=future,
|
||
timeline=timeline,
|
||
today=today,
|
||
get_country=agenda.get_country,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
)
|
||
|
||
|
||
@app.route("/conference/past")
|
||
def past_conference_list() -> str:
|
||
"""Page showing a list of conferences."""
|
||
today = date.today()
|
||
return flask.render_template(
|
||
"conference_list.html",
|
||
past=[conf for conf in build_conference_list() if conf["end_date"] < today],
|
||
today=today,
|
||
get_country=agenda.get_country,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
)
|
||
|
||
|
||
@app.route("/conference/ical")
|
||
def conference_ical() -> werkzeug.Response:
|
||
"""Return all conferences as an iCalendar feed."""
|
||
items = build_conference_list()
|
||
ical_data = build_conference_ical(items)
|
||
response = flask.Response(ical_data, mimetype="text/calendar")
|
||
response.headers["Content-Disposition"] = "inline; filename=conferences.ics"
|
||
return response
|
||
|
||
|
||
@app.route("/accommodation")
|
||
def accommodation_list() -> str:
|
||
"""Page showing a list of past, present and future accommodation."""
|
||
data_dir = app.config["PERSONAL_DATA"]
|
||
items = travel.parse_yaml("accommodation", data_dir)
|
||
|
||
# Create a dictionary to hold stats for each year
|
||
year_stats: defaultdict[int, dict[str, int]] = defaultdict(
|
||
lambda: {"total_nights": 0, "nights_abroad": 0}
|
||
)
|
||
|
||
# Calculate stats for each year
|
||
for stay in items:
|
||
current_date = stay["from"].date()
|
||
end_date = stay["to"].date()
|
||
while current_date < end_date:
|
||
year = current_date.year
|
||
year_stats[year]["total_nights"] += 1
|
||
if stay.get("country") != "gb":
|
||
year_stats[year]["nights_abroad"] += 1
|
||
current_date += timedelta(days=1)
|
||
|
||
# Sort the stats by year in descending order
|
||
sorted_year_stats = sorted(
|
||
year_stats.items(), key=lambda item: item[0], reverse=True
|
||
)
|
||
|
||
trip_lookup = {}
|
||
|
||
for trip in agenda.trip.build_trip_list():
|
||
for trip_stay in trip.accommodation:
|
||
key = (trip_stay["from"], trip_stay["name"])
|
||
trip_lookup[key] = trip
|
||
|
||
for item in items:
|
||
key = (item["from"], item["name"])
|
||
if this_trip := trip_lookup.get(key):
|
||
item["linked_trip"] = this_trip
|
||
|
||
now = uk_tz.localize(datetime.now())
|
||
|
||
past = [conf for conf in items if conf["to"] < now]
|
||
current = [conf for conf in items if conf["from"] <= now and conf["to"] >= now]
|
||
future = [conf for conf in items if conf["from"] > now]
|
||
|
||
return flask.render_template(
|
||
"accommodation.html",
|
||
past=past,
|
||
current=current,
|
||
future=future,
|
||
year_stats=sorted_year_stats,
|
||
get_country=agenda.get_country,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
)
|
||
|
||
|
||
@app.route("/trip")
|
||
def trip_list() -> werkzeug.Response:
|
||
"""Trip list to redirect to future trip list."""
|
||
return flask.redirect(flask.url_for("trip_future_list"))
|
||
|
||
|
||
@app.route("/trip/ical")
|
||
def trip_ical() -> werkzeug.Response:
|
||
"""Return trip components in iCal format."""
|
||
trips = get_trip_list()
|
||
ical_data = agenda.trip.build_trip_ical(trips)
|
||
response = flask.Response(ical_data, mimetype="text/calendar")
|
||
response.headers["Content-Disposition"] = "inline; filename=trips.ics"
|
||
return response
|
||
|
||
|
||
def calc_total_distance(trips: list[Trip]) -> float:
|
||
"""Total distance for trips."""
|
||
total = 0.0
|
||
for item in trips:
|
||
if dist := item.total_distance():
|
||
total += dist
|
||
|
||
return total
|
||
|
||
|
||
def calc_total_co2_kg(trips: list[Trip]) -> float:
|
||
"""Total CO₂ for trips."""
|
||
return sum(item.total_co2_kg() or 0.0 for item in trips)
|
||
|
||
|
||
def sum_distances_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
|
||
"""Sum distances by transport type."""
|
||
distances_by_transport_type: defaultdict[str, float] = defaultdict(float)
|
||
for trip in trips:
|
||
for transport_type, dist in trip.distances_by_transport_type():
|
||
distances_by_transport_type[transport_type] += dist
|
||
|
||
return list(distances_by_transport_type.items())
|
||
|
||
|
||
def get_home_weather() -> list[StrDict]:
|
||
"""Get Bristol home weather forecast, with date objects added for templates."""
|
||
from datetime import date as date_type
|
||
|
||
forecasts = agenda.weather.get_forecast(
|
||
app.config["DATA_DIR"],
|
||
app.config.get("OPENWEATHERMAP_API_KEY", ""),
|
||
app.config["HOME_LATITUDE"],
|
||
app.config["HOME_LONGITUDE"],
|
||
cache_only=True,
|
||
)
|
||
for f in forecasts:
|
||
f["date_obj"] = date_type.fromisoformat(f["date"])
|
||
return forecasts
|
||
|
||
|
||
def get_trip_list() -> list[Trip]:
|
||
"""Get trip list with route distances."""
|
||
route_distances = agenda.travel.load_route_distances(app.config["DATA_DIR"])
|
||
return agenda.trip.get_trip_list(route_distances)
|
||
|
||
|
||
def trip_school_holiday_map(trips: list[Trip]) -> dict[str, list[Event]]:
|
||
"""Map trip-start ISO date to overlapping UK school holidays."""
|
||
if not trips:
|
||
return {}
|
||
|
||
starts = [trip.start for trip in trips]
|
||
ends = [trip.end or trip.start for trip in trips]
|
||
school_holidays = agenda.holidays.get_school_holidays(
|
||
min(starts),
|
||
max(ends),
|
||
app.config["DATA_DIR"],
|
||
)
|
||
|
||
result: dict[str, list[Event]] = {}
|
||
for trip in trips:
|
||
trip_end = trip.end or trip.start
|
||
overlaps = [
|
||
school_holiday
|
||
for school_holiday in school_holidays
|
||
if school_holiday.as_date <= trip_end
|
||
and school_holiday.end_as_date >= trip.start
|
||
]
|
||
result[trip.start.isoformat()] = overlaps
|
||
|
||
return result
|
||
|
||
|
||
@app.route("/trip/past")
|
||
def trip_past_list() -> str:
|
||
"""Page showing a list of past trips."""
|
||
today = date.today()
|
||
past = [item for item in get_trip_list() if (item.end or item.start) < today]
|
||
coordinates, routes = agenda.trip.get_coordinates_and_routes(past)
|
||
|
||
return flask.render_template(
|
||
"trip/list.html",
|
||
heading="Past trips",
|
||
trips=reversed(past),
|
||
trip_school_holiday_map=trip_school_holiday_map(past),
|
||
coordinates=coordinates,
|
||
routes=routes,
|
||
today=today,
|
||
get_country=agenda.get_country,
|
||
format_list_with_ampersand=format_list_with_ampersand,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
total_distance=calc_total_distance(past),
|
||
total_co2_kg=calc_total_co2_kg(past),
|
||
distances_by_transport_type=sum_distances_by_transport_type(past),
|
||
)
|
||
|
||
|
||
@app.route("/fixture/map")
|
||
def fixture_trip_map() -> str:
|
||
"""Map of trips."""
|
||
trips = get_trip_list()
|
||
coordinates, routes = agenda.trip.get_coordinates_and_routes(trips)
|
||
|
||
return flask.render_template(
|
||
"fixture/map.html",
|
||
heading="Trip map",
|
||
trips=trips,
|
||
coordinates=coordinates,
|
||
routes=routes,
|
||
get_country=agenda.get_country,
|
||
)
|
||
|
||
|
||
@app.route("/trip/future")
|
||
def trip_future_list() -> str:
|
||
"""Page showing a list of future trips."""
|
||
trip_list = get_trip_list()
|
||
today = date.today()
|
||
|
||
current = [
|
||
item
|
||
for item in trip_list
|
||
if item.start <= today and (item.end or item.start) >= today
|
||
]
|
||
|
||
future = [item for item in trip_list if item.start > today]
|
||
|
||
coordinates, routes = agenda.trip.get_coordinates_and_routes(current + future)
|
||
|
||
shown = current + future
|
||
|
||
trip_weather_map = {
|
||
trip.start.isoformat(): agenda.weather.get_trip_weather(
|
||
app.config["DATA_DIR"],
|
||
app.config.get("OPENWEATHERMAP_API_KEY", ""),
|
||
trip,
|
||
cache_only=True,
|
||
)
|
||
for trip in shown
|
||
}
|
||
|
||
return flask.render_template(
|
||
"trip/list.html",
|
||
heading="Future trips",
|
||
trips=shown,
|
||
trip_school_holiday_map=trip_school_holiday_map(shown),
|
||
trip_weather_map=trip_weather_map,
|
||
coordinates=coordinates,
|
||
routes=routes,
|
||
today=today,
|
||
get_country=agenda.get_country,
|
||
format_list_with_ampersand=format_list_with_ampersand,
|
||
fx_rate=agenda.fx.get_rates(app.config),
|
||
total_distance=calc_total_distance(current + future),
|
||
total_co2_kg=calc_total_co2_kg(current + future),
|
||
distances_by_transport_type=sum_distances_by_transport_type(current + future),
|
||
)
|
||
|
||
|
||
@app.route("/trip/text")
|
||
def trip_list_text() -> str:
|
||
"""Page showing a list of trips."""
|
||
trip_list = agenda.trip.get_trip_list()
|
||
today = date.today()
|
||
future = [item for item in trip_list if item.start > today]
|
||
|
||
return flask.render_template(
|
||
"trip_list_text.html",
|
||
future=future,
|
||
today=today,
|
||
get_country=agenda.get_country,
|
||
format_list_with_ampersand=format_list_with_ampersand,
|
||
)
|
||
|
||
|
||
def get_prev_current_and_next_trip(
|
||
start: str, trip_list: list[Trip]
|
||
) -> tuple[Trip | None, Trip | None, Trip | None]:
|
||
"""Get previous trip, this trip and next trip."""
|
||
trip_iter = iter(trip_list)
|
||
prev_trip = None
|
||
current_trip = None
|
||
for trip in trip_iter:
|
||
if trip.start.isoformat() == start:
|
||
current_trip = trip
|
||
break
|
||
prev_trip = trip
|
||
next_trip = next(trip_iter, None)
|
||
|
||
return (prev_trip, current_trip, next_trip)
|
||
|
||
|
||
def _timezone_name_from_datetime(value: typing.Any) -> str | None:
|
||
"""Get IANA timezone name from a datetime value if available."""
|
||
if not isinstance(value, datetime) or value.tzinfo is None:
|
||
return None
|
||
|
||
key = getattr(value.tzinfo, "key", None)
|
||
if isinstance(key, str):
|
||
return key
|
||
|
||
zone = getattr(value.tzinfo, "zone", None)
|
||
if isinstance(zone, str):
|
||
return zone
|
||
|
||
return None
|
||
|
||
|
||
def _format_offset_from_uk(offset_minutes: int) -> str:
|
||
"""Format offset from UK in +/-HH:MM."""
|
||
if offset_minutes == 0:
|
||
return "No difference"
|
||
sign = "+" if offset_minutes > 0 else "-"
|
||
hours, mins = divmod(abs(offset_minutes), 60)
|
||
return f"{sign}{hours:02d}:{mins:02d} vs UK"
|
||
|
||
|
||
def _trip_offset_minutes(
|
||
trip_start: date, trip_end: date, destination_timezone: str
|
||
) -> list[int]:
|
||
"""Unique UTC offset differences vs UK across the trip date range."""
|
||
destination_tz = ZoneInfo(destination_timezone)
|
||
uk_timezone = ZoneInfo("Europe/London")
|
||
current = trip_start
|
||
offsets: set[int] = set()
|
||
|
||
while current <= trip_end:
|
||
instant = datetime(
|
||
current.year, current.month, current.day, 12, tzinfo=timezone.utc
|
||
)
|
||
destination_offset = instant.astimezone(destination_tz).utcoffset()
|
||
uk_offset = instant.astimezone(uk_timezone).utcoffset()
|
||
if destination_offset is not None and uk_offset is not None:
|
||
offsets.add(int((destination_offset - uk_offset).total_seconds() // 60))
|
||
current += timedelta(days=1)
|
||
|
||
return sorted(offsets)
|
||
|
||
|
||
def _format_trip_offset_display(offsets: list[int]) -> str:
|
||
"""Format trip-range offsets; include variation if DST changes during trip."""
|
||
if not offsets:
|
||
return "Timezone unknown"
|
||
if len(offsets) == 1:
|
||
return _format_offset_from_uk(offsets[0])
|
||
return (
|
||
"Varies during trip: "
|
||
f"{_format_offset_from_uk(offsets[0])} to {_format_offset_from_uk(offsets[-1])}"
|
||
)
|
||
|
||
|
||
def _timezone_from_coordinates(latitude: float, longitude: float) -> str | None:
|
||
"""Resolve IANA timezone name from coordinates."""
|
||
timezone_finder = _get_timezone_finder()
|
||
if timezone_finder is None:
|
||
return None
|
||
|
||
for method_name in ("timezone_at", "certain_timezone_at", "closest_timezone_at"):
|
||
finder_method = getattr(timezone_finder, method_name, None)
|
||
if not callable(finder_method):
|
||
continue
|
||
tz_name = finder_method(lng=longitude, lat=latitude)
|
||
if isinstance(tz_name, str):
|
||
return tz_name
|
||
return None
|
||
|
||
|
||
def _get_timezone_finder() -> typing.Any:
|
||
"""Get timezone finder instance if dependency is available."""
|
||
try:
|
||
timezonefinder_module = importlib.import_module("timezonefinder")
|
||
except ModuleNotFoundError:
|
||
return None
|
||
|
||
timezone_finder_cls = getattr(timezonefinder_module, "TimezoneFinder", None)
|
||
if timezone_finder_cls is None:
|
||
return None
|
||
|
||
return timezone_finder_cls()
|
||
|
||
|
||
def get_destination_timezones(trip: Trip) -> list[StrDict]:
|
||
"""Build destination timezone metadata for the trip page."""
|
||
per_location: dict[tuple[str, str], list[str]] = defaultdict(list)
|
||
location_coords: dict[tuple[str, str], tuple[float, float]] = {}
|
||
for item in trip.accommodation + trip.conferences + trip.events:
|
||
location = item.get("location")
|
||
country = item.get("country")
|
||
if not isinstance(location, str) or not isinstance(country, str):
|
||
continue
|
||
|
||
key = (location, country.lower())
|
||
timezone_name = item.get("timezone")
|
||
if isinstance(timezone_name, str):
|
||
per_location[key].append(timezone_name)
|
||
|
||
latitude = item.get("latitude")
|
||
longitude = item.get("longitude")
|
||
if isinstance(latitude, (int, float)) and isinstance(longitude, (int, float)):
|
||
location_coords[key] = (float(latitude), float(longitude))
|
||
|
||
for field in (
|
||
"from",
|
||
"to",
|
||
"date",
|
||
"start",
|
||
"end",
|
||
"attend_start",
|
||
"attend_end",
|
||
):
|
||
candidate = _timezone_name_from_datetime(item.get(field))
|
||
if candidate:
|
||
per_location[key].append(candidate)
|
||
|
||
# Also collect airport locations from flights, for transit countries
|
||
flight_locations: list[tuple[str, Country]] = []
|
||
seen_flight_keys: set[tuple[str, str]] = set()
|
||
for item in trip.travel:
|
||
if item.get("type") != "flight":
|
||
continue
|
||
for airport_key in ("from_airport", "to_airport"):
|
||
airport = item.get(airport_key)
|
||
if not isinstance(airport, dict):
|
||
continue
|
||
city = airport.get("city")
|
||
country_code = airport.get("country")
|
||
if not isinstance(city, str) or not isinstance(country_code, str):
|
||
continue
|
||
if country_code == "gb":
|
||
continue
|
||
key = (city, country_code.lower())
|
||
lat = airport.get("latitude")
|
||
lon = airport.get("longitude")
|
||
if isinstance(lat, (int, float)) and isinstance(lon, (int, float)):
|
||
location_coords.setdefault(key, (float(lat), float(lon)))
|
||
if key not in seen_flight_keys:
|
||
seen_flight_keys.add(key)
|
||
flight_country = agenda.get_country(country_code)
|
||
if flight_country:
|
||
flight_locations.append((city, flight_country))
|
||
|
||
existing_location_keys = {(loc, c.alpha_2.lower()) for loc, c in trip.locations()}
|
||
all_locations = list(trip.locations()) + [
|
||
(city, country)
|
||
for city, country in flight_locations
|
||
if (city, country.alpha_2.lower()) not in existing_location_keys
|
||
]
|
||
|
||
destination_times: list[StrDict] = []
|
||
trip_end = trip.end or trip.start
|
||
|
||
for location, country in all_locations:
|
||
country_code = country.alpha_2.lower()
|
||
key = (location, country_code)
|
||
timezone_name = None
|
||
|
||
for candidate in per_location.get(key, []):
|
||
try:
|
||
ZoneInfo(candidate)
|
||
timezone_name = candidate
|
||
break
|
||
except Exception:
|
||
continue
|
||
|
||
if not timezone_name and key in location_coords:
|
||
latitude, longitude = location_coords[key]
|
||
coordinate_timezone = _timezone_from_coordinates(latitude, longitude)
|
||
if coordinate_timezone:
|
||
timezone_name = coordinate_timezone
|
||
|
||
if not timezone_name:
|
||
country_timezones = pytz.country_timezones.get(country_code, [])
|
||
if len(country_timezones) == 1:
|
||
timezone_name = country_timezones[0]
|
||
|
||
offset_display = "Timezone unknown"
|
||
if timezone_name:
|
||
offset_display = _format_trip_offset_display(
|
||
_trip_offset_minutes(trip.start, trip_end, timezone_name)
|
||
)
|
||
|
||
destination_times.append(
|
||
{
|
||
"location": location,
|
||
"country_name": country.name,
|
||
"country_flag": country.flag,
|
||
"timezone": timezone_name,
|
||
"offset_display": offset_display,
|
||
}
|
||
)
|
||
|
||
grouped: list[StrDict] = []
|
||
grouped_index: dict[tuple[str, str, str | None], int] = {}
|
||
for item in destination_times:
|
||
key = (item["country_name"], item["country_flag"], item["timezone"])
|
||
if key in grouped_index:
|
||
existing = grouped[grouped_index[key]]
|
||
existing_locations = typing.cast(list[str], existing["locations"])
|
||
existing_locations.append(typing.cast(str, item["location"]))
|
||
existing["location_count"] = (
|
||
typing.cast(int, existing["location_count"]) + 1
|
||
)
|
||
continue
|
||
|
||
grouped_index[key] = len(grouped)
|
||
grouped.append(
|
||
{
|
||
**item,
|
||
"locations": [item["location"]],
|
||
"location_count": 1,
|
||
}
|
||
)
|
||
|
||
for item in grouped:
|
||
location_count = typing.cast(int, item["location_count"])
|
||
country_name = typing.cast(str, item["country_name"])
|
||
country_flag = typing.cast(str, item["country_flag"])
|
||
if location_count > 1:
|
||
label = f"{country_name} ({location_count} locations)"
|
||
else:
|
||
label = f"{item['location']} ({country_name})"
|
||
if trip.show_flags:
|
||
label = f"{label} {country_flag}"
|
||
item["destination_label"] = label
|
||
|
||
return grouped
|
||
|
||
|
||
@app.route("/trip/<start>")
|
||
def trip_page(start: str) -> str:
|
||
"""Individual trip page."""
|
||
trip_list = get_trip_list()
|
||
|
||
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
||
if not trip:
|
||
flask.abort(404)
|
||
|
||
# Add Schengen compliance information
|
||
trip = agenda.trip_schengen.add_schengen_compliance_to_trip(trip)
|
||
|
||
coordinates = agenda.trip.collect_trip_coordinates(trip)
|
||
routes = agenda.trip.get_trip_routes(trip, app.config["PERSONAL_DATA"])
|
||
|
||
agenda.trip.add_coordinates_for_unbooked_flights(
|
||
routes, coordinates, app.config["PERSONAL_DATA"]
|
||
)
|
||
|
||
for route in routes:
|
||
if "geojson_filename" in route:
|
||
route["geojson"] = agenda.trip.read_geojson(
|
||
app.config["PERSONAL_DATA"], route.pop("geojson_filename")
|
||
)
|
||
|
||
trip_weather = agenda.weather.get_trip_weather(
|
||
app.config["DATA_DIR"],
|
||
app.config.get("OPENWEATHERMAP_API_KEY", ""),
|
||
trip,
|
||
cache_only=True,
|
||
)
|
||
|
||
return flask.render_template(
|
||
"trip_page.html",
|
||
trip=trip,
|
||
prev_trip=prev_trip,
|
||
next_trip=next_trip,
|
||
today=date.today(),
|
||
coordinates=coordinates,
|
||
routes=routes,
|
||
get_country=agenda.get_country,
|
||
format_list_with_ampersand=format_list_with_ampersand,
|
||
holidays=agenda.holidays.get_trip_holidays(trip),
|
||
school_holidays=agenda.holidays.get_trip_school_holidays(trip),
|
||
destination_times=get_destination_timezones(trip),
|
||
human_readable_delta=agenda.utils.human_readable_delta,
|
||
trip_weather=trip_weather,
|
||
)
|
||
|
||
|
||
@app.route("/trip/<start>/debug")
|
||
def trip_debug_page(start: str) -> str:
|
||
"""Trip debug page showing raw trip object data."""
|
||
|
||
if not flask.g.user.is_authenticated:
|
||
flask.abort(401)
|
||
|
||
trip_list = get_trip_list()
|
||
|
||
prev_trip, trip, next_trip = get_prev_current_and_next_trip(start, trip_list)
|
||
if not trip:
|
||
flask.abort(404)
|
||
|
||
# Add Schengen compliance information
|
||
trip = agenda.trip_schengen.add_schengen_compliance_to_trip(trip)
|
||
|
||
# Convert trip object to dictionary for display
|
||
trip_dict = {
|
||
"start": trip.start.isoformat(),
|
||
"name": trip.name,
|
||
"private": trip.private,
|
||
"travel": trip.travel,
|
||
"accommodation": trip.accommodation,
|
||
"conferences": trip.conferences,
|
||
"events": trip.events,
|
||
"flight_bookings": trip.flight_bookings,
|
||
"computed_properties": {
|
||
"title": trip.title,
|
||
"end": trip.end.isoformat() if trip.end else None,
|
||
"countries": [
|
||
{"name": c.name, "alpha_2": c.alpha_2, "flag": c.flag}
|
||
for c in trip.countries
|
||
],
|
||
"locations": [
|
||
{
|
||
"location": loc,
|
||
"country": {"name": country.name, "alpha_2": country.alpha_2},
|
||
}
|
||
for loc, country in trip.locations()
|
||
],
|
||
"total_distance": trip.total_distance(),
|
||
"total_co2_kg": trip.total_co2_kg(),
|
||
"distances_by_transport_type": trip.distances_by_transport_type(),
|
||
"co2_by_transport_type": trip.co2_by_transport_type(),
|
||
},
|
||
"schengen_compliance": (
|
||
{
|
||
"total_days_used": trip.schengen_compliance.total_days_used,
|
||
"days_remaining": trip.schengen_compliance.days_remaining,
|
||
"is_compliant": trip.schengen_compliance.is_compliant,
|
||
"current_180_day_period": [
|
||
trip.schengen_compliance.current_180_day_period[0].isoformat(),
|
||
trip.schengen_compliance.current_180_day_period[1].isoformat(),
|
||
],
|
||
"days_over_limit": trip.schengen_compliance.days_over_limit,
|
||
}
|
||
if trip.schengen_compliance
|
||
else None
|
||
),
|
||
}
|
||
|
||
# Convert to JSON for pretty printing
|
||
trip_json = json.dumps(trip_dict, indent=2, default=str)
|
||
trip_yaml = yaml.safe_dump(json.loads(trip_json), sort_keys=False)
|
||
|
||
return flask.render_template(
|
||
"trip_debug.html",
|
||
trip=trip,
|
||
trip_json=trip_json,
|
||
trip_yaml=trip_yaml,
|
||
start=start,
|
||
)
|
||
|
||
|
||
@app.route("/holidays")
|
||
def holiday_list() -> str:
|
||
"""List of holidays."""
|
||
today = date.today()
|
||
data_dir = app.config["DATA_DIR"]
|
||
year_start = date(today.year, 1, 1)
|
||
next_year = today + timedelta(days=1 * 365)
|
||
items = agenda.holidays.get_all(today - timedelta(days=2), next_year, data_dir)
|
||
school_holidays = agenda.holidays.get_school_holidays(
|
||
year_start, next_year, data_dir
|
||
)
|
||
|
||
items.sort(key=lambda item: (item.date, item.country))
|
||
school_holidays.sort(key=lambda item: (item.as_date, item.end_as_date))
|
||
|
||
return flask.render_template(
|
||
"holiday_list.html",
|
||
items=items,
|
||
school_holidays=school_holidays,
|
||
school_holiday_page_url=agenda.uk_school_holiday.school_holiday_page_url,
|
||
get_country=agenda.get_country,
|
||
today=today,
|
||
)
|
||
|
||
|
||
@app.route("/birthdays")
|
||
def birthday_list() -> str:
|
||
"""List of birthdays."""
|
||
today = date.today()
|
||
if not flask.g.user.is_authenticated:
|
||
flask.abort(401)
|
||
data_dir = app.config["PERSONAL_DATA"]
|
||
entities_file = os.path.join(data_dir, "entities.yaml")
|
||
items = agenda.birthday.get_birthdays(today - timedelta(days=2), entities_file)
|
||
items.sort(key=lambda item: item.date)
|
||
return flask.render_template("birthday_list.html", items=items, today=today)
|
||
|
||
|
||
@app.route("/trip/stats")
|
||
def trip_stats() -> str:
|
||
"""Travel stats: distance and price by year and travel type."""
|
||
trip_list = get_trip_list()
|
||
|
||
conferences = sum(len(item.conferences) for item in trip_list)
|
||
|
||
yearly_stats = agenda.stats.calculate_yearly_stats(
|
||
trip_list, app.config.get("PREVIOUSLY_VISITED")
|
||
)
|
||
overall_stats = agenda.stats.calculate_overall_stats(yearly_stats)
|
||
|
||
return flask.render_template(
|
||
"trip/stats.html",
|
||
count=len(trip_list),
|
||
total_distance=calc_total_distance(trip_list),
|
||
distances_by_transport_type=sum_distances_by_transport_type(trip_list),
|
||
yearly_stats=yearly_stats,
|
||
overall_stats=overall_stats,
|
||
conferences=conferences,
|
||
previously_visited=app.config.get("PREVIOUSLY_VISITED", set),
|
||
)
|
||
|
||
|
||
@app.route("/schengen")
|
||
def schengen_report() -> str:
|
||
"""Schengen compliance report."""
|
||
return agenda.trip_schengen.flask_route_schengen_report()
|
||
|
||
|
||
@app.route("/login")
|
||
def login() -> werkzeug.Response:
|
||
"""Redirect to Authentik for OIDC login."""
|
||
next_url = flask.request.args.get("next", flask.url_for("index"))
|
||
flask.session["login_next"] = next_url
|
||
redirect_uri = flask.url_for("auth_callback", _external=True)
|
||
return oauth.authentik.authorize_redirect(redirect_uri)
|
||
|
||
|
||
@app.route("/callback")
|
||
def auth_callback() -> werkzeug.Response:
|
||
"""Handle OIDC callback from Authentik."""
|
||
try:
|
||
token = oauth.authentik.authorize_access_token()
|
||
except Exception:
|
||
return flask.redirect(flask.url_for("login"))
|
||
userinfo = token.get("userinfo") or oauth.authentik.userinfo()
|
||
username = userinfo.get("preferred_username") or userinfo.get("email")
|
||
flask.session["user"] = {
|
||
"sub": userinfo["sub"],
|
||
"username": userinfo.get("preferred_username"),
|
||
"email": userinfo.get("email"),
|
||
}
|
||
flask.flash(f"Logged in as {username}")
|
||
next_url = flask.session.pop("login_next", flask.url_for("index"))
|
||
return flask.redirect(next_url)
|
||
|
||
|
||
@app.route("/logout")
|
||
def logout() -> werkzeug.Response:
|
||
"""Log out and redirect to Authentik end-session endpoint."""
|
||
flask.session.pop("user", None)
|
||
end_session_url = f"{authentik_url}/application/o/agenda/end-session/"
|
||
return flask.redirect(end_session_url)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
app.run(host="0.0.0.0")
|