server-26/drb-c2-core/app/internal/area_context.py

"""
Area context — the ground truth an operator sets about where a channel operates.

One shape, used at two scopes (server-26#36):

    area_context: {
      municipality?, county?, state?,
      center?, radius_km?, resolved_from?, resolved_at?,   # backend-written
      local_knowledge?: [ { term, meaning } ]
    }

WHY EVERY FIELD IS NULLABLE. The system level is only meaningful when it is true
of *every* talkgroup on that system. White Plains PD — it is, so an operator
fills it once and every talkgroup inherits. A statewide Colorado system — it is
not, so they leave it null and fill each talkgroup. Which level someone fills IS
their declaration of how homogeneous the system is, which is what lets one
schema serve both without a `system_type` flag to get out of sync.

WHY `local_knowledge` REPLACED `roads[]`/`landmarks[]`. Radio traffic references
intersections, schools, housing developments, rail stations and nicknames ("the
flats"), none of which fit two lists. And a bare term is half the information:
`11-X-ray` tells a corrector nothing, `11-X-ray — MTA PD patrol unit` is what
lets it recognise the sound.

WHY THE ANCHOR CAN BE ABSENT ON PURPOSE. `center`/`radius_km` exist so a
geocoded place name can be sanity-checked against the area the channel actually
covers (server-26#37). If municipality/county/state only resolve to something as
wide as a state, that check would confirm anything inside it while appearing to
work — worse than useless. So an anchor wider than
`settings.area_anchor_max_radius_km` is not written at all, and an absent anchor
means SKIP THE CHECK, never "accept anything".

THE CLIENT DOES NOT WRITE THE DERIVED FIELDS. `center`, `radius_km`,
`resolved_from` and `resolved_at` are computed here and merged in by the server.
Taking them from the request body is the same defect as the `ten_codes` wipe
fixed in 58efdbd: the frontend does not decide what is in a system document.
"""

import asyncio
import math
from datetime import datetime, timezone
from typing import Any, Optional

from app.config import settings
from app.internal.logger import logger

# Fields an operator sets. Anything else in an incoming body is dropped.
CLIENT_FIELDS = ("municipality", "county", "state", "local_knowledge")
# Fields this module owns. Carried forward from the stored document on every
# write, never read from the request.
SERVER_FIELDS = ("center", "radius_km", "resolved_from", "resolved_at")
# The three that identify a place, in the order they are geocoded.
PLACE_FIELDS = ("municipality", "county", "state")

_anchor_cache: dict[str, Optional[dict]] = {}


def geo_dist_km(lat1: float, lon1: float, lat2: float, lon2: float) -> float:
    """Haversine distance in km between two lat/lon points."""
    R = 6371.0
    dlat = math.radians(lat2 - lat1)
    dlon = math.radians(lon2 - lon1)
    a = (
        math.sin(dlat / 2) ** 2
        + math.cos(math.radians(lat1)) * math.cos(math.radians(lat2)) * math.sin(dlon / 2) ** 2
    )
    return R * 2 * math.asin(math.sqrt(a))


# -- Normalisation -------------------------------------------------------------

def normalize_local_knowledge(raw: Any) -> list[dict]:
    """
    Coerce whatever arrived into [{term, meaning}], dropping junk.

    Accepts a bare string list too — that is what `roads[]`/`landmarks[]` and the
    old flat `vocabulary` look like, and a term with no meaning is still worth
    having in the reference list.
    """
    if not isinstance(raw, list):
        return []
    out: list[dict] = []
    seen: set[str] = set()
    for item in raw:
        if isinstance(item, str):
            term, meaning = item.strip(), None
        elif isinstance(item, dict):
            term = str(item.get("term") or "").strip()
            meaning = str(item.get("meaning") or "").strip() or None
        else:
            continue
        key = term.lower()
        if not term or key in seen:
            continue
        seen.add(key)
        out.append({"term": term, "meaning": meaning} if meaning else {"term": term})
    return out


def knowledge_of(area: Optional[dict]) -> list[dict]:
    """
    This scope's local knowledge, folding the pre-#36 shape forward.

    `roads[]` and `landmarks[]` were the original fields and real systems still
    have them stored. Reading them as bare terms means the corrector keeps the
    ground truth an operator already entered instead of silently losing it the
    day this shipped; they disappear from the document the next time that scope
    is saved.
    """
    area = area or {}
    legacy = list(area.get("roads") or []) + list(area.get("landmarks") or [])
    return normalize_local_knowledge(list(area.get("local_knowledge") or []) + legacy)


def normalize(raw: Any) -> dict:
    """Client-supplied area_context -> the stored shape, server fields excluded."""
    if not isinstance(raw, dict):
        return {}
    out: dict[str, Any] = {}
    for field in PLACE_FIELDS:
        value = raw.get(field)
        if isinstance(value, str) and value.strip():
            out[field] = value.strip()
    knowledge = knowledge_of(raw)
    if knowledge:
        out["local_knowledge"] = knowledge
    return out


def merge_server_fields(incoming: dict, existing: Optional[dict]) -> dict:
    """Carry the backend-owned anchor forward across a client write."""
    out = dict(incoming)
    for field in SERVER_FIELDS:
        if existing and existing.get(field) is not None:
            out[field] = existing[field]
    return out


def merge_config(incoming: Any, existing: Optional[dict]) -> Any:
    """
    Reconcile a client-sent config blob with what the server already owns.

    The systems form sends `config.talkgroups[]` in full, so writing it verbatim
    destroys everything the backend put there — the resolved anchor and the
    pending term queue. That is the same defect as the `ten_codes` wipe fixed in
    58efdbd, and the same fix applies: the backend merges its own fields back in
    rather than taking dictation from the frontend.
    """
    if not isinstance(incoming, dict):
        return incoming
    incoming_tgs = incoming.get("talkgroups")
    if not isinstance(incoming_tgs, list):
        return incoming

    by_id: dict[int, dict] = {}
    for tg in ((existing or {}).get("talkgroups") or []):
        if isinstance(tg, dict):
            try:
                by_id[int(tg.get("id", -1))] = tg
            except (TypeError, ValueError):
                continue

    merged: list[Any] = []
    for tg in incoming_tgs:
        if not isinstance(tg, dict):
            merged.append(tg)
            continue
        try:
            prior = by_id.get(int(tg.get("id", -1))) or {}
        except (TypeError, ValueError):
            prior = {}
        out = dict(tg)
        area = normalize(tg.get("area_context"))
        prior_area = prior.get("area_context") or {}
        if area:
            out["area_context"] = merge_server_fields(area, prior_area)
        else:
            out.pop("area_context", None)
        if prior.get(PENDING_KEY):
            out[PENDING_KEY] = prior[PENDING_KEY]
        merged.append(out)

    return {**incoming, "talkgroups": merged}


# -- Scope resolution ----------------------------------------------------------

def effective(system_area: Optional[dict], tg_area: Optional[dict]) -> dict:
    """
    Merge the two scopes: talkgroup wins where set, system fills the gaps.

    `local_knowledge` concatenates rather than replaces, talkgroup entries first
    so they survive any downstream truncation and outrank a system entry for the
    same term. A multi-county system whose talkgroup covers one municipality must
    not have that municipality's terms buried under a county-wide list.
    """
    system_area = system_area or {}
    tg_area = tg_area or {}
    out: dict[str, Any] = {}
    for field in PLACE_FIELDS:
        value = tg_area.get(field) or system_area.get(field)
        if value:
            out[field] = value
    knowledge = normalize_local_knowledge(knowledge_of(tg_area) + knowledge_of(system_area))
    if knowledge:
        out["local_knowledge"] = knowledge
    return out


def talkgroup_entry(system_doc: Optional[dict], talkgroup_id: Any) -> dict:
    """The `config.talkgroups[]` entry for this talkgroup, or `{}`."""
    if not system_doc or talkgroup_id is None:
        return {}
    talkgroups = (system_doc.get("config") or {}).get("talkgroups") or []
    idx = _tg_index(talkgroups, talkgroup_id)
    return talkgroups[idx] if idx >= 0 else {}


def anchor_key(area: Optional[dict]) -> str:
    """
    Stable identity of the place an anchor was resolved from.

    Stored as `resolved_from`, which is what makes "did this actually change?"
    decidable — so the geocode happens when someone edits a town name, not on
    every read or every five minutes.
    """
    area = area or {}
    return "|".join((area.get(f) or "").strip().lower() for f in PLACE_FIELDS)


def has_place(area: Optional[dict]) -> bool:
    return bool(anchor_key(area).strip("|"))


def anchor_for(system_area: Optional[dict], tg_area: Optional[dict]) -> Optional[dict]:
    """
    The anchor to sanity-check geocoded locations against, or None.

    None has one meaning and it is load-bearing: SKIP THE CHECK. It covers an
    unconfigured system, an area too wide to discriminate, and a stored anchor
    whose `resolved_from` no longer matches the place it was computed for (an
    edit landed and the refresh has not run). Accepting a stale or oversized
    anchor would rubber-stamp locations while looking like verification.
    """
    key = anchor_key(effective(system_area, tg_area))
    for area in (tg_area, system_area):
        if not area:
            continue
        center, radius = area.get("center"), area.get("radius_km")
        if area.get("resolved_from") == key and center and radius:
            try:
                return {
                    "lat": float(center["lat"]),
                    "lng": float(center["lng"]),
                    "radius_km": float(radius),
                }
            except (KeyError, TypeError, ValueError):
                continue
    return None


# -- Anchor geocoding ----------------------------------------------------------

def _query(area: dict) -> str:
    return ", ".join(area[f] for f in PLACE_FIELDS if area.get(f))


async def resolve_anchor(area: dict) -> Optional[dict]:
    """
    Geocode municipality/county/state into {center, radius_km}, or None.

    The radius comes from the result's own viewport — half its diagonal — so a
    village anchors tightly and a county loosely, which is the real difference
    we care about. Anything wider than `area_anchor_max_radius_km` is discarded
    rather than stored: see the module docstring.
    """
    if not has_place(area):
        return None
    query = _query(area)
    if query in _anchor_cache:
        return _anchor_cache[query]
    if not settings.google_maps_api_key:
        logger.warning("GOOGLE_MAPS_API_KEY not set — area anchors cannot be resolved")
        return None

    import httpx

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.get(
                "https://maps.googleapis.com/maps/api/geocode/json",
                params={"address": query, "region": "us", "key": settings.google_maps_api_key},
            )
            r.raise_for_status()
            data = r.json()
        if data.get("status") != "OK" or not data.get("results"):
            logger.warning(f"Area anchor: {query!r} did not geocode ({data.get('status')})")
            _anchor_cache[query] = None
            return None
        geometry = data["results"][0].get("geometry") or {}
        loc = geometry.get("location") or {}
        lat, lng = float(loc["lat"]), float(loc["lng"])
        viewport = geometry.get("viewport") or {}
        ne, sw = viewport.get("northeast"), viewport.get("southwest")
        if ne and sw:
            radius_km = geo_dist_km(sw["lat"], sw["lng"], ne["lat"], ne["lng"]) / 2
        else:
            radius_km = settings.geocode_max_km
    except Exception as e:
        logger.warning(f"Area anchor geocoding failed for {query!r}: {e}")
        return None  # not cached — a transient failure should be retried

    if radius_km > settings.area_anchor_max_radius_km:
        logger.info(
            f"Area anchor: {query!r} spans ~{radius_km:.0f}km, wider than "
            f"area_anchor_max_radius_km={settings.area_anchor_max_radius_km} — storing no "
            f"anchor, so verification skips rather than rubber-stamps"
        )
        _anchor_cache[query] = None
        return None

    anchor = {
        "center": {"lat": lat, "lng": lng},
        "radius_km": round(radius_km, 2),
        "resolved_from": anchor_key(area),
        "resolved_at": datetime.now(timezone.utc).isoformat(),
    }
    _anchor_cache[query] = anchor
    logger.info(f"Area anchor: {query!r} -> ({lat:.4f}, {lng:.4f}) r={radius_km:.1f}km")
    return anchor


def _apply(area: dict, anchor: Optional[dict], key: str) -> dict:
    """Write (or clear) the derived fields on one scope's area_context."""
    out = {k: v for k, v in area.items() if k not in SERVER_FIELDS}
    if anchor:
        out.update(anchor)
    elif key.strip("|"):
        # A place is set but produced no usable anchor. Record that we tried, so
        # the next write does not geocode it again for the same answer.
        out["resolved_from"] = key
        out["resolved_at"] = datetime.now(timezone.utc).isoformat()
    return out


async def refresh_anchors(system_doc: dict) -> dict:
    """
    Recompute anchors for a system and every talkgroup that sets a place.

    Returns a Firestore patch — `{}` when nothing needed resolving. Talkgroups
    are refreshed alongside the system because a talkgroup's anchor is derived
    from its EFFECTIVE place (its own fields over the system's), so editing the
    system's county silently changes what every talkgroup should be anchored to.

    Only scopes whose `resolved_from` no longer matches are geocoded, and the
    per-query cache means N talkgroups in one town cost one request.
    """
    system_area = dict(system_doc.get("area_context") or {})
    patch: dict[str, Any] = {}

    system_key = anchor_key(system_area)
    if system_area.get("resolved_from") != system_key:
        anchor = await resolve_anchor(system_area) if has_place(system_area) else None
        patch["area_context"] = _apply(system_area, anchor, system_key)
        system_area = patch["area_context"]

    config = system_doc.get("config") or {}
    talkgroups = config.get("talkgroups")
    if not isinstance(talkgroups, list):
        return patch

    updated: list[dict] = []
    changed = False
    for tg in talkgroups:
        if not isinstance(tg, dict):
            updated.append(tg)
            continue
        tg_area = tg.get("area_context") or {}
        # No place of its own means it inherits the system's anchor wholesale —
        # nothing to store here, and anchor_for() falls back to the system.
        if not has_place(tg_area):
            if any(tg_area.get(f) is not None for f in SERVER_FIELDS):
                tg = {**tg, "area_context": {k: v for k, v in tg_area.items() if k not in SERVER_FIELDS}}
                changed = True
            updated.append(tg)
            continue
        key = anchor_key(effective(system_area, tg_area))
        if tg_area.get("resolved_from") == key:
            updated.append(tg)
            continue
        anchor = await resolve_anchor(effective(system_area, tg_area))
        updated.append({**tg, "area_context": _apply(tg_area, anchor, key)})
        changed = True

    if changed:
        patch["config"] = {**config, "talkgroups": updated}
    return patch


# -- Talkgroup-level pending terms ---------------------------------------------
#
# Proposals land on the TALKGROUP and are never promoted to the system
# automatically (server-26#37). The argument is blast radius: a wrong term on a
# talkgroup misleads one channel, the same term at system level misleads every
# channel on that system — including one 400km away on a statewide system, which
# is exactly the context poisoning the scope rule exists to prevent. If a term
# genuinely applies system-wide, carrying it on several talkgroups costs almost
# nothing; auto-promoting a wrong one is expensive to notice.

PENDING_KEY = "local_knowledge_pending"


def _tg_index(talkgroups: list, talkgroup_id: Any) -> int:
    try:
        wanted = int(talkgroup_id)
    except (TypeError, ValueError):
        return -1
    for i, tg in enumerate(talkgroups):
        if not isinstance(tg, dict):
            continue
        try:
            if int(tg.get("id", -1)) == wanted:
                return i
        except (TypeError, ValueError):
            continue
    return -1


def _known_terms(tg: dict, system_doc: dict) -> set[str]:
    """Everything this talkgroup already knows, at either scope, plus pending."""
    known = {
        e["term"].lower()
        for e in effective(system_doc.get("area_context"), tg.get("area_context"))
        .get("local_knowledge", [])
    }
    known |= {str(t).lower() for t in (tg.get("vocabulary") or [])}
    known |= {str(t).lower() for t in (system_doc.get("vocabulary") or [])}
    known |= {str(p.get("term", "")).lower() for p in (tg.get(PENDING_KEY) or [])}
    return known


async def add_pending(system_id: str, talkgroup_id: Any, entries: list[dict]) -> int:
    """
    Queue proposed {term, meaning} entries on one talkgroup for human review.

    Returns how many were actually queued. Nothing here writes to
    `local_knowledge` — approval is a person's decision, always.
    """
    from app.internal import firestore as fstore

    if not system_id or talkgroup_id is None or not entries:
        return 0
    system_doc = await fstore.doc_get("systems", system_id)
    if not system_doc:
        return 0
    config = dict(system_doc.get("config") or {})
    talkgroups = list(config.get("talkgroups") or [])
    idx = _tg_index(talkgroups, talkgroup_id)
    if idx < 0:
        return 0

    tg = dict(talkgroups[idx])
    known = _known_terms(tg, system_doc)
    now = datetime.now(timezone.utc).isoformat()
    queued: list[dict] = []
    for entry in entries:
        term = str(entry.get("term") or "").strip()
        if not term or term.lower() in known:
            continue
        known.add(term.lower())
        queued.append({
            "term": term,
            "meaning": entry.get("meaning") or None,
            "source": entry.get("source") or "verifier",
            "added_at": now,
            "source_call_ids": entry.get("source_call_ids") or [],
        })
    if not queued:
        return 0

    tg[PENDING_KEY] = list(tg.get(PENDING_KEY) or []) + queued
    talkgroups[idx] = tg
    config["talkgroups"] = talkgroups
    await fstore.doc_update("systems", system_id, {"config": config})
    logger.info(
        f"Local knowledge: {len(queued)} term(s) proposed for talkgroup "
        f"{talkgroup_id} on system {system_id}: {[q['term'] for q in queued]}"
    )
    return len(queued)


async def resolve_pending(system_id: str, talkgroup_id: Any, term: str, approve: bool) -> bool:
    """Approve a pending term onto the talkgroup, or dismiss it. Never promotes."""
    from app.internal import firestore as fstore

    system_doc = await fstore.doc_get("systems", system_id)
    if not system_doc:
        return False
    config = dict(system_doc.get("config") or {})
    talkgroups = list(config.get("talkgroups") or [])
    idx = _tg_index(talkgroups, talkgroup_id)
    if idx < 0:
        return False

    tg = dict(talkgroups[idx])
    pending = list(tg.get(PENDING_KEY) or [])
    match = next((p for p in pending if str(p.get("term", "")).lower() == term.lower()), None)
    if match is None:
        return False
    tg[PENDING_KEY] = [p for p in pending if p is not match]
    if approve:
        area = dict(tg.get("area_context") or {})
        area["local_knowledge"] = normalize_local_knowledge(
            list(area.get("local_knowledge") or [])
            + [{"term": match["term"], "meaning": match.get("meaning")}]
        )
        tg["area_context"] = area
    talkgroups[idx] = tg
    config["talkgroups"] = talkgroups
    await fstore.doc_update("systems", system_id, {"config": config})
    return True


async def refresh_anchors_bg(system_id: str) -> None:
    """Fire-and-forget refresh, for callers that must not block on Maps."""
    from app.internal import firestore as fstore

    try:
        doc = await fstore.doc_get("systems", system_id)
        if not doc:
            return
        patch = await refresh_anchors(doc)
        if patch:
            await fstore.doc_update("systems", system_id, patch)
    except Exception as e:
        logger.warning(f"Area anchor refresh failed for system {system_id}: {e}")


def schedule_refresh(system_id: str) -> None:
    """Kick a refresh without making the caller wait for the geocoder."""
    try:
        asyncio.get_running_loop().create_task(refresh_anchors_bg(system_id))
    except RuntimeError:  # no loop (tests, scripts) — nothing to schedule
        pass