"""
import_tickets.py — Fireside Communications · INC ticket ingestion (raw-first)
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
Loads the client's field-ops INC ticket snapshots into the `tickets` schema — the
source of the FleetOps "Tickets" map.
    tickets.inc   — incidents / customer faults

STRICTLY INC: CRQ (new-installation) exports are out of scope and not processed
here. `tickets.crq` stays in the schema but is not fed by this pipeline.

RAW-FIRST: each row stores only `ticket_id` + `raw` (the source record as jsonb).
Everything downstream reads from `raw` (resilient to source schema drift). The DB
derives `geom` (see migrations): feed coords (raw lat/lng) -> location geocode
(tickets.geo_locations) -> cluster centroid (tickets.geo_clusters) -> none.

Source data: the n8n hourly S3 export (see n8n-hourly-s3-full-data-exports.md)
writes a full current-state snapshot CSV per hour to the `tickets` bucket at
    automations/inc/<EAT-timestamp>.csv   (e.g. 2026-06-15T17-00-00.csv)
There is NO latest pointer, NO metadata envelope, and NO deltas — each file is a
flat CSV (header + rows). We ingest the NEWEST file:
  - skip-if-unchanged: if its S3 ETag matches the last processed file's ETag we
    skip the DB write (the export re-emits byte-identical content most hours);
  - drop is_alarm=true rows + the "EXPORT STOPPED…" truncation-sentinel row;
  - drop derivable / provenance / zero-info columns (see DROP_FIELDS);
  - normalize region -> lowercase, raw_status -> UPPERCASE;
  - upsert on ticket_id (PRIMARY KEY → no duplication; never delete, so closure
    history accumulates), and record snapshot freshness in tickets.import_meta;
  - on success, MOVE the file to automations/inc/processed/ (copy + delete).

Geocoding (two layers, both via a KEYED provider — public Nominatim rate-limits):
  --geocode-clusters   one coordinate per cluster (coarse fallback; ~50 lookups)
  --geocode-locations  precise per-location for ACTIONABLE INC tickets: parses the
                       real place out of location_name (region+cluster+location_name,
                       network codes stripped), geocodes it, caches in
                       tickets.geo_locations, then re-resolves geoms.
Provider/key from env: GEOCODER_PROVIDER (locationiq|opencage), GEOCODER_API_KEY.

Usage (needs DATABASE_URL + RUSTFS_* + GEOCODER_* env; see .env.example):
    python import_tickets.py --from-bucket --apply
    python import_tickets.py --inc-csv 2026-06-15T17-00-00.csv --apply
    python import_tickets.py --geocode-clusters --apply
    python import_tickets.py --geocode-locations --apply

Pre-requisite: migration applied (run_migrations.py) — tickets.inc/crq +
geo_clusters + geo_locations + reporting.fn_tickets_for_map.
━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━
"""

from __future__ import annotations

import argparse
import csv
import io
import math
import os
import re
import time
from datetime import datetime, timezone, timedelta

import boto3
import requests
import psycopg2.extras
from botocore.config import Config as BotoConfig

from shared import clean, get_conn, get_logger

log = get_logger("import_tickets")

# ── INC ingestion config ──────────────────────────────────────────────────────
_TABLE = "tickets.inc"
_DATASET = "inc"
_BUCKET = os.getenv("TICKETS_BUCKET", "tickets")
_INC_PREFIX = "automations/inc/"
_PROCESSED_PREFIX = "automations/inc/processed/"
_EAT = timezone(timedelta(hours=3))  # Africa/Nairobi — filenames + data are EAT

# Garbage row the source leaks (commonly the first data line): its ticket_id is the
# message itself. Matched by prefix so position/exact-tail don't matter.
_SENTINEL_PREFIX = "EXPORT STOPPED"

# Columns dropped before building `raw`: derivable (week_*), the client's row-level
# export provenance (source_s3_*, source_snapshot_id), and zero-information columns
# (department=always FTTH, source_type=duplicate of service_type). We KEEP
# service_type and `bucket` (the latter is a closed/pending lifecycle flag).
DROP_FIELDS = frozenset({
    "week_start", "week_end",
    "source_s3_bucket", "source_s3_key", "source_snapshot_id",
    "department", "source_type",
})

# Only files matching automations/inc/<EAT-timestamp>.csv (NOT processed/, NOT the
# leftover latest.csv/, latest.json/, full/ prefixes).
_CSV_KEY_RE = re.compile(r"^automations/inc/(\d{4}-\d{2}-\d{2}T\d{2}-\d{2}-\d{2})\.csv$")

# Geocoder (keyed) — public Nominatim rate-limits bulk, so we use LocationIQ/OpenCage.
_PROVIDER = os.getenv("GEOCODER_PROVIDER", "locationiq").lower()
_API_KEY = os.getenv("GEOCODER_API_KEY", "")
_GEOCODE_INTERVAL_S = float(os.getenv("GEOCODER_MIN_INTERVAL_S", "1.1"))
_last_geocode_at = 0.0


# ── data loading (CSV · newest-file · ETag skip-if-unchanged) ───────────────────
# The n8n hourly export writes a full current-state CSV per hour to
# automations/inc/<EAT-timestamp>.csv (no latest pointer, no envelope, no deltas).
# We ingest the NEWEST file; if its S3 ETag matches the last processed file's ETag
# we skip the DB write (the export re-emits byte-identical content most hours).
# S3 access is via boto3 (no aws-CLI dependency → runs cleanly in a slim container).
def _s3_client():
    """boto3 S3 client for the rustfs endpoint (force path-style addressing)."""
    return boto3.client(
        "s3",
        endpoint_url=os.environ["RUSTFS_ENDPOINT"],
        aws_access_key_id=os.environ["RUSTFS_ACCESS_KEY"],
        aws_secret_access_key=os.environ["RUSTFS_SECRET_KEY"],
        region_name=os.getenv("RUSTFS_REGION", "us-east-1"),
        config=BotoConfig(s3={"addressing_style": "path"}, signature_version="s3v4",
                          retries={"max_attempts": 3, "mode": "standard"}),
    )


def _ts_from_key(key: str) -> datetime | None:
    """EAT timestamp embedded in an automations/inc/<ts>.csv key (or None)."""
    m = _CSV_KEY_RE.match(key)
    if not m:
        return None
    try:  # regex shape can match an impossible date (e.g. 9999-99-99T…) — don't crash the sort
        return datetime.strptime(m.group(1), "%Y-%m-%dT%H-%M-%S").replace(tzinfo=_EAT)
    except ValueError:
        return None


def _list_inc_csvs(s3) -> list[tuple[str, str]]:
    """[(key, etag)] for every automations/inc/<ts>.csv (excludes processed/ + dirs)."""
    out: list[tuple[str, str]] = []
    for page in s3.get_paginator("list_objects_v2").paginate(Bucket=_BUCKET, Prefix=_INC_PREFIX):
        for it in page.get("Contents", []):
            if _CSV_KEY_RE.match(it["Key"]):
                out.append((it["Key"], (it.get("ETag") or "").strip('"')))
    return out


def _get_text(s3, key: str) -> str:
    """Download an object's body as UTF-8 text."""
    return s3.get_object(Bucket=_BUCKET, Key=key)["Body"].read().decode("utf-8")


def _last_processed_etag() -> str | None:
    """ETag of the most recently ingested INC file (from tickets.import_meta)."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                "SELECT metadata->>'source_etag' FROM tickets.import_meta WHERE dataset = %s",
                (_DATASET,),
            )
            row = cur.fetchone()
            return row[0] if row else None


def _parse_csv(text: str) -> list[dict]:
    return list(csv.DictReader(io.StringIO(text)))


def _load_csv_local(path: str) -> list[dict]:
    with open(path, encoding="utf-8", newline="") as f:
        return list(csv.DictReader(f))


def _move_processed(s3, keys: list[str]) -> None:
    """Archive listed INC csv objects to automations/inc/processed/ (copy + delete)."""
    for key in keys:
        dst = _PROCESSED_PREFIX + key.rsplit("/", 1)[-1]
        s3.copy_object(Bucket=_BUCKET, CopySource={"Bucket": _BUCKET, "Key": key}, Key=dst)
        s3.delete_object(Bucket=_BUCKET, Key=key)
        log.info("archived %s -> %s", key, dst)


# ── row preparation (filter · drop columns · normalize) ─────────────────────────
def _keep_row(row: dict) -> bool:
    """Drop alarm rows + the truncation-sentinel; require a real ticket_id."""
    tid = clean(row.get("ticket_id"))
    if not tid or tid.startswith(_SENTINEL_PREFIX):
        return False
    return clean(row.get("is_alarm")) != "true"


def _prepare(row: dict) -> dict:
    """Strip DROP_FIELDS and normalize region/raw_status — returns the `raw` payload."""
    r = {k: v for k, v in row.items() if k not in DROP_FIELDS}
    if r.get("region"):
        r["region"] = r["region"].lower()
    if r.get("raw_status"):
        r["raw_status"] = r["raw_status"].upper()
    return r


# ── upsert (raw-first) ────────────────────────────────────────────────────────
def _record_meta(cur, meta: dict, records_ingested: int) -> None:
    """Upsert the INC snapshot metadata (powers map freshness + holds source_etag).

    Runs on the caller's cursor so the row upsert and the meta write commit
    together — a half-written state (rows in, meta stale) breaks skip-if-unchanged.
    """
    cur.execute(
        """INSERT INTO tickets.import_meta
             (dataset, export_type, exported_at, snapshot_date, source_schema,
              source_table, row_count, records_ingested, n8n_execution_id, metadata,
              ingested_at)
           VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, now())
           ON CONFLICT (dataset) DO UPDATE
             SET export_type = EXCLUDED.export_type, exported_at = EXCLUDED.exported_at,
                 snapshot_date = EXCLUDED.snapshot_date, source_schema = EXCLUDED.source_schema,
                 source_table = EXCLUDED.source_table, row_count = EXCLUDED.row_count,
                 records_ingested = EXCLUDED.records_ingested,
                 n8n_execution_id = EXCLUDED.n8n_execution_id, metadata = EXCLUDED.metadata,
                 ingested_at = now()""",
        (_DATASET, clean(meta.get("export_type")), clean(meta.get("exported_at")),
         clean(meta.get("snapshot_date")), clean(meta.get("source_schema")),
         clean(meta.get("source_table")), meta.get("row_count"), records_ingested,
         clean(meta.get("n8n_execution_id")), psycopg2.extras.Json(meta)),
    )


def upsert(rows: list[dict], apply: bool, meta: dict | None = None) -> int:
    meta = meta or {}
    kept = [r for r in rows if _keep_row(r)]
    payload = [(clean(r["ticket_id"]), psycopg2.extras.Json(_prepare(r))) for r in kept]
    log.info("%s: %d rows read, %d kept, %d dropped (alarm/sentinel/no-id)",
             _TABLE, len(rows), len(payload), len(rows) - len(payload))
    if not apply:
        log.info("DRY-RUN — nothing written to %s. Use --apply.", _TABLE)
        return len(payload)
    with get_conn() as conn:
        with conn.cursor() as cur:
            psycopg2.extras.execute_values(
                cur,
                f"INSERT INTO {_TABLE} (ticket_id, raw) VALUES %s "
                "ON CONFLICT (ticket_id) DO UPDATE SET raw = EXCLUDED.raw, ingested_at = now()",
                payload, page_size=500,
            )
            # same transaction as the upsert: rows + snapshot meta commit atomically
            _record_meta(cur, meta, len(payload))
    log.info("upserted %d rows into %s", len(payload), _TABLE)
    return len(payload)


def _capture_history() -> None:
    """Append new closures + upsert today's backlog snapshot (tickets.capture_history)."""
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT tickets.capture_history()")
            log.info("history: %s", cur.fetchone()[0])


def ingest(args) -> None:
    # Local-file path (dev): ingest a single CSV, no bucket / no archive.
    if args.inc_csv:
        rows = _load_csv_local(args.inc_csv)
        name = os.path.basename(args.inc_csv)
        ts = _ts_from_key(_INC_PREFIX + name)
        meta = {"export_type": "full", "source_s3_key": name, "row_count": len(rows)}
        if ts:
            meta["exported_at"] = ts.isoformat()
        upsert(rows, args.apply, meta=meta)
        return

    # --from-bucket: newest INC csv → skip-if-unchanged → ingest → archive.
    s3 = _s3_client()
    listing = _list_inc_csvs(s3)
    if not listing:
        log.info("no INC csv files under %s — nothing to do", _INC_PREFIX)
        return
    listing.sort(key=lambda ke: _ts_from_key(ke[0]) or datetime.min.replace(tzinfo=_EAT))
    all_keys = [k for k, _ in listing]
    newest_key, newest_etag = listing[-1]
    log.info("newest INC file: %s (etag=%s; %d file(s) present)",
             newest_key, newest_etag, len(listing))

    last_etag = _last_processed_etag()
    if newest_etag and newest_etag == last_etag:
        log.info("etag unchanged from last processed (%s) — skipping DB write", last_etag)
        if args.apply:
            _move_processed(s3, all_keys)
            _capture_history()   # still record today's snapshot even when unchanged
        else:
            log.info("DRY-RUN — would archive %d file(s) to %s", len(all_keys), _PROCESSED_PREFIX)
        return

    rows = _parse_csv(_get_text(s3, newest_key))
    ts = _ts_from_key(newest_key)
    meta = {"export_type": "full", "source_s3_key": newest_key,
            "source_etag": newest_etag, "row_count": len(rows)}
    if ts:
        meta["exported_at"] = ts.isoformat()
    upsert(rows, args.apply, meta=meta)
    if args.apply:
        _move_processed(s3, all_keys)
        _capture_history()
    else:
        log.info("DRY-RUN — would archive %d file(s) to %s", len(all_keys), _PROCESSED_PREFIX)


# ── place extraction (strip network codes, keep the real place) ───────────────
# Leading site-code prefixes (NW_, CO_, ADR_MNT_, COAST_, …) — applied repeatedly.
_PREFIX_RE = re.compile(r"^(?:NW|NE|NM|SW|SE|CO|COAST|ADR|MNT|CMT|DR|NAIROBI|FIBER\w*)[\s_]+")
# 'NW' is the one site-code that the source also glues straight onto the place with
# no separator (NWKIAMBU, NWRIDGE, NWTHE — ~1.7k rows in a single snapshot). Safe to
# split because no place/word starts with "NW"; the other codes (CO/NE/SE/DR…) begin
# real words (COAST, NEW, SEASONS, DRIVE) so we only strip THOSE when delimited above.
_GLUED_NW_RE = re.compile(r"^NW(?=[A-Z])")
# Inline network/work-order codes to drop wherever they appear.
_CODE_RE = re.compile(
    r"\b(?:SDUS|SDU|MDUS|MDU|FDT\s*\d*|AP|CLUSTER\s*\d*[A-Z]?|PHASE\s*\d+|CL\s*\d+|MNT|SITE|ADR)\b"
)
# Trailing '-<segment>' after the final hyphen: a unit/instruction code, not a place.
# Dropped only when it LOOKS like one — a unit number (37, F32, 3C, 302), a short
# code (<=3 chars: E, NB, KKK), or an instruction phrase (CALL ON ARRIVAL, TBC, NA).
# A real word tail (…-MALL) is kept.
_UNIT_TAIL_RE = re.compile(r"^[A-Z]{0,2}\d+[A-Z]{0,3}$")
_TAIL_INSTRUCTION_TOKENS = frozenset({
    "CALL", "TO", "NA", "NB", "TBC", "NULL", "NONE", "NIL", "OOO",
    "OBT", "PENDING", "CONFIRM", "CHECK", "CLIENT", "ON",
})


def extract_place(location_name: str | None) -> str:
    """Pull the human place/landmark out of a coded location_name string.

    e.g. 'NW_RUIRU KAMAKIS_DEEP EAST APARTMENT-37' -> 'RUIRU KAMAKIS DEEP EAST APARTMENT'
         'NWKIAMBU_KIRIGITI_MWANJA APARTMENTS-TBC' -> 'KIAMBU KIRIGITI MWANJA APARTMENTS'
    """
    s = (location_name or "").upper().strip()
    if not s:
        return ""
    # drop the trailing '-<segment>' only when it's a unit/instruction code, not a
    # real word (so '…-37'/'…-CALL ON ARRIVAL' drop but '…-MALL' is kept)
    if "-" in s:
        head, _, tail = s.rpartition("-")
        head, tail = head.strip(), tail.strip()
        first = tail.split()[0] if tail else ""
        if head and (not tail or len(tail) <= 3 or _UNIT_TAIL_RE.match(tail)
                     or first in _TAIL_INSTRUCTION_TOKENS):
            s = head
    s = s.replace("_", " ")
    # strip leading site-code prefixes (may be stacked: ADR MNT KAHAWA…; or glued: NWKIAMBU)
    prev = None
    while prev != s:
        prev = s
        s = _PREFIX_RE.sub("", s).strip()
        s = _GLUED_NW_RE.sub("", s).strip()
    s = _CODE_RE.sub(" ", s)
    s = re.sub(r"\s+", " ", s).strip(" ,-")
    return s


def compose_query(location_name: str | None, cluster: str | None, region: str | None) -> str:
    parts = [p for p in (extract_place(location_name), clean(cluster), clean(region), "Kenya") if p]
    return ", ".join(dict.fromkeys(parts))  # de-dupe while preserving order


def compose_queries(location_name: str | None, cluster: str | None,
                    region: str | None) -> list[str]:
    """Ordered geocode candidates, most → least specific (two-pass estate fallback).

    Building-level location_names (e.g. 'KAHAWA WENDANI ALVO HOUSE') aren't in OSM, so
    the precise query 404s. We then fall back to the estate (leading tokens of the
    place) — each still constrained to the cluster viewbox + distance check by the
    caller, so a coarse hit lands in the right neighbourhood (tighter than the bare
    cluster centroid). We deliberately do NOT add a pure-cluster candidate: that would
    just reproduce the cluster centroid while mislabelling it geo_source='location';
    a truly unmatchable ticket should keep its honest cluster-centroid fallback.
    e.g. 'KAHAWA WENDANI ALVO HOUSE' -> ['KAHAWA WENDANI ALVO HOUSE, WENDANI, nairobi,
             Kenya', 'KAHAWA WENDANI, nairobi, Kenya', 'KAHAWA, nairobi, Kenya']
    """
    region_part, cluster_part = clean(region), clean(cluster)
    place = extract_place(location_name)
    toks = place.split()
    out: list[str] = []

    def add(*parts: str | None) -> None:
        q = ", ".join(dict.fromkeys([p for p in parts if p] + ["Kenya"]))
        if q and q != "Kenya" and q not in out:
            out.append(q)

    add(place, cluster_part, region_part)                 # 1. full precise
    if len(toks) > 2:
        add(" ".join(toks[:2]), region_part)              # 2. estate (leading 2 tokens)
    if len(toks) > 1:
        add(toks[0], region_part)                         # 3. leading token (broad estate)
    return out


# ── keyed geocoder ────────────────────────────────────────────────────────────
def _throttle() -> None:
    global _last_geocode_at
    wait = _GEOCODE_INTERVAL_S - (time.monotonic() - _last_geocode_at)
    if wait > 0:
        time.sleep(wait)
    _last_geocode_at = time.monotonic()


def _haversine_km(lat1: float, lng1: float, lat2: float, lng2: float) -> float:
    dlat, dlng = math.radians(lat2 - lat1), math.radians(lng2 - lng1)
    a = (math.sin(dlat / 2) ** 2
         + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlng / 2) ** 2)
    return 2 * 6371.0 * math.asin(math.sqrt(a))


def geocode(query: str, viewbox: tuple | None = None) -> tuple[float, float, float | None] | None:
    """Forward-geocode via the configured keyed provider. (lat, lng, confidence) | None.

    `viewbox` = (min_lon, min_lat, max_lon, max_lat) constrains results to a box
    around the cluster centroid (bounded), which stops the geocoder matching a
    landmark name in the wrong city.
    """
    if not _API_KEY:
        log.error("GEOCODER_API_KEY is not set — cannot geocode (provider=%s)", _PROVIDER)
        return None
    _throttle()
    try:
        if _PROVIDER == "opencage":
            params = {"key": _API_KEY, "q": query, "limit": 1, "countrycode": "ke", "no_annotations": 1}
            if viewbox:
                params["bounds"] = "%s,%s,%s,%s" % viewbox
            r = requests.get("https://api.opencagedata.com/geocode/v1/json", params=params, timeout=15)
            r.raise_for_status()
            res = (r.json().get("results") or [])
            if res:
                g = res[0]["geometry"]
                return float(g["lat"]), float(g["lng"]), res[0].get("confidence")
        else:  # locationiq (default)
            params = {"key": _API_KEY, "q": query, "format": "json", "limit": 1, "countrycodes": "ke"}
            if viewbox:
                params["viewbox"] = "%s,%s,%s,%s" % viewbox
                params["bounded"] = 1
            r = requests.get("https://us1.locationiq.com/v1/search", params=params, timeout=15)
            if r.status_code == 404:  # LocationIQ returns 404 for "no matches"
                return None
            r.raise_for_status()
            hits = r.json()
            if hits:
                h = hits[0]
                return float(h["lat"]), float(h["lon"]), float(h.get("importance") or 0)
    except (requests.RequestException, ValueError, KeyError) as e:
        log.warning("geocode failed for %r: %s", query, e)
    return None


# ── cluster gazetteer (coarse fallback) ───────────────────────────────────────
def geocode_clusters(apply: bool) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT key, region FROM (
                  SELECT tickets.norm_cluster(raw->>'cluster') AS key,
                         (array_agg(raw->>'region'))[1]        AS region
                  FROM tickets.inc WHERE raw->>'cluster' IS NOT NULL GROUP BY 1
                  UNION
                  SELECT tickets.norm_cluster(raw->>'cluster'),
                         (array_agg(raw->>'region'))[1]
                  FROM tickets.crq WHERE raw->>'cluster' IS NOT NULL GROUP BY 1
                ) z
                WHERE key IS NOT NULL
                  AND NOT EXISTS (SELECT 1 FROM tickets.geo_clusters g
                                  WHERE g.cluster_key = z.key AND g.geom IS NOT NULL)
                """
            )
            todo = cur.fetchall()
    log.info("%d clusters to geocode", len(todo))
    if not apply:
        for key, region in todo:
            log.info("  would geocode cluster: %s (%s)", key, region)
        return
    written = 0
    for key, region in todo:
        hit = geocode(f"{key}, {region}, Kenya" if region else f"{key}, Kenya")
        if not hit:
            continue
        lat, lng, _ = hit
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """INSERT INTO tickets.geo_clusters (cluster_key, region, lat, lng, source, verified)
                       VALUES (%s, %s, %s, %s, %s, false)
                       ON CONFLICT (cluster_key) DO UPDATE
                         SET region = EXCLUDED.region, lat = EXCLUDED.lat,
                             lng = EXCLUDED.lng, source = EXCLUDED.source""",
                    (key, region, lat, lng, _PROVIDER),
                )
        written += 1
    _resolve()
    log.info("gazetteer: %d clusters written (unverified — review tickets.geo_clusters)", written)


# ── per-location geocoding (precise; actionable INC) ──────────────────────────
# A location geocode is only trusted if it lands within this radius of the
# cluster centroid; otherwise the geocoder matched the landmark in the wrong
# place and we fall back to the cluster centroid.
_MAX_KM_FROM_CLUSTER = float(os.getenv("GEOCODER_MAX_KM", "25"))
_VIEWBOX_DEG = 0.2  # ~22 km half-box around the cluster centroid


def geocode_locations(apply: bool) -> None:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute(
                """
                SELECT t.key, t.location_name, t.cluster, t.region, gc.lat AS clat, gc.lng AS clng
                FROM (
                  SELECT tickets.norm_cluster(raw->>'location_name')  AS key,
                         (array_agg(raw->>'location_name'))[1]        AS location_name,
                         (array_agg(raw->>'cluster'))[1]              AS cluster,
                         (array_agg(raw->>'region'))[1]               AS region,
                         tickets.norm_cluster((array_agg(raw->>'cluster'))[1]) AS ckey
                  FROM tickets.inc
                  WHERE (raw->>'is_actionable')::boolean
                    AND raw->>'location_name' IS NOT NULL
                    AND tickets.norm_cluster(raw->>'location_name') IS NOT NULL
                    AND NOT EXISTS (SELECT 1 FROM tickets.geo_locations gl
                                    WHERE gl.query_key = tickets.norm_cluster(raw->>'location_name')
                                      AND gl.geom IS NOT NULL)
                  GROUP BY 1
                ) t
                LEFT JOIN tickets.geo_clusters gc ON gc.cluster_key = t.ckey
                """
            )
            todo = cur.fetchall()
    log.info("%d actionable-INC locations to geocode (provider=%s)", len(todo), _PROVIDER)
    if not apply:
        for key, loc, cluster, region, clat, clng in todo[:50]:
            log.info("  %s  ->  %s", key, "  |  ".join(compose_queries(loc, cluster, region)))
        return
    written = missed = coarse = 0
    for key, loc, cluster, region, clat, clng in todo:
        viewbox = None
        if clat is not None and clng is not None:
            viewbox = (clng - _VIEWBOX_DEG, clat - _VIEWBOX_DEG, clng + _VIEWBOX_DEG, clat + _VIEWBOX_DEG)
        # two-pass: precise → estate → cluster; accept the FIRST in-range hit. A wrong-area
        # match (> MAX_KM from the cluster centroid) is skipped so we try a coarser query.
        hit = used = None
        for i, cand in enumerate(compose_queries(loc, cluster, region)):
            g = geocode(cand, viewbox)
            if not g:
                continue
            lat, lng, conf = g
            if (clat is not None and clng is not None
                    and _haversine_km(lat, lng, clat, clng) > _MAX_KM_FROM_CLUSTER):
                continue
            hit, used = g, cand
            if i > 0:
                coarse += 1
            break
        if not hit:
            missed += 1   # no match even coarsely — keeps cluster-centroid fallback
            continue
        lat, lng, conf = hit
        with get_conn() as conn:
            with conn.cursor() as cur:
                cur.execute(
                    """INSERT INTO tickets.geo_locations
                         (query_key, location_name, cluster, region, query, lat, lng, confidence, provider)
                       VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
                       ON CONFLICT (query_key) DO UPDATE
                         SET location_name = EXCLUDED.location_name, cluster = EXCLUDED.cluster,
                             region = EXCLUDED.region, query = EXCLUDED.query, lat = EXCLUDED.lat,
                             lng = EXCLUDED.lng, confidence = EXCLUDED.confidence, provider = EXCLUDED.provider""",
                    (key, loc, cluster, region, used, lat, lng, conf, _PROVIDER),
                )
        written += 1
        log.info("  geocoded %s -> %.5f, %.5f", used, lat, lng)
    n = _resolve()
    log.info("locations: %d accepted (%d via estate/cluster fallback), %d unmatched; "
             "re-resolved geom on %d tickets (unverified — review tickets.geo_locations)",
             written, coarse, missed, n)


def _resolve() -> int:
    with get_conn() as conn:
        with conn.cursor() as cur:
            cur.execute("SELECT tickets.resolve_ticket_geoms()")
            return cur.fetchone()[0]


# ── entrypoint ────────────────────────────────────────────────────────────────
def main() -> None:
    ap = argparse.ArgumentParser(description="Ingest INC tickets from CSV (raw-first) + geocode")
    ap.add_argument("--apply", action="store_true", help="Write to DB (default: dry-run)")
    ap.add_argument("--from-bucket", action="store_true",
                    help="Ingest the newest INC csv from the rustfs tickets bucket (aws CLI); "
                         "skips if unchanged (ETag) and archives processed files")
    ap.add_argument("--inc-csv", default=None, help="Local INC tickets CSV file (dev)")
    ap.add_argument("--geocode-clusters", action="store_true",
                    help="Geocode distinct clusters into the gazetteer, then re-resolve geoms")
    ap.add_argument("--geocode-locations", action="store_true",
                    help="Geocode actionable-INC location_names precisely (keyed provider), then re-resolve")
    ap.add_argument("--capture-history", action="store_true",
                    help="Run tickets.capture_history() standalone (closure_events + daily snapshot)")
    args = ap.parse_args()

    if args.geocode_clusters:
        geocode_clusters(apply=args.apply)
        return
    if args.geocode_locations:
        geocode_locations(apply=args.apply)
        return
    if args.capture_history:
        _capture_history()
        return
    if not (args.from_bucket or args.inc_csv):
        ap.error("provide --from-bucket, --inc-csv, --geocode-clusters, --geocode-locations, or --capture-history")
    ingest(args)


if __name__ == "__main__":
    main()