diff --git a/lib/holiday.py b/lib/holiday.py new file mode 100644 index 0000000..642c049 --- /dev/null +++ b/lib/holiday.py @@ -0,0 +1,141 @@ +import sys, os +sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) + +import yaml +import requests +import xml.etree.ElementTree as ET +from datetime import datetime, date +from sqlalchemy import select, insert, delete + +# config.yaml 경로 및 로딩 +BASE_DIR = os.path.abspath(os.path.join(os.path.dirname(__file__), '..')) +CONFIG_PATH = os.path.join(BASE_DIR, 'conf', 'config.yaml') + +with open(CONFIG_PATH, 'r', encoding='utf-8') as f: + cfg = yaml.safe_load(f) + +SERVICE_KEY = cfg['DATA_API']['serviceKey'] + +# DB 관련 +from conf import db, db_schema +holiday_table = db_schema.holiday + + +def fetch_holiday_api_xml(year): + """특일정보 API 호출 (XML)""" + url = "http://apis.data.go.kr/B090041/openapi/service/SpcdeInfoService/getRestDeInfo" + params = { + 'serviceKey': SERVICE_KEY, + 'solYear': str(year), + 'numOfRows': 100, + 'pageNo': 1, + 'type': 'xml', + } + + try: + resp = requests.get(url, params=params, timeout=10) + resp.raise_for_status() + + full_url = resp.url + xml_root = ET.fromstring(resp.content) + + items = [] + body = xml_root.find('.//body') + items_node = body.find('items') if body is not None else None + + if items_node is None: + print(f"[WARN] {year}년 특일정보 없음 or 응답 형식 이상") + return full_url, [] + + for item in items_node.findall('item'): + locdate = item.findtext('locdate') + dateName = item.findtext('dateName') + if locdate and dateName: + items.append({'locdate': locdate, 'dateName': dateName}) + + return full_url, items + + except requests.exceptions.RequestException as e: + print(f"[ERROR] {year}년 특일정보 API 요청 실패: {e}") + return None, [] + except ET.ParseError as e: + print(f"[ERROR] {year}년 XML 파싱 실패: {e}") + return None, [] + + +def holiday_exists(year): + """DB에 해당 연도 특일 정보 존재 여부 확인""" + session = db.get_session() + try: + stmt = select(holiday_table).where( + holiday_table.c.date.between(f"{year}0101", f"{year}1231") + ) + return session.execute(stmt).first() is not None + finally: + session.close() + + +def insert_holidays(items): + """특일정보 DB 삽입 (중복 제거 후 삽입)""" + session = db.get_session() + try: + for item in items: + date_str = item.get('locdate') + name = item.get('dateName') + if date_str and name: + session.execute( + delete(holiday_table).where(holiday_table.c.date == date_str) + ) + session.execute( + insert(holiday_table).values(date=date_str, name=name) + ) + session.commit() + except Exception as e: + session.rollback() + print(f"[ERROR] 특일정보 DB 삽입 실패: {e}") + finally: + session.close() + + +def update_holidays_for_year(year): + """해당 연도의 특일정보 API에서 조회 후 DB에 저장""" + url, items = fetch_holiday_api_xml(year) + if url: + print(f"📡 호출 URL: {url}") + if items: + insert_holidays(items) + print(f"✅ {year}년 특일 {len(items)}건 저장 완료") + else: + print(f"⚠️ {year}년 특일정보 없음 또는 실패") + + +def init_holidays(): + """2017년부터 현재 연도까지 특일정보 저장""" + this_year = datetime.now().year + + for y in range(2017, this_year): + if not holiday_exists(y): + print(f"📅 {y}년 → 특일정보 없음 → API 호출") + update_holidays_for_year(y) + else: + print(f"✅ {y}년 → 특일정보 이미 존재함") + + print(f"🔄 {this_year}년은 매번 최신 갱신") + update_holidays_for_year(this_year) + + +def is_korean_holiday(dt: date) -> bool: + """주어진 날짜가 특일인지 여부""" + session = db.get_session() + try: + date_str = dt.strftime("%Y%m%d") + stmt = select(holiday_table).where(holiday_table.c.date == date_str) + return session.execute(stmt).first() is not None + finally: + session.close() + + +if __name__ == "__main__": + print("📌 특일정보 초기화 시작") + init_holidays() + print("✅ 특일정보 초기화 완료")