From df7fd2fb9e3845eb7638ba95d985ac88c51b8a4d Mon Sep 17 00:00:00 2001 From: Edward Betts Date: Wed, 9 Sep 2026 11:58:50 +0100 Subject: [PATCH] Extract conference iCal generation into agenda module --- agenda/conference_ical.py | 90 ++++++++++++++++++++++++++++++++ web_view.py | 106 +++++--------------------------------- 2 files changed, 104 insertions(+), 92 deletions(-) create mode 100644 agenda/conference_ical.py diff --git a/agenda/conference_ical.py b/agenda/conference_ical.py new file mode 100644 index 0000000..4e4624a --- /dev/null +++ b/agenda/conference_ical.py @@ -0,0 +1,90 @@ +"""Generate iCalendar feeds for conferences.""" + +import hashlib +from datetime import datetime, timedelta, timezone + +import agenda +import agenda.utils +from agenda import ical +from agenda.types import StrDict + + +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: + if not conf["has_exact_dates"]: + continue + 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)) + ical.append_property(lines, "SUMMARY", ical.escape_text(conf["name"])) + 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") diff --git a/web_view.py b/web_view.py index 7a0a37c..a0f3968 100755 --- a/web_view.py +++ b/web_view.py @@ -4,7 +4,6 @@ import decimal import functools -import hashlib import importlib import inspect import json @@ -20,6 +19,7 @@ from zoneinfo import ZoneInfo import flask import pytz +from pycountry.db import Country import werkzeug import werkzeug.debug.tbtools import yaml @@ -27,6 +27,7 @@ from authlib.integrations.flask_client import OAuth from werkzeug.middleware.proxy_fix import ProxyFix import agenda.conference +import agenda.conference_ical import agenda.data import agenda.error_mail import agenda.fx @@ -39,20 +40,20 @@ import agenda.trip_schengen import agenda.uk_school_holiday import agenda.utils import agenda.weather -from agenda import calendar, format_list_with_ampersand, ical, travel, uk_tz +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) +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) +oauth = OAuth(app) # type: ignore[no-untyped-call] authentik_url = app.config["AUTHENTIK_URL"] -oauth.register( +oauth.register( # type: ignore[no-untyped-call] name="authentik", client_id=app.config["AUTHENTIK_CLIENT_ID"], client_secret=app.config["AUTHENTIK_CLIENT_SECRET"], @@ -488,51 +489,6 @@ def build_conference_series_list() -> list[StrDict]: return series_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[str, typing.Any] | None: @@ -609,42 +565,6 @@ def build_conference_timeline( } -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: - if not conf["has_exact_dates"]: - continue - 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)) - ical.append_property(lines, "SUMMARY", ical.escape_text(conf["name"])) - 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.""" @@ -739,7 +659,7 @@ def conference_series_page(series_id: str) -> str: def conference_ical() -> werkzeug.Response: """Return all conferences as an iCalendar feed.""" items = build_conference_list() - ical_data = build_conference_ical(items) + 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 @@ -1312,9 +1232,9 @@ def get_destination_timezones(trip: Trip) -> list[StrDict]: 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]] + group_key = (item["country_name"], item["country_flag"], item["timezone"]) + if group_key in grouped_index: + existing = grouped[grouped_index[group_key]] existing_locations = typing.cast(list[str], existing["locations"]) existing_locations.append(typing.cast(str, item["location"])) existing["location_count"] = ( @@ -1322,7 +1242,7 @@ def get_destination_timezones(trip: Trip) -> list[StrDict]: ) continue - grouped_index[key] = len(grouped) + grouped_index[group_key] = len(grouped) grouped.append( { **item, @@ -1551,7 +1471,9 @@ def login() -> werkzeug.Response: 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) + return typing.cast( + werkzeug.Response, oauth.authentik.authorize_redirect(redirect_uri) + ) @app.route("/callback")