diff --git a/etl/pipelines/google_places/__init__.py b/etl/pipelines/google_places/__init__.py new file mode 100644 index 0000000..ac4ff08 --- /dev/null +++ b/etl/pipelines/google_places/__init__.py @@ -0,0 +1 @@ +"""Google Places ETL pipelines.""" diff --git a/etl/pipelines/google_places/common.py b/etl/pipelines/google_places/common.py new file mode 100644 index 0000000..ecd434e --- /dev/null +++ b/etl/pipelines/google_places/common.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from dtos import Activity, Address, Availability, Contact, ItemCategory, Service + + +@dataclass(frozen=True) +class GooglePlacesQuery: + text_query: str + data_source: str + item_category: ItemCategory + activity: Activity + + +def extract_postcode(formatted_address: str | None) -> str | None: + if not formatted_address: + return None + parts = formatted_address.split() + for part in parts: + if len(part) == 5 and part.isdigit(): + return part + return None + + +def normalize_google_place( + raw: dict[str, Any], + *, + data_source: str, + data_source_id: str, + item_category: ItemCategory, + activity: Activity, +) -> dict[str, Any]: + display_name = raw.get("displayName") or {} + formatted_address = raw.get("formattedAddress") + location = raw.get("location") or {} + return { + "data_source_id": data_source_id, + "data_source": data_source, + "name": display_name.get("text") or data_source_id, + "lat": location.get("latitude", 0.0), + "lon": location.get("longitude", 0.0), + "address": Address( + street=formatted_address, + postcode=extract_postcode(formatted_address), + ), + "contact": Contact(), + "services": [Service(activity=activity, item_category=item_category)], + "availability": Availability(), + } diff --git a/etl/pipelines/google_places/normalizer.py b/etl/pipelines/google_places/normalizer.py new file mode 100644 index 0000000..f3c0f81 --- /dev/null +++ b/etl/pipelines/google_places/normalizer.py @@ -0,0 +1,34 @@ +from __future__ import annotations + +from dtos import Activity, ItemCategory, NormalizedLocation, RawLocation +from base.normalizer import BaseNormalizer + +from pipelines.google_places.common import normalize_google_place + + +class GooglePlacesRepairNormalizer(BaseNormalizer): + def normalize(self, raw_locations: list[RawLocation]) -> list[NormalizedLocation]: + return [ + NormalizedLocation(**normalize_google_place( + raw.payload, + data_source=raw.data_source, + data_source_id=raw.data_source_id, + item_category=ItemCategory.SHOES, + activity=Activity.REPAIR_PAID, + )) + for raw in raw_locations + ] + + +class GooglePlacesDonationNormalizer(BaseNormalizer): + def normalize(self, raw_locations: list[RawLocation]) -> list[NormalizedLocation]: + return [ + NormalizedLocation(**normalize_google_place( + raw.payload, + data_source=raw.data_source, + data_source_id=raw.data_source_id, + item_category=ItemCategory.CLOTHING, + activity=Activity.DONATION_DROP, + )) + for raw in raw_locations + ] diff --git a/etl/pipelines/google_places/querier.py b/etl/pipelines/google_places/querier.py new file mode 100644 index 0000000..ccc4525 --- /dev/null +++ b/etl/pipelines/google_places/querier.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from datetime import datetime, timezone +from urllib.error import HTTPError, URLError +from urllib.request import Request, urlopen +import json +import os +from typing import Any + +from base.querier import BaseQuerier +from dtos import RawLocation + + +GOOGLE_PLACES_ENDPOINT = "https://places.googleapis.com/v1/places:searchText" + + +class GooglePlacesQuerier(BaseQuerier): + def __init__(self, *, api_key: str | None = None, text_query: str, data_source: str): + self.api_key = api_key or os.environ.get("GOOGLE_API_KEY") + self.text_query = text_query + self.data_source = data_source + + def fetch(self) -> list[RawLocation]: + if not self.api_key: + raise RuntimeError("GOOGLE_API_KEY is required") + + payload = self._request({"textQuery": self.text_query}) + places = payload.get("places", []) + fetched_at = datetime.now(timezone.utc) + raw_locations: list[RawLocation] = [] + for index, place in enumerate(places): + raw_locations.append( + RawLocation( + data_source=self.data_source, + data_source_id=place.get("id") or f"{self.data_source}-{index}", + fetched_at=fetched_at, + payload=place, + ) + ) + return raw_locations + + def _request(self, body: dict[str, Any]) -> dict[str, Any]: + request = Request( + GOOGLE_PLACES_ENDPOINT, + data=json.dumps(body).encode("utf-8"), + headers={ + "Content-Type": "application/json", + "X-Goog-Api-Key": self.api_key, + "X-Goog-FieldMask": "places.id,places.displayName,places.formattedAddress,places.location,places.types", + }, + method="POST", + ) + try: + with urlopen(request, timeout=30) as response: + return json.loads(response.read().decode("utf-8")) + except (HTTPError, URLError) as exc: + raise RuntimeError(f"Google Places request failed: {exc}") from exc + + +class GooglePlacesRepairQuerier(GooglePlacesQuerier): + def __init__(self, api_key: str | None = None): + super().__init__(api_key=api_key, text_query="shoe repair Boston MA", data_source="google_places_repair") + + +class GooglePlacesDonationQuerier(GooglePlacesQuerier): + def __init__(self, api_key: str | None = None): + super().__init__(api_key=api_key, text_query="donation centers Boston MA", data_source="google_places_donations") diff --git a/etl/pipelines/google_places/test_pipeline.py b/etl/pipelines/google_places/test_pipeline.py new file mode 100644 index 0000000..a6d18e4 --- /dev/null +++ b/etl/pipelines/google_places/test_pipeline.py @@ -0,0 +1,54 @@ +from datetime import datetime, timezone +from pathlib import Path +import json + +from dtos import Activity, ItemCategory, RawLocation +from pipelines.google_places.normalizer import ( + GooglePlacesDonationNormalizer, + GooglePlacesRepairNormalizer, +) + + +SAMPLES = Path(__file__).resolve().parents[3] / "data-explorations" / "google-places" / "samples" + + +def _load_places(name: str): + return json.loads((SAMPLES / name).read_text())["places"] + + +def _raw_locations(data_source: str, prefix: str, filename: str): + fetched_at = datetime.now(timezone.utc) + return [ + RawLocation( + data_source=data_source, + data_source_id=f"{prefix}-{index}", + fetched_at=fetched_at, + payload=place, + ) + for index, place in enumerate(_load_places(filename)) + ] + + +def test_repair_normalizer_maps_google_places_sample(): + normalized = GooglePlacesRepairNormalizer().normalize( + _raw_locations("google_places_repair", "repair", "shoe-repair-00.json") + ) + + assert normalized + assert normalized[0].data_source == "google_places_repair" + assert normalized[0].services[0].activity == Activity.REPAIR_PAID + assert normalized[0].services[0].item_category == ItemCategory.SHOES + assert normalized[0].name == "David's Instant Shoe Repair" + assert normalized[0].address.street == "281 Franklin St, Boston, MA 02110, USA" + + +def test_donation_normalizer_maps_google_places_sample(): + normalized = GooglePlacesDonationNormalizer().normalize( + _raw_locations("google_places_donations", "donation", "donations-00.json") + ) + + assert normalized + assert normalized[0].data_source == "google_places_donations" + assert normalized[0].services[0].activity == Activity.DONATION_DROP + assert normalized[0].services[0].item_category == ItemCategory.CLOTHING + assert normalized[0].name == "Morgan Memorial Goodwill Industries"