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

"""
Place verification — is the name the corrector produced a real place *here*?

The transcript corrector (`transcript_correction.py`) substitutes sound-alikes
against a reference list. It has no way to tell whether its own output is a real
place, so "Cool Parts, Illinois" and "Shout out to Optum" are exactly as
acceptable to it as a genuine street name. This module is the check
(server-26#37).

MAPS AS A VERIFIER, NOT AS PROMPT STUFFING. Injecting every road and POI in a
town would be hundreds of names on a pass that runs on every transcribed call.
Instead we take the handful of location-shaped nouns a transcript actually
contains and ask one question per noun:

  1. Geocode it, bounded by the talkgroup's anchor.
  2. Inside the radius -> accept, done.
  3. Outside, or no result -> look for a sound-alike that DOES resolve inside.
  4. Found one -> correct to it, and propose {term, meaning} to that
     talkgroup's local_knowledge as pending.

Cost scales with location nouns, not call volume, and every verified miss
permanently improves the reference data for that channel.

NO ANCHOR MEANS SKIP, NOT ACCEPT. An anchor too wide to discriminate is not
stored at all (see `area_context`), and without one this module returns
immediately. A statewide radius would confirm anything inside it, which is worse
than not checking — it looks like verification and is not.

THE FREE TIER RUNS FIRST. A sound-alike among the terms the operator already
entered costs nothing and is more trustworthy than anything Maps guesses, so
`local_knowledge` and `vocabulary` are searched before any request goes out.
"""

import re
from difflib import SequenceMatcher
from typing import Any, Optional

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

# Soundex-style consonant classes. Letters that a vocoder + Whisper routinely
# swap land in the same bucket, so "Optum"/"Ossining" stay far apart while
# "Snowden"/"Snowdon" collapse together.
_CLASSES = {
    "b": "1", "f": "1", "p": "1", "v": "1",
    "c": "2", "g": "2", "j": "2", "k": "2", "q": "2", "s": "2", "x": "2", "z": "2",
    "d": "3", "t": "3",
    "l": "4",
    "m": "5", "n": "5",
    "r": "6",
}
_DIGRAPHS = (("ph", "f"), ("gh", "g"), ("ck", "k"), ("wr", "r"), ("kn", "n"), ("wh", "w"))


def _norm(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", " ", (text or "").lower()).strip()


def phonetic_key(text: str) -> str:
    """
    Consonant-class skeleton of a name. Vowels drop out; a run of the same class
    collapses unless a vowel separates it.
    """
    letters = re.sub(r"[^a-z]", "", (text or "").lower())
    for a, b in _DIGRAPHS:
        letters = letters.replace(a, b)
    out: list[str] = []
    prev = ""
    for ch in letters:
        code = _CLASSES.get(ch, "")
        if code and code != prev:
            out.append(code)
        prev = code if ch not in "aeiouyhw" else ""
    return "".join(out)


def sounds_like(heard: str, candidate: str) -> float:
    """
    0..1 similarity, the better of the phonetic and the literal comparison.

    Both are needed: Whisper errors are sometimes phonetic ("5 acre" for
    "5-baker") and sometimes near-spellings ("Croton Ave" for "Croton Avenue"),
    and a key comparison alone scores the second one poorly.
    """
    literal = SequenceMatcher(None, _norm(heard), _norm(candidate)).ratio()
    ka, kb = phonetic_key(heard), phonetic_key(candidate)
    phonetic = SequenceMatcher(None, ka, kb).ratio() if ka and kb else 0.0
    return max(literal, phonetic)


# -- Maps ----------------------------------------------------------------------

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


async def _geocode_in_anchor(query: str, anchor: dict) -> Optional[dict]:
    """Geocode `query` and return its coords only if they land inside the anchor."""
    from app.internal.intelligence import _geocode_location

    coords = await _geocode_location(query, anchor=anchor)
    return coords


async def _places_soundalike(heard: str, anchor: dict) -> Optional[dict]:
    """
    Ask Maps for places near the anchor matching the misheard text.

    Places Text Search does its own fuzzy matching against a biased region, which
    is usually enough — but "usually" is not a standard, so the result still has
    to pass `sounds_like` before it is allowed to rewrite a transcript. Without
    that guard the API happily returns the nearest gas station for any garbage
    string.
    """
    if not settings.google_maps_api_key:
        return None

    import httpx

    try:
        async with httpx.AsyncClient(timeout=5.0) as client:
            r = await client.get(
                "https://maps.googleapis.com/maps/api/place/textsearch/json",
                params={
                    "query": heard,
                    "location": f"{anchor['lat']},{anchor['lng']}",
                    "radius": int(anchor["radius_km"] * 1000),
                    "key": settings.google_maps_api_key,
                },
            )
            r.raise_for_status()
            data = r.json()
    except Exception as e:
        logger.warning(f"Place search failed for {heard!r}: {e}")
        return None

    if data.get("status") not in ("OK", "ZERO_RESULTS"):
        logger.warning(f"Place search for {heard!r} returned {data.get('status')}")
        return None

    for result in (data.get("results") or [])[:5]:
        name = (result.get("name") or "").strip()
        loc = (result.get("geometry") or {}).get("location") or {}
        if not name or "lat" not in loc:
            continue
        distance = area_context.geo_dist_km(
            anchor["lat"], anchor["lng"], float(loc["lat"]), float(loc["lng"])
        )
        if distance > anchor["radius_km"]:
            continue
        score = sounds_like(heard, name)
        if score >= settings.place_soundalike_min_ratio:
            return {"term": name, "meaning": result.get("formatted_address") or None, "score": score}
    return None


def _known_soundalike(heard: str, area: dict) -> Optional[dict]:
    """Best sound-alike among terms the operator already entered. Free."""
    best: Optional[dict] = None
    for entry in area.get("local_knowledge") or []:
        term = entry.get("term") or ""
        if not term or _norm(term) == _norm(heard):
            continue
        score = sounds_like(heard, term)
        if score >= settings.place_soundalike_min_ratio and (best is None or score > best["score"]):
            best = {"term": term, "meaning": entry.get("meaning"), "score": score, "known": True}
    return best


# -- Public --------------------------------------------------------------------

def _substitute(text: str, swaps: list[tuple[str, str]]) -> str:
    for heard, replacement in swaps:
        text = re.sub(rf"\b{re.escape(heard)}\b", replacement, text, flags=re.IGNORECASE)
    return text


async def verify(
    call_id: str,
    text: str,
    segments: Optional[list[dict]],
    locations: list[str],
    system_area: Optional[dict],
    tg_area: Optional[dict],
    system_id: Optional[str] = None,
    talkgroup_id: Optional[Any] = None,
) -> tuple[Optional[str], Optional[list[dict]]]:
    """
    Check the corrector's location nouns against the talkgroup's anchor.

    Returns (text, segments) with verified substitutions applied, or (None, None)
    when nothing changed. Like correction itself, this is an improvement and
    never a dependency: any failure leaves the transcript exactly as it was.
    """
    if not settings.place_verification_enabled or not locations:
        return None, None

    anchor = area_context.anchor_for(system_area, tg_area)
    if not anchor:
        return None, None  # load-bearing: no anchor means skip, never accept

    area = area_context.effective(system_area, tg_area)
    suffix = _place_suffix(area)
    swaps: list[tuple[str, str]] = []
    proposals: list[dict] = []

    for heard in locations[: settings.place_verify_max_per_call]:
        heard = (heard or "").strip()
        if not heard:
            continue
        query = f"{heard}, {suffix}" if suffix else heard
        try:
            if await _geocode_in_anchor(query, anchor):
                continue  # real place, in the right area — nothing to do
            candidate = _known_soundalike(heard, area) or await _places_soundalike(heard, anchor)
        except Exception as e:
            logger.warning(f"Place verification failed for {heard!r} on call {call_id}: {e}")
            continue
        if not candidate:
            logger.info(
                f"Place verification: {heard!r} (call {call_id}) does not resolve near the "
                f"anchor and has no sound-alike that does — leaving it alone"
            )
            continue
        swaps.append((heard, candidate["term"]))
        if not candidate.get("known"):
            proposals.append({
                "term": candidate["term"],
                "meaning": candidate.get("meaning"),
                "source": "place_verifier",
                "source_call_ids": [call_id],
            })
        logger.info(
            f"Place verification: {heard!r} -> {candidate['term']!r} "
            f"(score {candidate['score']:.2f}, call {call_id})"
        )

    if not swaps:
        return None, None

    if proposals and system_id and talkgroup_id is not None:
        try:
            await area_context.add_pending(system_id, talkgroup_id, proposals)
        except Exception as e:
            logger.warning(f"Could not queue verified terms for call {call_id}: {e}")

    new_text = _substitute(text or "", swaps)
    new_segments = None
    if segments:
        new_segments = [{**s, "text": _substitute(s.get("text", ""), swaps)} for s in segments]
        if all(a["text"] == b.get("text") for a, b in zip(new_segments, segments)):
            new_segments = None
    return (new_text if new_text != (text or "") else None), new_segments