# -*- coding: utf-8 -*- """ 날씨 서비스 모듈 날씨 정보를 가져와 업데이트합니다. """ import json import os from datetime import datetime, timedelta, date from pathlib import Path from typing import Optional, Dict, Any, List from PySide6.QtCore import QObject, QTimer, QThread, Signal from core.config import ConfigManager from core.signals import GlobalSignals from core.constants import WEATHER_UPDATE_INTERVAL, DATA_DIR from core.logger import get_logger from database.crud import CRUDManager logger = get_logger(__name__) # 날씨 HTML 파일 경로 WEATHER_HTML_FILE = DATA_DIR / "weather_debug.html" WEATHER_TIMESTAMP_FILE = DATA_DIR / "weather_timestamp.txt" WEATHER_CACHE_DURATION = timedelta(hours=2) # 2시간 class WeatherWorker(QObject): """날씨 정보 가져오기 워커""" finished = Signal(dict) error = Signal(str) def __init__(self, lat: float, lon: float, code: str, force_refresh: bool = False): super().__init__() self.lat = lat self.lon = lon self.code = code self.force_refresh = force_refresh self.crud = CRUDManager() def run(self): """날씨 정보 가져오기""" try: # HTML 파일이 있고 2시간 이내면 파일에서 로드 if not self.force_refresh and self._is_cache_valid(): logger.info("캐시된 날씨 데이터 사용") html_content = self._load_html_file() if html_content: weather_data = self._parse_weather_data(html_content) if weather_data: self.finished.emit(weather_data) return # 네트워크에서 새로 가져오기 logger.info("기상청 API에서 날씨 정보 가져오기") html_content = self._fetch_weather_html() if html_content: # HTML 파일 저장 self._save_html_file(html_content) # 타임스탬프 저장 self._save_timestamp() # 파싱 weather_data = self._parse_weather_data(html_content) if weather_data: self.finished.emit(weather_data) else: self.error.emit("날씨 데이터 파싱 실패") else: self.error.emit("날씨 정보를 가져올 수 없습니다") except Exception as e: logger.error(f"날씨 정보 가져오기 실패: {e}") self.error.emit(str(e)) def _is_cache_valid(self) -> bool: """캐시가 유효한지 확인 (2시간 이내)""" if not WEATHER_HTML_FILE.exists() or not WEATHER_TIMESTAMP_FILE.exists(): return False try: with open(WEATHER_TIMESTAMP_FILE, 'r', encoding='utf-8') as f: timestamp_str = f.read().strip() timestamp = datetime.fromisoformat(timestamp_str) elapsed = datetime.now() - timestamp return elapsed < WEATHER_CACHE_DURATION except Exception as e: logger.error(f"타임스탬프 확인 실패: {e}") return False def _load_html_file(self) -> Optional[str]: """HTML 파일 로드""" try: if WEATHER_HTML_FILE.exists(): with open(WEATHER_HTML_FILE, 'r', encoding='utf-8') as f: return f.read() except Exception as e: logger.error(f"HTML 파일 로드 실패: {e}") return None def _save_html_file(self, content: str): """HTML 파일 저장""" try: DATA_DIR.mkdir(parents=True, exist_ok=True) with open(WEATHER_HTML_FILE, 'w', encoding='utf-8') as f: f.write(content) logger.info(f"날씨 HTML 파일 저장: {WEATHER_HTML_FILE}") except Exception as e: logger.error(f"HTML 파일 저장 실패: {e}") def _save_timestamp(self): """타임스탬프 저장""" try: DATA_DIR.mkdir(parents=True, exist_ok=True) with open(WEATHER_TIMESTAMP_FILE, 'w', encoding='utf-8') as f: f.write(datetime.now().isoformat()) except Exception as e: logger.error(f"타임스탬프 저장 실패: {e}") def _get_timestamp(self) -> Optional[datetime]: """저장된 타임스탬프 가져오기""" try: if WEATHER_TIMESTAMP_FILE.exists(): with open(WEATHER_TIMESTAMP_FILE, 'r', encoding='utf-8') as f: timestamp_str = f.read().strip() return datetime.fromisoformat(timestamp_str) except Exception as e: logger.error(f"타임스탬프 읽기 실패: {e}") return None def _fetch_weather_html(self) -> Optional[str]: """기상청 API에서 HTML 가져오기""" try: from curl_cffi import requests url = "https://www.weather.go.kr/w/wnuri-fct2021/main/digital-forecast.do" params = { "code": self.code, "unit": "m/s", "hr1": "Y", "lat": str(self.lat), "lon": str(self.lon) } headers = { "User-Agent": "Mozilla/5.0", "Referer": "https://www.weather.go.kr/w/index.do", "X-Requested-With": "XMLHttpRequest" } response = requests.get(url, params=params, headers=headers, impersonate="chrome120", timeout=30) response.raise_for_status() return response.text except Exception as e: logger.error(f"기상청 API 요청 실패: {e}") return None def _parse_weather_data(self, html_content: str) -> Optional[Dict[str, Any]]: """HTML에서 날씨 데이터 파싱""" try: from selectolax.parser import HTMLParser tree = HTMLParser(html_content) slides = tree.css("div.slide-wrap div.slide") if not slides: logger.warning("날씨 데이터 슬라이드를 찾지 못했습니다") return None # 현재 시간 기준으로 가장 가까운 데이터 찾기 now = datetime.now() current_hour = now.hour weather_items = [] for slide in slides: daily_node = slide.css_first("div.daily") if not daily_node: continue date_str = daily_node.attributes.get('data-date', '').strip() ul_items = daily_node.css("ul") for ul in ul_items: if "item" not in ul.attributes.get("class", ""): continue w_data = { "time": "-", "weather": "-", "temp": "-", "feels_like": "-", "prob": "-", "wind": "-", "humid": "-" } if 'data-time' in ul.attributes: w_data['time'] = ul.attributes['data-time'] lis = ul.css("li") for li in lis: label_node = li.css_first("span.hid") if not label_node: continue label_text = label_node.text(strip=True) if "시각" in label_text and w_data['time'] == "-": val = li.css_first("span:not(.hid)") if val: w_data['time'] = val.text(strip=True) elif "날씨" in label_text: wic = li.css_first(".wic") if wic: w_data['weather'] = wic.attributes.get("title") or wic.text(strip=True) elif "기온" in label_text: feel_node = li.css_first(".feel") if feel_node: w_data['temp'] = feel_node.text(deep=False, strip=True) elif "체감온도" in label_text and "기온" not in label_text: spans = li.css("span") if len(spans) > 1: w_data['feels_like'] = spans[-1].text(strip=True) elif "강수확률" in label_text: spans = li.css("span") if len(spans) > 1: prob = spans[-1].text(strip=True) w_data['prob'] = prob if prob else "-" elif "바람" in label_text: wd_node = li.css_first(".wdic") ws_node = li.css_first(".wspd:not(.qwsd)") wd = wd_node.text(strip=True) if wd_node else "" ws = ws_node.text(strip=True) if ws_node else "" w_data['wind'] = f"{wd} {ws}".strip() elif "습도" in label_text: spans = li.css("span") if len(spans) > 1: w_data['humid'] = spans[-1].text(strip=True) if w_data['time'] != "-": weather_items.append({ 'date': date_str, 'time': w_data['time'], **w_data }) if not weather_items: return None # 현재 시간에 가장 가까운 데이터 찾기 current_data = self._find_current_weather(weather_items, current_hour) # 주간조(09~18시)일 경우 추가 정보 계산 if 9 <= current_hour <= 18: day_data = self._calculate_daytime_stats(weather_items, current_hour) current_data.update(day_data) # 타임스탬프 정보 추가 timestamp = self._get_timestamp() current_data['fetched_at'] = timestamp.isoformat() if timestamp else None # 모든 날씨 데이터를 DB에 저장 try: all_weather_data = WeatherService._parse_all_weather_items(html_content) WeatherService._save_weather_data_to_db_static(all_weather_data, self.code) except Exception as e: logger.error(f"날씨 데이터 DB 저장 실패: {e}") return current_data except Exception as e: logger.error(f"날씨 데이터 파싱 실패: {e}") return None def _find_current_weather(self, items: List[Dict], current_hour: int) -> Dict[str, Any]: """현재 시간에 가장 가까운 날씨 데이터 찾기""" if not items: return {} # 현재 시간과 가장 가까운 항목 찾기 best_item = items[0] min_diff = float('inf') for item in items: try: time_str = item['time'] # "24:00" 같은 경우 처리 if time_str.endswith(':00'): hour_str = time_str.split(':')[0] hour = int(hour_str) if hour == 24: hour = 0 diff = abs(hour - current_hour) if diff < min_diff: min_diff = diff best_item = item except (ValueError, KeyError): continue # 기본 데이터 구조 생성 result = { "temp": self._parse_temp(best_item.get('temp', '-')), "condition": best_item.get('weather', '정보 없음'), "icon": self._get_weather_icon_from_text(best_item.get('weather', '')), "humidity": self._parse_percentage(best_item.get('humid', '-')), "wind_speed": best_item.get('wind', '-'), "feels_like": self._parse_temp(best_item.get('feels_like', '-')), "precipitation_prob": self._parse_percentage(best_item.get('prob', '-')), } return result def _calculate_daytime_stats(self, items: List[Dict], current_hour: int) -> Dict[str, Any]: """주간조(09~18시) 통계 계산""" # 09~18시 데이터만 필터링 daytime_items = [] for item in items: try: time_str = item['time'] if time_str.endswith(':00'): hour_str = time_str.split(':')[0] hour = int(hour_str) if hour == 24: hour = 0 if 9 <= hour <= 18: daytime_items.append(item) except (ValueError, KeyError): continue if not daytime_items: return {} # 온도 추출 temps = [] feels_likes = [] precip_probs = [] for item in daytime_items: temp = self._parse_temp(item.get('temp', '-')) feels = self._parse_temp(item.get('feels_like', '-')) prob = self._parse_percentage(item.get('prob', '-')) if temp is not None: temps.append(temp) if feels is not None: feels_likes.append(feels) if prob is not None: precip_probs.append(prob) result = {} if temps: result['temp_min'] = min(temps) result['temp_max'] = max(temps) if feels_likes: result['feels_like_min'] = min(feels_likes) result['feels_like_max'] = max(feels_likes) if precip_probs: result['precipitation_prob_max'] = max(precip_probs) return result def _parse_temp(self, temp_str: str) -> Optional[int]: """온도 문자열 파싱 (예: "4℃" -> 4)""" try: if temp_str == "-" or not temp_str: return None # 숫자만 추출 import re match = re.search(r'-?\d+', temp_str) if match: return int(match.group()) except (ValueError, AttributeError): pass return None def _parse_percentage(self, percent_str: str) -> Optional[int]: """퍼센트 문자열 파싱 (예: "60%" -> 60)""" try: if percent_str == "-" or not percent_str: return None import re match = re.search(r'\d+', percent_str) if match: return int(match.group()) except (ValueError, AttributeError): pass return None @staticmethod def _get_weather_icon_from_text(weather_text: str) -> str: """날씨 텍스트에서 아이콘 추출""" if "맑음" in weather_text: return "☀️" elif "구름" in weather_text: if "많음" in weather_text: return "☁️" else: return "⛅" elif "흐림" in weather_text: return "☁️" elif "비" in weather_text: return "🌧️" elif "눈" in weather_text: return "❄️" elif "천둥" in weather_text or "번개" in weather_text: return "⛈️" elif "안개" in weather_text or " fog" in weather_text.lower(): return "🌫️" else: return "🌤️" class WeatherService(QObject): """ 날씨 서비스 클래스 주기적으로 날씨 정보를 가져와 업데이트합니다. 2시간마다 자동으로 업데이트하며, HTML 파일로 캐시합니다. Examples: >>> weather = WeatherService() >>> weather.start() >>> weather.refresh() # 즉시 새로고침 """ def __init__(self): super().__init__() self.config = ConfigManager() self.signals = GlobalSignals() # 2시간마다 체크하는 타이머 (1분마다 체크) self._check_timer = QTimer() self._check_timer.timeout.connect(self._check_and_update) self._thread: Optional[QThread] = None self._worker: Optional[WeatherWorker] = None self._last_data: Dict[str, Any] = {} logger.info("날씨 서비스 초기화 완료") def start(self): """서비스 시작""" if not self.config.get('weather', 'enabled', True): return # 즉시 한 번 업데이트 self.update_weather() # 1분마다 체크 (2시간 경과 확인) self._check_timer.start(60 * 1000) # 1분 logger.info("날씨 서비스 시작") def stop(self): """서비스 중지""" self._check_timer.stop() if self._thread and self._thread.isRunning(): self._thread.quit() self._thread.wait() logger.info("날씨 서비스 중지") def _check_and_update(self): """2시간 경과 확인 후 업데이트""" if not WEATHER_TIMESTAMP_FILE.exists(): # 타임스탬프 파일이 없으면 업데이트 self.update_weather() return try: with open(WEATHER_TIMESTAMP_FILE, 'r', encoding='utf-8') as f: timestamp_str = f.read().strip() timestamp = datetime.fromisoformat(timestamp_str) elapsed = datetime.now() - timestamp if elapsed >= WEATHER_CACHE_DURATION: logger.info("2시간 경과, 날씨 정보 업데이트") self.update_weather() except Exception as e: logger.error(f"타임스탬프 확인 실패: {e}") def update_weather(self, force_refresh: bool = False): """ 날씨 정보 업데이트 Args: force_refresh: True면 캐시 무시하고 강제 새로고침 """ # 이전 작업이 실행 중이면 건너뛰기 if self._thread and self._thread.isRunning(): return lat = self.config.get('weather', 'location_lat', 35.1796) lon = self.config.get('weather', 'location_lon', 129.0756) # 지역 코드 가져오기 (기본값: 부산) location_name = self.config.get('weather', 'location_name', '부산') code = self._get_location_code(location_name) # 워커 생성 self._thread = QThread() self._worker = WeatherWorker(lat, lon, code, force_refresh) self._worker.moveToThread(self._thread) # 시그널 연결 self._thread.started.connect(self._worker.run) self._worker.finished.connect(self._on_weather_received) self._worker.finished.connect(self._thread.quit) self._worker.error.connect(self._on_weather_error) # 스레드 시작 self._thread.start() def refresh(self): """즉시 날씨 정보 새로고침""" logger.info("날씨 정보 강제 새로고침") self.update_weather(force_refresh=True) def _get_location_code(self, location_name: str) -> str: """지역명으로 코드 가져오기""" # 설정에서 지역 코드 가져오기 code = self.config.get('weather', 'location_code', '') if code: return code # 설정에 없으면 기본 매핑 사용 location_codes = { '부산': '2638057200', '서울': '1168000000', '대구': '2720000000', '인천': '2810000000', '광주': '2911000000', '대전': '3011000000', '울산': '3117000000', '수원': '4111000000', '고양': '4128000000', '용인': '4146000000', '성남': '4113000000', '부천': '4119000000', '화성': '4159000000', '안산': '4127000000', '안양': '4117000000', '평택': '4122000000', '의정부': '4115000000', '시흥': '4153000000', '김포': '4157000000', '광명': '4121000000', '이천': '4150000000', } return location_codes.get(location_name, '2638057200') def _on_weather_received(self, data: dict): """날씨 정보 수신""" self._last_data = data # JSON으로 변환하여 시그널 발생 json_data = json.dumps(data, ensure_ascii=False) self.signals.weather_updated.emit(json_data) logger.debug(f"날씨 업데이트: {data.get('temp')}°C, {data.get('condition')}") def _on_weather_error(self, error_msg: str): """날씨 오류""" self.signals.weather_error.emit(error_msg) def get_last_weather(self) -> Dict[str, Any]: """마지막 날씨 정보 반환""" return self._last_data.copy() def get_fetched_time(self) -> Optional[datetime]: """가져온 시간 반환""" if not WEATHER_TIMESTAMP_FILE.exists(): return None try: with open(WEATHER_TIMESTAMP_FILE, 'r', encoding='utf-8') as f: timestamp_str = f.read().strip() return datetime.fromisoformat(timestamp_str) except Exception as e: logger.error(f"타임스탬프 읽기 실패: {e}") return None def get_weather_for_shift(self, shift_type: str) -> Dict[str, Any]: """ 현재 근무 형태에 따른 날씨 정보를 반환합니다. Args: shift_type: 근무 유형 ("주간" 또는 "야간") Returns: 날씨 통계 데이터 """ try: crud = CRUDManager() location_code = self._get_location_code(self.config.get('weather', 'location_name', '부산')) # 오늘 날짜 today = date.today() # 근무 시간대의 날씨 통계 가져오기 stats = crud.get_weather_stats_for_shift(shift_type, today, location_code) # 현재 날씨 정보와 통합 current_weather = self.get_last_weather() result = { "current_temp": current_weather.get("temp"), "current_condition": current_weather.get("condition", "정보 없음"), "current_icon": current_weather.get("icon", "🌤"), "shift_type": shift_type, "temp_min": stats.get("temp_min"), "temp_max": stats.get("temp_max"), "feels_like_min": stats.get("feels_like_min"), "feels_like_max": stats.get("feels_like_max"), "max_precipitation_prob": stats.get("max_precipitation_prob"), "data_points": stats.get("data_points", 0) } return result except Exception as e: logger.error(f"근무 형태 날씨 정보 조회 실패: {e}") # 오류 시 현재 날씨 정보만 반환 current_weather = self.get_last_weather() return { "current_temp": current_weather.get("temp"), "current_condition": current_weather.get("condition", "정보 없음"), "current_icon": current_weather.get("icon", "🌤"), "shift_type": shift_type, "temp_min": None, "temp_max": None, "feels_like_min": None, "feels_like_max": None, "max_precipitation_prob": None, "data_points": 0 } def get_all_weather_items(self) -> List[Dict[str, Any]]: """ 모든 날씨 아이템 반환 (상세 다이얼로그용) Returns: 날씨 아이템 리스트 (날짜, 시간, 온도, 체감온도, 강수확률, 바람, 습도, 날씨 포함) """ if not WEATHER_HTML_FILE.exists(): return [] try: with open(WEATHER_HTML_FILE, 'r', encoding='utf-8') as f: html_content = f.read() return WeatherService._parse_all_weather_items(html_content) except Exception as e: logger.error(f"날씨 아이템 로드 실패: {e}") return [] @staticmethod def _parse_temp(temp_str: str) -> Optional[int]: """온도 문자열 파싱 (예: "4℃" -> 4)""" try: if temp_str == "-" or not temp_str: return None # 숫자만 추출 import re match = re.search(r'-?\d+', temp_str) if match: return int(match.group()) except (ValueError, AttributeError): pass return None @staticmethod def _parse_percentage(percent_str: str) -> Optional[int]: """퍼센트 문자열 파싱 (예: "60%" -> 60)""" try: if percent_str == "-" or not percent_str: return None import re match = re.search(r'\d+', percent_str) if match: return int(match.group()) except (ValueError, AttributeError): pass return None @staticmethod def _get_weather_icon_from_text(weather_text: str) -> str: """날씨 텍스트에서 아이콘 추출""" if "맑음" in weather_text: return "☀️" elif "구름" in weather_text: if "많음" in weather_text: return "☁️" else: return "⛅" elif "흐림" in weather_text: return "☁️" elif "비" in weather_text: return "🌧️" elif "눈" in weather_text: return "❄️" elif "천둥" in weather_text or "번개" in weather_text: return "⛈️" elif "안개" in weather_text or " fog" in weather_text.lower(): return "🌫️" else: return "🌤️" @staticmethod def _parse_all_weather_items(html_content: str) -> List[Dict[str, Any]]: """HTML에서 모든 날씨 아이템 파싱""" try: from selectolax.parser import HTMLParser tree = HTMLParser(html_content) slides = tree.css("div.slide-wrap div.slide") if not slides: return [] weather_items = [] for slide in slides: daily_node = slide.css_first("div.daily") if not daily_node: continue date_str = daily_node.attributes.get('data-date', '').strip() ul_items = daily_node.css("ul") for ul in ul_items: if "item" not in ul.attributes.get("class", ""): continue w_data = { "time": "-", "weather": "-", "temp": "-", "feels_like": "-", "prob": "-", "wind": "-", "humid": "-" } if 'data-time' in ul.attributes: w_data['time'] = ul.attributes['data-time'] lis = ul.css("li") for li in lis: label_node = li.css_first("span.hid") if not label_node: continue label_text = label_node.text(strip=True) if "시각" in label_text and w_data['time'] == "-": val = li.css_first("span:not(.hid)") if val: w_data['time'] = val.text(strip=True) elif "날씨" in label_text: wic = li.css_first(".wic") if wic: w_data['weather'] = wic.attributes.get("title") or wic.text(strip=True) elif "기온" in label_text: feel_node = li.css_first(".feel") if feel_node: w_data['temp'] = feel_node.text(deep=False, strip=True) elif "체감온도" in label_text and "기온" not in label_text: spans = li.css("span") if len(spans) > 1: w_data['feels_like'] = spans[-1].text(strip=True) elif "강수확률" in label_text: spans = li.css("span") if len(spans) > 1: prob = spans[-1].text(strip=True) w_data['prob'] = prob if prob else "-" elif "바람" in label_text: wd_node = li.css_first(".wdic") ws_node = li.css_first(".wspd:not(.qwsd)") wd = wd_node.text(strip=True) if wd_node else "" ws = ws_node.text(strip=True) if ws_node else "" w_data['wind'] = f"{wd} {ws}".strip() elif "습도" in label_text: spans = li.css("span") if len(spans) > 1: w_data['humid'] = spans[-1].text(strip=True) if w_data['time'] != "-": # 날짜와 시간을 합쳐서 datetime 객체 생성 try: # date_str 형식: "2026-01-04" # time_str 형식: "23:00" 또는 "24:00" time_str = w_data['time'] if time_str.endswith(':00'): hour_str = time_str.split(':')[0] hour = int(hour_str) if hour == 24: hour = 0 # 다음 날로 처리 date_obj = datetime.strptime(date_str, "%Y-%m-%d") date_obj = date_obj + timedelta(days=1) else: date_obj = datetime.strptime(date_str, "%Y-%m-%d") dt = date_obj.replace(hour=hour, minute=0, second=0, microsecond=0) # 현재 시간 이후의 데이터만 포함 if dt >= datetime.now().replace(minute=0, second=0, microsecond=0): weather_items.append({ 'datetime': dt, 'date': date_str, 'time': w_data['time'], 'temp': WeatherService._parse_temp(w_data['temp']), 'feels_like': WeatherService._parse_temp(w_data['feels_like']), 'precipitation_prob': WeatherService._parse_percentage(w_data['prob']), 'wind': w_data['wind'], 'humidity': WeatherService._parse_percentage(w_data['humid']), 'weather': w_data['weather'], 'icon': WeatherService._get_weather_icon_from_text(w_data['weather']) }) except (ValueError, KeyError) as e: logger.debug(f"날짜/시간 파싱 실패: {date_str} {w_data['time']} - {e}") continue # datetime 기준으로 정렬 weather_items.sort(key=lambda x: x['datetime']) return weather_items except Exception as e: logger.error(f"날씨 아이템 파싱 실패: {e}") return [] @staticmethod def _save_weather_data_to_db_static(weather_items: List[Dict[str, Any]], location_code: str): """ 파싱된 날씨 데이터를 데이터베이스에 저장합니다. Args: weather_items: 날씨 아이템 리스트 location_code: 지역코드 """ if not weather_items: return try: # 지역명 가져오기 (설정에서) config = ConfigManager() location_name = config.get('weather', 'location_name', '부산') crud = CRUDManager() saved_count = 0 for item in weather_items: try: crud.upsert_weather( datetime=item['datetime'], location_name=location_name, location_code=location_code, temp=item.get('temp'), feels_like=item.get('feels_like'), humidity=item.get('humidity'), wind_speed=item.get('wind', ''), wind_direction='', # 풍향 정보는 파싱하지 않음 precipitation_prob=item.get('precipitation_prob'), weather_condition=item.get('weather', ''), weather_icon=item.get('icon', '') ) saved_count += 1 except Exception as e: logger.debug(f"날씨 데이터 저장 실패 ({item.get('datetime')}): {e}") continue logger.info(f"날씨 데이터 {saved_count}개 DB에 저장됨") # 오래된 데이터 정리 (7일 이상 된 데이터 삭제) try: crud.cleanup_old_weather_data(7) except Exception as e: logger.error(f"오래된 날씨 데이터 정리 실패: {e}") except Exception as e: logger.error(f"날씨 데이터 DB 저장 중 오류: {e}")