jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
375bc64f2fd2ae7d89bc22719b17705e6df19ed6
F-r-ldraledigplannerare/backend/app/scrapers/fk_scraper.py
T
jacobh2 02db5430eb Harden the API and the data layer after a security and performance audit 
Security
- Every route now requires a login and refuses ids outside the caller's
  workspace with 404: plan, leave-wishes, day-overrides and vacation-days
  were fully anonymous; every other router trusted client-supplied
  workspace/child/parent ids. Scraper sessions are bound to their owner.
- Invitations can no longer be created for a foreign workspace, and a
  redeemer who is already a member (or a full workspace) is refused.
- Production refuses the default JWT secret; API docs are hidden in
  production; a failed migration aborts startup instead of serving.
- Login and register no longer reveal whether an email exists; inactive
  accounts are rejected; /auth/refresh is rate limited.
- The FK scraper writes debug artefacts only when explicitly enabled,
  logs no scraped values or page text, clears the BankID QR as soon as it
  is scanned and purges finished sessions after five minutes.
- Real FK page captures are removed from the tree (history rewrite
  follows); the parser tests use a synthetic fixture.

Performance and correctness
- Plan apply/sync run inside one locked transaction with batched writes
  instead of one commit per row, so a failure mid-batch leaves nothing
  half-applied and concurrent syncs no longer corrupt each other.
- SQLite runs in WAL mode with a busy timeout; 14 indexes cover the
  filters that were full scans; argon2 hashing runs off the event loop;
  responses are gzipped.
- Frontend: a 401 now refreshes the token once and retries, so sessions
  no longer die silently after 15 minutes; every data hook discards
  stale responses; logout clears state even when the server call fails.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CsXCCcDfmV4Sps4ZQSvNEY
2026-09-18 07:54:21 +02:00

632 lines
27 KiB
Python
Raw Blame History

"""Försäkringskassan headless-browser scraper — Playwright implementation.
Legal/ToS review completed (2026-06). Legal has cleared automated read-only
access for users who have given explicit per-session consent. Operational
guardrails must be maintained (see §2.3):
- Sandboxed browser context, torn down after each session.
- No cookies, session tokens, or QR images ever persisted to disk or DB by
default. Debug artifacts (HTML + screenshot of the logged-in page) are
OFF unless both scraper_debug_enabled and scraper_debug_dir are set —
they contain personal data and are for local debugging only.
- No personal data (names, personnummer, day counts, raw page text) in
logs above DEBUG; DEBUG lines log shapes/counts, never scraped values.
- Per-session user consent required before starting a session.
- Only the extracted day-count numbers are persisted (see persist_callback).
- QR frames and results are retained in RAM only for
scraper_result_retention_seconds after a session ends, then purged; at
most one live session per user.
Enable via APP_SCRAPER_ENABLED=true in the environment or backend/.env.
Architecture:
- All FK-specific selectors live in _SELECTORS below. One-place edit if FK changes their DOM.
- One in-memory session store per process (sessions are ephemeral by design).
- Browser context is torn down on completion, error, or cancel — never persisted.
- Scraped numbers are written to the DB via the persist_callback injected at session start.
"""
from __future__ import annotations
import asyncio
import base64
import logging
import re
import uuid
from collections.abc import Awaitable, Callable
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path
from typing import Literal
from uuid import UUID
from app.config import settings
from app.entities.parental_leave import ParentalLeaveFetch, TierData
from app.services.fetch_data_service import FetchDataService
logger = logging.getLogger(__name__)
# ---------------------------------------------------------------------------
# Selectors — all FK DOM knowledge lives here
# ---------------------------------------------------------------------------
LOGIN_URL = (
"https://www.forsakringskassan.se/privatperson/logga-in"
"#/?goto=%2Fprivatperson%2Fmina-sidor%2Falla-e-tjanster"
"%2Fansok-och-planera-foraldrapenning"
)
# The data-page landing URL after successful BankID auth
POST_LOGIN_URL_FRAGMENT = "ansok-och-planera-foraldrapenning#/"
# The data page we navigate to after clicking Ansök
DATA_PAGE_URL_FRAGMENT = "ansok-och-planera-foraldrapenning#/valjdagar"
class _SEL:
# Login page
COOKIE_ACCEPT = "button:has-text('Ja, jag godkänner alla kakor')"
BANKID_QR_LINK = "[data-test='login-link'] a:has-text('BankID med QR-kod')"
# Maintenance alert banner on login page
MAINTENANCE_TITLE = "h2.rubrik:has-text('Tekniskt underhål')"
MAINTENANCE_MESSAGE = ".sv-text-portlet-content"
# QR page (idpx.forsakringskassan.se)
QR_SVG = "[data-testid='mobile-auth-qr-code']"
QR_BUTTON = "[data-testid='mobile-auth-qr-code-button']"
# Post-login landing (forsakringskassan.se/...#/)
ANSOK_BUTTON = "button.button--primary:has-text('Ansök för')"
# Data page (#/valjdagar) — "Återstående dagar" section
STEP_DAGAR_KVAR = "[data-test='stepDagarKvar']"
FOR_DIG = "[data-test='forDig']"
FOR_DIG_DAGAR_KVAR = "[data-test='forDig'] .visa-dagar__dagar-kvar"
FOR_DIG_DISCRETE = "[data-test='forDig'] .visa-dagar__discrete"
DUBBELDAG_KVAR = "[data-test='dubbeldag'] .visa-dagar__dagar-kvar"
# niva-statistik items inside forDig (desktop version, first = sjukpenning, second = lägstanivå)
NIVA_STATISTIK_DESKTOP = "[data-test='forDig'] .visa-dagar__desktop .niva-statistik__dagar span"
# ---------------------------------------------------------------------------
# Session state
# ---------------------------------------------------------------------------
SessionStatus = Literal[
"launching", "awaiting_scan", "authenticating", "fetching", "done", "error", "cancelled"
]
@dataclass
class _SessionState:
session_id: str
workspace_id: UUID
personal_data_uuid: UUID
child_id: UUID
# The User.id that started this session. `poll_session`/`cancel_session`
# treat a mismatched owner exactly like an unknown session_id — never a
# 403, so a foreign session_id can't be distinguished from a made-up one.
owner_user_id: UUID
status: SessionStatus = "launching"
qr_data: bytes | None = None # raw PNG bytes; never persisted to disk/DB
result: ParentalLeaveFetch | None = None
message: str | None = None
created_at: datetime = field(default_factory=datetime.now)
# Called after a successful extraction to persist the extracted numbers.
# The callback receives the validated ParentalLeaveFetch and must write it to the DB.
# Only the extracted numbers are persisted — never cookies, QR, or session state.
persist_callback: Callable[[ParentalLeaveFetch], Awaitable[None]] | None = field(
default=None, repr=False, compare=False
)
# asyncio task running the scrape; kept so we can cancel it
_task: asyncio.Task | None = field(default=None, repr=False, compare=False)
# In-memory store — ephemeral by design; never written to DB.
_sessions: dict[str, _SessionState] = {}
# ---------------------------------------------------------------------------
# FKScraperService
# ---------------------------------------------------------------------------
class FKScraperService(FetchDataService):
"""Playwright-backed session service for scraping FK föräldrapenning data.
The service lives outside the normal FetchDataService "get/list" contract
because it's session-oriented (async, long-lived, polled). The get/list
methods remain unimplemented — the planner reads data via ManualFetchService
after a successful scrape saves the snapshot.
"""
async def get_latest(
self, personal_data_uuid: UUID, child_id: UUID
) -> ParentalLeaveFetch | None:
raise NotImplementedError("Use ManualFetchService for reads; FK scraper is write-only.")
async def list_history(
self, personal_data_uuid: UUID, child_id: UUID
) -> list[ParentalLeaveFetch]:
raise NotImplementedError("Use ManualFetchService for reads; FK scraper is write-only.")
# ------------------------------------------------------------------
# Public session API
# ------------------------------------------------------------------
async def start_session(
self,
workspace_id: UUID,
personal_data_uuid: UUID,
child_id: UUID,
owner_user_id: UUID,
persist_callback: Callable[[ParentalLeaveFetch], Awaitable[None]] | None = None,
) -> str:
"""Launch a new headless browser session. Returns the session_id.
persist_callback, if provided, is called with the validated ParentalLeaveFetch
once extraction succeeds. Failure in the callback sets the session to 'error'.
`owner_user_id` binds the session to the caller that started it — see
`_SessionState.owner_user_id`.
At most one live (non-terminal) session per `owner_user_id` — starting
a new one cancels any session already in flight for the same caller
(see F5: unbounded growth of `_sessions` per owner).
"""
for existing_id, existing_state in list(_sessions.items()):
if existing_state.owner_user_id == owner_user_id and existing_state.status not in (
"done",
"error",
"cancelled",
):
await self.cancel_session(existing_id, owner_user_id)
session_id = str(uuid.uuid4())
state = _SessionState(
session_id=session_id,
workspace_id=workspace_id,
personal_data_uuid=personal_data_uuid,
child_id=child_id,
owner_user_id=owner_user_id,
status="launching",
persist_callback=persist_callback,
)
_sessions[session_id] = state
task = asyncio.create_task(_run_session(state))
state._task = task
# Schedule TTL watchdog
asyncio.create_task(_ttl_watchdog(session_id))
logger.info("scraper session started [id=%s]", session_id)
return session_id
async def poll_session(self, session_id: str, owner_user_id: UUID) -> dict:
"""Return current status, QR bytes (base64), or result.
A `session_id` owned by a different caller is treated exactly like
an unknown one — never a 403, so a foreign session_id can't be
distinguished from a made-up one.
"""
state = _sessions.get(session_id)
if state is None or state.owner_user_id != owner_user_id:
return {"status": "error", "message": "Unknown session_id"}
response: dict = {"status": state.status}
if state.status == "awaiting_scan" and state.qr_data:
response["qr_data"] = base64.b64encode(state.qr_data).decode()
if state.status == "done" and state.result:
response["result"] = state.result
if state.message:
response["message"] = state.message
return response
async def cancel_session(self, session_id: str, owner_user_id: UUID) -> None:
"""Cancel a running session and tear down the browser.
A `session_id` owned by a different caller is a no-op, exactly like
an unknown one — never a 403 (see `poll_session`).
"""
state = _sessions.get(session_id)
if state is None or state.owner_user_id != owner_user_id:
return
_set_status(state, "cancelled")
state.qr_data = None
if state._task and not state._task.done():
state._task.cancel()
logger.info("scraper session cancelled [id=%s]", session_id)
# ---------------------------------------------------------------------------
# Internal helpers
# ---------------------------------------------------------------------------
def _set_status(state: _SessionState, status: SessionStatus, message: str | None = None) -> None:
state.status = status
if message:
state.message = message
def _generic_error_message(exc: Exception) -> str:
"""Map an internal exception to a fixed, generic message safe to return
to the poll client (see F4). Never embeds `str(exc)` — a `ValueError`
raised while parsing the FK page may otherwise carry raw page text.
Logs carry the exception type name only; full detail is never emitted
above DEBUG and is not persisted unless debug artifacts are explicitly
enabled (`scraper_debug_enabled`).
"""
if isinstance(exc, ValueError):
return "Scraper error: could not read data from the FK page. Please try again."
if type(exc).__name__ == "TimeoutError":
return "Scraper error: timed out waiting for the FK page. Please try again."
return "Scraper error: an unexpected error occurred. Please try again."
async def _ttl_watchdog(session_id: str) -> None:
"""Auto-cancel sessions that sit in awaiting_scan/authenticating too long.
Also acts as a secondary safety net (F5): if the session has already
reached a terminal state by the time this fires, purge it immediately
rather than rely solely on `_purge_later`.
"""
await asyncio.sleep(settings.scraper_session_ttl_seconds)
state = _sessions.get(session_id)
if state is None:
return
if state.status in ("awaiting_scan", "authenticating", "launching"):
_set_status(state, "error", "Session timed out waiting for BankID scan")
state.qr_data = None
if state._task and not state._task.done():
state._task.cancel()
logger.info("scraper session timed out [id=%s]", session_id)
elif state.status in ("done", "error", "cancelled"):
_sessions.pop(session_id, None)
async def _purge_later(session_id: str) -> None:
"""Remove a finished session from `_sessions` after the retention window.
Bounds how long a scraped result or a stale (already-cleared) QR frame
slot can be read back via `/poll` after the session ended (see F5).
"""
await asyncio.sleep(settings.scraper_result_retention_seconds)
_sessions.pop(session_id, None)
async def _check_maintenance(page) -> str | None:
"""Check for FK maintenance message. Returns the maintenance message if found, else None."""
try:
maintenance_title = page.locator(_SEL.MAINTENANCE_TITLE)
if await maintenance_title.count() == 0:
return None
# Found maintenance notice; extract the message
# The message is in a nearby .sv-text-portlet-content div
message_el = page.locator(
"h2.rubrik:has-text('Tekniskt underhål') ~ div .sv-text-portlet-content"
)
if await message_el.count() > 0:
message_text = await message_el.first.inner_text(timeout=3_000)
return message_text.strip()
# Fallback: just return a generic message
return "FK is under maintenance. Please try again later."
except Exception:
return None
async def _save_debug_artifacts(page, session_id: str, console_messages: list[str]) -> None:
"""Save HTML, screenshot, and console log to scraper_debug_dir on error.
OFF by default and requires both `scraper_debug_enabled` and
`scraper_debug_dir` to be set — these artifacts contain personal data
(name, balances, payout amounts) from the logged-in FK page.
"""
debug_dir = settings.scraper_debug_dir
if not settings.scraper_debug_enabled or not debug_dir:
return
try:
out = Path(debug_dir)
out.mkdir(parents=True, exist_ok=True)
prefix = out / session_id
console_path = prefix.with_suffix(".console.txt")
console_path.write_text("\n".join(console_messages), encoding="utf-8")
html_path = prefix.with_suffix(".html")
html_path.write_text(await page.content(), encoding="utf-8")
screenshot_path = prefix.with_suffix(".png")
await page.screenshot(path=str(screenshot_path), full_page=True)
logger.info(
"scraper debug artifacts saved [id=%s] dir=%s",
session_id,
out.resolve(),
)
except Exception as artifact_exc:
logger.warning("failed to save debug artifacts [id=%s]: %s", session_id, artifact_exc)
async def _run_session(state: _SessionState) -> None:
"""Main browser coroutine. Drives FK login → QR relay → data extraction."""
from playwright.async_api import async_playwright # lazy import — optional dep
console_messages: list[str] = []
async with async_playwright() as pw:
browser = await pw.chromium.launch(headless=settings.scraper_headless)
context = await browser.new_context(locale="sv-SE")
page = await context.new_page()
page.on("console", lambda msg: console_messages.append(f"[{msg.type}] {msg.text}"))
try:
await _do_login(page, state)
if state.status in ("error", "cancelled"):
return
await _do_extract(page, state)
except asyncio.CancelledError:
_set_status(state, "cancelled")
except Exception as exc:
_set_status(state, "error", _generic_error_message(exc))
logger.error(
"scraper error [id=%s]: %s", state.session_id, type(exc).__name__, exc_info=False
)
await _save_debug_artifacts(page, state.session_id, console_messages)
finally:
try:
await context.close()
await browser.close()
except Exception:
pass # already torn down
# Session has reached a terminal state (done/error/cancelled) — schedule
# its removal from `_sessions` after the retention window (see F5).
asyncio.create_task(_purge_later(state.session_id))
async def _do_login(page, state: _SessionState) -> None:
"""Navigate to FK login, handle cookie consent, click BankID QR, relay QR."""
_set_status(state, "launching")
logger.debug("[%s] navigating to FK login URL", state.session_id)
await page.goto(LOGIN_URL, wait_until="networkidle", timeout=30_000)
logger.debug("[%s] page loaded: %s", state.session_id, page.url)
# Check for maintenance message before proceeding
maintenance_msg = await _check_maintenance(page)
if maintenance_msg:
error_msg = f"FK is under maintenance: {maintenance_msg}"
_set_status(state, "error", error_msg)
logger.info("[%s] FK maintenance detected; aborting session", state.session_id)
return
# Accept cookie consent banner if present (only appears once per browser profile)
try:
cookie_btn = page.locator(_SEL.COOKIE_ACCEPT)
if await cookie_btn.count() > 0:
logger.debug("[%s] accepting cookie consent", state.session_id)
await cookie_btn.first.click(timeout=5_000)
else:
logger.debug("[%s] no cookie consent banner", state.session_id)
except Exception:
pass # not present; fine
# Check if BankID button is disabled (another maintenance indicator)
try:
bankid_btn = page.locator(_SEL.BANKID_QR_LINK)
if await bankid_btn.count() > 0:
is_disabled = await bankid_btn.first.is_disabled(timeout=3_000)
if is_disabled:
error_msg = "FK login temporarily unavailable (BankID button disabled). Please try again later."
_set_status(state, "error", error_msg)
logger.info("[%s] BankID button is disabled; aborting session", state.session_id)
return
except Exception:
pass # proceed anyway; may not be disabled
# Click "BankID med QR-kod"
logger.debug("[%s] clicking BankID QR link", state.session_id)
await page.locator(_SEL.BANKID_QR_LINK).first.click(timeout=30_000)
# Wait for the QR SVG to appear (we're on the idpx auth page now)
logger.debug("[%s] waiting for QR code to appear (url=%s)", state.session_id, page.url)
await page.wait_for_selector(_SEL.QR_SVG, timeout=15_000)
logger.debug("[%s] QR code visible; relaying to client", state.session_id)
_set_status(state, "awaiting_scan")
# Relay the QR until FK navigates away (user scanned) or session cancelled/timed out
relay_count = 0
while state.status == "awaiting_scan":
try:
qr_el = page.locator(_SEL.QR_BUTTON)
state.qr_data = await qr_el.screenshot(timeout=3_000)
relay_count += 1
if relay_count % 5 == 1:
logger.debug("[%s] QR screenshot relayed (count=%d)", state.session_id, relay_count)
except Exception:
pass # transient; keep trying
await asyncio.sleep(1)
# Detect navigation to post-login FK page
if POST_LOGIN_URL_FRAGMENT in page.url:
logger.debug("[%s] detected post-login URL; scan complete", state.session_id)
break
# If QR element disappears without the post-login URL, we're in "authenticating"
if await page.locator(_SEL.QR_SVG).count() == 0:
logger.debug("[%s] QR element gone; BankID app likely confirming", state.session_id)
break
# Leaving awaiting_scan (scanned, cancelled, or timed out) — the last QR
# frame is no longer needed and must not linger in RAM (see F5).
state.qr_data = None
if state.status == "cancelled":
return
_set_status(state, "authenticating")
logger.debug(
"[%s] authenticating — waiting for Ansök button (url=%s)", state.session_id, page.url
)
# Wait for post-login landing (the Ansök button appears after JS hydrates the page)
await page.wait_for_selector(_SEL.ANSOK_BUTTON, timeout=30_000)
logger.debug("[%s] Ansök button visible; login complete", state.session_id)
async def _do_extract(page, state: _SessionState) -> None:
"""Click 'Ansök för …', wait for #/valjdagar, extract parental-leave figures."""
_set_status(state, "fetching")
logger.debug("[%s] clicking Ansök button", state.session_id)
# Click the child's "Ansök för …" button
await page.locator(_SEL.ANSOK_BUTTON).first.click(timeout=10_000)
# Wait for the valjdagar data page to load
logger.debug("[%s] waiting for valjdagar page (url=%s)", state.session_id, page.url)
await page.wait_for_selector(_SEL.STEP_DAGAR_KVAR, timeout=30_000)
logger.debug("[%s] valjdagar page loaded (url=%s)", state.session_id, page.url)
# The panel mounts with placeholder zeros, then hydrates from an async fetch.
# Wait until the "av N dagar" total is positive — total is always 240/480, never 0,
# so this is a reliable "real data has arrived" signal (days_left can legitimately be 0).
logger.debug("[%s] waiting for valjdagar data to hydrate", state.session_id)
await page.wait_for_function(
"""(sel) => {
const el = document.querySelector(sel);
if (!el) return false;
const m = el.textContent.match(/av\\s+(\\d+)\\s+dagar/);
return m && parseInt(m[1], 10) > 0;
}""",
arg=_SEL.FOR_DIG_DISCRETE,
timeout=30_000,
)
logger.debug("[%s] valjdagar data hydrated (url=%s)", state.session_id, page.url)
data = await _extract_parental_leave(page, state)
if state.persist_callback is not None:
try:
await state.persist_callback(data)
except Exception as exc:
_set_status(
state,
"error",
f"Scrape succeeded but failed to persist data: {type(exc).__name__}: {exc}",
)
logger.error("persist error [id=%s]: %s", state.session_id, exc, exc_info=False)
return
_set_status(state, "done")
state.result = data
logger.info("scraper session done [id=%s]", state.session_id)
async def _extract_parental_leave(page, state: _SessionState) -> ParentalLeaveFetch:
"""Read day figures from the #/valjdagar page and return a validated entity."""
# -- days left for this parent ("För dig" section) --
logger.debug("[%s] reading days_left (FOR_DIG_DAGAR_KVAR)", state.session_id)
days_left_text = await page.locator(_SEL.FOR_DIG_DAGAR_KVAR).first.inner_text(timeout=5_000)
days_left = int(days_left_text.strip())
logger.debug("[%s] days_left parsed", state.session_id)
# -- total days from "av 240 dagar kvar efter ansökan" --
logger.debug("[%s] reading total days (FOR_DIG_DISCRETE)", state.session_id)
discrete_text = await page.locator(_SEL.FOR_DIG_DISCRETE).first.inner_text(timeout=5_000)
total_match = re.search(r"av\s+(\d+)\s+dagar", discrete_text)
if not total_match:
raise ValueError("Could not parse total days from FK page text")
total_days = int(total_match.group(1))
if total_days == 0:
raise ValueError(
"FK panel returned total_days=0 — data had not hydrated when read "
"(placeholder state). The hydration wait in _do_extract should prevent this."
)
days_used = total_days - days_left
logger.debug("[%s] total/used parsed", state.session_id)
# -- sjukpenningnivå and lägstanivå remaining (desktop niva-statistik) --
# all_inner_texts() does not accept a timeout argument — wait for the first element first.
logger.debug("[%s] reading tier breakdown (NIVA_STATISTIK_DESKTOP)", state.session_id)
await page.locator(_SEL.NIVA_STATISTIK_DESKTOP).first.wait_for(timeout=5_000)
niva_texts = await page.locator(_SEL.NIVA_STATISTIK_DESKTOP).all_inner_texts()
logger.debug("[%s] niva items=%d", state.session_id, len(niva_texts))
# Each item is like "(119 dagar)" — filter to desktop section (first two)
niva_numbers = []
for t in niva_texts:
m = re.search(r"\((\d+)\s+dagar\)", t)
if m:
niva_numbers.append(int(m.group(1)))
if len(niva_numbers) == 2:
break
if len(niva_numbers) < 2:
raise ValueError(f"Expected 2 tier values, got {len(niva_numbers)}")
sjukpenning_left, lagsta_niva_left = niva_numbers[0], niva_numbers[1]
logger.debug("[%s] tier breakdown parsed", state.session_id)
# Cross-check: sjukpenning_left + lagsta_niva_left should equal days_left
if sjukpenning_left + lagsta_niva_left != days_left:
logger.warning("Tier sum mismatch [id=%s]", state.session_id)
# -- dubbeldagar remaining --
logger.debug("[%s] reading dubbeldagar (DUBBELDAG_KVAR)", state.session_id)
dubbeldag_text = await page.locator(_SEL.DUBBELDAG_KVAR).first.inner_text(timeout=5_000)
dubbeldagar_left = int(dubbeldag_text.strip())
logger.debug("[%s] dubbeldagar parsed", state.session_id)
# -- Derive tier totals (standard FK allocation per parent) --
# Standard: 195 sjukpenningnivå + 45 lägstanivå = 240 per parent.
# Transfers can change this, but the page doesn't expose the totals directly.
# Assumption: lägstanivå days are never used before sjukpenning (most common case).
lagsta_niva_total = 45 # standard per-parent lägstanivå allocation (= rules.lagsta_niva_days/2)
sjukpenning_total = total_days - lagsta_niva_total # typically 195 for standard 240-day split
lagsta_niva_used = lagsta_niva_total - lagsta_niva_left
sjukpenning_used = sjukpenning_total - sjukpenning_left
# -- Derive reserved / transferable split --
# Reserved days (90 per parent) cannot be transferred; used assumption: transferable taken first.
reserved_days_total = settings.rules.reserved_days_per_parent # 90
transferable_total = total_days - reserved_days_total # 150 for standard 240-day case
transferable_used = min(days_used, transferable_total)
reserved_used = days_used - transferable_used
reserved_days_left = reserved_days_total - reserved_used
transferable_days_left = transferable_total - transferable_used
# -- Validate before returning --
if days_left < 0:
raise ValueError(f"days_left is negative ({days_left}) — FK data inconsistency")
if reserved_days_left > reserved_days_total:
raise ValueError(
f"reserved_days_left ({reserved_days_left}) > reserved_days_total ({reserved_days_total})"
)
now = datetime.now()
tier_data = TierData(
sjukpenning_total=sjukpenning_total,
sjukpenning_used=sjukpenning_used,
sjukpenning_left=sjukpenning_left,
lagsta_niva_total=lagsta_niva_total,
lagsta_niva_used=lagsta_niva_used,
lagsta_niva_left=lagsta_niva_left,
)
return ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=state.personal_data_uuid,
workspace_id=state.workspace_id,
child_id=state.child_id,
source="scraped",
fetched_at=now,
total_days=total_days,
days_used=days_used,
days_left=days_left,
reserved_days_total=reserved_days_total,
reserved_days_left=reserved_days_left,
transferable_days_left=transferable_days_left,
dubbeldagar_left=dubbeldagar_left,
tier_data=tier_data,
created_at=now,
updated_at=now,
)
Reference in New Issue View Git Blame Copy Permalink