Spaces:
Running
Running
File size: 6,321 Bytes
008947a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 | """Location search with local China data, Nominatim, rate limiting, and cache."""
from __future__ import annotations
import hashlib
import json
import os
import re
from dataclasses import asdict
from typing import Any, Protocol, cast
from geopy.extra.rate_limiter import RateLimiter
from geopy.geocoders import Nominatim
from .cache import DiskCache
from .china import search_china
from .models import Coordinate, Location
COORDINATE_QUERY = re.compile(r"^\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*[,/ ]\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*$")
class GeocodingError(RuntimeError):
"""Raised when a place search cannot be completed."""
class GeocoderProvider(Protocol):
def search(self, query: str, language: str, limit: int) -> list[Location]: ...
def parse_coordinate_query(query: str) -> Location | None:
"""Parse a latitude/longitude search query."""
match = COORDINATE_QUERY.match(query)
if match is None:
return None
coordinate = Coordinate(float(match.group(1)), float(match.group(2)))
return Location(
display_name=f"{coordinate.latitude:.5f}, {coordinate.longitude:.5f}",
coordinate=coordinate,
provider="coordinates",
)
class NominatimProvider:
"""Thin Nominatim provider. Call only after an explicit user submission."""
def __init__(self) -> None:
user_agent = os.getenv(
"MAPTOPOSTER_USER_AGENT",
"MapToPoster/0.2 (+https://github.com/IsaacHuo/maptoposter)",
)
endpoint = os.getenv("MAPTOPOSTER_NOMINATIM_DOMAIN", "nominatim.openstreetmap.org")
geocoder = cast(Any, Nominatim)(user_agent=user_agent, domain=endpoint, timeout=15)
self._geocode = RateLimiter(
geocoder.geocode,
min_delay_seconds=1.0,
max_retries=2,
error_wait_seconds=2.0,
swallow_exceptions=False,
)
def search(self, query: str, language: str = "en", limit: int = 5) -> list[Location]:
try:
results = self._geocode(
query,
exactly_one=False,
limit=limit,
addressdetails=True,
language="zh-CN" if language == "cn" else "en",
)
except Exception as exc:
raise GeocodingError(f"Location service is unavailable: {exc}") from exc
locations: list[Location] = []
for result in results or []:
raw = getattr(result, "raw", {}) or {}
address = raw.get("address", {}) or {}
locations.append(
Location(
display_name=str(getattr(result, "address", query)),
coordinate=Coordinate(float(result.latitude), float(result.longitude)),
country=str(address.get("country", "")),
region=str(address.get("state") or address.get("region") or ""),
country_code=str(address.get("country_code", "")),
provider="nominatim",
)
)
return locations
class GeocodingService:
"""Search places in priority order and persist external results."""
def __init__(self, cache: DiskCache | None = None, provider: GeocoderProvider | None = None) -> None:
self.cache = cache or DiskCache()
self.provider = provider or NominatimProvider()
def search(self, query: str, language: str = "en", limit: int = 5) -> list[Location]:
query = query.strip()
if not query:
raise ValueError("Search query cannot be empty.")
coordinate = parse_coordinate_query(query)
if coordinate is not None:
return [coordinate]
local = list(search_china(query, limit=limit))
if local:
return local[:limit]
key_payload = json.dumps([query.casefold(), language, limit], ensure_ascii=False)
key = hashlib.sha256(key_payload.encode("utf-8")).hexdigest()
cache_path = self.cache.geocoding_path(key)
cached = self.cache.read_json(cache_path)
cached_results = cached.get("results") if cached else None
if isinstance(cached_results, list):
external = [self._from_dict(item) for item in cached_results if isinstance(item, dict)]
else:
with self.cache.lock("geocoding", key):
cached = self.cache.read_json(cache_path)
cached_results = cached.get("results") if cached else None
if isinstance(cached_results, list):
external = [self._from_dict(item) for item in cached_results if isinstance(item, dict)]
else:
external = self.provider.search(query, language, limit)
self.cache.write_json(
cache_path,
{"query": query, "results": [self._to_dict(item) for item in external]},
)
combined: list[Location] = []
seen: set[tuple[float, float]] = set()
for result in [*local, *external]:
coordinate_key = (
round(result.coordinate.latitude, 5),
round(result.coordinate.longitude, 5),
)
if coordinate_key not in seen:
combined.append(result)
seen.add(coordinate_key)
if len(combined) >= limit:
break
return combined
@staticmethod
def _to_dict(location: Location) -> dict[str, object]:
payload = asdict(location)
payload["coordinate"] = asdict(location.coordinate)
return payload
@staticmethod
def _from_dict(payload: dict[str, object]) -> Location:
coordinate = payload.get("coordinate")
if not isinstance(coordinate, dict):
raise GeocodingError("Invalid cached coordinate.")
return Location(
display_name=str(payload.get("display_name", "")),
coordinate=Coordinate(
float(coordinate["latitude"]),
float(coordinate["longitude"]),
),
country=str(payload.get("country", "")),
region=str(payload.get("region", "")),
country_code=str(payload.get("country_code", "")),
provider=str(payload.get("provider", "cache")),
)
|