jacobh2
fbb7a04580
- Add 8 missing API client functions (getBreakPeriods, createBreakPeriod, deleteBreakPeriod, getTips, getParentalLeave, getPlannedDays, createPlannedDay, deletePlannedDay) - Fix useAnalytics to accept optional childId parameter - Fix ApiError strict optional properties handling - Export HeaderProps from Header component - Update hook tests to use correct data structures and optional parameters npx tsc --noEmit now passes with zero errors. Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
512 lines
18 KiB
Python
512 lines
18 KiB
Python
"""Tips engine.
|
||
|
||
Architecture:
|
||
- `Tip` is a plain dataclass returned to the router.
|
||
- `RULE_REGISTRY` is a list of async functions; each takes a `TipsContext`
|
||
and returns a `Tip | None`. None means the rule doesn't apply right now.
|
||
- `TipsService.get_tips()` runs all rules and returns the non-None results.
|
||
|
||
All 12 rules from ARCHITECTURE.md §6.1 are registered; the stubs return None
|
||
until implemented. This lets the junior wire rules one at a time by editing the
|
||
corresponding function without touching the plumbing.
|
||
|
||
To add a new rule:
|
||
1. Write an async function `_rule_<id>(ctx: TipsContext) -> Tip | None`.
|
||
2. Append it to RULE_REGISTRY.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
from collections import defaultdict
|
||
from dataclasses import dataclass
|
||
from datetime import date, timedelta
|
||
from typing import Callable, Coroutine, TypeAlias
|
||
from uuid import UUID
|
||
|
||
from dateutil.relativedelta import relativedelta as _relativedelta
|
||
|
||
from app.config import settings
|
||
from app.entities.break_period import BreakPeriod
|
||
from app.entities.child import Child
|
||
from app.entities.parental_leave import ParentalLeaveFetch
|
||
from app.entities.personal_data import PersonalData
|
||
from app.entities.planned_leave import PlannedLeaveDay
|
||
from app.entities.vacation_days import VacationDays
|
||
from app.repositories.break_period_repository import BreakPeriodRepository
|
||
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.vacation_days_repository import VacationDaysRepository
|
||
from app.repositories.workspace_repository import WorkspaceRepository
|
||
from app.utils.tip_predicates import (
|
||
days_left_at_age_4_projection,
|
||
has_vacation_near_expiry,
|
||
is_child_over_age,
|
||
is_child_under_age,
|
||
preschool_buffer_needed,
|
||
red_day_clusters_in_range,
|
||
sgi_at_risk,
|
||
weekly_intensity,
|
||
)
|
||
|
||
|
||
@dataclass
|
||
class Tip:
|
||
id: str
|
||
severity: str # 'info' | 'warning' | 'action'
|
||
message_key: str # i18n key for frontend
|
||
data: dict # context for frontend rendering (e.g. {days: 5, period: "Jul–Aug"})
|
||
|
||
|
||
@dataclass
|
||
class TipsContext:
|
||
workspace_id: UUID
|
||
workspace_planning_goal: str # 'max_time' | 'max_money' | 'balanced'
|
||
children: list[Child]
|
||
parents: list[PersonalData]
|
||
leave_by_parent: dict[UUID, ParentalLeaveFetch | None] # personal_data_uuid → latest
|
||
planned_days: list[PlannedLeaveDay] # all future planned days for workspace
|
||
vacation_by_parent: dict[UUID, list[VacationDays]]
|
||
break_periods: list[BreakPeriod]
|
||
|
||
|
||
# Type alias for a rule function.
|
||
RuleFn: TypeAlias = Callable[[TipsContext], Coroutine[None, None, Tip | None]]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Rule implementations (stub → return None until fully wired)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def _rule_use_vacation_before_expiry(ctx: TipsContext) -> Tip | None:
|
||
for parent in ctx.parents:
|
||
for vd in ctx.vacation_by_parent.get(parent.uuid, []):
|
||
if has_vacation_near_expiry(vd.total_days, vd.saved_days_in):
|
||
return Tip(
|
||
id="use_vacation_before_expiry",
|
||
severity="warning",
|
||
message_key="tips.use_vacation_before_expiry",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"saved_days": vd.saved_days_in,
|
||
"year": vd.year,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_balance_between_parents(ctx: TipsContext) -> Tip | None:
|
||
if len(ctx.parents) < 2:
|
||
return None
|
||
usages = [
|
||
(p, ctx.leave_by_parent[p.uuid].days_used)
|
||
for p in ctx.parents
|
||
if ctx.leave_by_parent.get(p.uuid) is not None
|
||
]
|
||
if len(usages) < 2:
|
||
return None
|
||
leading, leading_used = max(usages, key=lambda x: x[1])
|
||
lagging, lagging_used = min(usages, key=lambda x: x[1])
|
||
gap = leading_used - lagging_used
|
||
if gap > 30:
|
||
return Tip(
|
||
id="balance_between_parents",
|
||
severity="info",
|
||
message_key="tips.balance_between_parents",
|
||
data={
|
||
"leading_parent": leading.name,
|
||
"lagging_parent": lagging.name,
|
||
"gap_days": gap,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_preserve_for_age_4(ctx: TipsContext) -> Tip | None:
|
||
if not ctx.children:
|
||
return None
|
||
child = ctx.children[0]
|
||
fourth_bday = child.birth_date + _relativedelta(years=4)
|
||
for parent in ctx.parents:
|
||
leave = ctx.leave_by_parent.get(parent.uuid)
|
||
if leave is None:
|
||
continue
|
||
planned_fp_before_b4 = sum(
|
||
1
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and d.day_type in ("sjukpenning", "lagsta")
|
||
and d.date < fourth_bday
|
||
)
|
||
_, expiring_at_b4 = days_left_at_age_4_projection(
|
||
leave.days_used, planned_fp_before_b4, leave.total_days
|
||
)
|
||
if expiring_at_b4 > 0:
|
||
return Tip(
|
||
id="preserve_for_age_4",
|
||
severity="warning",
|
||
message_key="tips.preserve_for_age_4",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"expiring_days": expiring_at_b4,
|
||
"fourth_birthday": fourth_bday.isoformat(),
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_use_dubbeldagar_strategically(ctx: TipsContext) -> Tip | None:
|
||
if len(ctx.parents) < 2 or not ctx.children:
|
||
return None
|
||
child = ctx.children[0]
|
||
window_end = child.birth_date + _relativedelta(months=settings.rules.dubbeldagar_months_limit)
|
||
|
||
# Map date → set of parent UUIDs with FP planned that day (within dubbeldagar window)
|
||
fp_by_date: dict[date, set[UUID]] = defaultdict(set)
|
||
for d in ctx.planned_days:
|
||
if d.day_type in ("sjukpenning", "lagsta") and d.date <= window_end:
|
||
fp_by_date[d.date].add(d.parent_personal_data_uuid)
|
||
|
||
for day_date, parent_uuids in fp_by_date.items():
|
||
if len(parent_uuids) < 2:
|
||
continue
|
||
already_marked = any(
|
||
d.is_dubbeldag
|
||
for d in ctx.planned_days
|
||
if d.date == day_date and d.parent_personal_data_uuid in parent_uuids
|
||
)
|
||
if not already_marked:
|
||
return Tip(
|
||
id="use_dubbeldagar_strategically",
|
||
severity="action",
|
||
message_key="tips.use_dubbeldagar_strategically",
|
||
data={"date": day_date.isoformat()},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_lagsta_niva_coverage(ctx: TipsContext) -> Tip | None:
|
||
for parent in ctx.parents:
|
||
leave = ctx.leave_by_parent.get(parent.uuid)
|
||
if leave is None:
|
||
continue
|
||
tier = leave.tier_data
|
||
if tier.sjukpenning_left < 30 and tier.lagsta_niva_left > 0:
|
||
return Tip(
|
||
id="lagsta_niva_coverage",
|
||
severity="info",
|
||
message_key="tips.lagsta_niva_coverage",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"sjukpenning_left": tier.sjukpenning_left,
|
||
"lagsta_left": tier.lagsta_niva_left,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_stretch_time_at_home(ctx: TipsContext) -> Tip | None:
|
||
if not ctx.children:
|
||
return None
|
||
child = ctx.children[0]
|
||
fourth_bday = child.birth_date + _relativedelta(years=4)
|
||
today = date.today()
|
||
window_end = today + timedelta(weeks=4)
|
||
|
||
for parent in ctx.parents:
|
||
parent_window_days = [
|
||
d
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and today <= d.date <= window_end
|
||
]
|
||
intensity = weekly_intensity(parent_window_days, 4.0)
|
||
if intensity < 7:
|
||
continue
|
||
leave = ctx.leave_by_parent.get(parent.uuid)
|
||
if leave is None:
|
||
continue
|
||
planned_fp_before_b4 = sum(
|
||
1
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and d.day_type in ("sjukpenning", "lagsta")
|
||
and d.date < fourth_bday
|
||
)
|
||
projected_left, _ = days_left_at_age_4_projection(
|
||
leave.days_used, planned_fp_before_b4, leave.total_days
|
||
)
|
||
if projected_left < 60:
|
||
return Tip(
|
||
id="stretch_time_at_home",
|
||
severity="info",
|
||
message_key="tips.stretch_time_at_home",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"weekly_intensity": round(intensity, 1),
|
||
"projected_left": projected_left,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_skip_fp_first_12_months(ctx: TipsContext) -> Tip | None:
|
||
if ctx.workspace_planning_goal != "max_time":
|
||
return None
|
||
for child in ctx.children:
|
||
if is_child_under_age(child.birth_date, months=12):
|
||
for parent in ctx.parents:
|
||
fp_planned = [
|
||
d
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and d.day_type in ("sjukpenning", "lagsta")
|
||
and d.date < child.birth_date + _relativedelta(months=12)
|
||
]
|
||
if fp_planned:
|
||
return Tip(
|
||
id="skip_fp_first_12_months",
|
||
severity="info",
|
||
message_key="tips.skip_fp_first_12_months",
|
||
data={"parent_name": parent.name},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_sgi_minimum_after_age_1(ctx: TipsContext) -> Tip | None:
|
||
today = date.today()
|
||
for child in ctx.children:
|
||
if not is_child_over_age(child.birth_date, 1):
|
||
continue
|
||
period_end = today + timedelta(weeks=4)
|
||
for parent in ctx.parents:
|
||
if sgi_at_risk(parent, ctx.planned_days, child.birth_date, today, period_end):
|
||
return Tip(
|
||
id="sgi_minimum_after_age_1",
|
||
severity="warning",
|
||
message_key="tips.sgi_minimum_after_age_1",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"min_days_per_week": settings.rules.sgi_min_days_per_week_after_age_1,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_wrap_weekend_with_vacation(ctx: TipsContext) -> Tip | None:
|
||
if ctx.workspace_planning_goal != "max_money":
|
||
return None
|
||
today = date.today()
|
||
look_ahead = today + timedelta(weeks=8)
|
||
for parent in ctx.parents:
|
||
if parent.salary_type != "monthly":
|
||
continue
|
||
fp_dates = {
|
||
d.date
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and d.day_type in ("sjukpenning", "lagsta")
|
||
and today <= d.date <= look_ahead
|
||
}
|
||
for fp_date in sorted(fp_dates):
|
||
if fp_date.weekday() == 4: # Friday
|
||
monday = fp_date + timedelta(days=3)
|
||
if monday in fp_dates:
|
||
return Tip(
|
||
id="wrap_weekend_with_vacation",
|
||
severity="info",
|
||
message_key="tips.wrap_weekend_with_vacation",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"friday": fp_date.isoformat(),
|
||
"monday": monday.isoformat(),
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_wrap_red_days_with_vacation(ctx: TipsContext) -> Tip | None:
|
||
if ctx.workspace_planning_goal != "max_money":
|
||
return None
|
||
today = date.today()
|
||
look_ahead = today + timedelta(weeks=12)
|
||
clusters = red_day_clusters_in_range(today, look_ahead)
|
||
if not clusters:
|
||
return None
|
||
for parent in ctx.parents:
|
||
if parent.salary_type != "monthly":
|
||
continue
|
||
fp_dates = {
|
||
d.date
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and d.day_type in ("sjukpenning", "lagsta")
|
||
and today <= d.date <= look_ahead
|
||
}
|
||
for cluster_start, cluster_end in clusters:
|
||
day_before = cluster_start - timedelta(days=1)
|
||
day_after = cluster_end + timedelta(days=1)
|
||
if day_before in fp_dates or day_after in fp_dates:
|
||
return Tip(
|
||
id="wrap_red_days_with_vacation",
|
||
severity="info",
|
||
message_key="tips.wrap_red_days_with_vacation",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"cluster_start": cluster_start.isoformat(),
|
||
"cluster_end": cluster_end.isoformat(),
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_long_break_parent_swap(ctx: TipsContext) -> Tip | None:
|
||
if len(ctx.parents) < 2 or not ctx.break_periods:
|
||
return None
|
||
today = date.today()
|
||
for bp in ctx.break_periods:
|
||
if bp.end_date < today:
|
||
continue
|
||
working_days = sum(
|
||
1
|
||
for i in range((bp.end_date - bp.start_date).days + 1)
|
||
if (bp.start_date + timedelta(days=i)).weekday() < 5
|
||
)
|
||
if working_days == 0:
|
||
continue
|
||
for parent in ctx.parents:
|
||
covered = sum(
|
||
1
|
||
for d in ctx.planned_days
|
||
if d.parent_personal_data_uuid == parent.uuid
|
||
and bp.start_date <= d.date <= bp.end_date
|
||
)
|
||
if covered < working_days * 0.5:
|
||
other = next(p for p in ctx.parents if p.uuid != parent.uuid)
|
||
return Tip(
|
||
id="long_break_parent_swap",
|
||
severity="info",
|
||
message_key="tips.long_break_parent_swap",
|
||
data={
|
||
"break_name": bp.name,
|
||
"start_date": bp.start_date.isoformat(),
|
||
"end_date": bp.end_date.isoformat(),
|
||
"parent_name": parent.name,
|
||
"other_parent_name": other.name,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
async def _rule_reserve_for_preschool_planning_days(ctx: TipsContext) -> Tip | None:
|
||
if not ctx.children:
|
||
return None
|
||
child = ctx.children[0]
|
||
buffer_needed = preschool_buffer_needed(
|
||
child, settings.rules.preschool_planning_days_per_year
|
||
)
|
||
if buffer_needed == 0:
|
||
return None
|
||
for parent in ctx.parents:
|
||
leave = ctx.leave_by_parent.get(parent.uuid)
|
||
if leave is None:
|
||
continue
|
||
if leave.days_left < buffer_needed * 2:
|
||
return Tip(
|
||
id="reserve_for_preschool_planning_days",
|
||
severity="info",
|
||
message_key="tips.reserve_for_preschool_planning_days",
|
||
data={
|
||
"parent_name": parent.name,
|
||
"buffer_days": buffer_needed,
|
||
"days_left": leave.days_left,
|
||
},
|
||
)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Registry — order matters (earlier rules shown first)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
RULE_REGISTRY: list[RuleFn] = [
|
||
_rule_use_vacation_before_expiry,
|
||
_rule_balance_between_parents,
|
||
_rule_preserve_for_age_4,
|
||
_rule_use_dubbeldagar_strategically,
|
||
_rule_lagsta_niva_coverage,
|
||
_rule_stretch_time_at_home,
|
||
_rule_skip_fp_first_12_months,
|
||
_rule_sgi_minimum_after_age_1,
|
||
_rule_wrap_weekend_with_vacation,
|
||
_rule_wrap_red_days_with_vacation,
|
||
_rule_long_break_parent_swap,
|
||
_rule_reserve_for_preschool_planning_days,
|
||
]
|
||
|
||
|
||
class TipsService:
|
||
def __init__(
|
||
self,
|
||
leave_repo: ParentalLeaveRepository,
|
||
planned_repo: PlannedLeaveRepository,
|
||
child_repo: ChildRepository,
|
||
pd_repo: PersonalDataRepository,
|
||
vacation_repo: VacationDaysRepository,
|
||
break_repo: BreakPeriodRepository,
|
||
workspace_repo: WorkspaceRepository,
|
||
) -> None:
|
||
self._leave = leave_repo
|
||
self._planned = planned_repo
|
||
self._children = child_repo
|
||
self._parents = pd_repo
|
||
self._vacation = vacation_repo
|
||
self._breaks = break_repo
|
||
self._workspace_repo = workspace_repo
|
||
|
||
async def get_tips(self, workspace_id: UUID) -> list[Tip]:
|
||
"""Load workspace context then run all rules, returning non-None tips."""
|
||
ctx = await self._build_context(workspace_id)
|
||
results = await asyncio.gather(*(rule(ctx) for rule in RULE_REGISTRY))
|
||
return [tip for tip in results if tip is not None]
|
||
|
||
async def _build_context(self, workspace_id: UUID) -> TipsContext:
|
||
|
||
parents = await self._parents.list_by_workspace(workspace_id)
|
||
children = await self._children.list_by_workspace(workspace_id)
|
||
break_periods = await self._breaks.get_by_workspace(workspace_id)
|
||
|
||
today = date.today()
|
||
all_planned: list[PlannedLeaveDay] = []
|
||
leave_by_parent: dict[UUID, ParentalLeaveFetch | None] = {}
|
||
vacation_by_parent: dict[UUID, list[VacationDays]] = {}
|
||
|
||
if children:
|
||
child = children[0] # MVP: single child
|
||
for pd in parents:
|
||
leave_by_parent[pd.uuid] = await self._leave.get_latest_by_parent(pd.uuid, child.id)
|
||
future = await self._planned.list_by_workspace_child(
|
||
workspace_id, child.id, from_date=today
|
||
)
|
||
all_planned.extend(future)
|
||
vacation_by_parent[pd.uuid] = await self._vacation.list(
|
||
personal_data_uuid=pd.uuid, workspace_id=workspace_id
|
||
)
|
||
|
||
ws = await self._workspace_repo.get_by_id(workspace_id)
|
||
planning_goal = ws.planning_goal if ws else "balanced"
|
||
|
||
return TipsContext(
|
||
workspace_id=workspace_id,
|
||
workspace_planning_goal=planning_goal,
|
||
children=children,
|
||
parents=parents,
|
||
leave_by_parent=leave_by_parent,
|
||
planned_days=all_planned,
|
||
vacation_by_parent=vacation_by_parent,
|
||
break_periods=break_periods,
|
||
)
|