firstgarden/fg-auto
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

캡처 예보에 초단기 예보 + 단기 예보 형식으로 입력되어 단기예보만을 활용하면 정확도가 떨어지므로, 초단기 예보를 혼합해서 출력하도록 코드를 수정함.

Browse Source 
유지보수용 주석 및 디버그 변수 추가.
This commit is contained in:
KWON
2025-07-01 09:50:27 +09:00
parent a0bb137d52
commit 339a6c4de4
1 changed files with 131 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files

174
autouploader/weather.py
View File

@ -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)
try:
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; box-sizing: border-box;">',
'<h3 style="font-size: 1.8em; text-align: center; margin: 20px 0;">[시간대별 예상 강수량]</h3>',
'<table style="border-collapse: collapse; width: 100%; max-width: 400px; margin: 0 auto; font-size: 1em;">',
'<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>'
'<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 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'<tr><td style="border: 1px solid #333; padding: 2px;text-align: center;">{time_str}</td><td style="border: 1px solid #333; padding: 2px;text-align: center;">{mm}mm</td></tr>')
total_rainfall += mm
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; padding: 2px;text-align: center; font-weight: bold;">영업시간 중 총 예상 강수량: {total_rainfall:.1f}mm</td></tr>')
lines.append('</tbody></table><p style="text-align:right; font-size: 0.8em;">08:00 파주 조리읍 기상청 단기예보 기준</div>')
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 저장 함수 호출
save_weather_to_sqlite(date=TODAY, time_precip_list=time_precip_list, total_rainfall=total_rainfall)
# SQLite DB에 강수량 데이터 저장
save_weather_to_sqlite(TODAY, time_precip_list, total_rainfall)
return html_summary
except json.decoder.JSONDecodeError:
if retry:
print("JSON 디코드 오류 발생, 재시도 중...")
time.sleep(3) # 3초 대기
return get_precipitation_summary(retry=False)
else:
print("응답이 JSON 형식이 아닙니다.")
return ''
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()