"""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
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

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"

    # "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"


# ---------------------------------------------------------------------------
# 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


# ---------------------------------------------------------------------------
# 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)
    # 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,
    ) -> 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
            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.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)
            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:
        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)

    # 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.info("[%s] family-panel deduction skipped: %s", state.session_id, type(exc).__name__)
        state.family = None
        state.partner_deduction = None

    if state.persist_callback is not None:
        try:
            await state.persist_callback(data, state.partner_deduction)
        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."
        )
    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 INFO 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.info("[%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.info(
                "[%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.info(
                "[%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.info(
                "[%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.info(
            "[%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,
    )