Spaces:
Running
Running
File size: 2,890 Bytes
b7c46f4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 | """
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
|