1005 lines
32 KiB
Python
Executable file
1005 lines
32 KiB
Python
Executable file
#!/usr/bin/python3
|
|
|
|
"""Web page to show upcoming events."""
|
|
|
|
import inspect
|
|
import json
|
|
import os.path
|
|
import sys
|
|
import time
|
|
import traceback
|
|
import typing
|
|
from collections import defaultdict
|
|
from datetime import date
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import flask
|
|
import werkzeug
|
|
import werkzeug.debug.tbtools
|
|
import yaml
|
|
from authlib.integrations.flask_client import OAuth
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
|
|
import agenda.conference
|
|
import agenda.conference_ical
|
|
import agenda.conference_list
|
|
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.trip_timezones
|
|
import agenda.uk_school_holiday
|
|
import agenda.utils
|
|
import agenda.weather
|
|
from agenda import 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) # type: ignore[method-assign]
|
|
|
|
agenda.error_mail.setup_error_mail(app)
|
|
|
|
oauth = OAuth(app) # type: ignore[no-untyped-call]
|
|
authentik_url = app.config["AUTHENTIK_URL"]
|
|
oauth.register( # type: ignore[no-untyped-call]
|
|
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),
|
|
)
|
|
|
|
|
|
@app.route("/conference")
|
|
def conference_list() -> str:
|
|
"""Page showing a list of conferences."""
|
|
today = date.today()
|
|
items = agenda.conference_list.build_conference_list(
|
|
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
|
)
|
|
country_filter = agenda.conference_list.normalize_country_filter(
|
|
flask.request.args.get("country")
|
|
)
|
|
country_options = agenda.conference_list.conference_country_options(
|
|
[conf for conf in items if conf["latest_date"] >= today]
|
|
)
|
|
items = agenda.conference_list.filter_conferences_by_country(items, country_filter)
|
|
|
|
current = [
|
|
conf
|
|
for conf in items
|
|
if conf["has_exact_dates"]
|
|
and conf["start_date"] <= today
|
|
and conf["end_date"] >= today
|
|
]
|
|
future = [
|
|
conf for conf in items if conf not in current and conf["latest_date"] >= today
|
|
]
|
|
|
|
timeline = agenda.conference_list.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,
|
|
selected_country=country_filter,
|
|
country_options=country_options,
|
|
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()
|
|
items = agenda.conference_list.build_conference_list(
|
|
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
|
)
|
|
country_filter = agenda.conference_list.normalize_country_filter(
|
|
flask.request.args.get("country")
|
|
)
|
|
country_options = agenda.conference_list.conference_country_options(
|
|
[conf for conf in items if conf["latest_date"] < today]
|
|
)
|
|
items = agenda.conference_list.filter_conferences_by_country(items, country_filter)
|
|
return flask.render_template(
|
|
"conference_list.html",
|
|
past=[conf for conf in items if conf["latest_date"] < today],
|
|
today=today,
|
|
get_country=agenda.get_country,
|
|
selected_country=country_filter,
|
|
country_options=country_options,
|
|
fx_rate=agenda.fx.get_rates(app.config),
|
|
)
|
|
|
|
|
|
@app.route("/conference/series")
|
|
def conference_series_list() -> str:
|
|
"""Page showing conference series."""
|
|
return flask.render_template(
|
|
"conference_series_list.html",
|
|
series_list=agenda.conference_list.build_conference_series_list(
|
|
agenda.conference.load_series(app.config["PERSONAL_DATA"]),
|
|
agenda.conference_list.build_conference_list(
|
|
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
|
),
|
|
date.today(),
|
|
),
|
|
get_country=agenda.get_country,
|
|
)
|
|
|
|
|
|
@app.route("/conference/series/<series_id>")
|
|
def conference_series_page(series_id: str) -> str:
|
|
"""Page showing one conference series."""
|
|
series_lookup = agenda.conference.load_series(app.config["PERSONAL_DATA"])
|
|
series = series_lookup.get(series_id)
|
|
if series is None:
|
|
flask.abort(404)
|
|
|
|
conferences = [
|
|
conf
|
|
for conf in agenda.conference_list.build_conference_list(
|
|
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
|
)
|
|
if conf.get("series") == series_id
|
|
]
|
|
return flask.render_template(
|
|
"conference_series.html",
|
|
series_id=series_id,
|
|
series=series,
|
|
conferences=conferences,
|
|
today=date.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 = agenda.conference_list.build_conference_list(
|
|
app.config["PERSONAL_DATA"], agenda.trip.build_trip_list()
|
|
)
|
|
ical_data = agenda.conference_ical.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 sum_co2_by_transport_type(trips: list[Trip]) -> list[tuple[str, float]]:
|
|
"""Sum CO₂ by transport type."""
|
|
co2_by_transport_type: defaultdict[str, float] = defaultdict(float)
|
|
for trip in trips:
|
|
for transport_type, co2_kg in trip.co2_by_transport_type():
|
|
co2_by_transport_type[transport_type] += co2_kg
|
|
|
|
return list(co2_by_transport_type.items())
|
|
|
|
|
|
def get_home_weather() -> list[StrDict]:
|
|
"""Get Bristol home weather forecast, with date objects added for templates."""
|
|
|
|
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.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),
|
|
co2_by_transport_type=sum_co2_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),
|
|
co2_by_transport_type=sum_co2_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)
|
|
|
|
|
|
@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"]
|
|
)
|
|
agenda.trip.add_coordinates_for_eurostar_home_feeder(
|
|
trip, 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,
|
|
)
|
|
agenda.trip_timezones.localize_trip_car_journey_display_times(
|
|
trip, app.config["PERSONAL_DATA"]
|
|
)
|
|
|
|
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=agenda.trip_timezones.get_destination_timezones(trip),
|
|
human_readable_delta=agenda.utils.human_readable_delta,
|
|
trip_weather=trip_weather,
|
|
conference_free_days=agenda.trip.conference_free_days(trip),
|
|
)
|
|
|
|
|
|
@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),
|
|
total_co2_kg=calc_total_co2_kg(trip_list),
|
|
distances_by_transport_type=sum_distances_by_transport_type(trip_list),
|
|
co2_by_transport_type=sum_co2_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 typing.cast(
|
|
werkzeug.Response, 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")
|