1168 lines
38 KiB
Python
Executable file
1168 lines
38 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()
|
|
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()
|
|
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
|
|
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."
|
|
)
|
|
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_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]
|
|
|
|
return flask.render_template(
|
|
"conference_list.html",
|
|
current=current,
|
|
future=future,
|
|
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"],
|
|
)
|
|
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,
|
|
)
|
|
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 _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)
|
|
|
|
home_now = datetime.now(ZoneInfo("Europe/London"))
|
|
destination_times: list[StrDict] = []
|
|
|
|
for location, country in trip.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"
|
|
current_time = None
|
|
if timezone_name:
|
|
dest_now = datetime.now(ZoneInfo(timezone_name))
|
|
dest_offset = dest_now.utcoffset()
|
|
home_offset = home_now.utcoffset()
|
|
if dest_offset is not None and home_offset is not None:
|
|
offset_minutes = int((dest_offset - home_offset).total_seconds() // 60)
|
|
offset_display = _format_offset_from_uk(offset_minutes)
|
|
current_time = dest_now.strftime("%a %H:%M:%S")
|
|
|
|
destination_times.append(
|
|
{
|
|
"location": location,
|
|
"country_name": country.name,
|
|
"country_flag": country.flag,
|
|
"timezone": timezone_name,
|
|
"offset_display": offset_display,
|
|
"current_time": current_time,
|
|
}
|
|
)
|
|
|
|
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)
|
|
|
|
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,
|
|
)
|
|
|
|
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")
|