Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion uk_bin_collection/tests/input.json
Original file line number Diff line number Diff line change
Expand Up @@ -1559,11 +1559,13 @@
},
"LondonBoroughLambeth": {
"LAD24CD": "E09000022",
"house_number": "5",
"postcode": "SW4 6BS",
"skip_get_url": true,
"uprn": "100021881738",
"url": "https://wasteservice.lambeth.gov.uk/WhitespaceComms/GetServicesByUprn",
"wiki_name": "Lambeth",
"wiki_note": "Pass the UPRN. You can find it using [FindMyAddress](https://www.findmyaddress.co.uk/search)."
"wiki_note": "Pass the postcode and house number, or a UPRN. The UPRN is used when both are given."
},
"LondonBoroughLewisham": {
"LAD24CD": "E09000023",
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,23 @@
from uk_bin_collection.uk_bin_collection.common import *
from uk_bin_collection.uk_bin_collection.get_bin_data import AbstractGetBinDataClass

ADDRESS_SEARCH_URL = "https://wasteservice.lambeth.gov.uk/LLPG/addressSearch"


def match_address(options, paon):
"""Pick the option whose label starts with (or contains) the house number/name.

options: list of (value, label). Prefers "<paon> <street>" over "<paon> <flat>",
then any whole-word occurrence. Raises ValueError if nothing matches.
"""
paon = str(paon).strip()
p = re.escape(paon)
for pattern in (rf"^{p}[\s,]+[A-Za-z]", rf"^{p}[\s,]", rf"(^|[\s,]){p}([\s,]|$)"):
for value, label in options:
if re.search(pattern, label.strip(), re.I):
return value
raise ValueError(f"No address matching '{paon}' found for this postcode")


# import the wonderful Beautiful Soup and the URL grabber
class CouncilClass(AbstractGetBinDataClass):
Expand All @@ -13,6 +30,24 @@ class CouncilClass(AbstractGetBinDataClass):
implementation.
"""

def find_uprn(self, postcode, paon):
"""Resolve a UPRN from postcode + house number via Lambeth's LLPG search."""
res = requests.post(ADDRESS_SEARCH_URL, json={"postcode": postcode}, timeout=30)
res.raise_for_status()
results = (res.json().get("payload") or {}).get("results") or []
if not results:
raise ValueError(f"No addresses found for postcode {postcode}")
# Whole properties first so "5" picks 5 Prescott Place, not a flat within it
results.sort(key=lambda r: bool(r.get("SAO_TEXT") or r.get("SAO_START_NUMBER")))
options = []
for r in results:
pao = f"{r.get('PAO_START_NUMBER', '')}{r.get('PAO_START_SUFFIX', '')}"
label = " ".join(
x for x in (pao, r.get("PAO_TEXT"), r.get("STREET"), r.get("SAO_TEXT")) if x
)
options.append((r["UPRN"], label))
return match_address(options, paon)

def parse_data(self, page: str, **kwargs) -> dict:

"""
Expand All @@ -22,8 +57,9 @@ def parse_data(self, page: str, **kwargs) -> dict:

Parameters:
page (str): HTML or page content provided to the parser (not used by this implementation).
uprn (str, optional, in kwargs): The UPRN to query; required in kwargs as "uprn".

uprn (str, optional, in kwargs): The UPRN to query.
postcode, paon (str, optional, in kwargs): used to look up the UPRN when none is given.

Returns:
dict: A dictionary with a single key "bins" mapping to a list of objects with:
- "type" (str): Bin type (e.g., "recycling", "refuse", or the container's DisplayPhrase).
Expand All @@ -33,6 +69,10 @@ def parse_data(self, page: str, **kwargs) -> dict:
ConnectionRefusedError: If the API response status code is not 200.
"""
user_uprn = kwargs.get("uprn")
user_postcode = kwargs.get("postcode")
user_paon = kwargs.get("paon") or kwargs.get("house_number")
if not user_uprn and user_postcode and user_paon:
user_uprn = self.find_uprn(user_postcode, user_paon)
check_uprn(user_uprn)
data = {"bins": []}

Expand All @@ -44,7 +84,7 @@ def parse_data(self, page: str, **kwargs) -> dict:
body = {"uprn": user_uprn, "includeEventTypes": False, "includeFlags": True}
json_data = json.dumps(body)

res = requests.post(url, headers=headers, data=json_data)
res = requests.post(url, headers=headers, data=json_data, timeout=30)

if res.status_code != 200:
raise ConnectionRefusedError("Cannot connect to API!")
Expand Down
Loading