204 lines
7.6 KiB
Python
204 lines
7.6 KiB
Python
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 = [
|
|
'<div class="weatherinfo" style="max-width: 100%; overflow-x: auto; padding: 10px;">',
|
|
'<h3 style="font-size: 1.8em; text-align: center; margin: 20px 0;">[10:00 ~ 22:00 예상 강수량]</h3>',
|
|
'<table style="border-collapse: collapse; width: 100%; max-width: 400px; margin: 0 auto;">',
|
|
'<thead><tr>',
|
|
'<th style="border: 1px solid #333; padding: 2px; background-color: #f0f0f0;">시간</th>',
|
|
'<th style="border: 1px solid #333; padding: 2px; background-color: #f0f0f0;">강수량</th>',
|
|
'</tr></thead><tbody>'
|
|
]
|
|
|
|
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'<tr><td style="border: 1px solid #333; text-align: center;">{time_str}</td>'
|
|
f'<td style="border: 1px solid #333; text-align: center;">{mm}mm</td></tr>'
|
|
)
|
|
total_rainfall += mm
|
|
|
|
lines.append(
|
|
f'<tr><td colspan="2" style="border: 1px solid #333; text-align: center; font-weight: bold;">'
|
|
f'영업시간 총 예상 강수량: {total_rainfall:.1f}mm</td></tr>'
|
|
)
|
|
lines.append('</tbody></table><p style="text-align:right; font-size: 0.8em;">08:50 초단기 + 08:00 단기 예보 기준</p></div>')
|
|
|
|
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()
|