133 lines
5.0 KiB
Python
133 lines
5.0 KiB
Python
"""Deterministic billing dates for Danish subscriptions."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from calendar import monthrange
|
|
from datetime import date, timedelta
|
|
from typing import Optional
|
|
|
|
from dateutil.easter import easter
|
|
from dateutil.relativedelta import relativedelta
|
|
|
|
|
|
MONTH_BASED_INTERVALS = {"monthly", "quarterly", "yearly"}
|
|
SCHEDULE_TYPES = {"fixed_day", "first_business_day", "last_business_day", "interval_anchor"}
|
|
|
|
|
|
def validate_billing_schedule(
|
|
interval: str,
|
|
schedule_type: str,
|
|
billing_day: Optional[int],
|
|
) -> tuple[str, int]:
|
|
"""Return a runnable schedule or reject a combination the invoice job cannot execute."""
|
|
if interval not in {"daily", "biweekly", *MONTH_BASED_INTERVALS}:
|
|
raise ValueError("invalid billing_interval")
|
|
schedule_type = (schedule_type or "fixed_day").strip().lower()
|
|
day = int(billing_day or 1)
|
|
if interval in {"daily", "biweekly"}:
|
|
return "interval_anchor", day
|
|
if schedule_type == "interval_anchor":
|
|
raise ValueError("interval_anchor is only valid for daily and biweekly subscriptions")
|
|
if schedule_type not in SCHEDULE_TYPES:
|
|
raise ValueError("invalid billing_schedule_type")
|
|
if schedule_type == "fixed_day" and not 1 <= day <= 28:
|
|
raise ValueError("billing_day must be between 1 and 28")
|
|
return schedule_type, day
|
|
|
|
|
|
def danish_bank_holidays(year: int) -> set[date]:
|
|
"""Return Nationalbanken's recurring Danish bank closing days."""
|
|
easter_sunday = easter(year)
|
|
return {
|
|
date(year, 1, 1),
|
|
easter_sunday - timedelta(days=3), # Maundy Thursday
|
|
easter_sunday - timedelta(days=2), # Good Friday
|
|
easter_sunday + timedelta(days=1), # Easter Monday
|
|
easter_sunday + timedelta(days=39), # Ascension Day
|
|
easter_sunday + timedelta(days=40), # Bank holiday after Ascension
|
|
easter_sunday + timedelta(days=50), # Whit Monday
|
|
date(year, 6, 5),
|
|
date(year, 12, 24),
|
|
date(year, 12, 25),
|
|
date(year, 12, 26),
|
|
date(year, 12, 31),
|
|
}
|
|
|
|
|
|
def is_danish_bank_day(value: date) -> bool:
|
|
return value.weekday() < 5 and value not in danish_bank_holidays(value.year)
|
|
|
|
|
|
def resolve_month_date(year: int, month: int, schedule_type: str, billing_day: Optional[int]) -> date:
|
|
if schedule_type == "first_business_day":
|
|
candidate = date(year, month, 1)
|
|
while not is_danish_bank_day(candidate):
|
|
candidate += timedelta(days=1)
|
|
return candidate
|
|
if schedule_type == "last_business_day":
|
|
candidate = date(year, month, monthrange(year, month)[1])
|
|
while not is_danish_bank_day(candidate):
|
|
candidate -= timedelta(days=1)
|
|
return candidate
|
|
day = int(billing_day or 1)
|
|
if not 1 <= day <= 28:
|
|
raise ValueError("billing_day must be between 1 and 28")
|
|
return date(year, month, day)
|
|
|
|
|
|
def add_interval(value: date, interval: str) -> date:
|
|
if interval == "daily":
|
|
return value + timedelta(days=1)
|
|
if interval == "biweekly":
|
|
return value + timedelta(days=14)
|
|
if interval == "quarterly":
|
|
return value + relativedelta(months=3)
|
|
if interval == "yearly":
|
|
return value + relativedelta(years=1)
|
|
return value + relativedelta(months=1)
|
|
|
|
|
|
def next_billing_date(
|
|
anchor: date,
|
|
interval: str,
|
|
schedule_type: str = "fixed_day",
|
|
billing_day: Optional[int] = 1,
|
|
) -> date:
|
|
"""Advance one interval, then resolve the configured date in its target month."""
|
|
target = add_interval(anchor, interval)
|
|
if interval not in MONTH_BASED_INTERVALS or schedule_type == "interval_anchor":
|
|
return target
|
|
if schedule_type not in SCHEDULE_TYPES:
|
|
raise ValueError("invalid billing_schedule_type")
|
|
return resolve_month_date(target.year, target.month, schedule_type, billing_day)
|
|
|
|
|
|
def billing_date_for_period(
|
|
period_start: date,
|
|
lead_months: int,
|
|
schedule_type: str = "fixed_day",
|
|
billing_day: Optional[int] = 1,
|
|
) -> date:
|
|
"""Resolve the invoice date N calendar months before a coverage period starts."""
|
|
target = period_start - relativedelta(months=max(0, int(lead_months or 0)))
|
|
if schedule_type == "interval_anchor":
|
|
return target
|
|
return resolve_month_date(target.year, target.month, schedule_type, billing_day)
|
|
|
|
|
|
def advance_billing_periods(value: date, interval: str, periods: int = 1) -> date:
|
|
"""Advance a coverage boundary by a number of complete billing periods."""
|
|
result = value
|
|
for _ in range(max(1, int(periods or 1))):
|
|
result = add_interval(result, interval)
|
|
return result
|
|
|
|
|
|
def prorated_30_day_factor(period_start: date, first_full_period_start: date) -> float:
|
|
"""30/360-style fraction for a short opening period, capped at one month."""
|
|
if period_start >= first_full_period_start:
|
|
return 0.0
|
|
months = (first_full_period_start.year - period_start.year) * 12 + first_full_period_start.month - period_start.month
|
|
synthetic_days = months * 30 + min(first_full_period_start.day, 30) - min(period_start.day, 30)
|
|
return max(0.0, min(float(synthetic_days) / 30.0, 1.0))
|