"""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 resolved = resolve_month_date(target.year, target.month, schedule_type, billing_day) # With no billing lead, an arbitrary period start (for example 31 August) # must never yield an invoice date that has already passed (1 August). # In that situation the first valid scheduled invoice date is in the next # calendar month. A positive lead intentionally permits a date before the # coverage period and is left unchanged. if int(lead_months or 0) == 0 and resolved < period_start: following_month = target + relativedelta(months=1) resolved = resolve_month_date( following_month.year, following_month.month, schedule_type, billing_day, ) return resolved 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))