diff --git a/agenda/data.py b/agenda/data.py index 7054a72..6b19e73 100644 --- a/agenda/data.py +++ b/agenda/data.py @@ -252,6 +252,7 @@ async def get_data(now: datetime, config: flask.config.Config) -> AgendaData: holiday_list = holidays.get_all(last_year, next_year, data_dir) events += holidays.combine_holidays(holiday_list) + events += holidays.get_school_holidays(last_year, next_year, data_dir) if flask.g.user.is_authenticated: events += birthday.get_birthdays( last_year, os.path.join(my_data, "entities.yaml") diff --git a/agenda/holidays.py b/agenda/holidays.py index f939216..175e798 100644 --- a/agenda/holidays.py +++ b/agenda/holidays.py @@ -7,6 +7,7 @@ import flask import agenda.uk_holiday import holidays +from agenda.uk_school_holiday import school_holiday_list from .event import Event from .types import Holiday, Trip @@ -29,6 +30,23 @@ def get_trip_holidays(trip: Trip) -> list[Holiday]: ) +def get_school_holidays(start_date: date, end_date: date, data_dir: str) -> list[Event]: + """Get UK school holidays from cache.""" + return school_holiday_list(start_date, end_date, data_dir) + + +def get_trip_school_holidays(trip: Trip) -> list[Event]: + """Get UK school holidays happening during trip.""" + if not trip.end: + return [] + + return get_school_holidays( + trip.start, + trip.end, + flask.current_app.config["DATA_DIR"], + ) + + def us_holidays(start_date: date, end_date: date) -> list[Holiday]: """Get US holidays.""" found: list[Holiday] = [] diff --git a/agenda/uk_school_holiday.py b/agenda/uk_school_holiday.py new file mode 100644 index 0000000..1215da9 --- /dev/null +++ b/agenda/uk_school_holiday.py @@ -0,0 +1,249 @@ +"""UK school holidays (Bristol) via iCalendar.""" + +from __future__ import annotations + +import datetime +import json +import os + +import httpx + +from .event import Event + +school_holiday_page_url = ( + "https://www.bristol.gov.uk/residents/schools-learning-and-early-years/" + "school-term-and-holiday-dates" +) +school_holiday_ics_url = ( + "https://www.bristol.gov.uk/files/documents/" + "4641-bristol-school-term-and-holiday-dates-2021-2022-and-2022-2023-and-2023-" + "2024-calendar" +) + + +def ics_filename(data_dir: str) -> str: + """Filename for cached school-holiday ICS.""" + assert os.path.exists(data_dir) + return os.path.join(data_dir, "bristol-school-holidays.ics") + + +def json_filename(data_dir: str) -> str: + """Filename for cached parsed school-holiday data.""" + assert os.path.exists(data_dir) + return os.path.join(data_dir, "bristol-school-holidays.json") + + +def _unescape_ics_text(value: str) -> str: + """Decode escaped ICS text values.""" + return ( + value.replace("\\n", " ") + .replace("\\N", " ") + .replace("\\,", ",") + .replace("\\;", ";") + .replace("\\\\", "\\") + ).strip() + + +def unfold_ics_lines(ics_text: str) -> list[str]: + """Unfold folded ICS lines (RFC5545).""" + unfolded: list[str] = [] + for raw_line in ics_text.splitlines(): + line = raw_line.rstrip("\r\n") + if not line: + continue + if unfolded and line[:1] in {" ", "\t"}: + unfolded[-1] += line[1:] + else: + unfolded.append(line) + return unfolded + + +def _parse_ics_date(value: str) -> datetime.date: + """Parse date/date-time values in ICS.""" + value = value.strip() + if "T" in value: + date_part = value.split("T", 1)[0] + return datetime.datetime.strptime(date_part, "%Y%m%d").date() + return datetime.datetime.strptime(value, "%Y%m%d").date() + + +def _is_school_holiday_summary(summary: str) -> bool: + """Return True if summary looks like a school holiday event.""" + lower = summary.lower() + if "holiday" not in lower: + return False + if "bank holiday" in lower: + return False + return True + + +def _clean_summary(summary: str) -> str: + """Normalise holiday summary text for display.""" + summary = _unescape_ics_text(summary) + # The feed embeds long policy notes in parentheses after the name. + if " (" in summary: + summary = summary.split(" (", 1)[0] + return summary.strip() + + +def parse_school_holidays_from_ics(ics_text: str) -> list[Event]: + """Parse school holiday ranges from an ICS file as Events.""" + events: list[Event] = [] + current: dict[str, str] = {} + + def flush_current() -> None: + summary = current.get("SUMMARY") + dtstart = current.get("DTSTART") + dtend = current.get("DTEND") + if not summary or not dtstart or not dtend: + return + + clean_summary = _clean_summary(summary) + if not _is_school_holiday_summary(clean_summary): + return + + start_date = _parse_ics_date(dtstart) + end_exclusive = _parse_ics_date(dtend) + end_date = end_exclusive - datetime.timedelta(days=1) + if end_date < start_date: + return + + events.append( + Event( + name="uk_school_holiday", + date=start_date, + end_date=end_date, + title=clean_summary, + url=school_holiday_page_url, + ) + ) + + for line in unfold_ics_lines(ics_text): + if line == "BEGIN:VEVENT": + current = {} + continue + if line == "END:VEVENT": + flush_current() + current = {} + continue + + if ":" not in line: + continue + + key_part, value = line.split(":", 1) + key = key_part.split(";", 1)[0].upper() + + if key in {"SUMMARY", "DTSTART", "DTEND"}: + current[key] = value.strip() + + # De-duplicate by title/date-range. + unique: dict[tuple[str, datetime.date, datetime.date], Event] = {} + for event in events: + end_date = event.end_as_date + unique[(event.title or event.name, event.as_date, end_date)] = event + + return sorted(unique.values(), key=lambda item: (item.as_date, item.end_as_date)) + + +def write_school_holidays_json(events: list[Event], data_dir: str) -> None: + """Write parsed school-holiday events to JSON cache.""" + filename = json_filename(data_dir) + payload: list[dict[str, str]] = [ + { + "name": event.name, + "title": event.title or event.name, + "start": event.as_date.isoformat(), + "end": event.end_as_date.isoformat(), + "url": event.url or "", + } + for event in events + ] + with open(filename, "w", encoding="utf-8") as out: + json.dump(payload, out, indent=2) + + +def read_school_holidays_json(data_dir: str) -> list[Event]: + """Read parsed school-holiday events from JSON cache.""" + filename = json_filename(data_dir) + if not os.path.exists(filename): + return [] + + with open(filename, encoding="utf-8") as in_file: + loaded = json.load(in_file) + if not isinstance(loaded, list): + return [] + + parsed_events: list[Event] = [] + for raw_item in loaded: + if not isinstance(raw_item, dict): + continue + title = raw_item.get("title") + start_value = raw_item.get("start") + end_value = raw_item.get("end") + if not ( + isinstance(title, str) + and isinstance(start_value, str) + and isinstance(end_value, str) + ): + continue + + try: + start_date = datetime.date.fromisoformat(start_value) + end_date = datetime.date.fromisoformat(end_value) + except ValueError: + continue + + event_url = raw_item.get("url") + parsed_events.append( + Event( + name="uk_school_holiday", + date=start_date, + end_date=end_date, + title=title, + url=event_url if isinstance(event_url, str) and event_url else None, + ) + ) + + return sorted(parsed_events, key=lambda item: (item.as_date, item.end_as_date)) + + +def school_holiday_list( + start_date: datetime.date, + end_date: datetime.date, + data_dir: str, +) -> list[Event]: + """Get cached school-holiday events overlapping the supplied range.""" + items = read_school_holidays_json(data_dir) + return [ + item + for item in items + if item.as_date <= end_date and item.end_as_date >= start_date + ] + + +async def get_holiday_list(data_dir: str) -> list[Event]: + """Download, parse and cache school-holiday data.""" + headers = { + "User-Agent": ( + "Mozilla/5.0 (X11; Linux x86_64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0 Safari/537.36" + ), + "Accept": "text/calendar,*/*;q=0.9", + "Referer": school_holiday_page_url, + } + + async with httpx.AsyncClient(timeout=60.0, follow_redirects=True) as client: + response = await client.get(school_holiday_ics_url, headers=headers) + response.raise_for_status() + + content_type = response.headers.get("content-type", "") + ics_text = response.text + if "text/calendar" not in content_type and "BEGIN:VCALENDAR" not in ics_text: + raise ValueError("School holiday ICS download did not return calendar content") + + with open(ics_filename(data_dir), "w", encoding="utf-8") as out: + out.write(ics_text) + + events = parse_school_holidays_from_ics(ics_text) + write_school_holidays_json(events, data_dir) + return events diff --git a/templates/event_list.html b/templates/event_list.html index 0869ad0..04191e2 100644 --- a/templates/event_list.html +++ b/templates/event_list.html @@ -27,6 +27,7 @@ "waste_schedule": "Waste schedule", "gwr_advance_tickets": "GWR advance tickets", "critical_mass": "Critical Mass", + "uk_school_holiday": "UK school holiday", } %} @@ -34,6 +35,7 @@ "bank_holiday": "bg-success-subtle", "conference": "bg-primary-subtle", "us_holiday": "bg-secondary-subtle", + "uk_school_holiday": "bg-warning-subtle", "birthday": "bg-info-subtle", "waste_schedule": "bg-danger-subtle", } %} @@ -132,7 +134,7 @@ end: {{event.end_date.strftime("%H:%M") }} (duration: {{duration}}) {% elif event.end_date != event.date %} - {{event.end_date}} + to {{ event.end_as_date.strftime("%a, %d, %b") }} {% endif %} {% endif %} diff --git a/templates/holiday_list.html b/templates/holiday_list.html index f82a54f..d12a7d5 100644 --- a/templates/holiday_list.html +++ b/templates/holiday_list.html @@ -4,7 +4,7 @@ {% block content %}
-

Holidays

+

Public holidays

{% for item in items %} {% set country = get_country(item.country) %} @@ -20,5 +20,17 @@ {% endfor %}
+ +

UK school holidays (Bristol)

+ + {% for item in school_holidays %} + + + + + + + {% endfor %} +
{{ display_date(item.as_date) }}to {{ display_date(item.end_as_date) }}in {{ (item.as_date - today).days }} days{{ item.title }}
{% endblock %} diff --git a/templates/launches.html b/templates/launches.html index 09be1f7..f2576dc 100644 --- a/templates/launches.html +++ b/templates/launches.html @@ -68,7 +68,10 @@
launch status: {{ launch.status.abbrev }} - {% if launch.probability %}{{ launch.probability }}%{% endif %} + {% if launch.is_active_crewed %} + In space + {% endif %} + {% if launch.is_future and launch.probability %}{{ launch.probability }}%{% endif %}
@@ -121,7 +124,7 @@ {% else %}

No description.

{% endif %} - {% if launch.weather_concerns %} + {% if launch.weather_concerns and launch.status.name != "Launch Successful" %}

Weather concerns

{% for line in launch.weather_concerns.splitlines() %}

{{ line }}

diff --git a/templates/macros.html b/templates/macros.html index c774901..dec35b4 100644 --- a/templates/macros.html +++ b/templates/macros.html @@ -403,6 +403,15 @@ https://www.flightradar24.com/data/flights/{{ flight.airline_detail.iata | lower

{{ trip_link(trip) }} ({{ display_date(trip.start) }})

+ {% set school_holidays = trip_school_holiday_map.get(trip.start.isoformat(), []) if trip_school_holiday_map is defined else [] %} + {% if school_holidays %} +
+ UK school holiday + {% for item in school_holidays %} + {{ item.title }} ({{ display_date_no_year(item.as_date) }} to {{ display_date_no_year(item.end_as_date) }}) + {% endfor %} +
+ {% endif %}
+
+

UK school holidays (Bristol)

+ {% if school_holidays %} + + {% for item in school_holidays %} + + + + + + {% endfor %} +
{{ display_date(item.as_date) }}to {{ display_date(item.end_as_date) }}{{ item.title }}
+ {% else %} +

No UK school holidays during trip.

+ {% endif %} +
+ {{ next_and_previous() }}
diff --git a/tests/test_school_holiday.py b/tests/test_school_holiday.py new file mode 100644 index 0000000..278d9cc --- /dev/null +++ b/tests/test_school_holiday.py @@ -0,0 +1,50 @@ +"""Tests for UK school holiday parsing.""" + +from __future__ import annotations + +import datetime +from pathlib import Path + +from agenda import uk_school_holiday + + +def test_parse_school_holidays_from_ics() -> None: + """Parse holiday ranges and skip bank-holiday events.""" + ics_text = """BEGIN:VCALENDAR +BEGIN:VEVENT +DTSTART;VALUE=DATE:20260525 +DTEND;VALUE=DATE:20260601 +SUMMARY;LANGUAGE=en-gb:Half Term Holiday (Bristol notes) +END:VEVENT +BEGIN:VEVENT +DTSTART;VALUE=DATE:20260831 +DTEND;VALUE=DATE:20260901 +SUMMARY;LANGUAGE=en-gb:Summer Bank Holiday +END:VEVENT +END:VCALENDAR +""" + + parsed = uk_school_holiday.parse_school_holidays_from_ics(ics_text) + assert len(parsed) == 1 + assert parsed[0].name == "uk_school_holiday" + assert parsed[0].title == "Half Term Holiday" + assert parsed[0].as_date == datetime.date(2026, 5, 25) + assert parsed[0].end_as_date == datetime.date(2026, 5, 31) + + +def test_school_holiday_list_filters_overlap(tmp_path: Path) -> None: + """Filter cached school holiday ranges by overlap.""" + filename = uk_school_holiday.json_filename(str(tmp_path)) + with open(filename, "w", encoding="utf-8") as out: + out.write( + '[{"name":"uk_school_holiday","title":"Easter Holiday",' + '"start":"2026-04-01","end":"2026-04-12","url":""}]' + ) + + events = uk_school_holiday.school_holiday_list( + datetime.date(2026, 4, 10), + datetime.date(2026, 4, 20), + str(tmp_path), + ) + assert len(events) == 1 + assert events[0].title == "Easter Holiday" diff --git a/update.py b/update.py index 3d2576c..5e5960a 100755 --- a/update.py +++ b/update.py @@ -21,6 +21,7 @@ import agenda.mail import agenda.thespacedevs import agenda.types import agenda.uk_holiday +import agenda.uk_school_holiday from agenda.event import Event from agenda.types import StrDict from web_view import app @@ -37,6 +38,17 @@ async def update_bank_holidays(config: flask.config.Config) -> None: print(f"took {time_taken:.1f} seconds") +async def update_school_holidays(config: flask.config.Config) -> None: + """Update cached copy of Bristol school holidays.""" + t0 = time() + events = await agenda.uk_school_holiday.get_holiday_list(config["DATA_DIR"]) + time_taken = time() - t0 + if not sys.stdin.isatty(): + return + print(len(events), "school holidays in list") + print(f"took {time_taken:.1f} seconds") + + async def update_bristol_bins(config: flask.config.Config) -> None: """Update waste schedule from Bristol City Council.""" t0 = time() @@ -247,9 +259,52 @@ def is_test_flight(launch: StrDict) -> bool: def get_launch_by_slug(data: StrDict, slug: str) -> StrDict | None: """Find last update for space launch.""" - return {item["slug"]: typing.cast(StrDict, item) for item in data["results"]}.get( - slug + results = data.get("results") + if not isinstance(results, list): + return None + + by_slug: dict[str, StrDict] = {} + for item in results: + if not isinstance(item, dict): + continue + item_slug = item.get("slug") + if isinstance(item_slug, str): + by_slug[item_slug] = typing.cast(StrDict, item) + + return by_slug.get(slug) + + +def send_thespacedevs_payload_alert( + config: flask.config.Config, + reason: str, + data: StrDict | None, +) -> None: + """Alert admin when SpaceDevs update payload is missing expected fields.""" + payload = data or {} + detail = payload.get("detail") + status_code = payload.get("status_code", payload.get("status")) + + detail_text = detail if isinstance(detail, str) else "" + is_rate_limited = ( + status_code == 429 + or "rate" in detail_text.lower() + or "thrott" in detail_text.lower() ) + alert_type = "rate-limit" if is_rate_limited else "error" + + subject = f"⚠️ SpaceDevs {alert_type}: {reason}" + body = f"""SpaceDevs update returned an unexpected payload. + +Reason: {reason} +Type: {alert_type} +Status: {status_code!r} +Detail: {detail!r} +Payload keys: {sorted(payload.keys())} + +Expected payload shape includes a top-level 'results' list. +Updater: /home/edward/src/agenda/update.py +""" + agenda.mail.send_mail(config, subject, body) def update_thespacedevs(config: flask.config.Config) -> None: @@ -283,12 +338,26 @@ def update_thespacedevs(config: flask.config.Config) -> None: t0 = time() data = agenda.thespacedevs.next_launch_api_data(rocket_dir) if not data: + send_thespacedevs_payload_alert( + config, + reason="API request failed or returned invalid JSON", + data=None, + ) return # thespacedevs API call failed + data_results = data.get("results") + if not isinstance(data_results, list): + send_thespacedevs_payload_alert( + config, + reason="response missing top-level results list", + data=data, + ) + return + # Identify test-flight slugs present in the current data cur_test_slugs: set[str] = { typing.cast(str, item["slug"]) - for item in data.get("results", []) + for item in data_results if is_test_flight(typing.cast(StrDict, item)) } @@ -317,7 +386,7 @@ def update_thespacedevs(config: flask.config.Config) -> None: time_taken = time() - t0 if not sys.stdin.isatty(): return - rockets = [agenda.thespacedevs.summarize_launch(item) for item in data["results"]] + rockets = [agenda.thespacedevs.summarize_launch(item) for item in data_results] print(len(rockets), "launches") print(len(active_crewed or []), "active crewed missions") print(f"took {time_taken:.1f} seconds") @@ -419,6 +488,7 @@ def main() -> None: with app.app_context(): if hour % 3 == 0: asyncio.run(update_bank_holidays(app.config)) + asyncio.run(update_school_holidays(app.config)) asyncio.run(update_bristol_bins(app.config)) update_gwr_advance_ticket_date(app.config) # TODO: debug why update gandi fails diff --git a/web_view.py b/web_view.py index ad04588..e29d690 100755 --- a/web_view.py +++ b/web_view.py @@ -32,6 +32,7 @@ import agenda.trip import agenda.trip_schengen import agenda.utils 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__) @@ -177,11 +178,22 @@ async def recent() -> str: @app.route("/launches") def launch_list() -> str: """Web page showing List of space launches.""" - now = datetime.now() + 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") @@ -577,6 +589,33 @@ def get_trip_list() -> list[Trip]: 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.""" @@ -588,6 +627,7 @@ def trip_past_list() -> str: "trip/list.html", heading="Past trips", trips=reversed(past), + trip_school_holiday_map=trip_school_holiday_map(past), coordinates=coordinates, routes=routes, today=today, @@ -632,10 +672,13 @@ def trip_future_list() -> str: coordinates, routes = agenda.trip.get_coordinates_and_routes(current + future) + shown = current + future + return flask.render_template( "trip/list.html", heading="Future trips", - trips=current + future, + trips=shown, + trip_school_holiday_map=trip_school_holiday_map(shown), coordinates=coordinates, routes=routes, today=today, @@ -715,6 +758,7 @@ def trip_page(start: str) -> str: 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), human_readable_delta=agenda.utils.human_readable_delta, ) @@ -800,11 +844,19 @@ def holiday_list() -> str: data_dir = app.config["DATA_DIR"] 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( + today - timedelta(days=2), 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, get_country=agenda.get_country, today=today + "holiday_list.html", + items=items, + school_holidays=school_holidays, + get_country=agenda.get_country, + today=today, )