294 lines
9.5 KiB
Python
294 lines
9.5 KiB
Python
import os
|
|
import requests
|
|
import json
|
|
import re
|
|
from flask import Flask, request, jsonify, send_from_directory, make_response
|
|
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
from config import serviceKey
|
|
import logging
|
|
|
|
# 로깅 설정
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
app = Flask(__name__)
|
|
|
|
# 환경 변수에서 설정값 불러오기
|
|
DB_PATH = '/data/weather.sqlite'
|
|
DOMAIN = os.getenv('DOMAIN', 'http://localhost:5000')
|
|
debug_env = os.getenv('FLASK_DEBUG', '0')
|
|
DEBUG_MODE = debug_env == '1'
|
|
|
|
def parse_rainfall_value(value):
|
|
"""강수량 텍스트를 숫자로 변환"""
|
|
if not value or value == '-':
|
|
return 0.0
|
|
elif '~' in value:
|
|
return 0.5
|
|
else:
|
|
try:
|
|
return float(re.search(r'[\d.]+', value).group())
|
|
except (AttributeError, ValueError):
|
|
return 0.0
|
|
|
|
def get_captured_rainfall(date):
|
|
"""
|
|
캡처된 강수량 데이터 조회 (당일 기준)
|
|
|
|
Args:
|
|
date: 'YYYYMMDD' 형식
|
|
|
|
Returns:
|
|
tuple: (시간별_강수량_목록, 총강수량, 캡처_시각)
|
|
"""
|
|
try:
|
|
conn = sqlite3.connect(DB_PATH)
|
|
curs = conn.cursor()
|
|
|
|
curs.execute('''
|
|
SELECT hour, rainfall FROM rainfall_capture
|
|
WHERE date = ? ORDER BY hour
|
|
''', (date,))
|
|
hourly_data = curs.fetchall()
|
|
|
|
curs.execute('''
|
|
SELECT total_rainfall, capture_time FROM rainfall_summary
|
|
WHERE date = ?
|
|
''', (date,))
|
|
summary = curs.fetchone()
|
|
|
|
conn.close()
|
|
|
|
total = summary[0] if summary else 0.0
|
|
capture_time = summary[1] if summary else None
|
|
|
|
return hourly_data, total, capture_time
|
|
except Exception as e:
|
|
logger.error(f"캡처 강수량 조회 실패: {e}")
|
|
return [], 0.0, None
|
|
|
|
def get_forecast_rainfall(date):
|
|
"""
|
|
기상청 API를 통한 강수량 예보 조회
|
|
|
|
Args:
|
|
date: 'YYYYMMDD' 형식
|
|
|
|
Returns:
|
|
tuple: (시간별_강수량_목록, 총강수량) 또는 ([], 0.0)
|
|
"""
|
|
try:
|
|
# 기상청 초단기 예보 API
|
|
url = "http://apis.data.go.kr/1360000/VilageFcstInfoService_2.0/getUltraSrtFcst"
|
|
|
|
# 예보는 08:50 발표 기준
|
|
params = {
|
|
'serviceKey': serviceKey,
|
|
'numOfRows': '1000',
|
|
'pageNo': '1',
|
|
'dataType': 'JSON',
|
|
'base_date': date,
|
|
'base_time': '0850',
|
|
'nx': '57',
|
|
'ny': '130'
|
|
}
|
|
|
|
response = requests.get(url, params=params, timeout=10)
|
|
data = response.json()
|
|
|
|
if data['response']['header']['resultCode'] != '00':
|
|
return [], 0.0
|
|
|
|
rainfall_by_hour = {}
|
|
for item in data['response']['body']['items']['item']:
|
|
if item['category'] == 'RN1': # 1시간 강수량
|
|
hour = int(item['fcstTime'][:2])
|
|
if 10 <= hour <= 21: # 10시~21시(오후 9시)
|
|
rainfall_by_hour[hour] = parse_rainfall_value(item['fcstValue'])
|
|
|
|
hourly_list = [(h, rainfall_by_hour.get(h, 0.0)) for h in range(10, 22)]
|
|
total = sum(rain for _, rain in hourly_list)
|
|
|
|
return hourly_list, total
|
|
except Exception as e:
|
|
logger.error(f"API 강수량 예보 조회 실패: {e}")
|
|
return [], 0.0
|
|
|
|
def get_rainfall_data(date_str):
|
|
"""
|
|
날짜별 강수량 데이터 조회
|
|
- 당일(오늘): 09:00 캡처된 실제 데이터
|
|
- 미래 날짜: 기상청 예보 API
|
|
|
|
Args:
|
|
date_str: 'YYYYMMDD' 형식
|
|
|
|
Returns:
|
|
dict: {
|
|
'date': 날짜,
|
|
'is_forecast': 예보 여부,
|
|
'hourly_data': [(hour, rainfall), ...],
|
|
'total': 총강수량,
|
|
'note': 추가 설명
|
|
}
|
|
"""
|
|
today = datetime.now().strftime('%Y%m%d')
|
|
is_forecast = date_str > today
|
|
|
|
if is_forecast:
|
|
# 미래 날짜: API 예보
|
|
hourly, total = get_forecast_rainfall(date_str)
|
|
note = "⚠️ 이는 기상청 08:00 발표 예보입니다. 실제 이벤트 적용 기준은 당일 09:00 캡처 데이터입니다."
|
|
else:
|
|
# 당일 이상: 캡처 데이터
|
|
hourly, total, timestamp = get_captured_rainfall(date_str)
|
|
note = None
|
|
|
|
return {
|
|
'date': date_str,
|
|
'is_forecast': is_forecast,
|
|
'hourly_data': hourly,
|
|
'total': total,
|
|
'note': note
|
|
}
|
|
|
|
# 정적 파일 서빙: /data/ 경로로 이미지 접근 가능하게 함
|
|
@app.route('/data/<path:filename>')
|
|
def serve_data_file(filename):
|
|
return send_from_directory('/data', filename)
|
|
|
|
@app.route('/webhook', methods=['POST'])
|
|
def webhook():
|
|
"""
|
|
카카오 챗봇 웹훅
|
|
|
|
사용자가 요청한 날짜의 강수량 정보 응답
|
|
- 당일: 09:00 캡처 데이터
|
|
- 미래: 기상청 API 예보
|
|
"""
|
|
try:
|
|
data = request.get_json(silent=True)
|
|
|
|
# 사용자 요청 날짜 파싱 (기본값: 오늘)
|
|
today = datetime.now().strftime('%Y%m%d')
|
|
query_date = today
|
|
|
|
# 사용자 발화에서 날짜 추출 시도
|
|
if data and 'userRequest' in data and 'utterance' in data['userRequest']:
|
|
utterance = data['userRequest']['utterance'].strip()
|
|
|
|
# 내일, 모레 등의 상대 날짜 파싱
|
|
if '내일' in utterance:
|
|
query_date = (datetime.now() + timedelta(days=1)).strftime('%Y%m%d')
|
|
elif '모레' in utterance:
|
|
query_date = (datetime.now() + timedelta(days=2)).strftime('%Y%m%d')
|
|
elif '오늘' in utterance or utterance in ['', None]:
|
|
query_date = today
|
|
else:
|
|
# YYYYMMDD 형식의 날짜 찾기
|
|
date_match = re.search(r'(\d{8})', utterance)
|
|
if date_match:
|
|
query_date = date_match.group(1)
|
|
|
|
rainfall_info = get_rainfall_data(query_date)
|
|
|
|
# 응답 메시지 구성
|
|
date_obj = datetime.strptime(query_date, '%Y%m%d')
|
|
date_str = date_obj.strftime('%m월 %d일(%a)')
|
|
|
|
# 강수량 상세 정보
|
|
lines = [f"📅 {date_str}"]
|
|
if rainfall_info['is_forecast']:
|
|
lines.append("📊 예보 강수량 (08:00 발표 기준)")
|
|
else:
|
|
lines.append("📊 실제 강수량 (09:00 캡처 기준)")
|
|
|
|
lines.append("")
|
|
|
|
if rainfall_info['hourly_data']:
|
|
for hour, rainfall in rainfall_info['hourly_data']:
|
|
if isinstance(hour, tuple): # (hour, rainfall) 튜플인 경우
|
|
hour, rainfall = hour
|
|
rain_str = f"{rainfall:.1f}mm" if rainfall > 0 else "☀️ 강수 없음"
|
|
lines.append(f"{hour:02d}:00 → {rain_str}")
|
|
lines.append("")
|
|
lines.append(f"💧 총 강수량: {rainfall_info['total']:.1f}mm")
|
|
|
|
# 이벤트 적용 여부
|
|
if rainfall_info['total'] > 10:
|
|
lines.append("✅ 레이니데이 이벤트 기준(10mm 초과) 충족")
|
|
else:
|
|
lines.append("❌ 이벤트 기준(10mm 초과)을 충족하지 않음")
|
|
else:
|
|
lines.append("데이터를 찾을 수 없습니다.")
|
|
|
|
if rainfall_info['note']:
|
|
lines.append("")
|
|
lines.append(rainfall_info['note'])
|
|
|
|
response_text = '\n'.join(lines)
|
|
|
|
# 이미지 포함 여부 확인 (당일만)
|
|
if not rainfall_info['is_forecast']:
|
|
image_filename = f"weather_capture_{query_date}.png"
|
|
image_path = f"/data/{image_filename}"
|
|
outputs = [{
|
|
"simpleText": {
|
|
"text": response_text
|
|
}
|
|
}]
|
|
if os.path.isfile(image_path):
|
|
image_url = f"{DOMAIN}/data/{image_filename}"
|
|
outputs.append({
|
|
"image": {
|
|
"imageUrl": image_url,
|
|
"altText": f"{date_str} 날씨 캡처"
|
|
}
|
|
})
|
|
else:
|
|
outputs = [{
|
|
"simpleText": {
|
|
"text": response_text
|
|
}
|
|
}]
|
|
|
|
response_body = {
|
|
"version": "2.0",
|
|
"template": {
|
|
"outputs": outputs
|
|
}
|
|
}
|
|
|
|
resp = make_response(jsonify(response_body))
|
|
resp.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return resp
|
|
|
|
except Exception as e:
|
|
logger.error(f"웹훅 처리 중 오류: {e}", exc_info=True)
|
|
error_body = {
|
|
"version": "2.0",
|
|
"template": {
|
|
"outputs": [{
|
|
"simpleText": {
|
|
"text": f"❌ 오류가 발생했습니다: {str(e)}\n관리자에게 문의하세요."
|
|
}
|
|
}]
|
|
}
|
|
}
|
|
resp = make_response(jsonify(error_body))
|
|
resp.headers['Content-Type'] = 'application/json; charset=utf-8'
|
|
return resp
|
|
|
|
# 헬스 체크 엔드포인트
|
|
@app.route('/health', methods=['GET'])
|
|
def health_check():
|
|
return jsonify({'status': 'ok', 'timestamp': datetime.now().isoformat()}), 200
|
|
|
|
if __name__ == '__main__':
|
|
logger.info("Flask 웹서버 시작 (포트 5000)")
|
|
app.run(host='0.0.0.0', port=5000, debug=DEBUG_MODE)
|