import requests import os import json from datetime import datetime, timedelta def valid_until_hours(cached, hours=2): ts = datetime.fromisoformat(cached['ts']) return datetime.now() - ts < timedelta(hours=hours) def parse_precip(value): if value == '강수없음': return 0.0 elif '1mm 미만' in str(value): return 0.5 else: try: return float(value) except: return 0.0 def ensure_cache_dir(): cache_dir = os.path.abspath(os.path.join(os.path.dirname(__file__), '..', 'data', 'cache')) os.makedirs(cache_dir, exist_ok=True) return cache_dir def get_cache_or_request(name, valid_until_fn, request_fn): cache_dir = ensure_cache_dir() today = datetime.now().strftime("%Y%m%d") cache_file = os.path.join(cache_dir, f"{name}_{today}.json") if os.path.exists(cache_file): with open(cache_file, 'r', encoding='utf-8') as f: cached = json.load(f) if valid_until_fn(cached): return cached['data'] data = request_fn() with open(cache_file, 'w', encoding='utf-8') as f: json.dump({'ts': datetime.now().isoformat(), 'data': data}, f, ensure_ascii=False) return data def get_latest_base_date_time(now=None): if now is None: now = datetime.now() base_times = ["2330", "0230", "0530", "0830", "1130", "1430", "1730", "2030"] candidate = None for bt in reversed(base_times): hour = int(bt[:2]) minute = int(bt[2:]) if (now.hour > hour) or (now.hour == hour and now.minute >= minute): candidate = bt break if candidate is None: candidate = "2330" now -= timedelta(days=1) base_date = now.strftime("%Y%m%d") return base_date, candidate def get_daily_ultra_forecast(serviceKey): def request(): base_date, base_time = get_latest_base_date_time() url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst" params = { 'serviceKey': serviceKey, 'numOfRows': '1000', 'pageNo': '1', 'dataType': 'JSON', 'base_date': base_date, 'base_time': base_time, 'nx': '57', 'ny': '130' } try: resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() items = resp.json()['response']['body']['items']['item'] except Exception as e: print(f"[ERROR] 초단기예보 호출 실패: {e}") return {} daily_data = {} for item in items: dt = item['fcstDate'] cat = item['category'] val = item['fcstValue'] if dt not in daily_data: daily_data[dt] = {'sumRn': 0, 'minTa': [], 'maxTa': [], 'rhm': []} if cat == 'RN1': daily_data[dt]['sumRn'] += parse_precip(val) elif cat == 'T3H': try: t = float(val) daily_data[dt]['minTa'].append(t) daily_data[dt]['maxTa'].append(t) except: pass elif cat == 'REH': try: daily_data[dt]['rhm'].append(float(val)) except: pass result = {} for dt, vals in daily_data.items(): minTa = min(vals['minTa']) if vals['minTa'] else 0 maxTa = max(vals['maxTa']) if vals['maxTa'] else 0 avgRhm = sum(vals['rhm']) / len(vals['rhm']) if vals['rhm'] else 0 sumRn = round(vals['sumRn'], 2) result[dt] = { 'sumRn': round(sumRn, 1), 'minTa': round(minTa, 1), 'maxTa': round(maxTa, 1), 'avgRhm': round(avgRhm, 1) } return result return get_cache_or_request('ultra_forecast', lambda cached: valid_until_hours(cached, 2), request) def get_daily_vilage_forecast(serviceKey): def request(): base_date, _ = get_latest_base_date_time() url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getVilageFcst" params = { 'serviceKey': serviceKey, 'numOfRows': '1000', 'pageNo': '1', 'dataType': 'JSON', 'base_date': base_date, 'base_time': '0200', 'nx': '57', 'ny': '130' } try: resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() items = resp.json()['response']['body']['items']['item'] except Exception as e: print(f"[ERROR] 단기예보 호출 실패: {e}") return {} daily_data = {} for item in items: dt = item['fcstDate'] cat = item['category'] val = item['fcstValue'] if dt not in daily_data: daily_data[dt] = {'sumRn': 0, 'minTa': [], 'maxTa': [], 'rhm': []} if cat == 'RN1': daily_data[dt]['sumRn'] += parse_precip(val) elif cat == 'TMN': try: daily_data[dt]['minTa'].append(float(val)) except: pass elif cat == 'TMX': try: daily_data[dt]['maxTa'].append(float(val)) except: pass elif cat == 'REH': try: daily_data[dt]['rhm'].append(float(val)) except: pass result = {} for dt, vals in daily_data.items(): minTa = min(vals['minTa']) if vals['minTa'] else 0 maxTa = max(vals['maxTa']) if vals['maxTa'] else 0 avgRhm = sum(vals['rhm']) / len(vals['rhm']) if vals['rhm'] else 0 sumRn = round(vals['sumRn'], 2) result[dt] = { 'sumRn': round(sumRn, 1), 'minTa': round(minTa, 1), 'maxTa': round(maxTa, 1), 'avgRhm': round(avgRhm, 1) } return result return get_cache_or_request('vilage_forecast', lambda cached: valid_until_hours(cached, 6), request) def get_midterm_forecast(serviceKey, regId='11B20305'): def request(): url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidLandFcst" now = datetime.now() if now.hour < 6: tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800" elif now.hour < 18: tmFc = now.strftime("%Y%m%d") + "0600" else: tmFc = now.strftime("%Y%m%d") + "1800" params = { 'serviceKey': serviceKey, 'regId': regId, 'tmFc': tmFc, 'numOfRows': 10, 'pageNo': 1, 'dataType': 'JSON', } try: resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() data = resp.json() items = data.get('response', {}).get('body', {}).get('items', {}).get('item', []) if not items: print(f"[ERROR] 중기예보 응답 item 없음. tmFc={tmFc}, regId={regId}") return {} item = items[0] except Exception as e: print(f"[ERROR] 중기예보 호출 실패: {e}") return {} precip_probs = {} for day in range(3, 11): key = f'rnSt{day}' try: precip_probs[day] = int(item.get(key, 0)) except: precip_probs[day] = 0 return precip_probs return get_cache_or_request('midterm_precip', lambda cached: valid_until_hours(cached, 12), request) def get_midterm_temperature_forecast(serviceKey, regId='11B20305'): def request(): url = "http://apis.data.go.kr/1360000/MidFcstInfoService/getMidTa" now = datetime.now() if now.hour < 6: tmFc = (now - timedelta(days=1)).strftime("%Y%m%d") + "1800" elif now.hour < 18: tmFc = now.strftime("%Y%m%d") + "0600" else: tmFc = now.strftime("%Y%m%d") + "1800" params = { 'serviceKey': serviceKey, 'regId': regId, 'tmFc': tmFc, 'pageNo': '1', 'numOfRows': '10', 'dataType': 'JSON' } try: resp = requests.get(url, params=params, timeout=10) resp.raise_for_status() data = resp.json() items = data.get("response", {}).get("body", {}).get("items", {}).get("item", []) if not items: print(f"[ERROR] 응답에 item 없음. tmFc={tmFc}, regId={regId}") return {} item = items[0] except Exception as e: print(f"[ERROR] 중기기온예보 호출 실패: {e}") return {} temps = {} for day in range(3, 11): min_key = f'taMin{day}' max_key = f'taMax{day}' min_val = item.get(min_key) max_val = item.get(max_key) try: temps[day] = { 'min': int(min_val) if min_val is not None else None, 'max': int(max_val) if max_val is not None else None } except Exception: temps[day] = {'min': None, 'max': None} return temps return get_cache_or_request('midterm_temp', lambda cached: valid_until_hours(cached, 12), request) def get_weekly_precip(serviceKey): from datetime import date today = date.today() sunday = today + timedelta(days=(6 - today.weekday())) ultra = get_daily_ultra_forecast(serviceKey) short = get_daily_vilage_forecast(serviceKey) mid_precip = get_midterm_forecast(serviceKey) mid_temp = get_midterm_temperature_forecast(serviceKey) results = {} for i in range((sunday - today).days + 1): dt = today + timedelta(days=i) dt_str = dt.strftime("%Y%m%d") results[dt_str] = { 'sumRn': 0, 'minTa': 0, 'maxTa': 0, 'avgRhm': 0 } if dt_str in ultra: results[dt_str]['sumRn'] = ultra[dt_str]['sumRn'] results[dt_str]['avgRhm'] = ultra[dt_str]['avgRhm'] if dt_str in short: if short[dt_str]['minTa'] != 0: results[dt_str]['minTa'] = short[dt_str]['minTa'] if short[dt_str]['maxTa'] != 0: results[dt_str]['maxTa'] = short[dt_str]['maxTa'] day_offset = (dt - today).days # 0부터 시작 if day_offset >= 3: # 중기예보 강수 우선 적용 if day_offset in mid_precip: mid_rain = float(mid_precip[day_offset]) / 100 * 5.0 if mid_rain > results[dt_str]['sumRn']: results[dt_str]['sumRn'] = mid_rain # 중기예보 기온 적용: 단, None이거나 0이면 단기예보로 대체 key = str(day_offset) if key in mid_temp: mid_min = mid_temp[key]['min'] mid_max = mid_temp[key]['max'] if mid_min not in (None, 0): results[dt_str]['minTa'] = mid_min elif dt_str in short and short[dt_str]['minTa'] != 0: results[dt_str]['minTa'] = short[dt_str]['minTa'] if mid_max not in (None, 0): results[dt_str]['maxTa'] = mid_max elif dt_str in short and short[dt_str]['maxTa'] != 0: results[dt_str]['maxTa'] = short[dt_str]['maxTa'] # 중기 기온 적용 이후, 습도 보완 if results[dt_str]['avgRhm'] == 0 and dt_str in short and short[dt_str]['avgRhm'] != 0: results[dt_str]['avgRhm'] = short[dt_str]['avgRhm'] results[dt_str] = { 'sumRn': round(results[dt_str]['sumRn'], 1), 'minTa': round(results[dt_str]['minTa'], 1), 'maxTa': round(results[dt_str]['maxTa'], 1), 'avgRhm': round(results[dt_str]['avgRhm'], 1), } return results def print_weekly_precip_table(data_dict): # 헤더 출력 header = f"{'날짜':<10} {'강수량(mm)':>10} {'최저기온(℃)':>12} {'최고기온(℃)':>12} {'평균습도(%)':>12}" print(header) print('-' * len(header)) # 날짜 순서대로 출력 for dt in sorted(data_dict.keys()): vals = data_dict[dt] print(f"{dt:<10} {vals['sumRn']:10.1f} {vals['minTa']:12.1f} {vals['maxTa']:12.1f} {vals['avgRhm']:12.1f}") if __name__ == '__main__': import os, sys sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) from lib.common import load_config serviceKey = load_config()['DATA_API']['serviceKey'] data = get_weekly_precip(serviceKey) print(get_weekly_precip(serviceKey)) print_weekly_precip_table(data) print(get_daily_vilage_forecast(serviceKey)) print(get_midterm_temperature_forecast(serviceKey))