Spaces:
Running
Running
| """ | |
| Loads the local Overture extract (overture_extract.py) into `places`: assigns | |
| each place to the LGA it falls in (which also drops the Benin/Cameroon/Niger | |
| points the bounding box caught), maps its category to a subsector, then derives | |
| locality aliases for the gazetteer. | |
| A release replaces the previous one wholesale — places close and reopen, and | |
| Overture ids are stable across releases, so a full refresh is simpler and more | |
| honest than trying to diff. | |
| """ | |
| import sqlite3 | |
| import duckdb | |
| from intel.freshness import record_attempt | |
| from intel.geo import PolygonIndex | |
| from intel.handoff import DATA_STREAM, INTERFACE, produce | |
| from intel.ingest.overture_extract import DEFAULT_RELEASE, extract_path | |
| from intel.locations import add_place_localities, lga_polygons | |
| from intel.subsector_map import subsector_for | |
| SOURCE_ID = "overture_places" | |
| def ingest(con: sqlite3.Connection, release: str = DEFAULT_RELEASE, run_id: str = "manual") -> dict: | |
| path = extract_path(release) | |
| try: | |
| if not path.exists(): | |
| raise FileNotFoundError(f"{path} missing — run `py -m intel.ingest.overture_extract --release {release}` first") | |
| rows = duckdb.connect().sql( | |
| f"""select id, name, basic_category, taxonomy_primary, taxonomy_hierarchy, confidence, | |
| operating_status, lon, lat, locality | |
| from read_parquet('{path.as_posix()}') where lon is not null and lat is not null""" | |
| ).fetchall() | |
| index = PolygonIndex(lga_polygons()) | |
| con.execute("delete from places where source_id = ?", (SOURCE_ID,)) | |
| kept = outside = mapped = 0 | |
| batch = [] | |
| for pid, name, basic, tax, hier, conf, status, lon, lat, locality in rows: | |
| lga = index.lookup(lon, lat) | |
| if lga is None: | |
| outside += 1 | |
| continue | |
| sub = subsector_for(basic, tax) | |
| mapped += sub is not None | |
| batch.append((pid, name, basic, hier, sub, conf, status, lat, lon, lga, locality, SOURCE_ID, release)) | |
| kept += 1 | |
| con.executemany( | |
| """insert into places (id, name, category, taxonomy, subsector, confidence, operating_status, | |
| lat, lon, location_id, locality, source_id, release) | |
| values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", | |
| batch, | |
| ) | |
| con.commit() | |
| aliases = add_place_localities(con) | |
| except Exception as e: | |
| record_attempt(con, SOURCE_ID, ok=False, error=str(e)) | |
| raise | |
| record_attempt(con, SOURCE_ID, ok=True, latest_data_period=release.split(".")[0]) | |
| summary = {"release": release, "places": kept, "outside_nigeria": outside, | |
| "with_subsector": mapped, "locality_aliases_added": aliases} | |
| produce(con, run_id, DATA_STREAM, INTERFACE, {"kind": "places", **summary}) | |
| return summary | |