Extract conference iCal generation into agenda module
This commit is contained in:
parent
2d224783b2
commit
df7fd2fb9e
2 changed files with 104 additions and 92 deletions
90
agenda/conference_ical.py
Normal file
90
agenda/conference_ical.py
Normal file
|
|
@ -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")
|
||||
106
web_view.py
106
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")
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue