|
|
|
@ -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()
|
|
|
|
|