# =================================================================== # services/weather/precipitation.py # 강수량 데이터 서비스 모듈 # =================================================================== # 시간별 강수량 예보 데이터를 조회하고 요약 정보를 생성합니다. # HTML 테이블 생성 및 SQLite/MySQL 저장을 지원합니다. # =================================================================== """ 강수량 데이터 서비스 모듈 시간별 강수량 예보 데이터를 조회하고 요약 정보를 생성합니다. 다양한 출력 형식(HTML, JSON, 텍스트)을 지원합니다. 사용 예시: from services.weather.precipitation import PrecipitationService service = PrecipitationService(service_key) summary = service.get_daily_summary() html = service.generate_html_table() """ import os import sqlite3 from datetime import datetime from typing import Dict, List, Optional, Tuple, Any from core.logging_utils import get_logger from core.config import get_config try: # 상대 import (패키지 내에서) from .forecast import parse_precip, get_ultra_forecast, get_vilage_forecast except ImportError: # 절대 import (직접 실행 시) from services.weather.forecast import parse_precip, get_ultra_forecast, get_vilage_forecast logger = get_logger(__name__) class PrecipitationService: """ 강수량 데이터 서비스 클래스 초단기예보와 단기예보를 조합하여 시간별 강수량 예보를 제공합니다. Attributes: service_key: 기상청 API 서비스 키 start_hour: 집계 시작 시간 (기본: 10) end_hour: 집계 종료 시간 (기본: 22) """ def __init__( self, service_key: Optional[str] = None, start_hour: int = 10, end_hour: int = 22 ): """ Args: service_key: API 키 (None이면 설정에서 로드) start_hour: 집계 시작 시간 end_hour: 집계 종료 시간 """ if service_key is None: config = get_config() service_key = config.weather_service.get('service_key') or config.data_api.get('service_key', '') self.service_key = service_key self.start_hour = start_hour self.end_hour = end_hour def get_hourly_ultra_data(self, target_date: Optional[str] = None) -> Dict[int, float]: """ 초단기예보에서 시간별 강수량 추출 Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 Returns: 시간별 강수량 딕셔너리 {시간: 강수량(mm)} """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') items = get_ultra_forecast(self.service_key) result = {} for item in items: if item.get('category') != 'RN1': continue if item.get('fcstDate') != target_date: continue try: hour = int(item['fcstTime'][:2]) if self.start_hour <= hour <= self.end_hour: result[hour] = parse_precip(item['fcstValue']) except (ValueError, KeyError): continue return result def get_hourly_vilage_data(self, target_date: Optional[str] = None) -> Dict[int, float]: """ 단기예보에서 시간별 강수량 추출 Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 Returns: 시간별 강수량 딕셔너리 {시간: 강수량(mm)} """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') items = get_vilage_forecast(self.service_key, base_date=target_date) result = {} for item in items: if item.get('category') != 'PCP': continue if item.get('fcstDate') != target_date: continue try: hour = int(item['fcstTime'][:2]) if self.start_hour <= hour <= self.end_hour: result[hour] = parse_precip(item['fcstValue']) except (ValueError, KeyError): continue return result def get_daily_summary( self, target_date: Optional[str] = None ) -> Tuple[List[Tuple[str, float]], float]: """ 일별 강수량 요약 조회 초단기예보를 우선으로 하고, 없는 시간대는 단기예보로 보완합니다. Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 Returns: (시간별 강수량 리스트, 총 강수량) 튜플 시간별: [('10:00', 0.5), ('11:00', 0.0), ...] """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') # 초단기예보 데이터 (우선) ultra_data = self.get_hourly_ultra_data(target_date) # 단기예보 데이터 (보완용) vilage_data = self.get_hourly_vilage_data(target_date) time_precip_list = [] total_rainfall = 0.0 for hour in range(self.start_hour, self.end_hour + 1): # 초단기예보 우선, 없으면 단기예보 사용 rainfall = ultra_data.get(hour, vilage_data.get(hour, 0.0)) time_str = f"{hour:02d}:00" time_precip_list.append((time_str, rainfall)) total_rainfall += rainfall return time_precip_list, round(total_rainfall, 1) def generate_html_table( self, target_date: Optional[str] = None, title: Optional[str] = None ) -> str: """ 강수량 요약 HTML 테이블 생성 Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 title: 테이블 제목 (None이면 기본 제목) Returns: HTML 문자열 """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') time_precip_list, total_rainfall = self.get_daily_summary(target_date) if title is None: title = f"[{self.start_hour}:00 ~ {self.end_hour}:00 예상 강수량]" lines = [ '
', f'

{title}

', '', '', '', '', '' ] for time_str, rainfall in time_precip_list: lines.append( f'' f'' ) lines.append( f'' ) lines.append('
시간강수량
{time_str}{rainfall}mm
' f'총 예상 강수량: {total_rainfall:.1f}mm
') lines.append('

초단기 + 단기 예보 기준

') lines.append('
') return ''.join(lines) def generate_text_summary( self, target_date: Optional[str] = None ) -> str: """ 강수량 요약 텍스트 생성 Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 Returns: 텍스트 요약 문자열 """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') time_precip_list, total_rainfall = self.get_daily_summary(target_date) lines = [ f"📅 {target_date} 예상 강수량", f"⏰ {self.start_hour}:00 ~ {self.end_hour}:00", "-" * 20 ] for time_str, rainfall in time_precip_list: lines.append(f"{time_str} → {rainfall}mm") lines.append("-" * 20) lines.append(f"☔ 총 예상 강수량: {total_rainfall:.1f}mm") return "\n".join(lines) def save_to_sqlite( self, target_date: Optional[str] = None, db_path: str = '/data/weather.sqlite' ): """ 강수량 데이터를 SQLite에 저장 Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 db_path: SQLite 데이터베이스 파일 경로 """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') time_precip_list, total_rainfall = self.get_daily_summary(target_date) # 디렉토리 생성 os.makedirs(os.path.dirname(db_path), exist_ok=True) conn = sqlite3.connect(db_path) cursor = conn.cursor() # 테이블 생성 cursor.execute(''' CREATE TABLE IF NOT EXISTS precipitation ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL, time TEXT NOT NULL, rainfall REAL NOT NULL ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS precipitation_summary ( id INTEGER PRIMARY KEY AUTOINCREMENT, date TEXT NOT NULL UNIQUE, total_rainfall REAL NOT NULL ) ''') # 기존 데이터 삭제 cursor.execute('DELETE FROM precipitation WHERE date = ?', (target_date,)) cursor.execute('DELETE FROM precipitation_summary WHERE date = ?', (target_date,)) # 시간별 데이터 삽입 cursor.executemany( 'INSERT INTO precipitation (date, time, rainfall) VALUES (?, ?, ?)', [(target_date, t, r) for t, r in time_precip_list] ) # 총 강수량 삽입 cursor.execute( 'INSERT INTO precipitation_summary (date, total_rainfall) VALUES (?, ?)', (target_date, total_rainfall) ) conn.commit() conn.close() logger.info(f"SQLite 저장 완료: {target_date}, 총 {total_rainfall}mm") def get_precipitation_info( self, target_date: Optional[str] = None, output_format: str = 'dict' ) -> Any: """ 강수량 정보 조회 (다양한 형식 지원) Args: target_date: 대상 날짜 (YYYYMMDD). None이면 오늘 output_format: 출력 형식 ('dict', 'html', 'text', 'json') Returns: 요청된 형식의 강수량 정보 """ if target_date is None: target_date = datetime.now().strftime('%Y%m%d') if output_format == 'html': return self.generate_html_table(target_date) elif output_format == 'text': return self.generate_text_summary(target_date) elif output_format == 'json': time_precip_list, total = self.get_daily_summary(target_date) return { 'date': target_date, 'hourly': [{'time': t, 'rainfall': r} for t, r in time_precip_list], 'total': total } else: time_precip_list, total = self.get_daily_summary(target_date) return { 'date': target_date, 'hourly': time_precip_list, 'total': total } if __name__ == '__main__': """ 강수량 데이터 서비스 모듈 테스트 사용법: python services/weather/precipitation.py """ logger = get_logger(__name__) logger.info("=== 강수량 데이터 서비스 모듈 테스트 ===") try: config = get_config() service_key = config.data_api['service_key'] or "TEST_KEY" logger.info(f"설정 로드 완료") logger.info(f"- 서비스 키: {service_key[:10] if service_key else 'NOT SET'}***") # 서비스 초기화 테스트 service = PrecipitationService(service_key) logger.info("\nPrecipitationService 초기화 완료") logger.info("\n제공 기능:") logger.info("- get_daily_summary: 일일 강수량 요약") logger.info("- generate_html_table: HTML 테이블 생성") logger.info("- get_output_format: 다양한 형식 제공 (JSON, HTML, 텍스트)") logger.info("\n✓ 강수량 데이터 서비스 모듈 테스트 완료") except Exception as e: logger.error(f"강수량 모듈 테스트 실패: {e}") import traceback logger.error(traceback.format_exc())