diff --git a/autouploader/weather.py b/autouploader/weather.py index 20c2b15..fb0b731 100644 --- a/autouploader/weather.py +++ b/autouploader/weather.py @@ -3,25 +3,71 @@ import json import re import sqlite3 import os -import time from datetime import datetime from config import serviceKey, TODAY +# 디버그 모드 여부 설정 +# True일 경우 DB 저장 및 HTML 반환 없이, 콘솔에 평문 출력만 수행 +debug = False + def parse_precip(value): + """ + 강수량 텍스트를 숫자(mm)로 변환하는 함수 + - '강수없음'은 0.0으로 처리 + - '1mm 미만'은 0.5로 간주 + - 그 외는 숫자 추출하여 float 반환 + """ if value == '강수없음': return 0.0 elif '1mm 미만' in value: return 0.5 else: - match = re.search(r"[\d.]+", value) - if match: - return float(match.group()) - else: - return 0.0 + match = re.search(r"[\d.]+", str(value)) + return float(match.group()) if match else 0.0 -def get_precipitation_summary(retry=True): +def get_ultra_data(): + """ + 기상청 초단기 예보 API 호출 및 처리 함수 + - base_time: 08:50 고정 (config로도 가능) + - 'RN1' 카테고리(1시간 강수량) 데이터 필터링 + - 오늘 날짜, 10시부터 22시 사이 시간별 강수량 수집 + - 반환: {시간(정수): 강수량(mm)} 딕셔너리 + """ + url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst" + params = { + 'serviceKey': serviceKey, + 'numOfRows': '1000', + 'pageNo': '1', + 'dataType': 'JSON', + 'base_date': TODAY, + 'base_time': '0850', + 'nx': '57', + 'ny': '130' + } + response = requests.get(url, params=params) + data = response.json() + + if debug: + print("[DEBUG] 초단기예보 응답 JSON:", json.dumps(data, ensure_ascii=False, indent=2)) + + result = {} + for item in data['response']['body']['items']['item']: + if item['category'] == 'RN1' and item['fcstDate'] == TODAY: + hour = int(item['fcstTime'][:2]) + if 10 <= hour <= 22: + result[hour] = parse_precip(item['fcstValue']) + return result + +def get_vilage_data(): + """ + 기상청 단기 예보 API 호출 및 처리 함수 + - base_time: 08:00 고정 (config로도 가능) + - 'PCP' 카테고리(강수량) 데이터 필터링 + - 오늘 날짜, 10시부터 22시 사이 시간별 강수량 수집 + - 초단기 예보에 없는 시간 보완용으로 활용 + - 반환: {시간(정수): 강수량(mm)} 딕셔너리 + """ url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getVilageFcst" - params = { 'serviceKey': serviceKey, 'numOfRows': '1000', @@ -32,64 +78,92 @@ def get_precipitation_summary(retry=True): 'nx': '57', 'ny': '130' } - response = requests.get(url, params=params) + data = response.json() - try: - data = response.json() - total_rainfall = 0.0 - time_precip_list = [] + if debug: + print("[DEBUG] 단기예보 응답 JSON:", json.dumps(data, ensure_ascii=False, indent=2)) - lines = [ - '
', - '

[시간대별 예상 강수량]

', - '', - '', - '', - '', - '', - '', - '', - '' - ] + result = {} + for item in data['response']['body']['items']['item']: + if item['category'] == 'PCP' and item['fcstDate'] == TODAY: + hour = int(item['fcstTime'][:2]) + # 초단기 데이터가 우선이므로 중복 시간은 제외 + if 10 <= hour <= 22 and hour not in result: + result[hour] = parse_precip(item['fcstValue']) + return result - for item in data['response']['body']['items']['item']: - if item['category'] == 'PCP' and item['fcstDate'] == TODAY: - time = item['fcstTime'] - if 900 < int(time) < 2300: - mm = parse_precip(item['fcstValue']) - time_str = f"{time[:2]}:{time[2:]}" - lines.append(f'') - total_rainfall += mm - time_precip_list.append((time_str, mm)) +def get_precipitation_summary(): + """ + 초단기예보와 단기예보를 혼합하여 10시부터 22시까지 시간별 예상 강수량 요약 생성 + - debug 모드일 경우 평문 출력만 수행하며 DB 저장 및 HTML 반환은 하지 않음 + - debug 모드가 아닐 경우 HTML 포맷으로 테이블 생성 후 DB 저장 및 HTML 반환 + """ + ultra = get_ultra_data() # 초단기 예보 데이터 + vilage = get_vilage_data() # 단기 예보 데이터 - lines.append(f'') - lines.append('
시간강수량
{time_str}{mm}mm
영업시간 중 총 예상 강수량: {total_rainfall:.1f}mm

08:00 파주 조리읍 기상청 단기예보 기준

') + total_rainfall = 0.0 + time_precip_list = [] - html_summary = ''.join(lines) + if debug: + # 디버그 모드: 콘솔에 시간별 강수량과 총합 출력 + print(f"[DEBUG MODE] {TODAY} 10:00 ~ 22:00 예상 강수량") + for hour in range(10, 23): + # 초단기예보 우선, 없으면 단기예보 사용 + mm = ultra.get(hour, vilage.get(hour, 0.0)) + time_str = f"{hour:02d}:00" + print(f"{time_str} → {mm}mm") + total_rainfall += mm + print(f"영업시간 총 예상 강수량: {total_rainfall:.1f}mm") + return None # DB 저장, HTML 반환 없이 종료 - # SQLite 저장 함수 호출 - save_weather_to_sqlite(date=TODAY, time_precip_list=time_precip_list, total_rainfall=total_rainfall) + # debug가 False인 경우 HTML 테이블 생성 및 DB 저장 + lines = [ + '
', + '

[10:00 ~ 22:00 예상 강수량]

', + '', + '', + '', + '', + '' + ] - return html_summary + for hour in range(10, 23): + mm = ultra.get(hour, vilage.get(hour, 0.0)) + time_str = f"{hour:02d}:00" + time_precip_list.append((time_str, mm)) + lines.append( + f'' + f'' + ) + total_rainfall += mm - except json.decoder.JSONDecodeError: - if retry: - print("JSON 디코드 오류 발생, 재시도 중...") - time.sleep(3) # 3초 대기 - return get_precipitation_summary(retry=False) - else: - print("응답이 JSON 형식이 아닙니다.") - return '' + lines.append( + f'' + ) + lines.append('
시간강수량
{time_str}{mm}mm
' + f'영업시간 총 예상 강수량: {total_rainfall:.1f}mm

08:50 초단기 + 08:00 단기 예보 기준

') + + html_summary = ''.join(lines) + + # SQLite DB에 강수량 데이터 저장 + save_weather_to_sqlite(TODAY, time_precip_list, total_rainfall) + + return html_summary def save_weather_to_sqlite(date, time_precip_list, total_rainfall): - db_path = '/data/weather.sqlite' - os.makedirs(os.path.dirname(db_path), exist_ok=True) + """ + 강수량 데이터를 SQLite DB에 저장하는 함수 + - 테이블이 없으면 생성 + - 동일 날짜 데이터는 기존 삭제 후 삽입 + """ + db_path = '/data/weather.sqlite' # DB 파일 경로 + os.makedirs(os.path.dirname(db_path), exist_ok=True) # 폴더 없으면 생성 conn = sqlite3.connect(db_path) curs = conn.cursor() - # 테이블 생성 (없으면) + # 강수량 상세 데이터 테이블 생성 curs.execute(''' CREATE TABLE IF NOT EXISTS precipitation ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -99,6 +173,7 @@ def save_weather_to_sqlite(date, time_precip_list, total_rainfall): ) ''') + # 강수량 요약 데이터 테이블 생성 curs.execute(''' CREATE TABLE IF NOT EXISTS precipitation_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, @@ -107,20 +182,21 @@ def save_weather_to_sqlite(date, time_precip_list, total_rainfall): ) ''') - # 기존 데이터 삭제 (동일 날짜) + # 동일 날짜 기존 데이터 삭제 curs.execute('DELETE FROM precipitation WHERE date = ?', (date,)) curs.execute('DELETE FROM precipitation_summary WHERE date = ?', (date,)) - # 데이터 삽입 + # 시간별 강수량 데이터 삽입 curs.executemany('INSERT INTO precipitation (date, time, rainfall) VALUES (?, ?, ?)', [(date, t, r) for t, r in time_precip_list]) + # 총 강수량 요약 데이터 삽입 curs.execute('INSERT INTO precipitation_summary (date, total_rainfall) VALUES (?, ?)', (date, total_rainfall)) conn.commit() conn.close() print(f"[DB 저장 완료] {date} 강수량 데이터 저장됨") -# 테스트용 if __name__ == "__main__": - print(get_precipitation_summary()) + # 메인 실행: 함수 호출만 (결과는 디버그 모드에 따라 다름) + get_precipitation_summary()