remove some complexity around writing with nested dicts

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

Author: johnseekins

default_data.py

@@ -28,21 +28,10 @@ def enrich_facility_data(self, facilities_data: dict) -> dict:
         enriched_data = copy.deepcopy(facilities_schema)
         total = len(facilities_data["facilities"])
 
-        for i, facility in enumerate(facilities_data["facilities"]):
-            logger.info("Processing facility %s/%s: %s...", i + 1, total, facility["name"])
-            enriched_facility = copy.deepcopy(facility)
-            base_enrichment = {
-                "wikipedia_page_url": "",
-                "wikipedia_search_query": "",
-                "wikidata_page_url": "",
-                "wikidata_search_query": "",
-                "osm_result_url": "",
-                "osm_search_query": "",
-            }
-
-            enriched_facility.update(base_enrichment)
-
+        for index, facility in enumerate(facilities_data["facilities"]):
             facility_name = facility["name"]
+            logger.info("Processing facility %s/%s: %s...", index + 1, total, facility_name)
+            enriched_facility = copy.deepcopy(facility)
 
             # Wikipedia search # todo refactor to method
             try:

enricher.py

@@ -1,10 +1,8 @@
+import copy
 import csv
+import flatdict  # type: ignore [import-untyped]
 import json
-from schemas import (
-    debug_schema,
-    facility_schema,
-    enrichment_schema,
-)
+from schemas import enrichment_print_schema
 from utils import logger
 
 
@@ -18,22 +16,17 @@ def export_to_file(
         return ""
 
     full_name = f"{filename}.{file_type}"
+    csv_filtered_keys = ["raw_scrape", "wikipedia_search_query", "wikidata_search_query", "osm_search_query"]
     try:
         with open(full_name, "w", newline="", encoding="utf-8") as f_out:
             if file_type == "csv":
-                base_fields: list = list(facility_schema.keys())
-                fieldnames: list = base_fields.copy()
-
-                if any(field in facilities_data["facilities"][0] for field in enrichment_schema):
-                    fieldnames.extend(enrichment_schema)
-
-                if any(field in facilities_data["facilities"][0] for field in debug_schema):
-                    fieldnames.extend(debug_schema)
+                flatdata = [flatdict.FlatDict(f, delimiter=".") for f in facilities_data["facilities"]]
+                fieldnames = [k for k in flatdata[0].keys() if k not in csv_filtered_keys]
 
                 writer = csv.DictWriter(f_out, fieldnames=fieldnames)
                 writer.writeheader()
-                for facility in facilities_data["facilities"]:
-                    row_data = {field: facility.get(field, "") for field in fieldnames}
+                for facility in flatdata:
+                    row_data = {field: facility.get(field, None) for field in fieldnames}
                     writer.writerow(row_data)
             elif file_type == "json":
                 json.dump(facilities_data, f_out, indent=2, sort_keys=True, default=str)
@@ -75,37 +68,34 @@ def print_summary(facilities_data: dict) -> None:
         logger.info("  %s: %s", office, count)
 
     # Check enrichment data if available
-    if "wikipedia_page_url" in facilities_data["facilities"][0]:
-        wiki_found = sum(
-            1 for f in facilities_data["facilities"] if f.get("wikipedia_page_url") and f["wikipedia_page_url"]
-        )
-        wikidata_found = sum(
-            1 for f in facilities_data["facilities"] if f.get("wikidata_page_url") and f["wikidata_page_url"]
-        )
-        osm_found = sum(1 for f in facilities_data["facilities"] if f.get("osm_result_url") and f["osm_result_url"])
+    enrich_data = copy.deepcopy(enrichment_print_schema)
+    enrich_data["wiki_found"] = sum(1 for f in facilities_data["facilities"] if f.get("wikipedia_page_url", None))
+    enrich_data["wikidata_found"] = sum(1 for f in facilities_data["facilities"] if f.get("wikidata_page_url", None))
+    enrich_data["osm_found"] = sum(1 for f in facilities_data["facilities"] if f.get("osm_result_url", None))
 
+    if any(v > 0 for v in enrich_data.values()):
         logger.info("\n=== External Data Enrichment Results ===")
         logger.info(
             "Wikipedia pages found: %s/%s (%s%%)",
-            wiki_found,
+            enrich_data["wiki_found"],
             total_facilities,
-            wiki_found / total_facilities * 100,
+            enrich_data["wiki_found"] / total_facilities * 100,
         )
         logger.info(
             "Wikidata entries found: %s/%s (%s%%)",
-            wikidata_found,
+            enrich_data["wikidata_found"],
             total_facilities,
-            wikidata_found / total_facilities * 100,
+            enrich_data["wikidata_found"] / total_facilities * 100,
         )
         logger.info(
             "OpenStreetMap results found: %s/%s (%s%%)",
-            osm_found,
+            enrich_data["osm_found"],
             total_facilities,
-            osm_found / total_facilities * 100,
+            enrich_data["osm_found"] / total_facilities * 100,
         )
 
         # Debug information if available
-        if "wikipedia_search_query" in facilities_data["facilities"][0]:
+        if facilities_data["facilities"][0].get("wikipedia_search_query", None):
             logger.info("\n=== Wikipedia Debug Information ===")
             false_positives = 0
             errors = 0
@@ -120,10 +110,10 @@ def print_summary(facilities_data: dict) -> None:
             logger.info("Search errors encountered: %s", errors)
             logger.info("Note: Review 'wikipedia_search_query' column for detailed search information")
 
-        if "wikidata_search_query" in facilities_data["facilities"][0]:
+        if facilities_data["facilities"][0].get("wikidata_search_query", None):
             logger.warning("Note: Review 'wikidata_search_query' column for detailed search information")
 
-        if "osm_search_query" in facilities_data["facilities"][0]:
+        if facilities_data["facilities"][0].get("osm_search_query", None):
             logger.warning("Note: Review 'osm_search_query' column for detailed search information")
 
     logger.info("\n=== ICE Detention Facilities Scraper: Run completed ===")

file_utils.py

@@ -24,7 +24,7 @@
 from file_utils import export_to_file, print_summary
 import default_data
 from enricher import ExternalDataEnricher
-from scraper import ICEFacilityScraper
+from scraper import ICEGovFacilityScraper
 from utils import logger
 
 # CLI, argument parsing, script orchestration
@@ -78,7 +78,7 @@ def main() -> None:
         exit(1)
 
     if args.scrape:
-        scraper = ICEFacilityScraper()
+        scraper = ICEGovFacilityScraper()
         facilities_data = scraper.scrape_facilities()
     elif args.load_existing:
         facilities_data = copy.deepcopy(default_data.facilities_data)

main.py

@@ -6,6 +6,7 @@ readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
     "beautifulsoup4>=4.13.5",
+    "flatdict>=4.0.1",
     "lxml>=6.0.1",
     "requests>=2.32.5",
 ]

pyproject.toml

@@ -23,18 +23,13 @@
     "phone": "",
     "raw_scrape": "",
     "source_url": "",
+    "wikipedia_page_url": "",
+    "wikidata_page_url": "",
+    "osm_result_url": "",
+    "wikipedia_search_query": "",
+    "wikidata_search_query": "",
+    "osm_search_query": "",
 }
-enrichment_schema = [
-    "wikipedia_page_url",
-    "wikidata_page_url",
-    "osm_result_url",
-]
-
-debug_schema = [
-    "wikipedia_search_query",
-    "wikidata_search_query",
-    "osm_search_query",
-]
 
 # enrichment response object
 resp_info_schema = {
@@ -44,3 +39,10 @@
     "url": "",
     "method": "none",
 }
+
+# enrichment print details
+enrichment_print_schema = {
+    "wiki_found": 0,
+    "wikidata_found": 0,
+    "osm_found": 0,
+}

schemas.py

@@ -15,7 +15,7 @@
 )
 
 
-class ICEFacilityScraper(object):
+class ICEGovFacilityScraper(object):
     # All methods for scraping ICE websites
 
     def __init__(self):
@@ -72,24 +72,19 @@ def _scrape_updated(self, url: str):
         timestamp = f"{timestamp}-+{tz}"
         return datetime.datetime.strptime(timestamp, timestamp_format)
 
-    def _scrape_page(self, url: str) -> list:
+    def _scrape_page(self, page_url: str) -> list:
         """Scrape a single page of facilities using BeautifulSoup"""
-        logger.debug("  Fetching: %s", url)
+        logger.debug("  Fetching: %s", page_url)
         try:
-            response = session.get(url, timeout=30)
+            response = session.get(page_url, timeout=30)
             response.raise_for_status()
         except Exception as e:
-            logger.error("  Error parsing %s: %s", url, e)
+            logger.error("  Error parsing %s: %s", page_url, e)
             return []
 
         # Parse HTML with BeautifulSoup
         soup = BeautifulSoup(response.content, "html.parser")
 
-        # Extract facilities from the parsed HTML
-        return self._extract_facilities_from_html(soup, url)
-
-    def _extract_facilities_from_html(self, soup, page_url: str) -> list:
-        """Extract facility data from BeautifulSoup parsed HTML"""
         facilities = []
 
         # Look for the main content area - ICE uses different possible containers
@@ -122,7 +117,7 @@ def _extract_facilities_from_html(self, soup, page_url: str) -> list:
             "article",  # Article elements
             "div.node",  # Drupal node containers
         ]
-        facility_elements = []
+        facility_elements: list = []
         for selector in facility_selectors:
             elements = content_container.select(selector)
             if elements: