40 lines
1.3 KiB
Python
40 lines
1.3 KiB
Python
from __future__ import annotations
|
|
|
|
import calendar
|
|
from datetime import datetime, time, timedelta
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from app.services.credit.utils import ensure_aware
|
|
|
|
BUSINESS_TZ = ZoneInfo("Asia/Shanghai")
|
|
|
|
|
|
def to_business_time(value: datetime) -> datetime:
|
|
return ensure_aware(value).astimezone(BUSINESS_TZ)
|
|
|
|
|
|
def add_natural_months(anchor_at: datetime, months: int) -> datetime:
|
|
"""始终基于传入锚点计算自然月，月底压缩后下月恢复原锚点日。"""
|
|
anchor = to_business_time(anchor_at)
|
|
month_index = anchor.year * 12 + anchor.month - 1 + int(months)
|
|
year, month_zero = divmod(month_index, 12)
|
|
month = month_zero + 1
|
|
target_day = min(anchor.day, calendar.monthrange(year, month)[1])
|
|
return anchor.replace(year=year, month=month, day=target_day)
|
|
|
|
|
|
def natural_month_period(anchor_at: datetime, sequence: int) -> tuple[datetime, datetime]:
|
|
start = add_natural_months(anchor_at, sequence)
|
|
end = add_natural_months(anchor_at, sequence + 1)
|
|
return start, end
|
|
|
|
|
|
def next_local_midnight(value: datetime) -> datetime:
|
|
local = to_business_time(value)
|
|
next_day = local.date() + timedelta(days=1)
|
|
return datetime.combine(next_day, time.min, tzinfo=BUSINESS_TZ)
|
|
|
|
|
|
def last_usable_at(expires_at: datetime) -> datetime:
|
|
return ensure_aware(expires_at) - timedelta(microseconds=1)
|