106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
"""ISO-week helpers and the weekly digest schedule clock.
|
|
|
|
The digest idempotency key is the ISO week id ("2026-W36", Monday..Sunday) - the role krx_date plays
|
|
in US2KR. A digest for week W is scheduled at the first run-weekday after W ends (default: Monday
|
|
07:00 KST, i.e. the morning right after the covered week's Sunday).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
from datetime import date, datetime, time, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from sebrief.config import KST
|
|
|
|
KST_TZ = ZoneInfo(KST)
|
|
|
|
WEEK_RE = re.compile(r"^(\d{4})-W(\d{2})$")
|
|
|
|
|
|
def week_id(d: date) -> str:
|
|
"""ISO week id for the week containing `d` (ISO year, not calendar year)."""
|
|
iso = d.isocalendar()
|
|
return f"{iso.year}-W{iso.week:02d}"
|
|
|
|
|
|
def week_range(week: str) -> tuple[date, date]:
|
|
"""(Monday, Sunday) of an ISO week id. Raises ValueError for malformed / non-existent weeks."""
|
|
m = WEEK_RE.match(week)
|
|
if not m:
|
|
raise ValueError(f"invalid week id {week!r}, expected YYYY-Wnn (e.g. 2026-W36)")
|
|
start = date.fromisocalendar(int(m.group(1)), int(m.group(2)), 1)
|
|
return start, start + timedelta(days=6)
|
|
|
|
|
|
def parse_hhmm(value: str) -> tuple[int, int]:
|
|
try:
|
|
hh, mm = value.strip().split(":")
|
|
hour, minute = int(hh), int(mm)
|
|
except ValueError as exc:
|
|
raise ValueError(f"invalid HH:MM time: {value!r}") from exc
|
|
if not (0 <= hour <= 23 and 0 <= minute <= 59):
|
|
raise ValueError(f"invalid HH:MM time: {value!r}")
|
|
return hour, minute
|
|
|
|
|
|
def _require_aware(now: datetime) -> datetime:
|
|
if now.tzinfo is None or now.utcoffset() is None:
|
|
raise ValueError("`now` must be a tz-aware datetime")
|
|
return now
|
|
|
|
|
|
class WeeklyClock:
|
|
"""Digest schedule calculator: plain weekday arithmetic in Asia/Seoul, no exchange calendars."""
|
|
|
|
def __init__(self, run_weekday: int = 0, run_time: str = "07:00") -> None:
|
|
if not (0 <= run_weekday <= 6):
|
|
raise ValueError("run_weekday must be 0 (Mon) .. 6 (Sun)")
|
|
self.run_weekday = run_weekday
|
|
hour, minute = parse_hhmm(run_time)
|
|
self.run_time = time(hour, minute)
|
|
|
|
# ------------------------------------------------------------------ mapping
|
|
def _run_at(self, d: date) -> datetime:
|
|
return datetime.combine(d, self.run_time, tzinfo=KST_TZ)
|
|
|
|
def covered_week(self, run_date: date) -> str:
|
|
"""Week a run on `run_date` covers: the ISO week before the one containing `run_date`."""
|
|
return week_id(run_date - timedelta(days=7))
|
|
|
|
def run_at_for(self, week: str) -> datetime:
|
|
"""Scheduled run time for the digest covering `week`: first run-weekday after the week ends."""
|
|
_, end = week_range(week)
|
|
d = end + timedelta(days=1)
|
|
while d.weekday() != self.run_weekday:
|
|
d += timedelta(days=1)
|
|
return self._run_at(d)
|
|
|
|
# ------------------------------------------------------------------ schedule
|
|
def next_run(self, now: datetime) -> tuple[str, datetime]:
|
|
"""(week, run_at KST) for the earliest digest run strictly after `now`."""
|
|
now = _require_aware(now)
|
|
d = now.astimezone(KST_TZ).date()
|
|
while d.weekday() != self.run_weekday:
|
|
d += timedelta(days=1)
|
|
if self._run_at(d) <= now:
|
|
d += timedelta(days=7)
|
|
return self.covered_week(d), self._run_at(d)
|
|
|
|
def last_run(self, now: datetime) -> tuple[str, datetime]:
|
|
"""(week, run_at KST) of the most recent scheduled run at or before `now`."""
|
|
_, next_at = self.next_run(now)
|
|
last_at = next_at - timedelta(days=7)
|
|
return self.covered_week(last_at.date()), last_at
|
|
|
|
def latest_completed_week(self, now: datetime) -> str:
|
|
"""Most recent fully elapsed ISO week (the default run-once target)."""
|
|
now = _require_aware(now)
|
|
return week_id(now.astimezone(KST_TZ).date() - timedelta(days=7))
|
|
|
|
def catchup_candidate(self, now: datetime, window_hours: int) -> str | None:
|
|
"""Week whose run time lies in (now - window, now]; None when nothing was missed."""
|
|
now = _require_aware(now)
|
|
week, run_at = self.last_run(now)
|
|
return week if now - run_at <= timedelta(hours=window_hours) else None
|