import requests import json import re import sqlite3 import os 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.]+", str(value)) return float(match.group()) if match else 0.0 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', 'pageNo': '1', 'dataType': 'JSON', 'base_date': TODAY, 'base_time': '0800', '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'] == '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 def get_precipitation_summary(): """ 초단기예보와 단기예보를 혼합하여 10시부터 22시까지 시간별 예상 강수량 요약 생성 - debug 모드일 경우 평문 출력만 수행하며 DB 저장 및 HTML 반환은 하지 않음 - debug 모드가 아닐 경우 HTML 포맷으로 테이블 생성 후 DB 저장 및 HTML 반환 """ ultra = get_ultra_data() # 초단기 예보 데이터 vilage = get_vilage_data() # 단기 예보 데이터 total_rainfall = 0.0 time_precip_list = [] 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 반환 없이 종료 # debug가 False인 경우 HTML 테이블 생성 및 DB 저장 lines = [ '
', '

[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" time_precip_list.append((time_str, mm)) lines.append( f'' f'' ) total_rainfall += mm 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) print(f"[HTML 생성 완료] {TODAY} 강수 요약 HTML 생성됨") return html_summary def save_weather_to_sqlite(date, time_precip_list, total_rainfall): """ 강수량 데이터를 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, date TEXT NOT NULL, time TEXT NOT NULL, rainfall REAL NOT NULL ) ''') # 강수량 요약 데이터 테이블 생성 curs.execute(''' CREATE TABLE IF NOT EXISTS precipitation_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL UNIQUE, total_rainfall REAL NOT NULL ) ''') # 동일 날짜 기존 데이터 삭제 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__": # 메인 실행: 함수 호출만 (결과는 디버그 모드에 따라 다름) get_precipitation_summary()