File size: 3,033 Bytes
42d88ae | 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 | import aiohttp
from typing import Optional
from dataclasses import dataclass
from Backend.core.logging import get_logger
logger = get_logger(__name__)
@dataclass
class LocationInfo:
city: Optional[str] = None
locality: Optional[str] = None
district: Optional[str] = None
state: Optional[str] = None
country: Optional[str] = None
pincode: Optional[str] = None
full_address: Optional[str] = None
class GeocodingService:
NOMINATIM_URL = "https://nominatim.openstreetmap.org/reverse"
async def reverse_geocode(self, latitude: float, longitude: float) -> LocationInfo:
params = {
"lat": latitude,
"lon": longitude,
"format": "json",
"addressdetails": 1,
"zoom": 18,
}
headers = {
"User-Agent": "CityIssueResolutionAgent/1.0"
}
try:
async with aiohttp.ClientSession() as session:
async with session.get(
self.NOMINATIM_URL,
params=params,
headers=headers,
timeout=aiohttp.ClientTimeout(total=10)
) as response:
if response.status == 200:
data = await response.json()
return self._parse_response(data)
else:
logger.warning(f"Geocoding failed: {response.status}")
return LocationInfo()
except Exception as e:
logger.error(f"Geocoding error: {e}")
return LocationInfo()
def _parse_response(self, data: dict) -> LocationInfo:
address = data.get("address", {})
city = (
address.get("city") or
address.get("town") or
address.get("municipality") or
address.get("village") or
address.get("suburb")
)
locality = (
address.get("suburb") or
address.get("neighbourhood") or
address.get("quarter") or
address.get("borough")
)
district = (
address.get("county") or
address.get("district") or
address.get("state_district")
)
state = address.get("state")
country = address.get("country")
pincode = address.get("postcode")
full_address = data.get("display_name")
return LocationInfo(
city=city,
locality=locality,
district=district,
state=state,
country=country,
pincode=pincode,
full_address=full_address,
)
async def get_city_from_coordinates(self, latitude: float, longitude: float) -> Optional[str]:
location = await self.reverse_geocode(latitude, longitude)
return location.city or location.locality or location.district
geocoding_service = GeocodingService()
|