77 lines
2.0 KiB
Python
77 lines
2.0 KiB
Python
import json
|
|
from typing import Any
|
|
|
|
import httpx
|
|
from pydantic import AliasChoices, BaseModel, Field
|
|
|
|
from app.core.config import settings
|
|
|
|
|
|
class TiandituGeocodeRequest(BaseModel):
|
|
keyword: str = Field(
|
|
...,
|
|
min_length=1,
|
|
validation_alias=AliasChoices("keyword", "keyWord"),
|
|
description="地理编码地址关键字",
|
|
)
|
|
|
|
|
|
class TiandituGeocodingConfigError(RuntimeError):
|
|
pass
|
|
|
|
|
|
class TiandituGeocodingAPIError(RuntimeError):
|
|
def __init__(self, status_code: int, detail: Any):
|
|
super().__init__("Tianditu Geocoding API request failed")
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
|
|
|
|
async def geocode_tianditu(
|
|
request: TiandituGeocodeRequest,
|
|
*,
|
|
client: httpx.AsyncClient | None = None,
|
|
) -> dict[str, Any]:
|
|
if not settings.TIANDITU_GEOCODER_TOKEN:
|
|
raise TiandituGeocodingConfigError("TIANDITU_GEOCODER_TOKEN is not configured")
|
|
|
|
params = {
|
|
"ds": json.dumps({"keyWord": request.keyword}, ensure_ascii=False),
|
|
"tk": settings.TIANDITU_GEOCODER_TOKEN,
|
|
}
|
|
|
|
if client is not None:
|
|
response = await client.get(settings.TIANDITU_GEOCODER_URL, params=params)
|
|
return _parse_response(response)
|
|
|
|
async with httpx.AsyncClient(
|
|
timeout=settings.TIANDITU_GEOCODER_TIMEOUT_SECONDS
|
|
) as managed_client:
|
|
response = await managed_client.get(
|
|
settings.TIANDITU_GEOCODER_URL,
|
|
params=params,
|
|
)
|
|
return _parse_response(response)
|
|
|
|
|
|
def _parse_response(response: httpx.Response) -> dict[str, Any]:
|
|
try:
|
|
response.raise_for_status()
|
|
except httpx.HTTPStatusError as exc:
|
|
raise TiandituGeocodingAPIError(
|
|
exc.response.status_code,
|
|
_response_detail(exc.response),
|
|
) from exc
|
|
|
|
data = response.json()
|
|
if str(data.get("status")) != "0":
|
|
raise TiandituGeocodingAPIError(502, data)
|
|
return data
|
|
|
|
|
|
def _response_detail(response: httpx.Response) -> Any:
|
|
try:
|
|
return response.json()
|
|
except ValueError:
|
|
return response.text
|