"""HR leave / time-off domain service.

Two responsibilities, split into a sync side and a read side:

* :func:`sync_leave_data` — pulls **all** of a person's SF TimeAccount
  buckets (法定年假 / 福利年假 / 带薪病假 / 调休假 / …, not just the two
  the legacy carryover knew about) plus every TimeAccountDetail line,
  mirrors them into ``hr_leave_accounts`` / ``hr_leave_postings``
  idempotently, computes per-account balance/entitled/used + per-line
  running balance, soft-deletes rows SF no longer returns, and finally
  derives the four legacy ``identity_persons.annual_leave_*`` snapshot
  columns from the freshly-synced data so the 年假计算器 keeps working
  off a single SF round-trip.

* :class:`HrLeaveService` — read side used by the router/MCP. Resolves a
  person's accounts (current year first) and an account's ledger, with
  the running ``balance_before`` derived on read.

SF date literals are OData v2 ``/Date(ms)/`` epochs; helpers below parse
them to ``date`` / aware ``datetime``.
"""
from __future__ import annotations

import logging
import re
from datetime import UTC, date, datetime
from decimal import Decimal, InvalidOperation
from typing import TYPE_CHECKING, Any

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.hr.leave_models import (
    HrLeaveAbsence,
    HrLeaveAccount,
    HrLeavePosting,
)
from app.identity.person_models import IdentityPerson

if TYPE_CHECKING:
    from app.domains.hr.service import SuccessFactorsLifecycleService

logger = logging.getLogger(__name__)

# bookingType values that count as leave actually taken (used days) —
# same contract the legacy carryover uses. ACCRUAL / MANUAL_ADJUSTMENT
# are entitlement entries and do NOT count as used.
USED_BOOKING_TYPES = frozenset({"EMPLOYEE_TIME", "DEDUCTION"})

# Fallback label map for the common tenant account-type codes, used only
# when SF's TimeAccountType entity is unavailable. The backend stores a
# value so MCP / API consumers get a human label; the frontend ALSO carries
# an i18n-aware ``leave.accountType.*`` map for the same codes (it renders
# per-locale, which a single stored Chinese string cannot).
#
# Keep this in sync with ``leave_insights_service``'s classification sets and
# with the frontend map. The 2026-08-09 audit found ``CHN_PSL`` and
# ``CHN_COMPL`` missing from BOTH this table and (at the time) the frontend,
# so employees saw the raw vendor codes on their own leave cards —
# ``CHN_COMPL`` was even already known to the insights module as a 调休 code.
_FALLBACK_TYPE_LABELS: dict[str, str] = {
    "CHN_AnnualLeave": "法定年假",
    "CHN_PL": "福利年假",
    "CHN_SickLeave": "带薪病假",
    "CHN_PaidSickLeave": "带薪病假",
    "CHN_PSL": "带薪病假",
    "CHN_CompLeave": "调休假",
    "CHN_COMPL": "调休假",
    "CHN_TOIL": "调休假",
    "MY_AnnuL": "法定年假",
}

# Localised labels for the same codes. ``_FALLBACK_TYPE_LABELS`` above is a
# single stored Chinese string (it's what lands in the DB column), which is
# fine for a UI that re-labels client-side — but NOT for outbound copy: a
# notification body is rendered server-side in the recipient's language, and
# splicing 「法定年假」 into an English mail is exactly the mixed-script
# residue the i18n rules exist to prevent.
#
# Only the vendor codes the tenant actually uses are listed. An unlisted code
# degrades to the SF-provided label (Chinese, but meaningful) and only then to
# the raw code — never the other way round, because a reader is better served
# by 「福利年假」 than by ``CHN_PL``.
#
# **These MUST match the frontend's ``leave.accountType.*`` i18n keys**, and
# ``tests/test_hr/test_leave_type_label_parity.py`` fails the build if they
# drift. The frontend strings are the incumbent — employees already read them
# on their own leave card — so an email that calls the same bucket something
# else is Helix contradicting itself to the same person.
LOCALIZED_TYPE_LABELS: dict[str, dict[str, str]] = {
    "CHN_AnnualLeave": {
        "zh": "法定年假",
        "en": "Statutory annual leave",
        "ja": "法定年次休暇",
    },
    "CHN_COMPL": {
        "zh": "调休假",
        "en": "Time off in lieu",
        "ja": "代休",
    },
    "CHN_CompLeave": {
        "zh": "调休假",
        "en": "Time off in lieu",
        "ja": "代休",
    },
    "CHN_PL": {
        "zh": "福利年假",
        "en": "Company annual leave",
        "ja": "福利年次休暇",
    },
    "CHN_PSL": {
        "zh": "带薪病假",
        "en": "Paid sick leave",
        "ja": "有給病気休暇",
    },
    "CHN_PaidSickLeave": {
        "zh": "带薪病假",
        "en": "Paid sick leave",
        "ja": "有給病気休暇",
    },
    "CHN_SickLeave": {
        "zh": "带薪病假",
        "en": "Paid sick leave",
        "ja": "有給病気休暇",
    },
    "CHN_TOIL": {
        "zh": "调休假",
        "en": "Time off in lieu",
        "ja": "代休",
    },
    "MY_AnnuL": {
        "zh": "法定年假",
        "en": "Statutory annual leave",
        "ja": "法定年次休暇",
    },
}


def localized_type_label(
    code: str | None, sf_label: str | None, lang: str,
) -> str:
    """A leave account-type label in ``lang`` (``zh`` / ``en`` / ``ja``).

    Chinese readers keep seeing SF's own label when it exists, so Helix never
    contradicts the HRIS wording the employee sees there. Other languages
    prefer the curated translation and fall back to SF's label, then the code.
    """
    variants = LOCALIZED_TYPE_LABELS.get(code or "")
    if lang == "zh":
        return sf_label or (variants or {}).get("zh") or (code or "")
    if variants and variants.get(lang):
        return variants[lang]
    return sf_label or (code or "")


_DATE_RE = re.compile(r"/Date\((-?\d+)(?:[+-]\d+)?\)/")


def _parse_sf_date(value: object) -> date | None:
    if not isinstance(value, str):
        return None
    m = _DATE_RE.match(value)
    if not m:
        return None
    return datetime.fromtimestamp(int(m.group(1)) / 1000, tz=UTC).date()


def _parse_sf_datetime(value: object) -> datetime | None:
    if not isinstance(value, str):
        return None
    m = _DATE_RE.match(value)
    if not m:
        return None
    return datetime.fromtimestamp(int(m.group(1)) / 1000, tz=UTC)


def _to_decimal(value: object) -> Decimal | None:
    if value is None:
        return None
    try:
        return Decimal(str(value))
    except (InvalidOperation, ValueError):
        return None


def _q2(value: Decimal | None) -> Decimal | None:
    return None if value is None else value.quantize(Decimal("0.01"))


def leave_fiscal_year_label(
    *, period_end: date | None, fiscal_year: int | None,
) -> str | None:
    """Return the HR leave FY label, such as ``FY26``.

    Leave's FY naming is deliberately different from the finance calendar:
    annual-leave grants are named for the year their leave period ends.  Thus
    an account granted on 2025-04-01 and usable through 2026-09-30 is FY26,
    even though the raw ``fiscal_year`` mirror column holds its start year
    (2025).  ``period_end`` must be preferred because late-FY joiners can
    have a 2026 start date while belonging to the same FY26 grant cycle.
    """
    year = period_end.year if period_end is not None else (
        fiscal_year + 1 if fiscal_year is not None else None
    )
    return f"FY{year % 100:02d}" if year is not None else None


async def sync_leave_data(
    service: SuccessFactorsLifecycleService,
    *,
    user_ids: list[str] | None = None,
    commit: bool = True,
) -> dict[str, Any]:
    """Mirror SF EC Time-Off into ``hr_leave_*`` + derive legacy columns.

    ``user_ids`` (SF ``userId``) scopes the refresh; ``None`` refreshes
    every active SF-sourced person in this service's HRIS instance.

    Returns a summary dict (``accounts_upserted`` / ``postings_upserted``
    / ``accounts_tombstoned`` / ``persons_touched`` / ``legacy_derived``)
    persisted as the scheduled-job ``last_result_summary``. Zero is a
    valid outcome (tenant without EC Time-Off, empty roster).
    """
    db = service.db
    today_date = service.today

    # 1. Resolve target persons (scoped to this HRIS instance), keyed by
    #    SF userId — mirrors sync_annual_leave_carryover's selection so CN
    #    and EU syncs never overwrite each other's rows.
    person_query = select(IdentityPerson).where(
        IdentityPerson.source_system == "successfactors",
        IdentityPerson.deleted_at.is_(None),
    )
    if service.hris_instance_id is not None:
        from sqlalchemy import or_
        person_query = person_query.where(
            or_(
                IdentityPerson.hris_instance_id == service.hris_instance_id,
                IdentityPerson.hris_instance_id.is_(None),
            )
        )
    if user_ids is not None:
        if not user_ids:
            return {"accounts_upserted": 0, "postings_upserted": 0,
                    "accounts_tombstoned": 0, "absences_upserted": 0,
                    "absences_tombstoned": 0, "persons_touched": 0,
                    "legacy_derived": 0}
        person_query = person_query.where(
            IdentityPerson.hris_person_id.in_(user_ids)
        )
    person_rows = (await db.execute(person_query)).scalars().all()
    person_by_user_id: dict[str, IdentityPerson] = {
        p.hris_person_id: p for p in person_rows if p.hris_person_id
    }
    if not person_by_user_id:
        return {"accounts_upserted": 0, "postings_upserted": 0,
                "accounts_tombstoned": 0, "absences_upserted": 0,
                "absences_tombstoned": 0, "persons_touched": 0,
                "legacy_derived": 0}

    client = await service._get_client()

    # 2. Pull ALL time-account buckets for these users (no type filter →
    #    every leave type), plus the type-label map (best-effort).
    accounts = await client.list_time_accounts(
        user_ids=list(person_by_user_id.keys()),
        account_types=None,
    )
    # An empty list is an authoritative 0-account snapshot for persons
    # explicitly owned by this service. It must still reach the tombstone
    # sweep below, while avoiding three pointless SF calls.
    if accounts:
        type_labels = await client.list_time_account_types()
        account_uuids = [
            a["externalCode"] for a in accounts if a.get("externalCode")
        ]
        details = await client.list_time_account_details(
            time_account_uuids=account_uuids,
        )
        requests = await client.list_employee_time(
            user_ids=list(person_by_user_id.keys()),
        )
    else:
        type_labels = {}
        details = []
        requests = []

    # 3. Index every detail line for those accounts.
    details_by_account: dict[str, list[dict]] = {}
    for d in details:
        acct_uuid = d.get("TimeAccount_externalCode")
        if acct_uuid:
            details_by_account.setdefault(acct_uuid, []).append(d)

    # 3b. Pull the time-off REQUESTS (EmployeeTime) so we can stamp the
    #     reason / type / approval onto each deduction posting — that
    #     context isn't on TimeAccountDetail. Indexed per userId as a list
    #     of (start, end, reason, type, status); matched by booking_date.
    #     Best-effort: empty when the entity is unavailable.
    requests_by_user: dict[str, list[dict]] = {}
    for r in requests:
        uid = r.get("userId")
        if not uid:
            continue
        requests_by_user.setdefault(uid, []).append({
            "start": _parse_sf_date(r.get("startDate")),
            "end": _parse_sf_date(r.get("endDate")),
            "reason": (r.get("comment") or None),
            "leave_type": (r.get("timeType") or None),
            "status": (r.get("approvalStatus") or None),
        })

    def _match_request(user_id: str, on: date | None) -> dict | None:
        """The time-off request covering ``on`` for ``user_id`` (the
        deduction's date falls within the leave span). Returns the single
        unambiguous match, else None (so we never mis-attribute a reason)."""
        if not user_id or on is None:
            return None
        hits = [
            req for req in requests_by_user.get(user_id, [])
            if req["start"] and req["end"] and req["start"] <= on <= req["end"]
        ]
        return hits[0] if len(hits) == 1 else None

    # 4. Load existing Helix rows for these persons so we upsert (and can
    #    tombstone rows SF stopped returning).
    person_ids = [str(p.id) for p in person_by_user_id.values()]
    existing_accounts = {
        a.external_code: a
        for a in (
            await db.execute(
                select(HrLeaveAccount).where(
                    HrLeaveAccount.person_id.in_(person_ids)
                )
            )
        ).scalars().all()
    }
    existing_postings = {
        p.external_code: p
        for p in (
            await db.execute(
                select(HrLeavePosting).where(
                    HrLeavePosting.person_id.in_(person_ids)
                )
            )
        ).scalars().all()
    }
    existing_absences = {
        a.external_code: a
        for a in (
            await db.execute(
                select(HrLeaveAbsence).where(
                    HrLeaveAbsence.person_id.in_(person_ids)
                )
            )
        ).scalars().all()
    }

    person_by_id = {str(person.id): person for person in person_rows}

    def _can_tombstone_account(row: HrLeaveAccount) -> bool:
        """Whether this SF service is authoritative for ``row``.

        Legacy single-instance mode owns the whole selected person set. In
        multi-instance mode, an account explicitly stamped by another
        instance is never ours. An unscoped legacy account is safe only when
        its person has since been pinned to this instance; an unowned person
        remains ambiguous and is preserved.
        """
        if service.hris_instance_id is None:
            return True
        if row.hris_instance_id == service.hris_instance_id:
            return True
        if row.hris_instance_id is not None:
            return False
        person = person_by_id.get(row.person_id)
        return bool(
            person is not None
            and person.hris_instance_id == service.hris_instance_id
        )

    now = datetime.now(UTC)
    accounts_upserted = 0
    postings_upserted = 0
    absences_upserted = 0
    absences_tombstoned = 0
    seen_absence_codes: set[str] = set()
    seen_account_codes: set[str] = set()
    seen_posting_codes: set[str] = set()

    # account_uuid → HrLeaveAccount (resolved/created), for posting FK.
    account_by_uuid: dict[str, HrLeaveAccount] = {}

    for acct in accounts:
        ext = acct.get("externalCode")
        user_id = acct.get("userId")
        atype = acct.get("accountType")
        if not ext or not user_id or not atype:
            continue
        person = person_by_user_id.get(user_id)
        if person is None:
            continue
        seen_account_codes.add(ext)

        start = _parse_sf_date(acct.get("startDate"))
        end = _parse_sf_date(acct.get("endDate"))
        lines = details_by_account.get(ext, [])

        # Per-account aggregates over the detail lines.
        balance = Decimal(0)
        entitled = Decimal(0)
        used = Decimal(0)
        for d in lines:
            amt = _to_decimal(d.get("bookingAmount"))
            if amt is None:
                continue
            balance += amt
            if d.get("bookingType") in USED_BOOKING_TYPES:
                # Consumption ledger: leave taken posts a negative booking;
                # cancelling approved leave posts a positive reversal of the
                # SAME booking type. Net them (used -= amt) so a cancelled
                # absence doesn't inflate used_days, and a positive reversal
                # isn't double-counted as both used AND entitled.
                used -= amt
            elif amt > 0:
                entitled += amt

        label = type_labels.get(str(atype)) or _FALLBACK_TYPE_LABELS.get(str(atype))
        is_current = bool(start and end and start <= today_date <= end)
        fiscal_year = (start or end).year if (start or end) else None

        row = existing_accounts.get(ext)
        if row is None:
            row = HrLeaveAccount(external_code=ext)
            db.add(row)
            existing_accounts[ext] = row
        row.person_id = str(person.id)
        row.hris_person_id = user_id
        row.hris_instance_id = service.hris_instance_id
        row.account_type = atype
        row.account_type_label = label
        row.fiscal_year = fiscal_year
        row.start_date = start
        row.end_date = end
        row.booking_start_date = _parse_sf_date(acct.get("bookingStartDate"))
        row.booking_end_date = _parse_sf_date(acct.get("bookingEndDate"))
        row.account_closed = bool(acct.get("accountClosed"))
        row.balance_days = _q2(balance)
        row.entitled_days = _q2(entitled)
        row.used_days = _q2(used)
        row.is_current = is_current
        row.synced_at = now
        row.deleted_at = None
        account_by_uuid[ext] = row
        accounts_upserted += 1

    # Need account ids to FK the postings — flush so server-default PKs
    # and the GUID defaults are assigned.
    await db.flush()

    # 5. Upsert postings, computing the running balance per account in SF
    #    chronological order (booking_date, then created timestamp, then
    #    a stable external_code tiebreak).
    for acct_uuid, lines in details_by_account.items():
        account = account_by_uuid.get(acct_uuid)
        if account is None:
            # Detail for an account we skipped (no matching person) — drop.
            continue

        def _sort_key(d: dict) -> tuple:
            bd = _parse_sf_date(d.get("bookingDate")) or date.min
            cd = _parse_sf_datetime(d.get("createdDateTime")) or datetime.min.replace(tzinfo=UTC)
            return (bd, cd, str(d.get("externalCode") or ""))

        running = Decimal(0)
        for d in sorted(lines, key=_sort_key):
            ext = d.get("externalCode")
            amt = _to_decimal(d.get("bookingAmount"))
            if not ext or amt is None:
                continue
            running += amt
            seen_posting_codes.add(ext)

            prow = existing_postings.get(ext)
            if prow is None:
                prow = HrLeavePosting(external_code=ext)
                db.add(prow)
                existing_postings[ext] = prow
            prow.leave_account_id = account.id
            prow.account_external_code = acct_uuid
            prow.person_id = account.person_id
            prow.booking_type = str(d.get("bookingType") or "UNKNOWN")
            prow.booking_amount = _q2(amt)
            prow.booking_unit = d.get("bookingUnit")
            prow.booking_date = (
                _parse_sf_date(d.get("postingDate"))
                or _parse_sf_date(d.get("bookingDate"))
            )
            prow.accrual_period_id = d.get("accrualPeriodId")
            prow.balance_after = _q2(running)
            prow.created_by_source = (
                str(d["createdBy"]) if d.get("createdBy") else None
            )
            prow.created_at_source = _parse_sf_datetime(d.get("createdDateTime"))
            prow.comment = d.get("comment") or None
            # Stamp reason/type/approval from the matching time-off request
            # (only leave-taken lines map to a request; accrual/adjustment
            # lines clear it so a re-sync never leaves a stale reason).
            req = (
                _match_request(account.hris_person_id, prow.booking_date)
                if prow.booking_type in USED_BOOKING_TYPES else None
            )
            prow.reason = req["reason"] if req else None
            prow.leave_type = req["leave_type"] if req else None
            prow.request_status = req["status"] if req else None
            prow.synced_at = now
            prow.deleted_at = None
            postings_upserted += 1

    # 6. Tombstone rows SF no longer returns (scoped to synced persons AND
    #    the current HRIS instance). A legacy NULL-owned person is selected
    #    into each fan-out for compatibility, but cannot grant either region
    #    authority to delete the other one's leave history.
    accounts_tombstoned = 0
    tombstone_account_ids: set[Any] = set()
    for code, row in existing_accounts.items():
        if not _can_tombstone_account(row):
            continue
        tombstone_account_ids.add(row.id)
        if code not in seen_account_codes and row.deleted_at is None:
            row.deleted_at = now
            accounts_tombstoned += 1
    for code, prow in existing_postings.items():
        if (
            prow.leave_account_id in tombstone_account_ids
            and code not in seen_posting_codes
            and prow.deleted_at is None
        ):
            prow.deleted_at = now

    # 6b. Upsert the time-off REQUESTS themselves (EmployeeTime) into
    #     hr_leave_absences. These are the spans the External API
    #     availability feed reads — postings (accounting deductions) can't
    #     answer "on leave on date X". Reuses the ``requests`` already
    #     fetched in step 3b; keyed on SF externalCode, soft-deletes rows SF
    #     stopped returning (scoped to synced persons + instance authority).
    for r in requests:
        ext = r.get("externalCode")
        user_id = r.get("userId")
        if not ext or not user_id:
            continue
        person = person_by_user_id.get(user_id)
        if person is None:
            continue
        seen_absence_codes.add(ext)
        arow = existing_absences.get(ext)
        if arow is None:
            arow = HrLeaveAbsence(external_code=ext)
            db.add(arow)
            existing_absences[ext] = arow
        arow.person_id = str(person.id)
        arow.hris_person_id = user_id
        arow.hris_instance_id = service.hris_instance_id
        arow.start_date = _parse_sf_date(r.get("startDate"))
        arow.end_date = _parse_sf_date(r.get("endDate"))
        arow.time_type = r.get("timeType") or None
        arow.quantity_days = _q2(_to_decimal(r.get("quantityInDays")))
        arow.approval_status = r.get("approvalStatus") or None
        arow.synced_at = now
        arow.deleted_at = None
        absences_upserted += 1

    # Tombstone absences SF no longer returns. Same instance-authority gate
    # as accounts: never delete another region's rows. person_id membership
    # in the synced set is the scope (an absence's person is always in
    # person_by_id because we only ever wrote rows for synced persons).
    for code, arow in existing_absences.items():
        if code in seen_absence_codes or arow.deleted_at is not None:
            continue
        person = person_by_id.get(arow.person_id)
        if service.hris_instance_id is not None:
            if arow.hris_instance_id == service.hris_instance_id:
                pass  # ours
            elif arow.hris_instance_id is not None:
                continue  # another region's row
            elif not (
                person is not None
                and person.hris_instance_id == service.hris_instance_id
            ):
                continue  # unowned + person not pinned here — ambiguous
        arow.deleted_at = now
        absences_tombstoned += 1

    # 7. Derive the four legacy snapshot columns from the synced data so
    #    the 年假计算器 stays correct without a second SF round-trip.
    legacy_derived = _derive_legacy_columns(
        person_by_user_id, existing_accounts, today_date,
    )

    # Full-roster cron / post-confirm callers retain the historical
    # per-instance commit. The per-person HTTP/MCP path passes ``False`` so
    # its caller audit can join these rows in one outer commit (#2467).
    if commit:
        await db.commit()
    else:
        await db.flush()
    return {
        "accounts_upserted": accounts_upserted,
        "postings_upserted": postings_upserted,
        "accounts_tombstoned": accounts_tombstoned,
        "absences_upserted": absences_upserted,
        "absences_tombstoned": absences_tombstoned,
        "persons_touched": len(person_by_user_id),
        "legacy_derived": legacy_derived,
    }


def _derive_legacy_columns(
    person_by_user_id: dict[str, IdentityPerson],
    accounts: dict[str, HrLeaveAccount],
    today_date: date,
) -> int:
    """Recompute ``identity_persons.annual_leave_*`` from synced accounts.

    carry_* = prior-FY balance of the legal/welfare account; used_* =
    current-FY used of same. Mirrors the legacy
    ``sync_annual_leave_carryover`` output so downstream readers (the
    年假计算器 prefill) are unaffected by the new data model. Reads the
    legal/welfare account-type codes from settings.
    """
    from app.core.config import settings

    legal = (settings.HR_TIMEOFF_LEGAL_ACCOUNT_TYPE or "").strip()
    welfare = (settings.HR_TIMEOFF_WELFARE_ACCOUNT_TYPE or "").strip()
    if not legal and not welfare:
        return 0

    # Index live accounts per (person_id, account_type, role).
    by_person: dict[str, list[HrLeaveAccount]] = {}
    for row in accounts.values():
        if row.deleted_at is not None:
            continue
        by_person.setdefault(row.person_id, []).append(row)

    def _pick(rows: list[HrLeaveAccount], atype: str, *, current: bool):
        best = None
        for r in rows:
            if r.account_type != atype:
                continue
            if current and not r.is_current:
                continue
            if not current:
                # prior = ended strictly before today, latest end wins
                if not (r.end_date and r.end_date < today_date):
                    continue
                if best is None or (r.end_date and best.end_date and r.end_date > best.end_date):
                    best = r
                continue
            best = r
        return best

    def _q1(v):
        return None if v is None else Decimal(str(v)).quantize(Decimal("0.1"))

    touched = 0
    for person in person_by_user_id.values():
        rows = by_person.get(str(person.id), [])
        if not rows:
            continue
        prior_legal = _pick(rows, legal, current=False) if legal else None
        prior_welfare = _pick(rows, welfare, current=False) if welfare else None
        cur_legal = _pick(rows, legal, current=True) if legal else None
        cur_welfare = _pick(rows, welfare, current=True) if welfare else None

        changed = False
        if prior_legal is not None and prior_legal.balance_days is not None:
            nv = _q1(prior_legal.balance_days)
            if person.annual_leave_carry_legal_days != nv:
                person.annual_leave_carry_legal_days = nv
                changed = True
        if prior_welfare is not None and prior_welfare.balance_days is not None:
            nv = _q1(prior_welfare.balance_days)
            if person.annual_leave_carry_welfare_days != nv:
                person.annual_leave_carry_welfare_days = nv
                changed = True
        if cur_legal is not None and cur_legal.used_days is not None:
            nv = _q1(cur_legal.used_days)
            if person.annual_leave_used_legal_days != nv:
                person.annual_leave_used_legal_days = nv
                changed = True
        if cur_welfare is not None and cur_welfare.used_days is not None:
            nv = _q1(cur_welfare.used_days)
            if person.annual_leave_used_welfare_days != nv:
                person.annual_leave_used_welfare_days = nv
                changed = True
        if changed:
            touched += 1
    return touched


class HrLeaveService:
    """Read side for the leave domain (router + MCP)."""

    def __init__(self, db: AsyncSession):
        self.db = db

    async def list_accounts(
        self, person_id: str, *, include_closed: bool = True,
    ) -> list[HrLeaveAccount]:
        """A person's leave-account buckets, current-year first, then by
        type then descending fiscal year (matches SF's card ordering)."""
        q = select(HrLeaveAccount).where(
            HrLeaveAccount.person_id == person_id,
            HrLeaveAccount.deleted_at.is_(None),
        )
        if not include_closed:
            q = q.where(HrLeaveAccount.account_closed.is_(False))
        rows = (await self.db.execute(q)).scalars().all()
        return sorted(
            rows,
            key=lambda r: (
                0 if r.is_current else 1,
                r.account_type or "",
                -(r.fiscal_year or 0),
            ),
        )

    async def get_account(self, account_id: str) -> HrLeaveAccount | None:
        return await self.db.get(HrLeaveAccount, account_id)

    async def list_postings(self, account_id: str) -> list[HrLeavePosting]:
        """An account's ledger lines, newest posting first (matches the
        「过去的假期记录」 grid)."""
        rows = (
            await self.db.execute(
                select(HrLeavePosting).where(
                    HrLeavePosting.leave_account_id == account_id,
                    HrLeavePosting.deleted_at.is_(None),
                )
            )
        ).scalars().all()
        # NB: ``created_at_source`` comes back NAIVE on a fresh MySQL/TiDB
        # load (DateTime(timezone=True) strips tzinfo). Mixing those naive
        # values with an aware fallback in the sort key raises
        # "can't compare offset-naive and offset-aware datetimes" → 500 →
        # the ledger drawer's 「加载假期记录失败」. Coerce every key element
        # to NAIVE so the comparison is always like-for-like.
        def _naive(dt: datetime | None) -> datetime:
            if dt is None:
                return datetime.min
            return dt.replace(tzinfo=None)

        return sorted(
            rows,
            key=lambda r: (r.booking_date or date.min, _naive(r.created_at_source)),
            reverse=True,
        )