"""Analytics service: days-left-at-age-4 and planned-vs-actual reconciliation. The headline calculation (§6.5 ARCHITECTURE.md): projected_used_by_B4 = days_used + planned_days(date < B4) projected_left_at_B4 = total_days - projected_used_by_B4 savable_after_B4 = min(projected_left_at_B4, CAP) # age-4 ceiling expiring_at_B4 = max(projected_left_at_B4 - CAP, 0) # LOST days Where `CAP` is 96 (per child) normally, or 48 (per parent) for a `couple` workspace with two parents — see `app.utils.tip_predicates. resolve_age4_savable_cap`, the single function that decides which applies. All rule constants come from config.LeaveRules — no magic numbers. dateutil.relativedelta handles leap-year safe 4th-birthday arithmetic. """ from __future__ import annotations from dataclasses import dataclass from datetime import date from uuid import UUID from dateutil.relativedelta import relativedelta from app.config import settings from app.repositories.child_repository import ChildRepository from app.repositories.parental_leave_repository import ParentalLeaveRepository from app.repositories.personal_data_repository import PersonalDataRepository from app.repositories.planned_leave_repository import PlannedLeaveRepository from app.repositories.workspace_repository import WorkspaceRepository from app.utils.errors import NotFoundError from app.utils.tip_predicates import days_left_at_age_4_projection, resolve_age4_savable_cap @dataclass class Age4Projection: """Per-parent projection of days remaining at the child's 4th birthday.""" personal_data_uuid: UUID child_id: UUID birth_date: date fourth_birthday: date total_days: int days_used: int # from latest FK snapshot planned_days_before_b4: int # from calendar projected_left_at_b4: int savable_after_b4: int # min(projected_left, savable_cap_at_b4) expiring_at_b4: int # max(projected_left - savable_cap_at_b4, 0) — LOST if not used # The cap actually applied — 96 (per child) or 48 (per parent, couple # workspace with 2 parents). See `resolve_age4_savable_cap`. Surfaced so # the frontend can render "X kvar / Y sparbara" without re-deriving the # rule itself. savable_cap_at_b4: int @dataclass class WorkspaceOverview: workspace_id: UUID child_id: UUID parent_projections: list[Age4Projection] combined_days_used: int combined_days_left: int combined_expiring_at_b4: int class AnalyticsService: def __init__( self, parental_leave_repo: ParentalLeaveRepository, planned_leave_repo: PlannedLeaveRepository, child_repo: ChildRepository, personal_data_repo: PersonalDataRepository, workspace_repo: WorkspaceRepository, ) -> None: self._leave = parental_leave_repo self._planned = planned_leave_repo self._children = child_repo self._parents = personal_data_repo self._workspace_repo = workspace_repo async def _resolve_savable_cap(self, workspace_id: UUID) -> int: """The single call site AnalyticsService uses for the age-4 cap — wraps `resolve_age4_savable_cap` with the DB reads it needs (workspace planning_mode + member count) so callers never resolve the cap any other way. """ workspace = await self._workspace_repo.get_by_id(workspace_id) planning_mode = workspace.planning_mode if workspace is not None else "solo" members = await self._parents.list_by_workspace(workspace_id) return resolve_age4_savable_cap(planning_mode, len(members)) async def get_days_left_at_age_4( self, personal_data_uuid: UUID, child_id: UUID ) -> Age4Projection: """Compute the headline projection for a single parent+child pair.""" rules = settings.rules child = await self._children.get_by_id(child_id) if child is None: raise NotFoundError(f"Child {child_id} not found") # Compute the 4th birthday with leap-year-safe arithmetic. b4 = child.birth_date + relativedelta(years=4) latest = await self._leave.get_latest_by_parent(personal_data_uuid, child_id) if latest is None: # No FK data yet — assume full entitlement with 0 used. days_used = 0 total_days = rules.days_per_parent else: days_used = latest.days_used total_days = latest.total_days # Count föräldrapenning planned days that fall before the 4th birthday. # Vacation days do NOT count against the FP pool. today = date.today() sjukpenning_before_b4 = await self._planned.count_by_parent_type( personal_data_uuid, child_id, "sjukpenning", from_date=today, to_date=b4, ) lagsta_before_b4 = await self._planned.count_by_parent_type( personal_data_uuid, child_id, "lagsta", from_date=today, to_date=b4, ) planned_fp_before_b4 = sjukpenning_before_b4 + lagsta_before_b4 savable_cap = await self._resolve_savable_cap(child.workspace_id) # Shared with the plan-proposal side (`allocation_service._Ledger. # build_age4_projections`) so the two surfaces can never re-derive # this arithmetic differently — see `days_left_at_age_4_projection`. projected_left, expiring = days_left_at_age_4_projection( days_used=days_used, planned_fp_before_b4=planned_fp_before_b4, total_days=total_days, max_savable=savable_cap, ) savable = min(projected_left, savable_cap) return Age4Projection( personal_data_uuid=personal_data_uuid, child_id=child_id, birth_date=child.birth_date, fourth_birthday=b4, total_days=total_days, days_used=days_used, planned_days_before_b4=planned_fp_before_b4, projected_left_at_b4=projected_left, savable_after_b4=savable, expiring_at_b4=expiring, savable_cap_at_b4=savable_cap, ) async def get_overview(self, workspace_id: UUID, child_id: UUID) -> WorkspaceOverview: """Compute projections for all parents in a workspace for one child.""" members = await self._parents.list_by_workspace(workspace_id) if not members: raise NotFoundError(f"No parents found in workspace {workspace_id}") projections = [] for pd in members: proj = await self.get_days_left_at_age_4(pd.uuid, child_id) projections.append(proj) combined_used = sum(p.days_used + p.planned_days_before_b4 for p in projections) combined_left = sum(p.projected_left_at_b4 for p in projections) combined_expiring = sum(p.expiring_at_b4 for p in projections) return WorkspaceOverview( workspace_id=workspace_id, child_id=child_id, parent_projections=projections, combined_days_used=combined_used, combined_days_left=combined_left, combined_expiring_at_b4=combined_expiring, ) async def validate_dubbeldagar(self, child_id: UUID, new_date: date) -> None: """Raise DubbeldagError if adding a dubbeldagar day would break rules. Called from the planning service before persisting a new day. """ from app.utils.errors import DubbeldagError child = await self._children.get_by_id(child_id) if child is None: raise NotFoundError(f"Child {child_id} not found") # Window: child's birth_date ... birth_date + 15 months window_end = child.birth_date + relativedelta( months=settings.rules.dubbeldagar_months_limit ) if new_date > window_end: raise DubbeldagError( f"Dubbeldagar only allowed within the child's first " f"{settings.rules.dubbeldagar_months_limit} months " f"(deadline {window_end})", message_key="planApply.rejection.dubbeldagWindowExceeded", data={ "months_limit": settings.rules.dubbeldagar_months_limit, "deadline": window_end.isoformat(), }, ) # Quota: count already-planned dubbeldagar from child's birth_date onward. already_planned = await self._planned.count_dubbeldagar( child_id, from_date=child.birth_date ) if already_planned >= settings.rules.max_dubbeldagar: raise DubbeldagError( f"Dubbeldagar quota ({settings.rules.max_dubbeldagar} days) already reached", message_key="planApply.rejection.dubbeldagQuotaReached", data={"max_dubbeldagar": settings.rules.max_dubbeldagar}, )