324 lines
12 KiB
Python
324 lines
12 KiB
Python
"""Resolver that matches a query against a local JSON registry.
|
|
|
|
Registry schema (a single JSON file):
|
|
|
|
{
|
|
"locations": [
|
|
{
|
|
"canonical_name": "...",
|
|
"aliases": ["...", "..."],
|
|
"operator": "...",
|
|
"site": "...",
|
|
"city": "...",
|
|
"country": "...",
|
|
"region": "...",
|
|
"latitude": 0.0,
|
|
"longitude": 0.0,
|
|
"precision": "precise" | "site" | "city",
|
|
"confidence": 0.0,
|
|
"verification_status": "verified",
|
|
"source_note": "...",
|
|
"verified_at": "YYYY-MM-DD"
|
|
}
|
|
],
|
|
"city_fallbacks": [ {city, country, latitude, longitude, ...} ]
|
|
}
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from functools import lru_cache
|
|
from pathlib import Path
|
|
from typing import Any, Callable, Iterable
|
|
|
|
from ..models import (
|
|
RENDERABLE_PRECISIONS,
|
|
LocationCandidate,
|
|
LocationQuery,
|
|
ResolverOutput,
|
|
)
|
|
from ..text import (
|
|
city_key,
|
|
normalize_country_text,
|
|
normalize_text,
|
|
parse_float,
|
|
)
|
|
|
|
# Field-priority weights when scoring "this query field text contains this
|
|
# alias text". Tuned to match the legacy compute-center ordering — name beats
|
|
# site beats operator beats city — which generalizes well to other domains.
|
|
_DEFAULT_FIELD_PRIORITY = {
|
|
"name": 8,
|
|
"site": 6,
|
|
"operator": 5,
|
|
"city": 3,
|
|
}
|
|
|
|
|
|
def default_score_alias_match(
|
|
alias_field: str, record_field: str, alias_text: str
|
|
) -> int:
|
|
score = max(0, len(alias_text))
|
|
score += _DEFAULT_FIELD_PRIORITY.get(alias_field, 1)
|
|
if alias_field == record_field:
|
|
score += 4
|
|
if alias_field == "name" and record_field in {"name", "name_short", "alias"}:
|
|
score += 6
|
|
if alias_field == "site" and record_field in {"site", "organization"}:
|
|
score += 4
|
|
if alias_field == "operator" and record_field in {"operator", "organization"}:
|
|
score += 4
|
|
return score
|
|
|
|
|
|
@lru_cache(maxsize=32)
|
|
def _load_registry_file(path: str) -> dict[str, Any]:
|
|
with Path(path).open("r", encoding="utf-8") as handle:
|
|
return json.load(handle)
|
|
|
|
|
|
@lru_cache(maxsize=32)
|
|
def _build_alias_index(
|
|
path: str,
|
|
) -> tuple[tuple[dict[str, Any], tuple[tuple[str, str], ...]], ...]:
|
|
index: list[tuple[dict[str, Any], tuple[tuple[str, str], ...]]] = []
|
|
for entry in _load_registry_file(path).get("locations", []):
|
|
aliases: list[tuple[str, str]] = []
|
|
seen: set[str] = set()
|
|
for alias in [entry.get("canonical_name"), *(entry.get("aliases") or [])]:
|
|
normalized = normalize_text(alias)
|
|
if normalized and normalized not in seen:
|
|
aliases.append(("name", normalized))
|
|
seen.add(normalized)
|
|
for field_name in ("operator", "site", "city"):
|
|
value = entry.get(field_name)
|
|
normalized = normalize_text(value)
|
|
if normalized and normalized not in seen:
|
|
aliases.append((field_name, normalized))
|
|
seen.add(normalized)
|
|
index.append((entry, tuple(aliases)))
|
|
return tuple(index)
|
|
|
|
|
|
def _query_corpus(query: LocationQuery) -> dict[str, str]:
|
|
"""Map a query into normalized strings keyed by source field."""
|
|
fields: dict[str, str] = {
|
|
"name": query.name or "",
|
|
"city": query.city or "",
|
|
"country": query.country or "",
|
|
}
|
|
for alias in query.aliases:
|
|
if alias and alias != query.name:
|
|
fields["name_short"] = alias
|
|
break
|
|
extra = query.extra or {}
|
|
for key in ("site", "operator", "organization"):
|
|
value = extra.get(key)
|
|
if value:
|
|
fields[key] = str(value)
|
|
return {key: normalize_text(value) for key, value in fields.items() if value}
|
|
|
|
|
|
def _country_compatible(entry: dict[str, Any], query: LocationQuery) -> bool:
|
|
record_country = normalize_country_text(query.country)
|
|
entry_country = normalize_country_text(entry.get("country"))
|
|
if not record_country or not entry_country:
|
|
return True
|
|
return normalize_text(record_country) == normalize_text(entry_country)
|
|
|
|
|
|
def _normalized_alias_matches(alias_normalized: str, record_text: str) -> bool:
|
|
alias_tokens = alias_normalized.split()
|
|
record_tokens = record_text.split()
|
|
if not alias_tokens or not record_tokens:
|
|
return False
|
|
if len(alias_tokens) == 1:
|
|
return alias_tokens[0] in record_tokens
|
|
window_size = len(alias_tokens)
|
|
return any(
|
|
record_tokens[index : index + window_size] == alias_tokens
|
|
for index in range(0, len(record_tokens) - window_size + 1)
|
|
)
|
|
|
|
|
|
def _entry_to_candidate(
|
|
entry: dict[str, Any],
|
|
*,
|
|
matched_alias: str,
|
|
matched_fields: Iterable[str],
|
|
source: str,
|
|
score_explainer: str,
|
|
confidence_floor: float,
|
|
) -> LocationCandidate:
|
|
canonical_name = entry.get("canonical_name") or matched_alias
|
|
# Registry entries are treated as candidates unless explicitly verified.
|
|
# This prevents migrated hard-coded hints from appearing as factual
|
|
# location evidence.
|
|
is_verified = entry.get("verification_status") == "verified"
|
|
precision = entry.get("precision") or "city"
|
|
if precision not in RENDERABLE_PRECISIONS:
|
|
precision = "city"
|
|
fields_summary = ", ".join(sorted(set(matched_fields))) or "name"
|
|
confidence_value = parse_float(entry.get("confidence"))
|
|
confidence = (
|
|
float(confidence_value)
|
|
if confidence_value is not None
|
|
else confidence_floor
|
|
)
|
|
return LocationCandidate(
|
|
latitude=float(parse_float(entry.get("latitude")) or 0.0),
|
|
longitude=float(parse_float(entry.get("longitude")) or 0.0),
|
|
display_name=canonical_name,
|
|
precision=precision,
|
|
confidence=confidence,
|
|
query=f"local_registry::{matched_alias or canonical_name}",
|
|
source=source,
|
|
source_note=entry.get("source_note")
|
|
or f"{score_explainer}: matched {fields_summary}",
|
|
matched_fields=tuple(sorted(set(matched_fields))) or ("name",),
|
|
needs_confirmation=bool(entry.get("needs_confirmation")) or not is_verified,
|
|
city=entry.get("city"),
|
|
region=entry.get("region"),
|
|
country=entry.get("country"),
|
|
matched_location_name=canonical_name,
|
|
location_verified_at=entry.get("verified_at") if is_verified else None,
|
|
suggested_registry_entry=None,
|
|
)
|
|
|
|
|
|
class RegistryResolver:
|
|
"""Match a query against a JSON registry (plus its city_fallbacks table)."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
registry_path: Path | str,
|
|
name: str = "local_registry",
|
|
city_fallback_source: str = "local_registry_city",
|
|
city_fallback_confidence_default: float = 0.65,
|
|
confidence_default: float = 0.85,
|
|
score_alias_match: Callable[[str, str, str], int] = default_score_alias_match,
|
|
) -> None:
|
|
self.name = name
|
|
self._registry_path = str(Path(registry_path))
|
|
self._city_fallback_source = city_fallback_source
|
|
self._city_fallback_confidence_default = city_fallback_confidence_default
|
|
self._confidence_default = confidence_default
|
|
self._score = score_alias_match
|
|
|
|
def reload(self) -> None:
|
|
"""Drop the cached registry — useful when the JSON file is edited."""
|
|
_load_registry_file.cache_clear()
|
|
_build_alias_index.cache_clear()
|
|
|
|
def resolve(self, query: LocationQuery) -> ResolverOutput:
|
|
candidates: list[LocationCandidate] = []
|
|
candidates.extend(self._registry_candidates(query))
|
|
city_candidate = self._city_fallback_candidate(query)
|
|
if city_candidate is not None:
|
|
candidates.append(city_candidate)
|
|
return ResolverOutput(candidates=tuple(candidates))
|
|
|
|
# ── internals ──────────────────────────────────────────────
|
|
|
|
def _registry_candidates(
|
|
self, query: LocationQuery
|
|
) -> list[LocationCandidate]:
|
|
corpus = _query_corpus(query)
|
|
if not corpus:
|
|
return []
|
|
|
|
# When the query carries a name (a record-specific identifier), require
|
|
# at least one alias match against a name-class field — otherwise a
|
|
# generic shared field like operator="RIPE NCC" would promote every
|
|
# registry entry that lists that operator, regardless of whether the
|
|
# name matches.
|
|
query_has_name = bool(corpus.get("name") or corpus.get("name_short"))
|
|
|
|
results: list[LocationCandidate] = []
|
|
for entry, aliases in _build_alias_index(self._registry_path):
|
|
best_alias = ""
|
|
best_score = 0
|
|
matched_fields: list[str] = []
|
|
matched_via_name_alias = False
|
|
for alias_field, alias_normalized in aliases:
|
|
for record_field, record_text in corpus.items():
|
|
if not _normalized_alias_matches(alias_normalized, record_text):
|
|
continue
|
|
score = self._score(
|
|
alias_field, record_field, alias_normalized
|
|
)
|
|
if score > best_score or (
|
|
score == best_score
|
|
and len(alias_normalized) > len(best_alias)
|
|
):
|
|
best_score = score
|
|
best_alias = alias_normalized
|
|
if record_field not in matched_fields:
|
|
matched_fields.append(record_field)
|
|
if alias_field == "name" and record_field in {"name", "name_short"}:
|
|
matched_via_name_alias = True
|
|
if not matched_fields or best_score <= 0:
|
|
continue
|
|
if query_has_name and not matched_via_name_alias:
|
|
continue
|
|
if not _country_compatible(entry, query):
|
|
continue
|
|
results.append(
|
|
_entry_to_candidate(
|
|
entry,
|
|
matched_alias=best_alias,
|
|
matched_fields=matched_fields,
|
|
source=self.name,
|
|
score_explainer="Registry alias match",
|
|
confidence_floor=self._confidence_default,
|
|
)
|
|
)
|
|
return results
|
|
|
|
def _city_fallback_candidate(
|
|
self, query: LocationQuery
|
|
) -> LocationCandidate | None:
|
|
country = normalize_country_text(query.country)
|
|
city = city_key(query.city)
|
|
if not country or not city:
|
|
return None
|
|
|
|
for fallback in _load_registry_file(self._registry_path).get(
|
|
"city_fallbacks", []
|
|
):
|
|
fallback_country = normalize_country_text(fallback.get("country"))
|
|
fallback_city = city_key(fallback.get("city"))
|
|
if fallback_country != country or fallback_city != city:
|
|
continue
|
|
confidence_value = parse_float(fallback.get("confidence"))
|
|
confidence = (
|
|
float(confidence_value)
|
|
if confidence_value is not None
|
|
else self._city_fallback_confidence_default
|
|
)
|
|
return LocationCandidate(
|
|
latitude=float(parse_float(fallback.get("latitude")) or 0.0),
|
|
longitude=float(parse_float(fallback.get("longitude")) or 0.0),
|
|
display_name=fallback.get("city") or "",
|
|
precision="city",
|
|
confidence=confidence,
|
|
query=(
|
|
f"city_fallback::{fallback.get('city')}, "
|
|
f"{fallback.get('country')}"
|
|
),
|
|
source=self._city_fallback_source,
|
|
source_note=fallback.get("source_note")
|
|
or f"City fallback for {fallback.get('city')}, {fallback.get('country')}",
|
|
matched_fields=("city", "country"),
|
|
needs_confirmation=False,
|
|
city=fallback.get("city"),
|
|
region=fallback.get("region"),
|
|
country=fallback.get("country"),
|
|
matched_location_name=fallback.get("city"),
|
|
location_verified_at=fallback.get("verified_at"),
|
|
suggested_registry_entry=None,
|
|
)
|
|
return None
|