F-r-ldraledigplannerare/backend/app/scrapers/fk_scraper.py

"""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 asdict, dataclass, field, replace
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
from app.services.fetch_data_service import FetchDataService
from app.services.leave_snapshot import build_snapshot
from app.utils.errors import ValidationError

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"

# The "Min föräldrapenning" e-service page. Reports the transferable-days
# split ("Dagar ni kan föra över") directly, per-tier, for both parents —
# figures that cannot be reconstructed from #/valjdagar alone (see
# `_extract_transferable_days`'s docstring).
MIN_FP_URL = (
    "https://www.forsakringskassan.se/privatperson/mina-sidor/alla-e-tjanster/min-foraldrapenning"
)


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"

    # "För dig och annan förälder" family panel — shows the CHILD's combined
    # total ("Totalt för <barn>"), from which the other parent's remaining
    # days can be deduced (family - mine). The exact DOM here is ASSUMED to
    # mirror forDig's structure (same visa-dagar/niva-statistik classes)
    # pending a real capture — see `fk_overview_synthetic.html`'s header
    # comment. Parsing against these selectors must stay fully defensive
    # (see `_extract_family_summary`): if FK's real markup differs, the
    # feature degrades to "no partner deduction" rather than breaking the
    # caller's own extraction.
    FAMILY = "[data-test='forDigOchAnnanForalder']"
    FAMILY_DAGAR_KVAR = FAMILY + " .visa-dagar__dagar-kvar"
    FAMILY_DISCRETE = FAMILY + " .visa-dagar__discrete"
    FAMILY_NIVA_STATISTIK = FAMILY + " .visa-dagar__desktop .niva-statistik__dagar span"

    # Min föräldrapenning page — "Dagar ni kan föra över" panel (per child).
    # See `MIN_FP_URL`'s comment and `_extract_transferable_days`.
    MIN_FP_BARN_SEKTION = ".min-fp .barn-sektion"
    BARN_RUBRIK = ".barn-rubrik strong"
    PANEL_FORA_OVER_DAGAR = ".panel-fora-over-dagar"
    FORA_OVER_GROUP = ".fora-over-group"
    MINFP_LABEL = "h4.minfp-label"
    EXPANDABLE_BUTTON = ".expandable-paragraph__button"


# ---------------------------------------------------------------------------
# Family-panel summary and partner deduction
# ---------------------------------------------------------------------------


@dataclass(frozen=True)
class FamilySummary:
    """Parsed "Totalt för <barn>" family-panel figures (the CHILD's combined pool)."""

    total_days: int
    days_left: int
    sjukpenning_left: int | None
    lagsta_niva_left: int | None


@dataclass(frozen=True)
class PartnerDeduction:
    """The other parent's balance, deduced as family totals minus mine."""

    total_days: int
    days_left: int
    sjukpenning_left: int
    lagsta_niva_left: int
    dubbeldagar_left: int
    transferable_days_left: int | None = None


@dataclass(frozen=True)
class TransferableDays:
    """FK's own reported "Dagar ni kan föra över" split, per tier and parent.

    Read from the `min-foraldrapenning` e-service page (see `MIN_FP_URL`),
    not derived — this is the source of truth `_apply_transferable` feeds
    back into `build_snapshot` as `transferable_days_left`. Partner fields
    are `None` for a solo parent (no "Annan förälder" group on the page).
    """

    mine_sjukpenning: int
    mine_lagsta_niva: int
    partner_sjukpenning: int | None = None
    partner_lagsta_niva: int | None = None

    @property
    def mine_total(self) -> int:
        return self.mine_sjukpenning + self.mine_lagsta_niva

    @property
    def partner_total(self) -> int | None:
        if self.partner_sjukpenning is None or self.partner_lagsta_niva is None:
            return None
        return self.partner_sjukpenning + self.partner_lagsta_niva


# ---------------------------------------------------------------------------
# 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
    # The child's display name, when the caller had it loaded already (see
    # `start_session`). Used only to pick the right `.barn-sektion` on the
    # min-foraldrapenning page when a workspace has more than one child —
    # see `_extract_transferable_days`. `None` disables that feature for
    # multi-child workspaces (never guesses a section) but never affects
    # the caller's own #/valjdagar extraction.
    child_name: str | None = None
    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)
    # Family-panel summary and the deduced partner balance, when the family
    # panel was present and its numbers were internally consistent with
    # `result` — see `_extract_family_summary`/`_deduce_partner`. Both stay
    # None whenever the family panel is absent, unparsable, or inconsistent;
    # that never affects `result` (the caller's own extraction).
    family: FamilySummary | None = None
    partner_deduction: PartnerDeduction | None = None
    # Called after a successful extraction to persist the extracted numbers.
    # The callback receives the validated ParentalLeaveFetch (mine) and the
    # deduced partner balance (or None). Only the extracted numbers are
    # persisted — never cookies, QR, or session state.
    persist_callback: (
        Callable[[ParentalLeaveFetch, PartnerDeduction | None], 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, PartnerDeduction | None], Awaitable[None]] | None
        ) = None,
        child_name: str | 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`.

        `child_name`, if the caller already has the `Child` loaded, is used
        only to pick the right section on the min-foraldrapenning page when
        a workspace has more than one child — see `_extract_transferable_days`.
        The scraper never looks it up itself (no repository access from here).

        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,
            child_name=child_name,
            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.debug("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
            response["partner_deduction"] = (
                asdict(state.partner_deduction) if state.partner_deduction else None
            )
        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.debug("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.debug("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.debug(
            "scraper debug artifacts saved [id=%s] dir=%s",
            session_id,
            out.resolve(),
        )
    except Exception as artifact_exc:
        logger.warning("failed to save debug artifacts: %s", 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)
            if state.status == "done" and settings.scraper_debug_dump_on_success:
                await _save_debug_artifacts(page, state.session_id, console_messages)
        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:
        _set_status(state, "error", "FK is currently under maintenance. Please try again later.")
        logger.debug("[%s] FK maintenance banner text: %s", state.session_id, maintenance_msg)
        logger.info("FK maintenance detected; aborting session")
        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("BankID button is disabled; aborting session")
                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)

    # Family panel ("för dig och annan förälder") -> deduce the partner's
    # balance. Best-effort only: any failure here must never affect `data`,
    # which is already a fully validated, ready-to-persist result.
    try:
        family = await _extract_family_summary(page, state)
        state.family = family
        if family is not None:
            state.partner_deduction = _deduce_partner(data, family)
    except Exception as exc:
        logger.debug(
            "[%s] family-panel deduction skipped: %s", state.session_id, type(exc).__name__
        )
        state.family = None
        state.partner_deduction = None

    # "Dagar ni kan föra över" panel on the min-foraldrapenning page -> FK's
    # own reported transferable-days split, replacing the transferable-first
    # assumption in `data`/`state.partner_deduction`. Best-effort only, and
    # must run after the family-panel block above: the family panel only
    # exists on #/valjdagar, and this navigates away from it.
    try:
        transferable = await _extract_transferable_days(page, state, state.child_name)
        if transferable is not None:
            data = _apply_transferable(data, transferable)
            if state.partner_deduction is not None and transferable.partner_total is not None:
                state.partner_deduction = replace(
                    state.partner_deduction, transferable_days_left=transferable.partner_total
                )
    except Exception as exc:
        logger.debug(
            "[%s] transferable-days parse skipped: %s", state.session_id, type(exc).__name__
        )

    if state.persist_callback is not None:
        try:
            await state.persist_callback(data, state.partner_deduction)
        except Exception as exc:
            _set_status(
                state,
                "error",
                "Scrape succeeded but failed to persist the data. Please try again.",
            )
            logger.error(
                "persist error [id=%s]: %s: %s",
                state.session_id,
                type(exc).__name__,
                exc,
                exc_info=False,
            )
            return

    _set_status(state, "done")
    state.result = data
    logger.info("scraper session done")


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."
        )
    logger.debug("[%s] total days 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, reserved/transferable split, and validate --
    # The page doesn't expose tier totals or the reserved/transferable split
    # directly, only totals-per-tier-left and the grand total; everything
    # else is derived by the single shared function also used by manual
    # entry, so the two paths can never disagree (see
    # `app.services.leave_snapshot.build_snapshot`'s docstring). This
    # raises `ValidationError` (not `ValueError`) on bad data — see
    # `tests/unit/test_fk_scraper.py::TestValidation` for the exact
    # messages a caller may match on.
    return build_snapshot(
        personal_data_uuid=state.personal_data_uuid,
        workspace_id=state.workspace_id,
        child_id=state.child_id,
        source="scraped",
        total_days=total_days,
        sjukpenning_left=sjukpenning_left,
        lagsta_niva_left=lagsta_niva_left,
        dubbeldagar_left=dubbeldagar_left,
    )


async def _extract_family_summary(page, state: _SessionState) -> FamilySummary | None:
    """Read the "för dig och annan förälder" family panel, if present and parsable.

    Defensive by design: the real FK DOM for this panel is unconfirmed (see
    `_SEL.FAMILY*`'s comment), so any missing element or unexpected text is
    treated as "no family data available" — logged at DEBUG and returned as
    None — never raised. The caller's own `data` (this parent's snapshot)
    must succeed regardless of what happens here.
    """
    try:
        family_panel = page.locator(_SEL.FAMILY)
        if await family_panel.count() == 0:
            logger.debug("[%s] family panel not present; skipping deduction", state.session_id)
            return None

        dagar_kvar_loc = page.locator(_SEL.FAMILY_DAGAR_KVAR)
        if await dagar_kvar_loc.count() == 0:
            logger.debug(
                "[%s] family panel dagar-kvar element not found; skipping deduction",
                state.session_id,
            )
            return None
        dagar_kvar_text = await dagar_kvar_loc.first.inner_text(timeout=2_000)
        days_left = int(dagar_kvar_text.strip())

        discrete_loc = page.locator(_SEL.FAMILY_DISCRETE)
        if await discrete_loc.count() == 0:
            logger.debug(
                "[%s] family panel discrete element not found; skipping deduction",
                state.session_id,
            )
            return None
        discrete_text = await discrete_loc.first.inner_text(timeout=2_000)
        total_match = re.search(r"av\s+(\d+)\s+dagar", discrete_text)
        if not total_match:
            logger.debug(
                "[%s] could not parse 'av N dagar' from family panel text; skipping deduction",
                state.session_id,
            )
            return None
        total_days = int(total_match.group(1))

        # Tier spans are optional — best-effort, absence doesn't invalidate the summary.
        sjukpenning_left: int | None = None
        lagsta_niva_left: int | None = None
        niva_loc = page.locator(_SEL.FAMILY_NIVA_STATISTIK)
        if await niva_loc.count() > 0:
            niva_texts = await niva_loc.all_inner_texts()
            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:
                sjukpenning_left, lagsta_niva_left = niva_numbers[0], niva_numbers[1]

        return FamilySummary(
            total_days=total_days,
            days_left=days_left,
            sjukpenning_left=sjukpenning_left,
            lagsta_niva_left=lagsta_niva_left,
        )
    except Exception as exc:
        logger.debug(
            "[%s] family panel parsing failed (%s); skipping deduction",
            state.session_id,
            type(exc).__name__,
        )
        return None


def _deduce_partner(mine: ParentalLeaveFetch, family: FamilySummary) -> PartnerDeduction | None:
    """Deduce the other parent's balance as family totals minus mine.

    Returns None (logging a warning) whenever the numbers are internally
    inconsistent — e.g. the family panel reports fewer days than this parent
    alone already accounts for, which can only mean a stale read or a DOM
    assumption that doesn't match FK's real markup.
    """
    total = family.total_days - mine.total_days
    left = family.days_left - mine.days_left

    if total <= 0 or left < 0 or left > total:
        logger.warning(
            "family/partner deduction inconsistent: family_total=%d family_left=%d "
            "mine_total=%d mine_left=%d -> deduced_total=%d deduced_left=%d",
            family.total_days,
            family.days_left,
            mine.total_days,
            mine.days_left,
            total,
            left,
        )
        return None

    rules = settings.rules

    if family.sjukpenning_left is not None and family.lagsta_niva_left is not None:
        sjuk_left = family.sjukpenning_left - mine.tier_data.sjukpenning_left
        lagsta_left = family.lagsta_niva_left - mine.tier_data.lagsta_niva_left
        if sjuk_left < 0 or lagsta_left < 0:
            logger.warning(
                "family/partner tier deduction inconsistent (sjuk=%d, lagsta=%d); "
                "falling back to proportional split",
                sjuk_left,
                lagsta_left,
            )
            lagsta_share = (
                rules.lagsta_niva_days * total // (rules.sjukpenning_days + rules.lagsta_niva_days)
            )
            lagsta_left = min(left, lagsta_share)
            sjuk_left = left - lagsta_left
    else:
        # No family-level tier breakdown available — assume lägstanivå is
        # untouched first (mirrors the same "transferable spent first"
        # assumption `build_snapshot` makes for the reserved/transferable split).
        lagsta_share = (
            rules.lagsta_niva_days * total // (rules.sjukpenning_days + rules.lagsta_niva_days)
        )
        lagsta_left = min(left, lagsta_share)
        sjuk_left = left - lagsta_left

    return PartnerDeduction(
        total_days=total,
        days_left=left,
        sjukpenning_left=sjuk_left,
        lagsta_niva_left=lagsta_left,
        dubbeldagar_left=mine.dubbeldagar_left,
    )


# ---------------------------------------------------------------------------
# Transferable days ("Dagar ni kan föra över") — FK's own reported split
# ---------------------------------------------------------------------------

_TRANSFERABLE_RE = re.compile(r"(\d+)\s+dagar?\s+på\s+(sjukpenningnivå|lägstanivå)")


async def _extract_transferable_days(
    page, state: _SessionState, child_name: str | None
) -> TransferableDays | None:
    """Read the "Dagar ni kan föra över" panel on the min-foraldrapenning
    e-service page and return FK's own reported transferable-days split.

    This figure cannot be reconstructed from #/valjdagar (see
    `app.services.leave_snapshot.build_snapshot`'s module docstring) — FK's
    own accounting there is not reproducible by any plausible model, so it
    must be read off this second page directly.

    Fully defensive by design, mirroring `_extract_family_summary`: any
    missing element, unmatched child, or unexpected text is "no transferable
    data available" — logged at DEBUG and returned as None — never raised.
    The caller's own `data` (already validated) must be unaffected either way.
    """
    try:
        logger.debug("[%s] navigating to min-foraldrapenning page", state.session_id)
        await page.goto(MIN_FP_URL, wait_until="domcontentloaded")

        logger.debug("[%s] waiting for transfer panel", state.session_id)
        await page.wait_for_selector(_SEL.PANEL_FORA_OVER_DAGAR, timeout=30_000)

        # The panel mounts before the async fetch lands — wait for real
        # per-tier figures rather than trusting first paint (mirrors the
        # #/valjdagar hydration guard in `_do_extract`).
        logger.debug("[%s] waiting for transfer panel data to hydrate", state.session_id)
        await page.wait_for_function(
            """(sel) => {
            const els = document.querySelectorAll(sel);
            for (const el of els) {
                if (/\\d+\\s+dagar?\\s+på/.test(el.textContent)) return true;
            }
            return false;
        }""",
            arg=_SEL.FORA_OVER_GROUP + " p",
            timeout=30_000,
        )
        logger.debug("[%s] transfer panel data hydrated", state.session_id)

        sections = page.locator(_SEL.MIN_FP_BARN_SEKTION)
        section_count = await sections.count()
        if section_count == 0:
            logger.debug(
                "[%s] no barn-sektion found on min-foraldrapenning page; skipping "
                "transferable-days parse",
                state.session_id,
            )
            return None

        if section_count == 1:
            chosen = sections.first
        else:
            if not child_name or not child_name.strip():
                logger.debug(
                    "[%s] multiple children on min-foraldrapenning page and no "
                    "child_name to match; skipping transferable-days parse",
                    state.session_id,
                )
                return None
            target = child_name.strip().split()[0].casefold()
            chosen = None
            for i in range(section_count):
                section = sections.nth(i)
                rubrik_loc = section.locator(_SEL.BARN_RUBRIK)
                if await rubrik_loc.count() == 0:
                    continue
                rubrik_text = (await rubrik_loc.first.inner_text(timeout=2_000)).strip().casefold()
                if rubrik_text == target:
                    chosen = section
                    break
            if chosen is None:
                logger.debug(
                    "[%s] no barn-sektion matched child_name; skipping transferable-days parse",
                    state.session_id,
                )
                return None

        panel = chosen.locator(_SEL.PANEL_FORA_OVER_DAGAR)
        if await panel.count() == 0:
            logger.debug(
                "[%s] chosen barn-sektion has no transfer panel; skipping transferable-days parse",
                state.session_id,
            )
            return None

        # Content is present in the DOM even when the accordion is collapsed,
        # so text reads work as-is. Still click through as insurance against
        # FK switching to lazy rendering — a same-page toggle, never a dialog.
        button = panel.locator(_SEL.EXPANDABLE_BUTTON)
        if await button.count() > 0:
            aria_expanded = await button.first.get_attribute("aria-expanded")
            if aria_expanded == "false":
                logger.debug("[%s] expanding collapsed transfer panel", state.session_id)
                await button.first.click(timeout=5_000)

        groups = panel.locator(_SEL.FORA_OVER_GROUP)
        group_count = await groups.count()

        mine_sjuk: int | None = None
        mine_lagsta: int | None = None
        partner_sjuk: int | None = None
        partner_lagsta: int | None = None

        for i in range(group_count):
            group = groups.nth(i)
            label_loc = group.locator(_SEL.MINFP_LABEL)
            if await label_loc.count() == 0:
                continue
            label_text = (await label_loc.first.inner_text(timeout=2_000)).strip()

            # Direct children only — the tooltip inside "Du kan föra över"
            # holds decoy "45 dagar"/"90 dagar" text that a descendant
            # selector would also pick up.
            p_texts = await group.locator(":scope > p").all_inner_texts()
            sjuk: int | None = None
            lagsta: int | None = None
            for t in p_texts:
                m = _TRANSFERABLE_RE.search(t)
                if not m:
                    continue
                value = int(m.group(1))
                if m.group(2) == "sjukpenningnivå":
                    sjuk = value
                else:
                    lagsta = value

            if label_text.startswith("Du kan"):
                mine_sjuk, mine_lagsta = sjuk, lagsta
            elif label_text.startswith("Annan förälder"):
                partner_sjuk, partner_lagsta = sjuk, lagsta

        if mine_sjuk is None or mine_lagsta is None:
            logger.debug(
                "[%s] 'Du kan föra över' group missing or incomplete; skipping "
                "transferable-days parse",
                state.session_id,
            )
            return None

        return TransferableDays(
            mine_sjukpenning=mine_sjuk,
            mine_lagsta_niva=mine_lagsta,
            partner_sjukpenning=partner_sjuk,
            partner_lagsta_niva=partner_lagsta,
        )
    except Exception as exc:
        logger.debug(
            "[%s] transferable-days panel parsing failed (%s); skipping",
            state.session_id,
            type(exc).__name__,
        )
        return None


def _apply_transferable(
    data: ParentalLeaveFetch, transferable: TransferableDays
) -> ParentalLeaveFetch:
    """Re-run `build_snapshot` with FK's own reported transferable-days
    figure in place of the transferable-first assumption used when `data`
    was first built.

    An FK-reported figure that fails `build_snapshot`'s validation (should
    never happen, but FK's markup is not a contract) degrades to the
    original `data` unchanged rather than failing the whole fetch.
    """
    try:
        return build_snapshot(
            personal_data_uuid=data.personal_data_uuid,
            workspace_id=data.workspace_id,
            child_id=data.child_id,
            source=data.source,
            total_days=data.total_days,
            sjukpenning_left=data.tier_data.sjukpenning_left,
            lagsta_niva_left=data.tier_data.lagsta_niva_left,
            dubbeldagar_left=data.dubbeldagar_left,
            transferable_days_left=transferable.mine_total,
            fetched_at=data.fetched_at,
        )
    except ValidationError as exc:
        logger.warning(
            "FK-reported transferable_days_left=%d implausible against derived "
            "days_left=%d (child_id=%s); keeping derived reserved/transferable split: %s",
            transferable.mine_total,
            data.days_left,
            data.child_id,
            exc,
        )
        return data