"""Month-end book-close calendar engine — pure, deterministic, no DB / no I/O. This module turns a *closing period* (the month being closed, e.g. ``2026-06``) plus a calendar config (public holidays, makeup workdays 补班日, weekend-work rule, target workday) into: * the **book-close target date** — the Nth business day of the *following* month on which the standalone book closes (WD5 by default); * the **close window** — the ordered run of calendar days from Day+1 (first day of the following month) through the close date, each classified (workday / weekend-work / weekend-off / holiday / makeup) and, for the workdays, numbered WD1, WD2, …; and * the concrete **deadline date** for any task anchor token (``Day-5`` … ``Day-1``, ``Day+1`` … ``Day+N``, or ``On going``). Anchoring rules (reverse-engineered from, and unit-tested against, the PingCAP finance team's live close checklist): * **Day-1** = the last calendar day of the closing month (an editable anchor). **Day-N** (N≥1) counts *calendar* days backward from Day-1, so a Day-3 deadline can legitimately land on a weekend — these are pre-close prep tasks keyed to the month boundary, not to business days. * **Day+1** = the first calendar day of the following month (an editable anchor). **Day+N** (N≥1) is the date of the **Nth business day** (WDn) of the close window — Day+1→WD1, Day+5→WD5, etc. — so Day+N skips weekends and holidays. * **Weekend-work promotion**: when the close window (Day+1 → close date) overlaps a *full* Saturday+Sunday weekend, the team works the configured weekend-work day (Saturday by default), so that day counts as a business day toward WD numbering. The paired day stays off. Promotion applies only inside the close window; weekends after the close date are ordinary days off when extending WD numbering for Day+N beyond the target. All functions take plain ``datetime.date`` / ``set[date]`` inputs and return plain values, so the engine is trivially unit-testable and free of any SQLAlchemy / request context. The DB-backed wrapper that loads the persisted config lives in :mod:`app.domains.finance.close_service`. """ from __future__ import annotations import calendar import re from dataclasses import dataclass, field from datetime import date, timedelta # ── day-kind classification for the close-window strip ── KIND_WORKDAY = "workday" # ordinary Mon–Fri business day KIND_WEEKEND_WORK = "weekend_work" # weekend day worked under the overlap rule KIND_MAKEUP = "makeup" # 补班日 — weekend officially designated a workday KIND_WEEKEND_OFF = "weekend_off" # Sat/Sun not worked KIND_HOLIDAY = "holiday" # public holiday (off, even on a weekday) _WORKING_KINDS = frozenset({KIND_WORKDAY, KIND_WEEKEND_WORK, KIND_MAKEUP}) # How many business days past the target we still resolve, so Day+N anchors # beyond the close date (the checklist uses up to Day+10) map to a real date. _WD_LOOKAHEAD = 15 # Hard walk bound — defends against an infinite loop if the config were ever # pathological (e.g. every day a holiday). 200 calendar days >> any real window. _MAX_WALK_DAYS = 200 _SATURDAY = 5 # date.weekday(): Mon=0 … Sat=5, Sun=6 _SUNDAY = 6 _DAY_TOKEN_RE = re.compile(r"^Day([+-])(\d+)$") ONGOING = "On going" def parse_period(period: str) -> tuple[int, int]: """``"2026-06"`` → ``(2026, 6)``. Raises ``ValueError`` on bad input.""" m = re.fullmatch(r"(\d{4})-(\d{2})", period or "") if not m: raise ValueError(f"period must be YYYY-MM, got {period!r}") year, month = int(m.group(1)), int(m.group(2)) if not 1 <= month <= 12: raise ValueError(f"month out of range in period {period!r}") return year, month def last_day_of_month(year: int, month: int) -> date: return date(year, month, calendar.monthrange(year, month)[1]) def first_day_of_following_month(year: int, month: int) -> date: return date(year + 1, 1, 1) if month == 12 else date(year, month + 1, 1) def default_anchors(period: str) -> tuple[date, date]: """Default (Day-1, Day+1) anchors for a closing period. Day-1 = last day of the closing month; Day+1 = first day of the following month. Both are overridable per close run. """ year, month = parse_period(period) return last_day_of_month(year, month), first_day_of_following_month(year, month) def _is_work_weekend_weekday(d: date, work_weekend_day: str) -> bool: """Is ``d`` the configured weekend-work weekday (Sat or Sun)?""" if work_weekend_day == "sunday": return d.weekday() == _SUNDAY return d.weekday() == _SATURDAY # default: saturday @dataclass(frozen=True) class WindowDay: """One calendar day inside (or adjacent to) the close window.""" day: date kind: str # 1-based business-day index (WDn) for working kinds; None for off days. workday_number: int | None = None # True when this is the configured book-close target workday. is_close: bool = False @dataclass(frozen=True) class CloseWindow: """Computed close window for one closing period.""" period: str anchor_minus_1: date anchor_plus_1: date target_workday: int work_weekend_day: str close_date: date # Window strip from Day+1 through the close date (inclusive), in order. days: list[WindowDay] = field(default_factory=list) # WDn → date, extended past the target by _WD_LOOKAHEAD for Day+N anchors. workday_dates: dict[int, date] = field(default_factory=dict) # Dates of weekend days promoted to workdays under the overlap rule. weekend_work_dates: list[date] = field(default_factory=list) def deadline_for(self, anchor: str) -> date | None: """Resolve a task anchor token to its deadline date (None if 未定).""" return resolve_anchor( anchor, anchor_minus_1=self.anchor_minus_1, workday_dates=self.workday_dates, ) def _classify( d: date, holidays: frozenset[date], makeup: frozenset[date], work_weekend_day: str, promoted: frozenset[date], ) -> str: """Classify a single calendar day. Makeup beats holiday beats weekday.""" if d in makeup: return KIND_MAKEUP if d in holidays: return KIND_HOLIDAY if d.weekday() < _SATURDAY: # Mon–Fri return KIND_WORKDAY # Weekend: worked only if promoted by the overlap rule. if d in promoted: return KIND_WEEKEND_WORK return KIND_WEEKEND_OFF def _walk_close_date( anchor_plus_1: date, target_workday: int, holidays: frozenset[date], makeup: frozenset[date], work_weekend_day: str, promoted: frozenset[date], ) -> date: """Walk forward from Day+1, counting working days, return the WD=target date.""" wd = 0 cur = anchor_plus_1 for _ in range(_MAX_WALK_DAYS): if _classify(cur, holidays, makeup, work_weekend_day, promoted) in _WORKING_KINDS: wd += 1 if wd == target_workday: return cur cur += timedelta(days=1) raise RuntimeError( f"close target WD{target_workday} not reached within {_MAX_WALK_DAYS} days " f"from {anchor_plus_1} — calendar config likely pathological" ) def compute_close_window( period: str, *, anchor_minus_1: date | None = None, anchor_plus_1: date | None = None, target_workday: int, work_weekend_day: str, holidays: set[date] | frozenset[date], makeup_workdays: set[date] | frozenset[date], ) -> CloseWindow: """Compute the book-close target + window for ``period``. See the module docstring for the anchoring + weekend-promotion rules. The weekend promotion is resolved by a monotone fixpoint: promoting a weekend-work day only pulls the close date *earlier*, the promoted set only grows, and it is bounded by the candidate weekend days in the window — so the loop converges in a couple of passes. """ if target_workday < 1: raise ValueError("target_workday must be >= 1") if work_weekend_day not in ("saturday", "sunday"): raise ValueError("work_weekend_day must be 'saturday' or 'sunday'") d_minus_1, d_plus_1 = default_anchors(period) anchor_minus_1 = anchor_minus_1 or d_minus_1 anchor_plus_1 = anchor_plus_1 or d_plus_1 holidays = frozenset(holidays) makeup = frozenset(makeup_workdays) # ── fixpoint over the weekend-work promotion set ── promoted: frozenset[date] = frozenset() close_date = anchor_plus_1 for _ in range(8): close_date = _walk_close_date( anchor_plus_1, target_workday, holidays, makeup, work_weekend_day, promoted, ) # Promote every configured weekend-work weekday whose FULL weekend # (it + its paired off-day) lies inside [Day+1, close_date]. new_promoted = set(promoted) cur = anchor_plus_1 while cur < close_date: if ( _is_work_weekend_weekday(cur, work_weekend_day) and cur not in holidays and (cur + timedelta(days=1)) <= close_date ): new_promoted.add(cur) cur += timedelta(days=1) new_promoted_fs = frozenset(new_promoted) if new_promoted_fs == promoted: break promoted = new_promoted_fs else: # pragma: no cover - safety net, fixpoint converges well under 8 raise RuntimeError("weekend-work promotion did not converge") # ── build the window strip + WDn map (extended for Day+N lookahead) ── days: list[WindowDay] = [] workday_dates: dict[int, date] = {} wd = 0 cur = anchor_plus_1 max_wd = target_workday + _WD_LOOKAHEAD for _ in range(_MAX_WALK_DAYS): kind = _classify(cur, holidays, makeup, work_weekend_day, promoted) wd_num: int | None = None if kind in _WORKING_KINDS: wd += 1 wd_num = wd workday_dates[wd] = cur is_close = cur == close_date # The strip only covers Day+1 → close date (inclusive). if cur <= close_date: days.append(WindowDay(day=cur, kind=kind, workday_number=wd_num, is_close=is_close)) if wd >= max_wd and cur >= close_date: break cur += timedelta(days=1) return CloseWindow( period=period, anchor_minus_1=anchor_minus_1, anchor_plus_1=anchor_plus_1, target_workday=target_workday, work_weekend_day=work_weekend_day, close_date=close_date, days=days, workday_dates=workday_dates, weekend_work_dates=sorted(promoted), ) def resolve_anchor( anchor: str, *, anchor_minus_1: date, workday_dates: dict[int, date], ) -> date | None: """Resolve a task anchor token to its deadline date. * ``"On going"`` / empty → ``None`` (no deadline). * ``"Day-N"`` → ``anchor_minus_1 - (N-1)`` calendar days. * ``"Day+N"`` → date of business day WDn (``None`` if beyond lookahead). """ if not anchor or anchor == ONGOING: return None m = _DAY_TOKEN_RE.match(anchor.strip()) if not m: return None sign, n_str = m.group(1), m.group(2) n = int(n_str) if n < 1: return None if sign == "-": return anchor_minus_1 - timedelta(days=n - 1) return workday_dates.get(n)