diff --git a/agenda/data.py b/agenda/data.py index 8ae6305..b7f9cb8 100644 --- a/agenda/data.py +++ b/agenda/data.py @@ -16,7 +16,6 @@ from . import ( accommodation, birthday, busy, - calendar, carnival, conference, domains, diff --git a/agenda/waste_schedule.py b/agenda/waste_schedule.py index 6f0349b..947eb6c 100644 --- a/agenda/waste_schedule.py +++ b/agenda/waste_schedule.py @@ -44,7 +44,6 @@ async def get_html(data_dir: str, postcode: str, uprn: str) -> str: filename = f"{waste_dir}/{now_str}.html" forms_base_url = "https://forms.n-somerset.gov.uk" - # url2 = "https://forms.n-somerset.gov.uk/Waste/CollectionSchedule/ViewSchedule" url = "https://forms.n-somerset.gov.uk/Waste/CollectionSchedule" async with httpx.AsyncClient() as client: r = await client.post( @@ -69,11 +68,10 @@ async def get_html(data_dir: str, postcode: str, uprn: str) -> str: def parse_waste_schedule_date(day_and_month: str) -> date: """Parse waste schedule date.""" today = date.today() - this_year = today.year - date_format = "%A %d %B %Y" - d = datetime.strptime(f"{day_and_month} {this_year}", date_format).date() + fmt = "%A %d %B %Y" + d = datetime.strptime(f"{day_and_month} {today.year}", fmt).date() if d < today: - d = datetime.strptime(f"{day_and_month} {this_year + 1}", date_format).date() + d = datetime.strptime(f"{day_and_month} {today.year + 1}", fmt).date() return d @@ -188,25 +186,38 @@ async def get_bristol_gov_uk_data(uprn: str) -> httpx.Response: return response +def bristol_service(item: dict[str, typing.Any]) -> str: + """Bristol waste service name.""" + service: str = item["containerName"] + return "Recycling" if "Recycling" in service else service.partition(" ")[2] + + +def bristol_collections(item: dict[str, typing.Any]) -> typing.Iterable[date]: + """Bristol dates from collections.""" + for collection in item["collection"]: + for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: + yield date.fromisoformat(collection[collection_date_key][:10]) + + +def bristol_by_date(data: BristolSchedule, start_date: date) -> dict[date, list[str]]: + """Build list of Bristol services by date.""" + by_date: defaultdict[date, list[str]] = defaultdict(list) + + for item in data: + service = bristol_service(item) + for d in bristol_collections(item): + if d < start_date and service not in by_date[d]: + by_date[d].append(service) + return by_date + + async def get_bristol_gov_uk( start_date: date, data_dir: str, uprn: str, refresh: bool = False ) -> list[Event]: """Get waste collection schedule from Bristol City Council.""" data = await get_bristol_data(data_dir, uprn, refresh) - by_date: defaultdict[date, list[str]] = defaultdict(list) - - for item in data: - service = item["containerName"] - service = "Recycling" if "Recycling" in service else service.partition(" ")[2] - for collection in item["collection"]: - for collection_date_key in ["nextCollectionDate", "lastCollectionDate"]: - d = date.fromisoformat(collection[collection_date_key][:10]) - if d < start_date: - continue - if service not in by_date[d]: - by_date[d].append(service) - + by_date = bristol_by_date(data, start_date) return [ Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services)) for d, services in by_date.items()