"""
One-time (per Overture release) extract of Nigeria's places from Overture's public
S3 parquet into a local parquet file. ingest/places.py then loads that file into
the intel store — the slow network read happens once, not on every ingest.

    py -m intel.ingest.overture_extract [--release 2026-09-23.1]

Filtered by Nigeria's bounding box only; the exact country boundary is applied at
ingest (point-in-LGA-polygon), which also drops the bits of Benin/Cameroon/Niger the
box catches.
"""
import argparse
import time
from pathlib import Path

import duckdb

DEFAULT_RELEASE = "2026-09-23.1"
# Nigeria's extent with a small margin (lon 2.67–14.68, lat 4.27–13.89).
BBOX = {"xmin": 2.6, "xmax": 14.7, "ymin": 4.2, "ymax": 13.95}
RAW_DIR = Path(__file__).resolve().parent.parent / "data" / "raw"


def extract_path(release: str) -> Path:
    return RAW_DIR / f"overture_places_ng_{release}.parquet"


def extract(release: str = DEFAULT_RELEASE) -> Path:
    out = extract_path(release)
    RAW_DIR.mkdir(parents=True, exist_ok=True)
    con = duckdb.connect()
    con.sql("INSTALL spatial; LOAD spatial; INSTALL httpfs; LOAD httpfs; SET s3_region='us-west-2';")
    src = f"s3://overturemaps-us-west-2/release/{release}/theme=places/type=place/*"
    started = time.time()
    con.sql(f"""
        COPY (
          SELECT
            id,
            names.primary            AS name,
            basic_category,
            taxonomy.primary         AS taxonomy_primary,
            array_to_string(taxonomy.hierarchy, '>') AS taxonomy_hierarchy,
            confidence,
            operating_status,
            ST_X(geometry)           AS lon,
            ST_Y(geometry)           AS lat,
            addresses[1].locality    AS locality,
            addresses[1].region      AS region,
            addresses[1].country     AS country
          FROM read_parquet('{src}', hive_partitioning=1)
          WHERE bbox.xmin BETWEEN {BBOX['xmin']} AND {BBOX['xmax']}
            AND bbox.ymin BETWEEN {BBOX['ymin']} AND {BBOX['ymax']}
        ) TO '{out.as_posix()}' (FORMAT PARQUET)
    """)
    n = con.sql(f"SELECT count(*) FROM read_parquet('{out.as_posix()}')").fetchone()[0]
    print(f"Extracted {n} places to {out} in {time.time() - started:.0f}s")
    return out


if __name__ == "__main__":
    p = argparse.ArgumentParser()
    p.add_argument("--release", default=DEFAULT_RELEASE)
    extract(p.parse_args().release)