Split out code for reading events from YAML

This commit is contained in:
Edward Betts 2024-05-16 15:20:58 +01:00
parent 18d8fa6b7c
commit 7bb6110f45
2 changed files with 87 additions and 76 deletions

View file

@ -10,10 +10,8 @@ from time import time
import dateutil.rrule
import dateutil.tz
import flask
import isodate # type: ignore
import lxml
import pytz
import yaml
from . import (
accommodation,
@ -23,6 +21,7 @@ from . import (
conference,
domains,
economist,
events_yaml,
gandi,
gwr,
hn,
@ -61,25 +60,6 @@ def timezone_transition(
]
def midnight(d: date) -> datetime:
"""Convert from date to midnight on that day."""
return datetime.combine(d, datetime.min.time())
def dates_from_rrule(
rrule: str, start: date, end: date
) -> typing.Sequence[datetime | date]:
"""Generate events from an RRULE between start_date and end_date."""
all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"])
return [
i.date() if all_day else uk_tz.localize(i)
for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between(
midnight(start), midnight(end)
)
]
async def waste_collection_events(
data_dir: str, postcode: str, uprn: str
) -> list[Event]:
@ -97,60 +77,6 @@ async def bristol_waste_collection_events(
return await waste_schedule.get_bristol_gov_uk(start_date, data_dir, uprn)
def get_yaml_event_date_field(item: dict[str, str]) -> str:
"""Event date field name."""
return (
"end_date"
if item["name"] == "travel_insurance"
else ("start_date" if "start_date" in item else "date")
)
def get_yaml_event_end_date_field(item: dict[str, str]) -> str:
"""Event date field name."""
return (
"end_date"
if item["name"] == "travel_insurance"
else ("start_date" if "start_date" in item else "date")
)
def read_events_yaml(
data_dir: str, start: date, end: date, skip_trips: bool = False
) -> list[Event]:
"""Read eventes from YAML file."""
events: list[Event] = []
for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
if "trip" in item and skip_trips:
continue
duration = (
isodate.parse_duration(item["duration"]) if "duration" in item else None
)
dates = (
dates_from_rrule(item["rrule"], start, end)
if "rrule" in item
else [item[get_yaml_event_date_field(item)]]
)
for dt in dates:
e = Event(
name=item["name"],
date=dt,
end_date=(
dt + duration
if duration
else (
item.get("end_date")
if item["name"] != "travel_insurance"
else None
)
),
title=item.get("title"),
url=item.get("url"),
)
events.append(e)
return events
def find_events_during_stay(
accommodation_events: list[Event], markets: list[Event]
) -> list[Event]:
@ -426,7 +352,7 @@ async def get_data(
for key in "backwell_bins", "bristol_bins":
if results[key]:
events += results[key]
events += read_events_yaml(my_data, last_year, next_year)
events += events_yaml.read(my_data, last_year, next_year)
events += subscription.get_events(os.path.join(my_data, "subscriptions.yaml"))
events += gandi.get_events(data_dir)
events += economist.publication_dates(last_week, next_year)

85
agenda/events_yaml.py Normal file
View file

@ -0,0 +1,85 @@
"""Read events from YAML."""
import os
import typing
from datetime import date, datetime
import dateutil.rrule
import isodate # type: ignore
import yaml
from . import uk_tz
from .types import Event
def midnight(d: date) -> datetime:
"""Convert from date to midnight on that day."""
return datetime.combine(d, datetime.min.time())
def dates_from_rrule(
rrule: str, start: date, end: date
) -> typing.Sequence[datetime | date]:
"""Generate events from an RRULE between start_date and end_date."""
all_day = not any(param in rrule for param in ["BYHOUR", "BYMINUTE", "BYSECOND"])
return [
i.date() if all_day else uk_tz.localize(i)
for i in dateutil.rrule.rrulestr(rrule, dtstart=midnight(start)).between(
midnight(start), midnight(end)
)
]
def get_yaml_event_date_field(item: dict[str, str]) -> str:
"""Event date field name."""
return (
"end_date"
if item["name"] == "travel_insurance"
else ("start_date" if "start_date" in item else "date")
)
def get_yaml_event_end_date_field(item: dict[str, str]) -> str:
"""Event date field name."""
return (
"end_date"
if item["name"] == "travel_insurance"
else ("start_date" if "start_date" in item else "date")
)
def read(
data_dir: str, start: date, end: date, skip_trips: bool = False
) -> list[Event]:
"""Read eventes from YAML file."""
events: list[Event] = []
for item in yaml.safe_load(open(os.path.join(data_dir, "events.yaml"))):
if "trip" in item and skip_trips:
continue
duration = (
isodate.parse_duration(item["duration"]) if "duration" in item else None
)
dates = (
dates_from_rrule(item["rrule"], start, end)
if "rrule" in item
else [item[get_yaml_event_date_field(item)]]
)
for dt in dates:
e = Event(
name=item["name"],
date=dt,
end_date=(
dt + duration
if duration
else (
item.get("end_date")
if item["name"] != "travel_insurance"
else None
)
),
title=item.get("title"),
url=item.get("url"),
)
events.append(e)
return events