dekun
44 lines
1.2 KiB
Python
44 lines
1.2 KiB
Python
from datetime import datetime, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
TZ = ZoneInfo("Asia/Shanghai")
|
|
DAY_CUTOFF_HOUR = 8
|
|
|
|
|
|
def now_shanghai() -> datetime:
|
|
return datetime.now(TZ)
|
|
|
|
|
|
def _align_cutoff(dt: datetime) -> datetime:
|
|
cutoff = dt.replace(hour=DAY_CUTOFF_HOUR, minute=0, second=0, microsecond=0)
|
|
if dt < cutoff:
|
|
cutoff -= timedelta(days=1)
|
|
return cutoff
|
|
|
|
|
|
def get_yesterday_period(now: datetime | None = None) -> tuple[datetime, datetime]:
|
|
"""[D-1 08:00, D 08:00) in Shanghai time."""
|
|
now = now or now_shanghai()
|
|
end = _align_cutoff(now)
|
|
start = end - timedelta(days=1)
|
|
return start, end
|
|
|
|
|
|
def get_daybefore_period(now: datetime | None = None) -> tuple[datetime, datetime]:
|
|
"""前日周期 [D-2 08:00, D-1 08:00) in Shanghai time."""
|
|
now = now or now_shanghai()
|
|
end = _align_cutoff(now) - timedelta(days=1)
|
|
start = end - timedelta(days=1)
|
|
return start, end
|
|
|
|
|
|
def get_today_period(now: datetime | None = None) -> tuple[datetime, datetime]:
|
|
"""[D 08:00, now) in Shanghai time."""
|
|
now = now or now_shanghai()
|
|
start = _align_cutoff(now)
|
|
return start, now
|
|
|
|
|
|
def to_ms(dt: datetime) -> int:
|
|
return int(dt.timestamp() * 1000)
|