solorpower/crawler/time_utils.py
haneulai a716dbef96
Some checks are pending
CI / Crawler (Python ${{ matrix.python-version }}) (3.10) (push) Waiting to run
CI / Crawler (Python ${{ matrix.python-version }}) (3.11) (push) Waiting to run
CI / API (Python 3.11) (push) Waiting to run
CI / Database migration (push) Waiting to run
CI / App web build (Node 20) (push) Waiting to run
feat: harden solar monitoring through stage 7
2026-08-07 14:07:22 +09:00

45 lines
1.4 KiB
Python

"""KST 기준 시간 판단과 Supabase 조회 경계를 제공한다."""
from datetime import date, datetime, time, timedelta, timezone
from typing import Tuple
KST = timezone(timedelta(hours=9), name="KST")
UTC = timezone.utc
def ensure_kst(value: datetime) -> datetime:
"""datetime을 KST aware 값으로 변환한다.
기존 SQLite에 저장된 naive 값은 서버 로컬 시간이 아니라 KST로 기록된
값이므로 KST로 간주한다.
"""
if value.tzinfo is None:
return value.replace(tzinfo=KST)
return value.astimezone(KST)
def now_kst() -> datetime:
"""서버 OS 시간대와 무관한 현재 KST 시각을 반환한다."""
return datetime.now(KST)
def today_kst() -> date:
return now_kst().date()
def yesterday_kst() -> date:
return today_kst() - timedelta(days=1)
def parse_legacy_kst_datetime(value: str) -> datetime:
"""SQLite의 기존 naive ISO 값과 신규 aware ISO 값을 모두 KST로 읽는다."""
return ensure_kst(datetime.fromisoformat(value))
def kst_day_bounds_utc(target_date: date) -> Tuple[datetime, datetime]:
"""KST 하루의 UTC 반개구간 [start, next_start)을 반환한다."""
start_kst = datetime.combine(target_date, time.min, tzinfo=KST)
next_start_kst = start_kst + timedelta(days=1)
return start_kst.astimezone(UTC), next_start_kst.astimezone(UTC)