F-r-ldraledigplannerare/backend/app/utils/tip_predicates.py

"""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.
    """
    return vacation_saved_in > 0 and vacation_total > threshold_remaining_days


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()