import time import math import datetime from typing import Dict, Any, Optional import httpx # 주요 한국 도시 좌표 목록 KOREAN_CITIES = [ {"name": "서울", "lat": 37.5665, "lon": 126.9780}, {"name": "인천", "lat": 37.4563, "lon": 126.7052}, {"name": "수원", "lat": 37.2636, "lon": 127.0286}, {"name": "춘천", "lat": 37.8813, "lon": 127.7298}, {"name": "강릉", "lat": 37.7519, "lon": 128.8761}, {"name": "대전", "lat": 36.3504, "lon": 127.3845}, {"name": "세종", "lat": 36.4800, "lon": 127.2890}, {"name": "청주", "lat": 36.6424, "lon": 127.4890}, {"name": "대구", "lat": 35.8714, "lon": 128.6014}, {"name": "포항", "lat": 36.0190, "lon": 129.3435}, {"name": "전주", "lat": 35.8242, "lon": 127.1480}, {"name": "광주", "lat": 35.1595, "lon": 126.8526}, {"name": "부산", "lat": 35.1796, "lon": 129.0756}, {"name": "울산", "lat": 35.5384, "lon": 129.3114}, {"name": "창원", "lat": 35.2280, "lon": 128.6811}, {"name": "제주", "lat": 33.4996, "lon": 126.5312}, ] def find_nearest_city(lat: float, lon: float) -> str: closest = min( KOREAN_CITIES, key=lambda c: (c["lat"] - lat) ** 2 + (c["lon"] - lon) ** 2 ) dist_sq = (closest["lat"] - lat) ** 2 + (closest["lon"] - lon) ** 2 # 약 50km 이내이면 해당 도시명 반환 if dist_sq < 0.25: return closest["name"] return "현재 위치" # WMO Weather interpretation code (WW) # https://open-meteo.com/en/docs WMO_MAP = { 0: ("맑음", "mdi-weather-sunny", "mdi-weather-night", "amber-darken-1", "indigo-lighten-2"), 1: ("대체로 맑음", "mdi-weather-partly-cloudy", "mdi-weather-night-partly-cloudy", "amber-darken-2", "indigo-lighten-3"), 2: ("구름 조금", "mdi-weather-partly-cloudy", "mdi-weather-partly-cloudy", "blue-grey-lighten-1", "blue-grey-lighten-1"), 3: ("흐림", "mdi-weather-cloudy", "mdi-weather-cloudy", "grey-darken-1", "grey-darken-1"), 45: ("안개", "mdi-weather-fog", "mdi-weather-fog", "blue-grey", "blue-grey"), 48: ("서리 안개", "mdi-weather-fog", "mdi-weather-fog", "blue-grey", "blue-grey"), 51: ("약한 이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "light-blue", "light-blue"), 53: ("이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "light-blue-darken-1", "light-blue-darken-1"), 55: ("강한 이슬비", "mdi-weather-rainy", "mdi-weather-rainy", "blue-darken-1", "blue-darken-1"), 56: ("약한 어는 이슬비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan", "cyan"), 57: ("강한 어는 이슬비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan-darken-1", "cyan-darken-1"), 61: ("약한 비", "mdi-weather-rainy", "mdi-weather-rainy", "blue", "blue"), 63: ("비", "mdi-weather-rainy", "mdi-weather-rainy", "blue-darken-1", "blue-darken-1"), 65: ("강한 비", "mdi-weather-pouring", "mdi-weather-pouring", "blue-darken-3", "blue-darken-3"), 66: ("약한 어는 비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan", "cyan"), 67: ("강한 어는 비", "mdi-weather-snowy-rainy", "mdi-weather-snowy-rainy", "cyan-darken-2", "cyan-darken-2"), 71: ("약한 눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-3", "light-blue-lighten-3"), 73: ("눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-2", "light-blue-lighten-2"), 75: ("강한 눈", "mdi-weather-snowy-heavy", "mdi-weather-snowy-heavy", "blue-lighten-4", "blue-lighten-4"), 77: ("싸락눈", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue-lighten-1", "light-blue-lighten-1"), 80: ("약한 소나기", "mdi-weather-partly-rainy", "mdi-weather-partly-rainy", "blue", "blue"), 81: ("소나기", "mdi-weather-partly-rainy", "mdi-weather-partly-rainy", "blue-darken-1", "blue-darken-1"), 82: ("강한 소나기", "mdi-weather-pouring", "mdi-weather-pouring", "blue-darken-2", "blue-darken-2"), 85: ("약한 눈 소나기", "mdi-weather-snowy", "mdi-weather-snowy", "light-blue", "light-blue"), 86: ("강한 눈 소나기", "mdi-weather-snowy-heavy", "mdi-weather-snowy-heavy", "blue-lighten-3", "blue-lighten-3"), 95: ("뇌우", "mdi-weather-lightning", "mdi-weather-lightning", "deep-purple", "deep-purple"), 96: ("뇌우 및 우박", "mdi-weather-lightning-rainy", "mdi-weather-lightning-rainy", "deep-purple-darken-2", "deep-purple-darken-2"), 99: ("강한 뇌우 및 우박", "mdi-weather-lightning-rainy", "mdi-weather-lightning-rainy", "deep-purple-darken-3", "deep-purple-darken-3"), } def resolve_weather_style(code: int, is_day: int = 1): info = WMO_MAP.get(code) if not info: return { "text": "흐림", "icon": "mdi-weather-cloudy", "color": "grey" } text, day_icon, night_icon, day_color, night_color = info return { "text": text, "icon": day_icon if is_day else night_icon, "color": day_color if is_day else night_color, } # 캐시 저장소: (lat, lon) -> (timestamp, data) _cache: Dict[str, Any] = {} CACHE_TTL = 900 # 15분 KOREAN_WEEKDAYS = ["월", "화", "수", "목", "금", "토", "일"] async def get_weather_forecast(lat: float = 37.5665, lon: float = 126.9780) -> Dict[str, Any]: cache_key = f"{round(lat, 2)}_{round(lon, 2)}" now = time.time() if cache_key in _cache: cached_time, cached_data = _cache[cache_key] if now - cached_time < CACHE_TTL: return cached_data url = ( f"https://api.open-meteo.com/v1/forecast" f"?latitude={lat}&longitude={lon}" f"¤t=temperature_2m,relative_humidity_2m,apparent_temperature,is_day,precipitation,weather_code,wind_speed_10m" f"&daily=weather_code,temperature_2m_max,temperature_2m_min,precipitation_probability_max" f"&timezone=auto" ) try: async with httpx.AsyncClient(timeout=10.0) as client: resp = await client.get(url) resp.raise_for_status() data = resp.json() except Exception as e: # 캐시가 만료되었더라도 기존 캐시가 있으면 반환 if cache_key in _cache: return _cache[cache_key][1] raise e current_raw = data.get("current", {}) daily_raw = data.get("daily", {}) is_day = current_raw.get("is_day", 1) current_code = current_raw.get("weather_code", 0) current_style = resolve_weather_style(current_code, is_day) current_formatted = { "temperature": round(current_raw.get("temperature_2m", 0.0), 1), "apparent_temperature": round(current_raw.get("apparent_temperature", 0.0), 1), "humidity": current_raw.get("relative_humidity_2m", 0), "wind_speed": round(current_raw.get("wind_speed_10m", 0.0), 1), "precipitation": current_raw.get("precipitation", 0.0), "weather_code": current_code, "weather_text": current_style["text"], "icon": current_style["icon"], "color": current_style["color"], "is_day": bool(is_day), "time": current_raw.get("time"), } daily_list = [] dates = daily_raw.get("time", []) weather_codes = daily_raw.get("weather_code", []) temps_max = daily_raw.get("temperature_2m_max", []) temps_min = daily_raw.get("temperature_2m_min", []) precip_probs = daily_raw.get("precipitation_probability_max", []) for i in range(len(dates)): date_str = dates[i] try: dt = datetime.date.fromisoformat(date_str) weekday_kr = KOREAN_WEEKDAYS[dt.weekday()] if i == 0: day_label = "오늘" elif i == 1: day_label = "내일" else: day_label = f"{weekday_kr}요일" formatted_date = f"{dt.month}.{dt.day} ({weekday_kr})" except Exception: day_label = date_str formatted_date = date_str code = weather_codes[i] if i < len(weather_codes) else 0 style = resolve_weather_style(code, 1) daily_list.append({ "date": date_str, "day_label": day_label, "formatted_date": formatted_date, "weather_code": code, "weather_text": style["text"], "icon": style["icon"], "color": style["color"], "temp_max": round(temps_max[i], 1) if i < len(temps_max) else 0.0, "temp_min": round(temps_min[i], 1) if i < len(temps_min) else 0.0, "precipitation_prob": precip_probs[i] if i < len(precip_probs) else 0, }) city_name = find_nearest_city(lat, lon) result = { "location": { "name": city_name, "lat": lat, "lon": lon, }, "current": current_formatted, "daily": daily_list, "updated_at": datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S") } _cache[cache_key] = (now, result) return result