129 lines
4.2 KiB
Python
129 lines
4.2 KiB
Python
"""Domain-neutral data structures for the location pipeline."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import Any, Mapping
|
|
|
|
# Renderable precision tiers, ordered from most precise to least.
|
|
RENDERABLE_PRECISIONS: tuple[str, ...] = ("precise", "site", "city")
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocationQuery:
|
|
"""Domain-neutral input for the resolution pipeline.
|
|
|
|
``name`` and ``aliases`` are matched against registry alias indexes;
|
|
``city`` / ``country`` / ``region`` provide geographic context for both
|
|
registry lookups and Nominatim queries; ``source_latitude`` /
|
|
``source_longitude`` short-circuit when the record already carries
|
|
coordinates; ``extra`` carries domain-specific fields (operator, site,
|
|
organization, asn, peer_ip, …) that resolvers can opt into.
|
|
"""
|
|
|
|
name: str | None = None
|
|
aliases: tuple[str, ...] = ()
|
|
city: str | None = None
|
|
country: str | None = None
|
|
region: str | None = None
|
|
source_latitude: float | None = None
|
|
source_longitude: float | None = None
|
|
extra: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class LocationCandidate:
|
|
"""A resolved location candidate produced by a resolver."""
|
|
|
|
latitude: float
|
|
longitude: float
|
|
display_name: str
|
|
precision: str # "precise" | "site" | "city" | (rejected: country/unknown)
|
|
confidence: float
|
|
query: str
|
|
source: str
|
|
source_note: str | None
|
|
matched_fields: tuple[str, ...]
|
|
needs_confirmation: bool
|
|
city: str | None = None
|
|
region: str | None = None
|
|
country: str | None = None
|
|
matched_location_name: str | None = None
|
|
location_verified_at: str | None = None
|
|
suggested_registry_entry: dict[str, Any] | None = None
|
|
raw_payload: dict[str, Any] | None = None
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"latitude": self.latitude,
|
|
"longitude": self.longitude,
|
|
"display_name": self.display_name,
|
|
"precision": self.precision,
|
|
"confidence": self.confidence,
|
|
"query": self.query,
|
|
"source": self.source,
|
|
"source_note": self.source_note,
|
|
"matched_fields": list(self.matched_fields),
|
|
"needs_confirmation": self.needs_confirmation,
|
|
"city": self.city,
|
|
"region": self.region,
|
|
"country": self.country,
|
|
"matched_location_name": self.matched_location_name,
|
|
"location_verified_at": self.location_verified_at,
|
|
"suggested_registry_entry": self.suggested_registry_entry,
|
|
"raw_payload": self.raw_payload,
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolverOutput:
|
|
"""What a single resolver returns from one ``resolve()`` call."""
|
|
|
|
candidates: tuple[LocationCandidate, ...] = ()
|
|
attempted_queries: tuple[str, ...] = ()
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolutionDiagnostic:
|
|
"""Why we could not resolve, plus what we tried."""
|
|
|
|
failure_reason: str
|
|
attempted_queries: tuple[str, ...] = ()
|
|
record_id: int | None = None
|
|
source: str | None = None
|
|
source_id: str | None = None
|
|
name: str | None = None
|
|
country: str | None = None
|
|
city: str | None = None
|
|
site: str | None = None
|
|
operator: str | None = None
|
|
extra: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
def to_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"failure_reason": self.failure_reason,
|
|
"attempted_queries": list(self.attempted_queries),
|
|
"record_id": self.record_id,
|
|
"source": self.source,
|
|
"source_id": self.source_id,
|
|
"name": self.name,
|
|
"country": self.country,
|
|
"city": self.city,
|
|
"site": self.site,
|
|
"operator": self.operator,
|
|
**({"extra": dict(self.extra)} if self.extra else {}),
|
|
}
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ResolutionResult:
|
|
"""Pipeline output: best candidate (if any) + diagnostic on miss."""
|
|
|
|
location: LocationCandidate | None
|
|
diagnostic: ResolutionDiagnostic | None
|
|
attempted_queries: tuple[str, ...] = ()
|
|
|
|
@property
|
|
def is_resolved(self) -> bool:
|
|
return bool(self.location)
|