agenda/data.py

@@ -16,6 +16,7 @@ from . import (
     accommodation,
     birthday,
     busy,
+    calendar,
     carnival,
     conference,
     domains,

agenda/waste_schedule.py

@@ -44,6 +44,7 @@ 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(
@@ -68,10 +69,11 @@ 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()
-    fmt = "%A %d %B %Y"
+    this_year = today.year
-    d = datetime.strptime(f"{day_and_month} {today.year}", fmt).date()
+    date_format = "%A %d %B %Y"
+    d = datetime.strptime(f"{day_and_month} {this_year}", date_format).date()
     if d < today:
-        d = datetime.strptime(f"{day_and_month} {today.year + 1}", fmt).date()
+        d = datetime.strptime(f"{day_and_month} {this_year + 1}", date_format).date()
     return d
 
 
@@ -186,38 +188,25 @@ 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 = bristol_by_date(data, start_date)
+    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)
+
     return [
         Event(name="waste_schedule", date=d, title="Bristol: " + ", ".join(services))
         for d, services in by_date.items()