close to child objects

Signed-off-by: John Seekins <[email protected]>

Author: johnseekins

enricher.py

@@ -5,7 +5,10 @@
     wikipedia,
     openstreetmap,
 )
-from schemas import facilities_schema
+from schemas import (
+    default_coords,
+    facilities_schema,
+)
 import time
 from utils import logger
 
@@ -37,15 +40,20 @@ def enrich_facility(facility_data: tuple) -> tuple:
     logger.info("Enriching facility %s...", facility_name)
     enriched_facility = copy.deepcopy(facility)
 
-    wiki = wikipedia.search(facility_name)
-    wd = wikidata.search(facility_name)
-    osm = openstreetmap.search(facility_name, facility.get("address", {}))
-    enriched_facility["wikipedia_page_url"] = wiki.get("url", "")
-    enriched_facility["wikipedia_search_query"] = wiki.get("search_query_steps", "")
-    enriched_facility["wikidata_page_url"] = wd.get("url", "")
-    enriched_facility["wikidata_search_query"] = wd.get("search_query_steps", "")
-    enriched_facility["osm_result_url"] = osm.get("url", "")
-    enriched_facility["osm_search_query"] = osm.get("search_query_steps", "")
+    wiki = wikipedia.Wikipedia({"facility_name": facility_name})
+    wiki_res = wiki.search()
+    wd = wikidata.Wikidata({"facility_name": facility_name})
+    wd_res = wd.search()
+    osm = openstreetmap.OpenStreetMap({"facility_name": facility_name, "address": facility.get("address", {})})
+    osm_res = osm.search()
+    enriched_facility["wikipedia"]["page_url"] = wiki_res.get("url", "")
+    enriched_facility["wikipedia"]["search_query"] = wiki_res.get("search_query_steps", "")
+    enriched_facility["wikidata"]["page_url"] = wd_res.get("url", "")
+    enriched_facility["wikidata"]["search_query"] = wd_res.get("search_query_steps", "")
+    enriched_facility["osm"]["latitude"] = osm_res.get("details", {}).get("latitude", default_coords["latitude"])
+    enriched_facility["osm"]["longitude"] = osm_res.get("details", {}).get("longitude", default_coords["longitude"])
+    enriched_facility["osm"]["url"] = osm_res.get("url", "")
+    enriched_facility["osm"]["search_query"] = osm_res.get("search_query_steps", "")
 
     logger.debug(enriched_facility)
     return facility_id, enriched_facility

enrichers/__init__.py

@@ -0,0 +1,98 @@
+import copy
+import requests
+from schemas import enrich_resp_schema
+import time
+from utils import (
+    session,
+)
+
+
+class Enrichment(object):
+    required_keys = [
+        "facility_name",
+    ]
+
+    def __init__(self, **kwargs):
+        self.resp_info = copy.deepcopy(enrich_resp_schema)
+        for k in self.required_keys:
+            if k not in kwargs.keys():
+                raise KeyError("Missing required key %s in %s", k, kwargs)
+        self.search_args = copy.deepcopy(kwargs)
+        self.wait_time = int(kwargs.get("wait_time", 1))
+
+    def search(self) -> dict:
+        """Child objects should implement this"""
+        return {}
+
+    def _req(self, url: str, params: dict = {}, timeout: int = 10) -> requests.Response:
+        """requests response wrapper to ensure we honor waits"""
+        response = session.get(url, allow_redirects=True, timeout=timeout)
+        response.raise_for_status()
+        time.sleep(self.wait_time)
+        return response
+
+    def _minimal_clean_facility_name(self, name: str) -> str:
+        """Minimal cleaning that preserves important context like 'County Jail'"""
+        cleaned = name
+
+        # Remove pipe separators and take the main name
+        if "|" in cleaned:
+            parts = cleaned.split("|")
+            cleaned = max(parts, key=len).strip()
+
+        # Only remove very generic suffixes, keep specific ones like "County Jail"
+        generic_suffixes = [
+            "Service Processing Center",
+            "ICE Processing Center",
+            "Immigration Processing Center",
+            "Contract Detention Facility",
+            "Adult Detention Facility",
+        ]
+
+        for suffix in generic_suffixes:
+            if cleaned.endswith(suffix):
+                cleaned = cleaned[: -len(suffix)].strip()
+                break
+
+        return cleaned
+
+    def _clean_facility_name(self, name: str) -> str:
+        """Clean facility name for better search results"""
+        # Remove common suffixes and prefixes that might interfere with search
+        # This function may not be helpful - may be counterproductive.
+        cleaned = name
+
+        # Remove pipe separators and take the main name
+        if "|" in cleaned:
+            parts = cleaned.split("|")
+            # Take the longer part as it's likely the full name
+            cleaned = max(parts, key=len).strip()
+
+        # Remove common facility type suffixes for broader search
+        suffixes_to_remove = [
+            "Detention Center",
+            "Processing Center",
+            "Correctional Center",
+            "Correctional Facility",
+            "Detention Facility",
+            "Service Processing Center",
+            "ICE Processing Center",
+            "Immigration Processing Center",
+            "Adult Detention Facility",
+            "Contract Detention Facility",
+            "Regional Detention Center",
+            "County Jail",
+            "County Detention Center",
+            "Sheriff's Office",
+            "Justice Center",
+            "Safety Center",
+            "Jail Services",
+            "Correctional Complex",
+            "Public Safety Complex",
+        ]
+
+        for suffix in suffixes_to_remove:
+            if cleaned.endswith(suffix):
+                cleaned = cleaned[: -len(suffix)].strip()
+                break
+        return cleaned

enrichers/openstreetmap.py

@@ -1,105 +1,100 @@
-import copy
-from enrichers.utils import (
-    clean_facility_name,
-    NOMINATIM_DELAY,
-)
-from schemas import enrich_resp_schema
-import time
-from utils import (
-    logger,
-    session,
-)
+from schemas import default_coords
+from enrichers import Enrichment
+from utils import logger
 
 
-def search(facility_name: str, address: dict) -> dict:
-    search_name = clean_facility_name(facility_name)
-    search_url = "https://nominatim.openstreetmap.org/search"
-    resp_info = copy.deepcopy(enrich_resp_schema)
-    resp_info["enrichment_type"] = "openstreetmap"
-    data = []
-    if not address:
-        logger.debug("No address for %s, simply searching for name", facility_name)
-        params = {
-            "q": search_name,
-            "format": "json",
-            "limit": 5,
-            "dedupe": 1,
-        }
-        logger.debug("Searching OSM for %s", search_name)
-        resp_info["search_query_steps"].append(search_name)  # type: ignore [attr-defined]
-        try:
-            response = session.get(search_url, params=params, timeout=15)  # type: ignore [arg-type]
-            response.raise_for_status()
-            data = response.json()
-            time.sleep(NOMINATIM_DELAY)
-        except Exception as e:
-            logger.debug(" OSM search error for '%s': %s", facility_name, e)
-            resp_info["search_query_steps"].append(f"(Failed -> {e})")  # type: ignore [attr-defined]
-            return resp_info
-    else:
-        full_address = (
-            f"{address['street']} {address['locality']}, {address['administrative_area']} {address['postal_code']}"
-        )
-        locality = f"{address['locality']}, {address['administrative_area']} {address['postal_code']}"
+class OpenStreetMap(Enrichment):
+    def search(self) -> dict:
+        facility_name = self.search_args["facility_name"]
+        address = self.search_args.get("address", {})
+        search_name = self._clean_facility_name(facility_name)
         search_url = "https://nominatim.openstreetmap.org/search"
-        search_params = {
-            "facility_name": {
-                "q": f"{search_name} {full_address}",
+        self.resp_info["enrichment_type"] = "openstreetmap"
+        data = []
+        if not address:
+            logger.debug("No address for %s, simply searching for name", facility_name)
+            params = {
+                "q": search_name,
                 "format": "json",
                 "limit": 5,
                 "dedupe": 1,
-            },
-            "street_address": {
-                "q": f"{full_address}",
-                "format": "json",
-                "limit": 5,
-                "dedupe": 1,
-            },
-            "locality": {
-                "q": f"{locality}",
-                "format": "json",
-                "limit": 5,
-                "dedupe": 1,
-            },
-        }
-        for search_name, params in search_params.items():
-            logger.debug("Searching OSM for %s", params["q"])
-            resp_info["search_query_steps"].append(params["q"])  # type: ignore [attr-defined]
+            }
+            logger.debug("Searching OSM for %s", search_name)
+            self.resp_info["search_query_steps"].append(search_name)  # type: ignore [attr-defined]
             try:
-                response = session.get(search_url, params=params, timeout=15)  # type: ignore [arg-type]
-                response.raise_for_status()
+                response = self._req(search_url, params=params, timeout=15)
                 data = response.json()
-                time.sleep(NOMINATIM_DELAY)
             except Exception as e:
                 logger.debug(" OSM search error for '%s': %s", facility_name, e)
-                resp_info["search_query_steps"].append(f"(Failed -> {e})")  # type: ignore [attr-defined]
-                continue
-            if data:
-                return resp_info
-    # when the URL result is a "way" this is usually correct.
-    # checks top five results.
-    match_terms = ["prison", "detention", "correctional", "jail"]
-    for result in data:
-        osm_type = result.get("type", "").lower()
-        display_name = result.get("display_name", "").lower()
-        if any(term in osm_type for term in match_terms) or any(term in display_name for term in match_terms):
-            # todo courthouse could be added, or other tags such as "prison:for=migrant" as a clear positive search result.
-            osm_id = result.get("osm_id", "")
-            osm_type_prefix = result.get("osm_type", "")
-            title = result.get("display_name", "")
-            if osm_id and osm_type_prefix:
-                resp_info["url"] = f"https://www.openstreetmap.org/{osm_type_prefix}/{osm_id}"
-                resp_info["title"] = title
-                return resp_info
-    # fallback to first result
-    first_result = data[0]
-    logger.debug("Address searches didn't directly find anything, just using the first result: %s", first_result)
-    # default to Washington, D.C.?
-    lat = first_result.get("lat", "38.89511000")
-    lon = first_result.get("lon", "-77.03637000")
-    title = first_result.get("display_name", "")
-    resp_info["search_query_steps"].append(f"{lat}&{lon}")  # type: ignore [attr-defined]
-    if lat and lon:
-        resp_info["url"] = f"https://www.openstreetmap.org/?mlat={lat}&mlon={lon}&zoom=15"
-        resp_info["title"] = title
-    return resp_info
+                self.resp_info["search_query_steps"].append(f"(Failed -> {e})")  # type: ignore [attr-defined]
+                return self.resp_info
+        else:
+            full_address = (
+                f"{address['street']} {address['locality']}, {address['administrative_area']} {address['postal_code']}"
+            )
+            locality = f"{address['locality']}, {address['administrative_area']} {address['postal_code']}"
+            search_url = "https://nominatim.openstreetmap.org/search"
+            search_params = {
+                "facility_name": {
+                    "q": f"{search_name} {full_address}",
+                    "format": "json",
+                    "limit": 5,
+                    "dedupe": 1,
+                },
+                "street_address": {
+                    "q": f"{full_address}",
+                    "format": "json",
+                    "limit": 5,
+                    "dedupe": 1,
+                },
+                "locality": {
+                    "q": f"{locality}",
+                    "format": "json",
+                    "limit": 5,
+                    "dedupe": 1,
+                },
+            }
+            for search_name, params in search_params.items():
+                logger.debug("Searching OSM for %s", params["q"])
+                self.resp_info["search_query_steps"].append(params["q"])  # type: ignore [attr-defined]
+                try:
+                    response = self._req(search_url, params=params, timeout=15)
+                    data = response.json()
+                except Exception as e:
+                    logger.debug(" OSM search error for '%s': %s", facility_name, e)
+                    self.resp_info["search_query_steps"].append(f"(Failed -> {e})")  # type: ignore [attr-defined]
+                    continue
+                if data:
+                    return self.resp_info
+        # when the URL result is a "way" this is usually correct.
+        # checks top five results.
+        match_terms = ["prison", "detention", "correctional", "jail"]
+        for result in data:
+            osm_type = result.get("type", "").lower()
+            lat = result.get("lat", default_coords["latitude"])
+            lon = result.get("lon", default_coords["longitude"])
+            display_name = result.get("display_name", "").lower()
+            if any(term in osm_type for term in match_terms) or any(term in display_name for term in match_terms):
+                # todo courthouse could be added, or other tags such as "prison:for=migrant" as a clear positive search result.
+                osm_id = result.get("osm_id", "")
+                osm_type_prefix = result.get("osm_type", "")
+                title = result.get("display_name", "")
+                if osm_id and osm_type_prefix:
+                    self.resp_info["url"] = f"https://www.openstreetmap.org/{osm_type_prefix}/{osm_id}"
+                    self.resp_info["details"]["latitude"] = lat  # type: ignore [index]
+                    self.resp_info["details"]["longitude"] = lon  # type: ignore [index]
+                    self.resp_info["title"] = title
+                    return self.resp_info
+        # fallback to first result
+        first_result = data[0]
+        logger.debug("Address searches didn't directly find anything, just using the first result: %s", first_result)
+        title = first_result.get("display_name", "")
+        lat = first_result.get("lat", default_coords["latitude"])
+        lon = first_result.get("lon", default_coords["longitude"])
+        self.resp_info["search_query_steps"].append(f"{lat}&{lon}")  # type: ignore [attr-defined]
+        if lat and lon:
+            self.resp_info["url"] = f"https://www.openstreetmap.org/?mlat={lat}&mlon={lon}&zoom=15"
+            self.resp_info["details"]["latitude"] = lat  # type: ignore [index]
+            self.resp_info["details"]["longitude"] = lon  # type: ignore [index]
+            self.resp_info["title"] = title
+        return self.resp_info