"""Predicate helpers for the tips engine. These are pure functions (no I/O) that the tips service calls after loading the workspace context. Each takes pre-fetched data and returns a bool or a supporting value. Keeping them pure makes them trivial to unit-test. """ from __future__ import annotations import hashlib import json from datetime import date, timedelta from dateutil.relativedelta import relativedelta from app.config import LeaveRules, settings from app.entities.child import Child from app.entities.personal_data import PersonalData from app.entities.planned_leave import PlannedLeaveDay from app.utils.holidays import is_non_working_day def weekly_intensity(planned_days: list[PlannedLeaveDay], period_weeks: float) -> float: """Average föräldrapenning days per week over a given period. Only sjukpenning+lagsta count; vacation days are excluded (they don't draw from the FP pool and are not the planning knob that affects SGI). """ if period_weeks <= 0: return 0.0 fp_days = [d for d in planned_days if d.day_type in ("sjukpenning", "lagsta")] return len(fp_days) / period_weeks def default_tier_pool_share(rules: LeaveRules, parent_count: int) -> tuple[int, int]: """The per-parent (sjukpenning_left, lagsta_left) assumed when no FK snapshot exists yet. `rules.sjukpenning_days`/`lagsta_niva_days` are the WHOLE CHILD's combined tier totals (390 + 90). A sole parent (`parent_count <= 1`) gets the full pool; with a partner (`parent_count >= 2`) it's split in half per parent (195 + 45) exactly as FK does. Shared by `AllocationInputLoader.load` and `PlanningService._tier_pool_left` so they can never derive it differently (see "Bug A": they once did, 390/90 in one place vs. 195/45 in the other). """ if parent_count <= 1: return rules.sjukpenning_days, rules.lagsta_niva_days return rules.sjukpenning_days // 2, rules.lagsta_niva_days // 2 def resolve_age4_savable_cap(parent_count: int) -> int: """Single source of truth for which age-4 savable-days ceiling applies. `rules.max_savable_days_at_age_4` (96) is the per-CHILD ceiling, pooled across however many parents the child legally has. The per-parent cut is strictly **non-poolable**: each parent's own savable balance is capped at `rules.max_savable_days_per_parent_at_age_4` (48) whenever the child has two (or more) parents — 96 each would let the family save 192 combined, double the legal figure. A genuine single parent keeps the full 96. This is a fact about the family (how many parents the child has), not a workspace display preference — it is purely a function of parent count. A workspace with two real members must always get the 48/48 split, never 96/96. `AnalyticsService` and `allocation_service`'s `_Ledger` both call this so they can never disagree on which cap applies. """ rules = settings.rules if parent_count >= 2: return rules.max_savable_days_per_parent_at_age_4 return rules.max_savable_days_at_age_4 def days_left_at_age_4_projection( days_used: int, planned_fp_before_b4: int, total_days: int, max_savable: int, ) -> tuple[int, int]: """Return (projected_left_at_b4, expiring_at_b4) without hitting the DB. `max_savable` is the resolved cap (see `resolve_age4_savable_cap`) — the caller decides which figure applies; this function only does the shared arithmetic so `AnalyticsService` and `allocation_service._Ledger` can never derive it differently. """ projected_left = total_days - (days_used + planned_fp_before_b4) expiring = max(projected_left - max_savable, 0) return projected_left, expiring def age4_pooling_loss( parent_projected_lefts: list[int], per_parent_cap: int, per_child_cap: int, ) -> int: """Days a couple loses because the age-4 cap applies per parent, not pooled. Each parent's own projected balance at the child's 4th birthday is capped individually at `per_parent_cap` (see `resolve_age4_savable_cap`) before summing, even though the family's shared ceiling is `per_child_cap`. When one parent is over their own cap while another still has headroom under it, the family ends up keeping less than `per_child_cap` would otherwise allow — e.g. 60 and 10 days left (cap 48, ceiling 96): the family could keep 70 pooled but only keeps 48 + 10 = 58, losing 12. A parent with no known projection should simply be left out of `parent_projected_lefts` — omitting them is the conservative treatment on both sides of the comparison (as if their balance were 0), so a caller missing one parent's data still gets a safe (never-overstated) answer. Returns the shortfall, 0 when nothing is lost. """ if not parent_projected_lefts: return 0 reachable_pooled = min(sum(parent_projected_lefts), per_child_cap) reachable_per_parent = sum(min(left, per_parent_cap) for left in parent_projected_lefts) return max(reachable_pooled - reachable_per_parent, 0) def is_child_over_age(birth_date: date, years: int) -> bool: """True if the child has turned `years`.""" return date.today() >= birth_date + relativedelta(years=years) def sgi_at_risk( personal_data: PersonalData, planned_days: list[PlannedLeaveDay], child_birth_date: date, period_start: date, period_end: date, ) -> bool: """True if the parent risks losing SGI (< 5 FP days/week after age 1). SGI is protected automatically in the first 12 months (§6). After that a non-working, non-jobseeking parent must take ≥5 days/week to keep full SGI. """ rules = settings.rules if not is_child_over_age(child_birth_date, 1): return False # still in the protected first year if personal_data.employment_status in ("employed", "jobseeking"): return False # working/registered jobseeker is protected regardless period_days = [ d for d in planned_days if d.parent_personal_data_uuid == personal_data.uuid and period_start <= d.date <= period_end and d.day_type in ("sjukpenning", "lagsta") ] weeks = ((period_end - period_start).days + 1) / 7 intensity = weekly_intensity(period_days, weeks) return intensity < rules.sgi_min_days_per_week_after_age_1 def red_day_clusters_in_range( from_date: date, to_date: date, year: int | None = None ) -> list[tuple[date, date]]: """Return (start, end) pairs for clusters of ≥2 consecutive non-working days. Uses `app.utils.holidays.is_non_working_day` — the single source of truth for "is this day free" (calendar cache + `holidays` library fallback), so weekends, public holidays AND de facto days (julafton, midsommarafton, nyårsafton, ...) all count. `year` is accepted for backward compatibility but unused: `is_non_working_day` resolves each date's own year, so a range spanning a year boundary is handled correctly without it. """ del year # unused — is_non_working_day resolves per-date clusters: list[tuple[date, date]] = [] cluster_start: date | None = None current = from_date while current <= to_date: is_holiday = is_non_working_day(current) if is_holiday: if cluster_start is None: cluster_start = current else: if cluster_start is not None and (current - cluster_start).days >= 2: clusters.append((cluster_start, current - timedelta(days=1))) cluster_start = None current += timedelta(days=1) if cluster_start is not None and (to_date - cluster_start).days >= 2: clusters.append((cluster_start, to_date)) return clusters def preschool_buffer_needed(child: Child, default_days_per_year: int) -> int: """Days that should be reserved for preschool planning-day closures. Returns the default_days_per_year unless the child has no preschool_start_date yet (in which case no reservation is needed). """ if child.preschool_start_date is None: return 0 if date.today() < child.preschool_start_date: return 0 return default_days_per_year def has_vacation_near_expiry( vacation_total: int, vacation_saved_in: int, threshold_remaining_days: int = 5, ) -> bool: """True if saved vacation days are approaching expiry (simple heuristic). Swedish law: vacation days above 20 may be saved for up to 5 years. This predicate flags when saved days exist — the tips service adds richer expiry logic when it has the full vacation history per year. Kept as-is (still called by `allocation_service`'s per-budget check) — `has_genuine_vacation_forfeiture` below is the real replacement used by `tips_service._rule_use_vacation_before_expiry`, which has the actual computed per-year breakdown instead of this coarse heuristic. """ return vacation_saved_in > 0 and vacation_total > threshold_remaining_days def has_genuine_vacation_forfeiture(expiring_age: int, expiring_over_cap: int) -> bool: """True if a parent's vacation balance for a given year actually lost days. Takes `VacationYearBalance.expiring_age`/`.expiring_over_cap` (see `AnalyticsService._vacation_year_series`) — the real, already-computed per-year forfeiture breakdown — rather than guessing from a raw total like `has_vacation_near_expiry` does. Fires on genuine loss from either cause (age-out or over-cap), never on days that are merely saved. """ return (expiring_age + expiring_over_cap) > 0 def hash_tip_data(data: dict[str, object]) -> str: """Stable fingerprint of a tip's `data` bag. Used to decide whether a previously-dismissed tip's underlying situation has changed materially: if a rule fires again under the same `Tip.id` but with different `data` (a bigger `expiring_days` count, a moved `fourth_birthday`, a new `gap_days`, ...), the old dismissal describes a fact that's no longer true and must not silence the new one. See `app.entities.tip_dismissal.TipDismissal` and `app.services.tips_service.TipsService._drop_stale_dismissed`. `data` only ever holds JSON scalars (see `app.models.tips.TipDataValue`), so `sort_keys=True` alone gives a stable, order-independent digest. """ canonical = json.dumps(data, sort_keys=True, default=str) return hashlib.sha256(canonical.encode("utf-8")).hexdigest()