887 lines
34 KiB
Python
887 lines
34 KiB
Python
# -*- 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}")
|
|
|
|
|