Spaces:
Running
Running
| """ | |
| The gazetteer: country → state → LGA (→ town, later), each with a centre point, | |
| plus an alias table that turns free text ("flooding in Ajah", "Port Harcourt | |
| traders") into a location id. | |
| State and LGA *names* come from the app's own list (nigeria_locations.json, | |
| exported from data/nigeria-locations.ts) so they're exactly what users pick in | |
| onboarding. *Shapes* come from geoBoundaries, whose LGA spellings differ ("Ado | |
| Odo/Ota" vs "Ado-Odo/Ota") and whose ADM2 file carries no state, so each | |
| boundary LGA is first placed in a state by point-in-polygon, then matched to the | |
| app's name within that state. | |
| resolve() is deliberately simple — exact alias matching on word boundaries, | |
| with state context used to disambiguate LGA names that exist in several states | |
| (Surulere, Obi, Bassa, Ifelodun, ...). It's the seed of the Layer 5 search | |
| service, not a replacement for it. | |
| """ | |
| import difflib | |
| import json | |
| import re | |
| import sqlite3 | |
| from pathlib import Path | |
| from intel.freshness import record_attempt | |
| from intel.geo import PolygonIndex, centroid | |
| HERE = Path(__file__).resolve().parent | |
| STATES_JSON = HERE / "nigeria_locations.json" | |
| RAW_DIR = HERE / "data" / "raw" | |
| ADM1_PATH = RAW_DIR / "geoBoundaries-NGA-ADM1_simplified.geojson" | |
| ADM2_PATH = RAW_DIR / "geoBoundaries-NGA-ADM2_simplified.geojson" | |
| COUNTRY_ID = "ng" | |
| FCT_APP_NAME = "FCT (Abuja)" | |
| ADM1_NAME_TO_APP = {"Abuja Federal Capital Territory": FCT_APP_NAME} | |
| # Boundary-file LGA spellings that normalise differently from the app's name and | |
| # are too far apart for fuzzy matching to pair safely. Keyed by | |
| # (app state, normalised boundary name) → app LGA name. | |
| BOUNDARY_LGA_TO_APP: dict[tuple[str, str], str] = { | |
| (FCT_APP_NAME, "municipal area council"): "Abuja Municipal (AMAC)", | |
| } | |
| # Common place names in news text that aren't state or LGA names. Kept short and | |
| # only for unambiguous cases; place-data localities add many more automatically. | |
| CURATED_ALIASES = { | |
| "abuja": "ng.fct", | |
| "lekki": "ng.lagos.eti-osa", | |
| "ajah": "ng.lagos.eti-osa", | |
| "victoria island": "ng.lagos.eti-osa", | |
| "yaba": "ng.lagos.lagos-mainland", | |
| "ibadan": "ng.oyo", | |
| "benin city": "ng.edo.oredo", | |
| "onitsha": "ng.anambra", | |
| "aba": "ng.abia", | |
| "warri": "ng.delta", | |
| "jos": "ng.plateau", | |
| "owerri": "ng.imo", | |
| "calabar": "ng.cross-river", | |
| "ilorin": "ng.kwara", | |
| "enugu": "ng.enugu", | |
| "kano": "ng.kano", | |
| # National bodies: a policy from these applies to the whole country. | |
| "federal government": "ng", | |
| "fg": "ng", | |
| "cbn": "ng", | |
| "central bank of nigeria": "ng", | |
| "firs": "ng", | |
| "national assembly": "ng", | |
| "nigerian": "ng", | |
| } | |
| def normalise(name: str) -> str: | |
| s = name.lower().replace("&", " and ") | |
| s = re.sub(r"\(.*?\)", " ", s) | |
| s = re.sub(r"[^a-z0-9]+", " ", s) | |
| return re.sub(r"\s+", " ", s).strip() | |
| def _key(name: str) -> str: | |
| return normalise(name).replace(" ", "") | |
| def slug(name: str) -> str: | |
| return normalise(name).replace(" ", "-") | |
| def state_id(app_state: str) -> str: | |
| return f"{COUNTRY_ID}.fct" if app_state == FCT_APP_NAME else f"{COUNTRY_ID}.{slug(app_state)}" | |
| def lga_id(app_state: str, lga: str) -> str: | |
| return f"{state_id(app_state)}.{slug(lga)}" | |
| def _load_geojson(path: Path) -> list[dict]: | |
| return json.loads(path.read_text(encoding="utf8"))["features"] | |
| def match_boundaries(states: list[dict], adm1: list[dict], adm2: list[dict]) -> tuple[dict, list[str]]: | |
| """Returns ({(app_state, app_lga): adm2 feature}, problems). Every app LGA | |
| must end up with exactly one boundary — anything else is reported, not | |
| guessed.""" | |
| app_states = {s["name"] for s in states} | |
| state_index = PolygonIndex([ | |
| (ADM1_NAME_TO_APP.get(f["properties"]["shapeName"], f["properties"]["shapeName"]), f["geometry"]) for f in adm1 | |
| ]) | |
| unknown = {k for k, _, _ in state_index._items} - app_states | |
| if unknown: | |
| raise ValueError(f"Boundary states not in the app list: {sorted(unknown)}") | |
| by_state: dict[str, list[dict]] = {} | |
| problems = [] | |
| for f in adm2: | |
| lon, lat = centroid(f["geometry"]) | |
| st = state_index.lookup(lon, lat) | |
| if st is None: | |
| problems.append(f"boundary LGA {f['properties']['shapeName']!r}: centre falls in no state") | |
| continue | |
| by_state.setdefault(st, []).append(f) | |
| matched: dict[tuple[str, str], dict] = {} | |
| for s in states: | |
| st, wanted = s["name"], list(s["lgas"]) | |
| pool = list(by_state.get(st, [])) | |
| def take(app_lga: str, feature: dict) -> None: | |
| matched[(st, app_lga)] = feature | |
| wanted.remove(app_lga) | |
| pool.remove(feature) | |
| for f in list(pool): | |
| k = _key(f["properties"]["shapeName"]) | |
| curated = BOUNDARY_LGA_TO_APP.get((st, normalise(f["properties"]["shapeName"]))) | |
| exact = [w for w in wanted if _key(w) == k] | |
| if curated and curated in wanted: | |
| take(curated, f) | |
| elif len(exact) == 1: | |
| take(exact[0], f) | |
| # Fuzzy pass for what's left, one-to-one, best pairs first. | |
| pairs = sorted( | |
| ((difflib.SequenceMatcher(None, _key(w), _key(f["properties"]["shapeName"])).ratio(), w, f) | |
| for w in wanted for f in pool), | |
| key=lambda p: -p[0], | |
| ) | |
| for ratio, w, f in pairs: | |
| if ratio < 0.6: | |
| break | |
| if w in wanted and f in pool: | |
| take(w, f) | |
| for w in wanted: | |
| problems.append(f"{st}: app LGA {w!r} has no boundary") | |
| for f in pool: | |
| problems.append(f"{st}: boundary LGA {f['properties']['shapeName']!r} matched no app LGA") | |
| return matched, problems | |
| def _add_alias(con: sqlite3.Connection, alias: str, location_id: str, origin: str) -> None: | |
| alias = normalise(alias) | |
| if alias: | |
| con.execute("insert or ignore into location_aliases (alias, location_id, origin) values (?, ?, ?)", | |
| (alias, location_id, origin)) | |
| def build(con: sqlite3.Connection) -> dict: | |
| """(Re)builds country/state/LGA rows and their aliases. Idempotent.""" | |
| states = json.loads(STATES_JSON.read_text(encoding="utf8")) | |
| adm1, adm2 = _load_geojson(ADM1_PATH), _load_geojson(ADM2_PATH) | |
| matched, problems = match_boundaries(states, adm1, adm2) | |
| if problems: | |
| raise ValueError("Boundary matching failed:\n" + "\n".join(problems)) | |
| def upsert(lid, level, name, parent, lon, lat): | |
| con.execute( | |
| """insert into locations (id, level, name, parent_id, lat, lon, source) values (?, ?, ?, ?, ?, ?, 'geoboundaries') | |
| on conflict (id) do update set level = excluded.level, name = excluded.name, | |
| parent_id = excluded.parent_id, lat = excluded.lat, lon = excluded.lon""", | |
| (lid, level, name, parent, lat, lon), | |
| ) | |
| country = {"type": "MultiPolygon", "coordinates": []} | |
| for f in adm1: | |
| g = f["geometry"] | |
| country["coordinates"] += [g["coordinates"]] if g["type"] == "Polygon" else g["coordinates"] | |
| upsert(COUNTRY_ID, "country", "Nigeria", None, *centroid(country)) | |
| _add_alias(con, "Nigeria", COUNTRY_ID, "gazetteer") | |
| adm1_by_app = {ADM1_NAME_TO_APP.get(f["properties"]["shapeName"], f["properties"]["shapeName"]): f for f in adm1} | |
| for s in states: | |
| sid = state_id(s["name"]) | |
| upsert(sid, "state", s["name"], COUNTRY_ID, *centroid(adm1_by_app[s["name"]]["geometry"])) | |
| _add_alias(con, s["name"], sid, "gazetteer") | |
| _add_alias(con, f"{s['name']} state", sid, "gazetteer") | |
| _add_alias(con, adm1_by_app[s["name"]]["properties"]["shapeName"], sid, "boundary") | |
| for lga in s["lgas"]: | |
| lid = lga_id(s["name"], lga) | |
| f = matched[(s["name"], lga)] | |
| upsert(lid, "lga", lga, sid, *centroid(f["geometry"])) | |
| _add_alias(con, lga, lid, "gazetteer") | |
| _add_alias(con, f["properties"]["shapeName"], lid, "boundary") | |
| # "Abuja Municipal (AMAC)" → also findable as "AMAC". | |
| for inner in re.findall(r"\((.*?)\)", lga): | |
| _add_alias(con, inner, lid, "gazetteer") | |
| _add_alias(con, "FCT", state_id(FCT_APP_NAME), "gazetteer") | |
| for alias, lid in CURATED_ALIASES.items(): | |
| _add_alias(con, alias, lid, "curated") | |
| con.commit() | |
| # The boundary files are downloaded by hand (backend/intel/README.md); the | |
| # build is what proves they're present and usable, so it reports to Layer 1. | |
| # No data period: boundaries have no "newest day", only the fetch clock. | |
| record_attempt(con, "geoboundaries", ok=True) | |
| n = con.execute("select level, count(*) from locations group by level").fetchall() | |
| return {r[0]: r[1] for r in n} | |
| def lga_polygons() -> list[tuple[str, dict]]: | |
| """(lga id, geometry) for point-in-LGA lookups (places ingest).""" | |
| states = json.loads(STATES_JSON.read_text(encoding="utf8")) | |
| matched, problems = match_boundaries(states, _load_geojson(ADM1_PATH), _load_geojson(ADM2_PATH)) | |
| if problems: | |
| raise ValueError("Boundary matching failed:\n" + "\n".join(problems)) | |
| return [(lga_id(st, lga), f["geometry"]) for (st, lga), f in matched.items()] | |
| def _alias_rows(con: sqlite3.Connection) -> list[sqlite3.Row]: | |
| """alias, location id, its level, and the state it belongs to (itself for a state).""" | |
| return con.execute( | |
| """select a.alias, l.id, l.level, | |
| case when l.level = 'state' then l.id when p.level = 'state' then p.id end as state | |
| from location_aliases a join locations l on l.id = a.location_id | |
| left join locations p on p.id = l.parent_id""" | |
| ).fetchall() | |
| def resolve(con: sqlite3.Connection, text: str | None) -> str | None: | |
| """Best location id mentioned in text, or None when there's no mention or | |
| the mention is genuinely ambiguous. Prefers the most specific level, with | |
| two rules: | |
| - a name that is both a state and an LGA ("Bauchi", "Ekiti") means the | |
| state — that's how people write; | |
| - an LGA name shared by several states ("Surulere") only resolves when the | |
| text also names one of those states; otherwise it falls back to their | |
| common state if they share one, else None. | |
| """ | |
| if not text: | |
| return None | |
| hay = f" {normalise(text)} " | |
| hits = [r for r in _alias_rows(con) if f" {r['alias']} " in hay] | |
| state_aliases = {r["alias"] for r in hits if r["level"] == "state"} | |
| hits = [r for r in hits if not (r["level"] in ("lga", "town") and r["alias"] in state_aliases)] | |
| named_states = {r["id"] for r in hits if r["level"] == "state"} | |
| for level in ("town", "lga"): | |
| cands = {(r["id"], r["state"]) for r in hits if r["level"] == level} | |
| if not cands: | |
| continue | |
| narrowed = {c for c in cands if c[1] in named_states} | |
| cands = narrowed or cands | |
| if len(cands) == 1: | |
| return next(iter(cands))[0] | |
| states = {st for _, st in cands} | |
| return next(iter(states)) if len(states) == 1 else None | |
| if len(named_states) == 1: | |
| return next(iter(named_states)) | |
| if named_states: | |
| return None | |
| return COUNTRY_ID if any(r["level"] == "country" for r in hits) else None | |
| def add_place_localities(con: sqlite3.Connection, min_places: int = 20, share: float = 0.8) -> int: | |
| """Turns place-data localities into aliases, by majority vote: a locality | |
| string with enough places, most of them inside one LGA, becomes an alias of | |
| that LGA; failing that, of their common state. Data-derived, so "Ikorodu" or | |
| "Agbara" become resolvable without anyone typing them in.""" | |
| rows = con.execute( | |
| """select lower(trim(locality)) loc, location_id, count(*) n from places | |
| where locality is not null and location_id is not null group by 1, 2""" | |
| ).fetchall() | |
| by_loc: dict[str, list[tuple[str, int]]] = {} | |
| for r in rows: | |
| by_loc.setdefault(r["loc"], []).append((r["location_id"], r["n"])) | |
| added = 0 | |
| for loc, counts in by_loc.items(): | |
| total = sum(n for _, n in counts) | |
| if total < min_places or not normalise(loc): | |
| continue | |
| best, n = max(counts, key=lambda c: c[1]) | |
| if n / total >= share: | |
| target = best | |
| else: | |
| state_counts: dict[str, int] = {} | |
| for lid, c in counts: | |
| state_counts[lid.rsplit(".", 1)[0]] = state_counts.get(lid.rsplit(".", 1)[0], 0) + c | |
| st, sn = max(state_counts.items(), key=lambda c: c[1]) | |
| if sn / total < share: | |
| continue | |
| target = st | |
| # Never let a derived alias shadow a gazetteer name (e.g. "Lagos" the | |
| # locality must keep meaning Lagos State, not whichever LGA dominates). | |
| if con.execute("select 1 from location_aliases where alias = ? and origin != 'places'", (normalise(loc),)).fetchone(): | |
| continue | |
| _add_alias(con, loc, target, "places") | |
| added += 1 | |
| con.commit() | |
| return added | |