""" 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