"""年假到期(清零)提醒 —— 圈人、渲染、群发。

背景:``hr_leave_accounts`` 里每个桶都带 ``booking_end_date``(SF 的最后可申请
日),员工端假期卡片早就会显示「还有 N 天到期」,管理端洞察板也早有「临期未休」
汇总 —— 但**没有任何主动提醒**。员工必须自己打开页面才看得到,于是每年 9 月
30 日都有人的法定年假静悄悄清零。这个模块补上最后一段:HR 按自己的口径圈人、
改文案、过目、群发。

三条设计红线,来自平台既有的事故与规矩:

1. **预览即发送**。预览和发送共用 :func:`resolve_cohort`,并且 HR 确认时看到
   的每个人的数字会**冻结**进批次行(``recipients_snapshot``),后台任务只读
   快照。否则 06:30 的 HRIS 同步 cron 在「预览」和「发送」之间跑一次,发出去
   的数字就和 HR 点确认时看到的不是一回事。
2. **群发必须走 fanout**。``enqueue_then_drain`` 先把每一行 outbox 提交成
   durable 记录再逐条投递;直接在一个事务里 dispatch N 次会把 N 个
   ``identity_persons`` 行锁扣在 N 次 HTTP 往返上,那正是 2026-06 的登录雪崩
   (``reference_outbox_fk_lock_login_outage``)。
3. **不新增自动推送**。平台的既定规矩是不给员工加自动推送(唯一例外是年度
   报告)。这里**只有** HR 手动触发的路径 —— 没有 cron、没有定时任务、没有
   「到 T-14 天自动发」。要自动化必须另立 ADR。

去重语义:「近期已提醒」= 在 ``resend_window_days`` 内收到过**非测试**批次的
卡片。默认从选择里排除,但 HR 可以 ``allow_resend`` 强制再发一轮(9 月冲刺期
再催一次是合理需求),所以这是护栏、不是硬闸。
"""
from __future__ import annotations

import hashlib
import logging
import uuid
from collections.abc import Iterable
from dataclasses import dataclass, field
from datetime import UTC, date, datetime, timedelta
from decimal import Decimal
from typing import Any, NamedTuple
from urllib.parse import urlparse
from zoneinfo import ZoneInfo

from sqlalchemy import func, select
from sqlalchemy import update as sa_update
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.exceptions import ConflictError, NotFoundError, ValidationError
from app.core.locale import DEFAULT_LANG, normalize_lang, resolve_lang
from app.domains.hr.leave_models import HrLeaveAccount
from app.domains.hr.leave_reminder_models import (
    BATCH_RESOURCE_TYPE,
    BATCH_STATUS_COMPLETED,
    BATCH_STATUS_FAILED,
    BATCH_STATUS_PARTIAL,
    BATCH_STATUS_SENDING,
    HrLeaveReminderBatch,
    HrLeaveReminderExclusion,
)
from app.domains.hr.leave_reminder_schemas import (
    ALLOWED_CHANNELS,
    AccountSnapshot,
    AccountTypeOption,
    CandidateRow,
    CohortSummary,
    PreviewResponse,
    ReminderFilters,
    RenderedMessage,
    TemplateCopy,
)
from app.domains.hr.leave_service import localized_type_label
from app.identity.bu_helpers import bu_from_org_path
from app.identity.person_models import IdentityOrgMembership, IdentityPerson

logger = logging.getLogger(__name__)

#: 通知模板 key。HR 的文案覆盖存在 ``email_template.<这个 key>``。
TEMPLATE_KEY = "hr_leave.expiry_reminder"
TEMPLATE_SETTING_KEY = f"email_template.{TEMPLATE_KEY}"

SETTING_HRIS_URL = "hr.leave_expiry_reminder.hris_url"
SETTING_RESEND_WINDOW = "hr.leave_expiry_reminder.resend_window_days"
SETTING_DEFAULTS = "hr.leave_expiry_reminder.defaults"
SETTING_DEFAULT_CHANNELS = "hr.leave_expiry_reminder.default_channels"
SETTING_SIGNOFF = "hr.leave_expiry_reminder.signoff"

DEFAULT_RESEND_WINDOW_DAYS = 7

#: Helix 里「我的假期」的深链 —— 站内通知的点击目标,以及 HR 没配 HRIS 入口时
#: 卡片按钮的降级落点。
MY_LEAVE_PATH = "/me/settings?tab=leave"

#: 请假入口的出厂值 —— HR 指定的飞书 applink(2026-08-18)。
#:
#: 上一版是 SF 的 Time Off 页(``https://hcm-cn10.hr.sapcloud.cn/sf/timeoff``);
#: HR 改成 applink 是因为主渠道是飞书卡片,applink 直接在飞书客户端内打开。
#: 注意 HR 最初给的 SF 链接带 ``?_s.crb=…`` **会话令牌** —— 那是发链接的人自己
#: 那次登录的令牌,原样发给 500 个人只会各自失效;下次换入口时同样要先看清
#: query 里带的是不是某个人的会话态。
#:
#: ⚠️ ``client/message/link/open?token=…`` 打开的是**一条飞书消息**,不是一个
#: 应用页 —— 能不能打开取决于收件人对那条消息所在会话有没有可见性。这是 HR
#: 的选择,Helix 侧无法校验;仍然是配置项,换入口不需要改代码。
DEFAULT_HRIS_URL = (
    "https://applink.feishu.cn/client/message/link/open"
    "?token=AmOb0W9JQUABaoP4VzlBC9k%3D"
)

#: HR 可以在文案里使用的占位符。发送路径逐人填充(见 :func:`build_context`)。
#: 白名单化不是形式主义:``_safe_format`` 容忍缺失的 key 却会把它**原样**留在
#: 正文里,于是一个手滑的 ``{name}`` 会带着大括号发给 500 个人。
ALLOWED_PLACEHOLDERS: tuple[str, ...] = (
    "person_name",
    "account_lines",
    "total_balance_days",
    # 「最早那个清零日会带走多少天」。没有它,HR 想写一句区分「合计」和「即将
    # 清零的那部分」的真话都写不出来(白名单会拒掉任何未知占位符)。
    "expiring_balance_days",
    "deadline_date",
    "deadline_month",
    "expires_in_days",
    # 余额的口径日期 —— 通知里的数字是发送时的快照,不写出来员工就会拿它和
    # 实时卡片对比然后来问 HR。
    "as_of_date",
    "hris_url",
)
#: 发送路径自动注入的两个,HR 可以用但不需要自己提供。
AUTO_PLACEHOLDERS: tuple[str, ...] = ("cta_line", "portal_url")

#: HR 可编辑的文案字段(= ``TemplateCopy`` 的字段,也是 email_template 的存储
#: 契约的子集)。不在这里的字段(``body_html`` 等)不接受 HR 写入。
EDITABLE_COPY_FIELDS: tuple[str, ...] = tuple(TemplateCopy.model_fields.keys())

#: 卡片里最多放几个 stat 格子(Feishu 是两列网格,>6 就该换成正文明细了)。
_MAX_CARD_STATS = 6

#: 一次群发的收件人硬上限。组织约 500 人、有假期账户的约 280 人,这个数字是
#: 防手滑/防脚本失控的天花板,不是业务限制。
MAX_RECIPIENTS_PER_BATCH = 1000

#: 试发的收件人上限。试发**不计入去重**,所以它必须小到不可能被当成群发用
#: —— 否则「勾 281 人 + test_mode」就是一条绕开「近期已提醒」护栏的通道。
MAX_TEST_RECIPIENTS = 20

#: 心跳停多久算「进程真的没了」。``run_batch`` 每 ``_HEARTBEAT_EVERY`` 行刷一次
#: ``updated_at``,所以这个阈值只需要覆盖**单行**的最坏投递耗时 —— 而通知域自己
#: 对那个上限的判断是 ``service.DELIVERY_LEASE`` = 15 分钟(注释写明「leaves room
#: for configured transport timeouts」)。取两倍留余量:窗口比单行最坏耗时还短,
#: reaper 就会把一个**活着**的 drain 判死,in-flight 闸随之放行第二批,而第一批的
#: 尾部还没投出去 ⇒ 同一批人两张卡。
_STALE_HEARTBEAT = timedelta(minutes=30)

#: 每投多少行刷一次批次心跳。1 会给每封信多一次 UPDATE+commit;5 在「心跳足够密」
#: 与「写放大可接受」之间。
_HEARTBEAT_EVERY = 5

_MONTH_NAMES_EN = (
    "January", "February", "March", "April", "May", "June",
    "July", "August", "September", "October", "November", "December",
)


def today_sh() -> date:
    """口径锚点:公司运营时区的今天。

    与员工假期卡片(``hr/router._leave_view_context``)和洞察板同一个锚点 ——
    三个面必须对同一个账户说同一个「还剩几天」。
    """
    return datetime.now(UTC).astimezone(ZoneInfo("Asia/Shanghai")).date()


def _f(value: Decimal | float | None) -> float:
    return 0.0 if value is None else float(value)


def fmt_days(value: float | None) -> str:
    """``5.0`` → ``5``;``0.69`` → ``0.69``。整数不带小数点。"""
    num = round(_f(value), 2)
    if num == int(num):
        return str(int(num))
    return f"{num:.2f}".rstrip("0").rstrip(".")


def _month_label(d: date, lang: str) -> str:
    if lang == "en":
        return _MONTH_NAMES_EN[d.month - 1]
    if lang == "ja":
        return f"{d.month}月"
    return f"{d.month} 月"


def fiscal_label(acct: AccountSnapshot) -> str:
    """HR 口中的财年标签(``FY26``)—— 从**周期结束年**推,不是从 ``fiscal_year``。

    两个坑叠在一起,所以这个函数值得单独存在:

    ① **HR 的 FY 编号和 ``fiscal_year.py`` 的差一年。** 平台的财务口径是
    「FY2026 = 2026-04 … 2027-03」(按**开始**年命名);HR 把 2025-04-01 发放、
    2026-03-31 结束的那个周期叫 **FY26**(按**结束**年命名)。两边都对,只是不是
    同一套编号 —— 所以这里**不要**去复用 ``current_fiscal_year()``。

    ② **``fiscal_year`` 列本身就不可靠。** 生产实测(2026-08-18):清零日同为
    2026-09-30 的桶里,261 个是 ``fiscal_year=2025``(2025-04-01 ~ 2026-03-31),
    另有 **7 个是 ``fiscal_year=2026``**(2026-01-12 ~ 2026-03-31,1~3 月入职的
    同事)。周期和清零日完全一样,``fiscal_year`` 却差 1 —— 按 ``fiscal_year + 1``
    命名会把那 7 个人的同一份额度标成「FY27」。

    所以判据是 ``end_date`` 落在哪一年:两组都得到 FY26。``end_date`` 缺失时退回
    ``fiscal_year + 1``(至少对 261 个人是对的),两个都没有就不加后缀。
    """
    year: int | None = None
    if acct.period_end is not None:
        year = acct.period_end.year
    elif acct.fiscal_year is not None:
        year = acct.fiscal_year + 1
    if year is None:
        return ""
    return f"FY{year % 100:02d}"


# ── 圈人 ────────────────────────────────────────────────────────────────


@dataclass(slots=True)
class CohortResult:
    candidates: list[CandidateRow]
    summary: CohortSummary
    #: person_id → 该人的 IdentityPerson(发送时要用来解析语言、邮箱)
    persons: dict[str, IdentityPerson] = field(default_factory=dict)


def _deadline_of(account: HrLeaveAccount) -> date | None:
    """清零日 = ``booking_end_date`` 优先,退到 ``end_date``。

    与 ``hr/router._leave_forfeit_signals`` 完全一致 —— 员工卡片上写的那个
    日期和提醒里写的那个日期必须是同一个,否则员工会拿着两个日期来问 HR。
    """
    return account.booking_end_date or account.end_date


async def resolve_cohort(
    db: AsyncSession,
    filters: ReminderFilters,
    *,
    resend_window_days: int | None = None,
    as_of: date | None = None,
) -> CohortResult:
    """按口径解析出候选名单 + 汇总。

    只查在职、有工作邮箱的人:没有邮箱两个渠道都投不出去(飞书也是按邮箱找人
    的),把他们留在名单里只会制造一批必然失败的 outbox 行。他们的人数会如实
    出现在 ``summary.skipped_no_email`` 里,而不是被静默吞掉。
    """
    from app.domains.hr.leave_insights_service import (
        forfeits_at_period_end,
        load_type_overrides,
    )

    today = as_of or today_sh()
    overrides = await load_type_overrides(db)
    if resend_window_days is None:
        resend_window_days = await get_resend_window_days(db)

    # 1. 在职人员 + 主部门(与洞察板同一个 join)。
    person_rows = (await db.execute(
        select(
            IdentityPerson,
            IdentityOrgMembership.org_node_name,
            IdentityOrgMembership.org_path,
        )
        .outerjoin(
            IdentityOrgMembership,
            (IdentityOrgMembership.person_id == IdentityPerson.id)
            & IdentityOrgMembership.is_primary.is_(True)
            & IdentityOrgMembership.deleted_at.is_(None),
        )
        .where(
            IdentityPerson.employment_status == "active",
            IdentityPerson.deleted_at.is_(None),
        )
    )).all()
    persons: dict[str, IdentityPerson] = {}
    dept_of: dict[str, tuple[str | None, str | None]] = {}
    for person, dept, org_path in person_rows:
        pid = str(person.id)
        persons[pid] = person
        dept_of[pid] = (dept, org_path)
    active_person_count = len(persons)

    # 2. 命中口径的假期账户。人少(约 500)、桶少(每人个位数),所以按人筛完
    #    再在 Python 里算日期窗口 —— 让「清零日 = coalesce(booking_end, end)」
    #    这条口径只有一个实现,而不是 SQL 里再写一遍。
    acct_q = select(HrLeaveAccount).where(
        HrLeaveAccount.deleted_at.is_(None),
        HrLeaveAccount.account_type.in_(filters.account_types),
    )
    if not filters.include_closed_accounts:
        acct_q = acct_q.where(HrLeaveAccount.account_closed.is_(False))
    if filters.fiscal_years:
        acct_q = acct_q.where(
            HrLeaveAccount.fiscal_year.in_(filters.fiscal_years),
        )
    accounts = (await db.execute(acct_q)).scalars().all()

    # 覆盖面口径:有**任何**假期账户的在职人数(不受本次口径影响),用来诚实
    # 地告诉 HR「这个功能能覆盖到多少人」。
    persons_with_leave_data = int((await db.execute(
        select(func.count(func.distinct(HrLeaveAccount.person_id))).where(
            HrLeaveAccount.deleted_at.is_(None),
            HrLeaveAccount.person_id.in_(_active_person_ids_subquery()),
        )
    )).scalar() or 0)

    by_person: dict[str, list[AccountSnapshot]] = {}
    synced_of: dict[str, list[datetime]] = {}
    by_account_type: dict[str, int] = {}
    synced_times: list[datetime] = []
    # 「因为没邮箱投不出去」必须在**其它口径都过了**之后才统计(见下面的候选
    # 循环)—— 否则一个余额 0.1 天、本来也不入选的无邮箱同事也会被算进这个给
    # HR 看的数字里。
    no_email_seen: set[str] = set()

    for a in accounts:
        pid = str(a.person_id)
        person = persons.get(pid)
        if person is None:
            continue  # 离职 / 软删 —— 永远不提醒
        balance = _f(a.balance_days)
        deadline = _deadline_of(a)
        expires_in = None if deadline is None else (deadline - today).days
        if expires_in is not None and expires_in < 0 and not filters.include_expired:
            continue
        if filters.expiring_within_days is not None:
            # 没有清零日的桶无法声称「即将到期」—— 有窗口条件时排除。
            if expires_in is None or expires_in > filters.expiring_within_days:
                continue
        by_person.setdefault(pid, []).append(AccountSnapshot(
            account_id=a.id,
            account_type=a.account_type,
            account_type_label=a.account_type_label,
            fiscal_year=a.fiscal_year,
            balance_days=round(balance, 2),
            entitled_days=round(_f(a.entitled_days), 2),
            used_days=round(_f(a.used_days), 2),
            deadline=deadline,
            period_end=a.end_date,
            expires_in_days=expires_in,
            forfeits=forfeits_at_period_end(a.account_type, overrides),
            is_current=bool(a.is_current),
            account_closed=bool(a.account_closed),
        ))
        if a.synced_at is not None:
            synced_of.setdefault(pid, []).append(a.synced_at)

    # 3. 去重信号:每人最近一次收到过**非测试**提醒 + HR 的长期排除决定。
    last_reminded = await _last_reminded_map(db, list(by_person.keys()))
    excluded_notes = await load_exclusions(db)
    window_start = (
        datetime.now(UTC) - timedelta(days=resend_window_days)
        if resend_window_days > 0 else None
    )

    candidates: list[CandidateRow] = []
    for pid, snaps in by_person.items():
        # ``min_balance_days`` 是**人**的门槛,不是账户的门槛。按账户筛会把
        # 「法定 15 天 + 福利 0 天」的人只留下法定那一格,而 HR 要的是两个 9/30
        # 的桶都出现在卡片上(0 也照显示,见 ``build_card_stats``)—— 同时
        # 「两个桶都是 0」的人仍然一个都不发,因为合计 0 过不了门槛。
        if round(sum(s.balance_days for s in snaps), 2) < filters.min_balance_days:
            continue
        person = persons[pid]
        # 没有工作邮箱两个渠道都投不出去。判在门槛**之后**:一个本来就不入选的
        # 人不该出现在给 HR 看的 ``skipped_no_email`` 里。
        if not (person.work_email or "").strip():
            no_email_seen.add(pid)
            continue
        dept, org_path = dept_of.get(pid, (None, None))
        snaps.sort(key=lambda s: (
            s.expires_in_days if s.expires_in_days is not None else 10**6,
            -s.balance_days,
        ))
        for s in snaps:
            by_account_type[s.account_type] = (
                by_account_type.get(s.account_type, 0) + 1
            )
        # 日期信号只从**还有天数**的桶里取。一个 0 天的桶是展示项(卡片要显示
        # 「福利年假 2025 · 0 天」),但它没有任何东西会在那天失效 —— 让它决定
        # 「即将到期」的日期,标题就会指向一个什么都不会损失的日子。
        # ``build_context`` 用的是同一条判据,否则名单里的日期和信里的日期会打架。
        deadlines = [
            s.deadline for s in snaps
            if s.deadline is not None and s.balance_days > 0
        ]
        earliest = min(deadlines) if deadlines else None
        acct_synced = synced_of.get(pid, [])
        synced_times.extend(acct_synced)
        reminded_at, reminded_batch = last_reminded.get(pid, (None, None))
        candidates.append(CandidateRow(
            person_id=person.id,
            display_name=person.display_name or person.work_email or pid,
            employee_no=person.employee_no,
            work_email=person.work_email,
            department=dept,
            bu=bu_from_org_path(org_path, fallback_name=dept),
            language=resolve_lang(person),
            accounts=snaps,
            total_balance_days=round(sum(s.balance_days for s in snaps), 2),
            earliest_deadline=earliest,
            expires_in_days=(
                None if earliest is None else (earliest - today).days
            ),
            data_synced_at=min(acct_synced) if acct_synced else None,
            # HR 的长期排除:照常出现在名单里、带标记、默认不勾选。
            excluded=pid in excluded_notes,
            excluded_note=excluded_notes.get(pid),
            last_reminded_at=reminded_at,
            last_reminded_batch_id=reminded_batch,
            reminded_recently=bool(
                reminded_at is not None
                and window_start is not None
                and _as_utc(reminded_at) >= window_start
            ),
        ))

    candidates.sort(key=lambda c: (
        c.expires_in_days if c.expires_in_days is not None else 10**6,
        -c.total_balance_days,
        c.display_name,
    ))

    reminded_recently_count = sum(1 for c in candidates if c.reminded_recently)
    # 「默认会发给几个人」必须和前端的默认勾选**同一口径**,否则 HR 看到的
    # 数字和他按下发送时的人数对不上。两条默认不勾的理由:近期已提醒、HR 排除。
    default_selected = sum(
        1 for c in candidates if not c.reminded_recently and not c.excluded
    )
    summary = CohortSummary(
        recipient_count=len(candidates),
        account_count=sum(len(c.accounts) for c in candidates),
        total_at_risk_days=round(
            sum(c.total_balance_days for c in candidates), 2,
        ),
        reminded_recently_count=reminded_recently_count,
        default_selected_count=default_selected,
        excluded_count=sum(1 for c in candidates if c.excluded),
        by_account_type=by_account_type,
        active_person_count=active_person_count,
        persons_with_leave_data=persons_with_leave_data,
        skipped_no_email=len(no_email_seen),
        oldest_synced_at=min(synced_times) if synced_times else None,
        newest_synced_at=max(synced_times) if synced_times else None,
        as_of_date=today,
    )
    return CohortResult(candidates=candidates, summary=summary, persons=persons)


def _as_utc(value: datetime) -> datetime:
    """MySQL/TiDB 取回的 ``DateTime(timezone=True)`` 常常是 naive 的。

    和一个 aware 的窗口下界比较会当场 ``TypeError``
    (``reference_naive_aware_datetime_mix_in_composed_queries``),所以统一
    按 UTC 补上 tzinfo。
    """
    return value if value.tzinfo is not None else value.replace(tzinfo=UTC)


async def _last_reminded_map(
    db: AsyncSession, person_ids: list[str],
) -> dict[str, tuple[datetime | None, uuid.UUID | None]]:
    """person_id → (最近一次非测试提醒的时间, 批次 id)。

    只认挂在**非测试**批次上的 outbox 行:HR 发给自己试水的那一条不该让任何
    真实收件人被判成「已提醒」。
    """
    from app.notifications.models import (
        STATUS_DELIVERED as STATUS_DELIVERED_CONST,
    )
    from app.notifications.models import (
        STATUS_PARTIAL as STATUS_PARTIAL_CONST,
    )
    from app.notifications.models import NotificationOutbox

    if not person_ids:
        return {}
    # 批次行数量级是「每年几次」,所以先把非测试批次的 id 取成字符串列表再
    # IN,而不是写成子查询:``related_resource_id`` 是 CHAR(36),和 GUID 列做
    # 子查询比较要依赖方言的隐式转换,显式取值反而更稳。
    real_batch_ids = [
        str(bid) for bid in (await db.execute(
            select(HrLeaveReminderBatch.id).where(
                HrLeaveReminderBatch.test_mode.is_(False),
            )
        )).scalars().all()
    ]
    if not real_batch_ids:
        return {}
    rows = (await db.execute(
        select(
            NotificationOutbox.recipient_person_id,
            NotificationOutbox.created_at,
            NotificationOutbox.related_resource_id,
        ).where(
            NotificationOutbox.template_key == TEMPLATE_KEY,
            NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE,
            NotificationOutbox.recipient_person_id.in_(person_ids),
            NotificationOutbox.related_resource_id.in_(real_batch_ids),
            # 「已提醒」= outbox 判定为 delivered/partial,不是「排过队」。一次因
            # 集成故障全军覆没的批次不该让整个 cohort 在去重窗口内被默认排除 ——
            # 明明没人收到。
            #
            # 边界(如实记下来):通知域把「员工把这个类别的两个推送渠道都静音了」
            # 也收敛成 DELIVERED(``service._deliver`` 的 inbox-only 语义),所以
            # 那种人会被算成「已提醒」。这是平台既定语义 —— 站内信箱确实有那一行,
            # 而且改它会让去重口径和重试口径打架(PARTIAL 行被刻意摘出重试池)。
            NotificationOutbox.status.in_(
                [STATUS_DELIVERED_CONST, STATUS_PARTIAL_CONST],
            ),
        )
    )).all()
    out: dict[str, tuple[datetime | None, uuid.UUID | None]] = {}
    for pid, created_at, batch_id in rows:
        key = str(pid)
        prev = out.get(key)
        if (
            prev is None
            or prev[0] is None
            or (created_at is not None and _as_utc(created_at) > _as_utc(prev[0]))
        ):
            out[key] = (
                created_at,
                uuid.UUID(str(batch_id)) if batch_id else None,
            )
    return out


# ── 设置(HR 自己维护的那几项) ─────────────────────────────────────────


async def get_resend_window_days(db: AsyncSession) -> int:
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(
        SETTING_RESEND_WINDOW, default=None, session=db,
    )
    try:
        value = int(raw)  # type: ignore[arg-type]
    except (TypeError, ValueError):
        return DEFAULT_RESEND_WINDOW_DAYS
    return max(0, min(365, value))


async def get_hris_url(db: AsyncSession) -> str:
    """卡片按钮的落点。HR 可以在页面上改,没设置过就用 :data:`DEFAULT_HRIS_URL`。

    出厂值不为空是刻意的:按钮只剩这一个(见 :func:`build_card_actions`),
    留空就意味着一张提醒卡把员工送回 Helix 的只读余额页,而 HR 要的是他直接
    落在能提申请的地方。仍然是配置项 —— 换 HRIS / 换深链不需要改代码。
    """
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(SETTING_HRIS_URL, default="", session=db)
    return str(raw or "").strip() or DEFAULT_HRIS_URL


def default_filters() -> ReminderFilters:
    """出厂口径:两种会清零的年假,合计剩余 ≥ 0.5 天,120 天内清零。

    含福利年假(``CHN_PL``):HR 确认它和法定年假一样是一年半有效期、同样在
    9/30 清零(2026-08-18),所以它属于「不提醒就真的会损失」的那一类。
    """
    return ReminderFilters(
        account_types=["CHN_AnnualLeave", "CHN_PL", "MY_AnnuL"],
        fiscal_years=[],
        min_balance_days=0.5,
        expiring_within_days=120,
        include_expired=False,
        include_closed_accounts=False,
    )


async def get_default_filters(db: AsyncSession) -> ReminderFilters:
    """HR 上次用的口径(存下来,下次进页面不用重新勾一遍)。"""
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(SETTING_DEFAULTS, default=None, session=db)
    if isinstance(raw, dict):
        try:
            return ReminderFilters.model_validate(raw)
        except Exception:  # noqa: BLE001 — 坏掉的存量设置不该让页面打不开
            logger.warning(
                "leave_expiry_reminder: stored defaults are invalid, "
                "falling back to the factory filters",
            )
    return default_filters()


async def get_default_channels(db: AsyncSession) -> list[str]:
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(
        SETTING_DEFAULT_CHANNELS, default=None, session=db,
    )
    if isinstance(raw, list):
        picked = [c for c in dict.fromkeys(str(x) for x in raw) if c in ALLOWED_CHANNELS]
        if picked:
            return picked
    return list(ALLOWED_CHANNELS)


# ── 排除名单(HR 的长期决定) ────────────────────────────────────────────


async def load_exclusions(db: AsyncSession) -> dict[str, str | None]:
    """``person_id(str) → note``。圈人时用它给候选行打「默认不勾」标记。"""
    rows = (await db.execute(select(HrLeaveReminderExclusion))).scalars().all()
    return {str(r.person_id): r.note for r in rows}


async def list_exclusions(db: AsyncSession) -> list[tuple[HrLeaveReminderExclusion, IdentityPerson | None]]:
    """排除名单 + 每行对应的人,按加入时间倒序(最近改的在最上面)。"""
    rows = (await db.execute(
        select(HrLeaveReminderExclusion)
        .order_by(HrLeaveReminderExclusion.created_at.desc())
    )).scalars().all()
    if not rows:
        return []
    persons = {
        str(p.id): p
        for p in (await db.execute(
            select(IdentityPerson).where(
                IdentityPerson.id.in_([str(r.person_id) for r in rows]),
            )
        )).scalars().all()
    }
    return [(r, persons.get(str(r.person_id))) for r in rows]


async def set_exclusion(
    db: AsyncSession,
    *,
    person_id: uuid.UUID,
    note: str | None,
    actor_person_id: uuid.UUID,
) -> HrLeaveReminderExclusion:
    """加入(或就地改备注)一条排除决定。

    人必须存在 —— 一个打错的 id 会安静地躺在表里、永远匹配不到任何人,而 HR 以为
    自己已经把某人排除掉了。
    """
    person = await db.get(IdentityPerson, str(person_id))
    if person is None or person.deleted_at is not None:
        raise NotFoundError("Person", str(person_id))
    clean = (note or "").strip() or None
    if clean is not None and len(clean) > 500:
        raise ValidationError("note is too long (max 500 characters).")
    row = (await db.execute(
        select(HrLeaveReminderExclusion).where(
            HrLeaveReminderExclusion.person_id == str(person_id),
        )
    )).scalars().first()
    if row is None:
        row = HrLeaveReminderExclusion(
            person_id=str(person_id),
            note=clean,
            created_by=str(actor_person_id),
        )
        db.add(row)
        await db.flush()
    else:
        row.note = clean
    return row


async def remove_exclusion(db: AsyncSession, person_id: uuid.UUID) -> bool:
    """删掉一条排除决定。返回「本来是否存在」,让调用方能如实回报。"""
    row = (await db.execute(
        select(HrLeaveReminderExclusion).where(
            HrLeaveReminderExclusion.person_id == str(person_id),
        )
    )).scalars().first()
    if row is None:
        return False
    await db.delete(row)
    return True


# ── 结尾寄语(HR 维护的一组话) ─────────────────────────────────────────

#: 出厂寄语。替换掉平台通用的「— Helix IT 平台」——这封信是催人**去休假**的,
#: 落款是一行公司签名就把它读成了行政通知。HR 可以整组改。
DEFAULT_SIGNOFF_LINES: tuple[dict[str, str], ...] = (
    {
        "zh": "工作再忙,也别忘了给自己放个假 🌴",
        "en": "However busy work gets, remember to take your time off 🌴",
        "ja": "どんなに忙しくても、休暇を取ることを忘れずに 🌴",
    },
    {
        "zh": "休息,是为了走更长的路 ☕",
        "en": "Rest is part of the work, not a break from it ☕",
        "ja": "休むことも仕事のうちです ☕",
    },
    {
        "zh": "平衡工作与生活,才能走得更远 🌿",
        "en": "A balanced life is a sustainable one 🌿",
        "ja": "仕事と生活のバランスが、長く働く力になります 🌿",
    },
    {
        "zh": "把假期用完,是对自己最好的投资 🌊",
        "en": "Using your leave is an investment in yourself 🌊",
        "ja": "休暇を使いきることは、自分への投資です 🌊",
    },
)

SIGNOFF_MODE_RANDOM = "random"
SIGNOFF_MODE_FIXED = "fixed"
_MAX_SIGNOFF_LINES = 20
_SIGNOFF_LINE_MAX = 200


def default_signoff() -> dict[str, Any]:
    return {
        "mode": SIGNOFF_MODE_RANDOM,
        "fixed_index": 0,
        "lines": [dict(line) for line in DEFAULT_SIGNOFF_LINES],
    }


def validate_signoff(raw: Any) -> dict[str, Any]:
    """校验 HR 提交的寄语配置,返回归一化后的形状。

    和文案校验同一个理由:这段东西会落进 ``platform_settings`` 再复制进每个批次
    的快照,然后原样发给几百个人。三语任一为空就退回该行的其它语言(而不是发出
    一个空签名),整行全空则拒收。
    """
    if not isinstance(raw, dict):
        raise ValidationError("signoff must be an object.")
    mode = str(raw.get("mode") or SIGNOFF_MODE_RANDOM).strip()
    if mode not in (SIGNOFF_MODE_RANDOM, SIGNOFF_MODE_FIXED):
        raise ValidationError(
            f"signoff.mode must be '{SIGNOFF_MODE_RANDOM}' or "
            f"'{SIGNOFF_MODE_FIXED}'.",
        )
    raw_lines = raw.get("lines")
    if not isinstance(raw_lines, list) or not raw_lines:
        raise ValidationError("signoff.lines must be a non-empty list.")
    if len(raw_lines) > _MAX_SIGNOFF_LINES:
        raise ValidationError(
            f"signoff.lines holds at most {_MAX_SIGNOFF_LINES} lines.",
        )
    lines: list[dict[str, str]] = []
    for i, item in enumerate(raw_lines):
        if not isinstance(item, dict):
            raise ValidationError(f"signoff.lines[{i}] must be an object.")
        vals = {
            lang: str(item.get(lang) or "").strip()
            for lang in ("zh", "en", "ja")
        }
        for lang, v in vals.items():
            if len(v) > _SIGNOFF_LINE_MAX:
                raise ValidationError(
                    f"signoff.lines[{i}].{lang} is too long "
                    f"(max {_SIGNOFF_LINE_MAX} characters).",
                )
        if not any(vals.values()):
            raise ValidationError(
                f"signoff.lines[{i}] is empty in all three languages.",
            )
        lines.append(vals)
    idx = raw.get("fixed_index", 0)
    try:
        fixed_index = int(idx)
    except (TypeError, ValueError):
        raise ValidationError("signoff.fixed_index must be an integer.") from None
    if not (0 <= fixed_index < len(lines)):
        raise ValidationError(
            f"signoff.fixed_index must be between 0 and {len(lines) - 1}.",
        )
    return {"mode": mode, "fixed_index": fixed_index, "lines": lines}


async def get_signoff(db: AsyncSession) -> dict[str, Any]:
    """HR 存的寄语配置;没存过 / 存坏了就退回出厂那组。"""
    from app.integrations.service import get_platform_setting

    raw = await get_platform_setting(SETTING_SIGNOFF, default=None, session=db)
    if raw is None:
        return default_signoff()
    try:
        return validate_signoff(raw)
    except ValidationError:
        logger.warning(
            "leave_expiry_reminder: stored signoff is invalid, "
            "falling back to the factory lines",
        )
        return default_signoff()


def pick_signoff(config: dict[str, Any], *, seed: str) -> dict[str, str]:
    """选出这一批要用的那一行。

    **每个批次一行,不是每个人一行**:预览即发送是这个功能的核心语义 —— HR 在
    「文案与预览」里看到的那句话必须就是发出去的那句话。逐人随机会让预览变成一句
    只对第一个收件人成立的话。

    随机也是**可复现**的:种子取批次 id,所以同一批次重渲染(站内通知回看、
    resend-failed 克隆行)拿到的还是同一句,而不是每次读都换一句。
    """
    lines = config.get("lines") or []
    if not lines:
        lines = [dict(line) for line in DEFAULT_SIGNOFF_LINES]
    if config.get("mode") == SIGNOFF_MODE_FIXED:
        idx = int(config.get("fixed_index") or 0)
        return dict(lines[idx % len(lines)])
    digest = hashlib.sha256(seed.encode("utf-8")).digest()
    return dict(lines[int.from_bytes(digest[:8], "big") % len(lines)])


def signoff_text(line: dict[str, str], lang: str) -> str:
    """按收件人语言取一行,缺失时沿 ja→en→zh 阶梯退 —— 和文案渲染同一个阶梯。"""
    lang = normalize_lang(lang) or DEFAULT_LANG
    for key in ({"ja": ("ja", "en", "zh"), "en": ("en", "zh", "ja")}
                .get(lang, ("zh", "en", "ja"))):
        v = str(line.get(key) or "").strip()
        if v:
            return v
    return ""


# ── 文案 ────────────────────────────────────────────────────────────────


def builtin_copy() -> TemplateCopy:
    from app.notifications.templates import BUILTIN_TEMPLATES

    tmpl = BUILTIN_TEMPLATES.get(TEMPLATE_KEY, {})
    return TemplateCopy(**{
        f: str(tmpl.get(f) or "") for f in EDITABLE_COPY_FIELDS
    })


async def effective_copy(db: AsyncSession) -> tuple[TemplateCopy, bool]:
    """(生效文案, 是否被 HR 覆盖过)。

    与运行时 ``_load_template`` 同一套合并语义:覆盖是**字段级**的,HR 只改了
    中文正文不会把英日文案一起清空。
    """
    from app.integrations.service import get_platform_setting

    base = builtin_copy().model_dump()
    override = await get_platform_setting(
        TEMPLATE_SETTING_KEY, default=None, session=db,
    )
    overridden = False
    if isinstance(override, dict):
        for f in EDITABLE_COPY_FIELDS:
            value = override.get(f)
            if isinstance(value, str) and value.strip():
                base[f] = value
                overridden = True
    return TemplateCopy(**base), overridden


def parse_fields(text: str) -> list[tuple[str, str | None, str | None]]:
    """枚举文案里**每一个**替换字段 → ``(field_name, spec, conversion)``。

    用 ``string.Formatter().parse()`` 而不是正则,因为正则只认「纯标识符 + 紧跟
    右括号」这一种形状,于是 ``{a.b}`` / ``{a[0]}`` / ``{ person_name }`` 会**完全
    绕过**白名单 —— 而 ``_safe_format`` 遇到 ``{a.b}`` 会 return 原文,也就是这个
    字段里每一个占位符都以字面量形态发给全部收件人(``{person_name.title}`` 更
    难看:一个内存地址进员工邮件)。正则同时看不懂 ``{{`` 转义,把
    ``literal {{braces}}`` 误判成未知占位符 ``braces`` 而拒绝写入。

    ``ValueError`` = 括号本身不配对,由调用方翻成校验错误。
    """
    from string import Formatter

    return [
        (name, spec, conv)
        for _literal, name, spec, conv in Formatter().parse(text or "")
        if name is not None
    ]


def extract_placeholders(text: str) -> list[str]:
    """文案引用到的占位符名(只含合法标识符形态的那些)。

    渲染后的「还有没有没替换掉的」检测也用它 —— 那时文本里只该剩字面量。
    """
    try:
        fields = parse_fields(text)
    except ValueError:
        return []
    return [name for name, _spec, _conv in fields if name.isidentifier()]


def validate_copy(copy: TemplateCopy) -> None:
    """挡住三类会伤到 500 个收件人的文案错误。

    1. 未知占位符 —— ``_safe_format`` 会把它**原样**留在正文里,员工看到的是
       带大括号的 ``{name}``。
    2. 位置参数 / 格式说明符(``{0}`` / ``{days:d}``)—— 这两个 ``_safe_format``
       是会**抛异常**的,一个字符就能让整批投递失败。
    3. 三语任一为空的必填字段 —— 空标题的通知等于没通知。
    """
    allowed = set(ALLOWED_PLACEHOLDERS) | set(AUTO_PLACEHOLDERS)
    required = ("subject", "subject_en", "subject_ja",
                "body_text", "body_text_en", "body_text_ja")
    data = copy.model_dump()
    for f in required:
        if not str(data.get(f) or "").strip():
            raise ValidationError(
                f"'{f}' must not be empty — every recipient language needs "
                "a subject and a body.",
            )
    for f in EDITABLE_COPY_FIELDS:
        text = str(data.get(f) or "")
        if not text:
            continue
        try:
            fields = parse_fields(text)
        except ValueError as exc:
            raise ValidationError(
                f"'{f}' has unbalanced braces ({exc}). Use {{{{ }}}} to write a "
                "literal brace.",
            ) from exc
        for name, spec, conv in fields:
            if spec or conv:
                raise ValidationError(
                    f"'{f}' uses a format spec or conversion "
                    f"(\u007b{name}{'!' + conv if conv else ''}"
                    f"{':' + spec if spec else ''}\u007d). Those raise at send "
                    "time and would break delivery for EVERY recipient. Use a "
                    "plain {placeholder} name.",
                )
            if not name.isidentifier():
                raise ValidationError(
                    f"'{f}' uses a positional / attribute / index field "
                    f"(\u007b{name}\u007d). Those either raise at send time or "
                    "ship the whole paragraph with literal braces. Use a plain "
                    "{placeholder} name.",
                )
            if name not in allowed:
                raise ValidationError(
                    f"'{f}' references unknown placeholder \u007b{name}\u007d. "
                    "Allowed: "
                    f"{', '.join('{' + a + '}' for a in sorted(allowed))}.",
                )


def validate_hris_url(raw: str) -> str:
    """空串 = 清掉入口;非空必须是 http(s) 绝对地址。

    卡片按钮的 URL 是要发给 500 个人点的,一个 ``javascript:`` 或者拼错协议的
    值在这里挡掉,比在员工的飞书里发现要便宜得多。
    """
    url = (raw or "").strip()
    if not url:
        return ""
    if len(url) > 500:
        raise ValidationError("hris_url is too long (max 500 characters).")
    if any(ch.isspace() or ord(ch) < 0x20 for ch in url):
        raise ValidationError("hris_url must not contain whitespace or control characters.")
    # ``startswith`` 大小写敏感,会拒掉合法的 ``HTTPS://…``;而它又会放过裸
    # ``https://``(``"https://".startswith("https://")`` 为真),几百张卡片上
    # 就会挂一个 href="https://" 的死按钮。urlparse 两个都治。
    parsed = urlparse(url)
    scheme = parsed.scheme.lower()
    if scheme not in ("http", "https") or not parsed.netloc:
        raise ValidationError(
            "hris_url must be an absolute http(s) URL with a host (or empty to "
            "remove the button).",
        )
    if "@" in parsed.netloc:
        # ``https://helix.pingcap.net@evil.com/`` 的真实主机是 evil.com ——
        # 经典视觉欺骗,而这个 URL 会挂在几百张卡片的按钮上。
        raise ValidationError(
            "hris_url must not contain a userinfo (…@…) part — the real host "
            "is whatever follows the '@', which makes the link deceptive.",
        )
    # 归一化协议大小写。消费侧 ``NotificationService._resolve_absolute_url`` 与
    # 前端的 ``isAbsoluteHttp`` 都用**大小写敏感**的 startswith 判绝对地址,
    # 所以 ``HTTPS://…`` 会被当成相对路径拼成 ``https://helix…/HTTPS://…``,
    # 几百张卡片挂一个死按钮 —— 正是这段校验想避免的失败形状,只是换了个入口。
    # 注意判据要用**原串**:``urlparse`` 自己就把 ``parsed.scheme`` 小写化了,
    # 拿它和小写比永远相等,归一化就成了死代码。
    head, sep, rest = url.partition("://")
    if sep and head != scheme:
        url = f"{scheme}://{rest}"
    return url


# ── 渲染 ────────────────────────────────────────────────────────────────


def build_context(
    *,
    display_name: str,
    accounts: list[AccountSnapshot],
    lang: str,
    hris_url: str,
    today: date,
    as_of: date | None = None,
) -> dict[str, Any]:
    """逐人、**按其语言**渲染的模板 context。

    一个 envelope 只对应一个收件人,所以 context 可以是本地化后的成品 ——
    ``account_lines`` 这种多行明细没法用占位符在模板里拼出来,而在这里拼就
    天然是收件人自己的语言。站内通知之后重渲染读的也是这份 context,所以
    员工回头看那条通知,语言和数字都还是当时那份。
    """
    lang = normalize_lang(lang) or DEFAULT_LANG
    lines: list[str] = []
    for acct in accounts:
        label = localized_type_label(acct.account_type, acct.account_type_label, lang)
        fy = f" FY{acct.fiscal_year}" if acct.fiscal_year else ""
        deadline = acct.deadline.isoformat() if acct.deadline else "—"
        # **只有会清零的桶才说「清零」。** ``forfeits`` 一路传到这里却从不被读,
        # 于是 HR 勾一个不清零的类型之后,几百个员工会收到「XX 2027-09-30 清零」
        # ——一句机器生成的假话。(福利年假曾经是这个例子,2026-08-18 HR 确认它
        # 确实会清零,判据随之改了 —— 但读 ``forfeits`` 这件事和口径无关。)
        # 已关闭的桶同理:员工自己的假期卡片把它画成灰色历史、连倒计时都不显示
        # (``hr/router._leave_forfeit_signals`` 对 closed 无条件降级),提醒却在催
        # 他「赶紧休掉」。这两类都改成陈述可申请的最后一天,那是真话。
        expiring = acct.forfeits and not acct.account_closed
        # ``entitled - used`` 与 ``balance`` 对不上时**不写**这两个派生数:
        # ``balance_days`` 是 SF 的直接真值(Σ bookingAmount),另两个是 Helix 从
        # posting 派生的,一条非消耗类型的负向 posting(期末冲销、结转出、手工负调)
        # 就能让三个数互相矛盾 —— 与其把矛盾发给员工,不如少说两个数。
        breakdown_ok = (
            acct.entitled_days is not None and acct.used_days is not None
            and round(acct.entitled_days - acct.used_days - acct.balance_days, 2) == 0
        )
        if lang == "en":
            head = f"• {label}{fy}: {fmt_days(acct.balance_days)} day(s) left"
            mid = (
                f" (granted {fmt_days(acct.entitled_days)} · "
                f"used {fmt_days(acct.used_days)})" if breakdown_ok else ""
            )
            tail = (
                f" · forfeited {deadline}" if expiring
                else f" · last bookable {deadline}"
            )
        elif lang == "ja":
            head = f"• {label}{fy}:残り {fmt_days(acct.balance_days)} 日"
            mid = (
                f"(付与 {fmt_days(acct.entitled_days)} · "
                f"使用 {fmt_days(acct.used_days)})" if breakdown_ok else ""
            )
            tail = f"· {deadline} 失効" if expiring else f"· {deadline} まで取得可"
        else:
            head = f"• {label}{fy}:剩余 {fmt_days(acct.balance_days)} 天"
            mid = (
                f"(累计授予 {fmt_days(acct.entitled_days)} · "
                f"已用 {fmt_days(acct.used_days)})" if breakdown_ok else ""
            )
            tail = f"· {deadline} 清零" if expiring else f"· 最晚可用于 {deadline}"
        lines.append(f"{head}{mid}{tail}")
    # 与 ``resolve_cohort`` **同一条判据**:0 天的桶只是展示项(卡片上要出现
    # 「福利年假 2025 · 0 天」),它没有任何东西会在那天失效,所以不让它决定
    # 「最早清零日」—— 否则标题里的日期会指向一个什么都不会损失的日子,而且
    # 名单行(``CandidateRow.earliest_deadline``)和信里的日期还会对不上。
    deadlines = [
        a.deadline for a in accounts
        if a.deadline is not None and a.balance_days > 0
    ]
    earliest = min(deadlines) if deadlines else None
    total = round(sum(a.balance_days for a in accounts), 2)
    # 「最早那个清零日会带走多少天」—— 主句里 ``total_balance_days`` 是**全部**桶
    # 的和,而 ``expires_in_days`` 是**最早**那个桶的天数;两个数并列而不加限定语
    # 时,员工会读成「合计 13 天全部在 9/30 清零」。给 HR 一个能说真话的占位符。
    expiring_total = round(
        sum(a.balance_days for a in accounts if a.deadline == earliest), 2,
    ) if earliest else 0.0
    return {
        "person_name": display_name,
        "account_lines": "\n".join(lines),
        "total_balance_days": fmt_days(total),
        "expiring_balance_days": fmt_days(expiring_total),
        "deadline_date": earliest.isoformat() if earliest else "—",
        "deadline_month": _month_label(earliest, lang) if earliest else "—",
        # 不 clamp:一个负数说明上游让已过期的桶漏进了发送路径,那是必须显形的
        # bug,而不是应该被 ``max(0, …)`` 抹平成「还有 0 天」的东西。
        "expires_in_days": (
            str((earliest - today).days) if earliest else "—"
        ),
        # 余额是**发送那一刻**的快照,员工三天后对比自己的卡片必然差几天。
        # 把口径日期写进信里,这个差就不再是「谁算错了」。
        "as_of_date": (as_of or today).isoformat(),
        "hris_url": hris_url,
    }


def build_card_stats(accounts: list[AccountSnapshot], lang: str):
    """余额卡片本体 —— 截图里那张卡的飞书版。

    一个账户一格,**只有余额**:标题是 SF 里的叫法(「法定年假 2025」——
    员工在 SF 上看到的就是这个,卡片跟着叫同一个名字才不用他做翻译),
    主数值是放大的剩余天数。HR 定的形状(2026-08-18):额度区不再写清零日、
    已用、累计授予 —— 那些细节挤在两列网格里只会把「你还剩几天」这一个数
    淹掉,而清零日在标题和正文里各说了一遍。

    ``[h]`` 是「放大一号」的行内标记(见 ``_apply_inline_markup``);飞书渲染成
    ``<font size='heading_3'>``,纯文本渠道降级成加粗,不会漏出标记本身。

    余额为 0 的桶**照样出现**(HR 要求):卡片是「这一轮 9/30 到期的两个桶」的
    全貌,少一格会让员工以为自己没有那个额度。两个桶都是 0 的人根本不会进名单
    (``resolve_cohort`` 的人级门槛),所以不存在「一张全 0 的卡」。

    超过 6 个账户就不再堆格子(网格会糊成一片),正文的 ``{account_lines}``
    可以把完整明细写全。
    """
    from app.notifications.service import CardStat

    lang = normalize_lang(lang) or DEFAULT_LANG
    unit = "day(s)" if lang == "en" else ("日" if lang == "ja" else "天")
    stats = []
    for acct in accounts[:_MAX_CARD_STATS]:
        label = localized_type_label(acct.account_type, acct.account_type_label, lang)
        # 格子标题说的是「这个数是什么」,不只是额度叫什么 —— HR 定的措辞
        # (2026-08-18 第二轮):「截至目前未休法定年假(FY26)」。财年后缀走
        # ``fiscal_label``(周期结束年,不是 ``fiscal_year`` 列),和标题/正文里
        # 的 FY26 对得上。
        fy = fiscal_label(acct)
        if lang == "en":
            # 「未休的 X」而不是「Unused X」:``localized_type_label`` 给的是
            # ``Statutory annual leave``(首字母大写),塞进 ``Unused …`` 就成了
            # 句中大写。把标签放句首就没这个问题。
            title = f"{label} still unused ({fy})" if fy else f"{label} still unused"
        elif lang == "ja":
            title = f"未消化の{label}({fy})" if fy else f"未消化の{label}"
        else:
            title = f"截至目前未休{label}({fy})" if fy else f"截至目前未休{label}"
        stats.append(CardStat(
            label=title,
            value=f"[h]{fmt_days(acct.balance_days)}[/h] {unit}",
        ))
    return stats


def build_card_actions(hris_url: str, lang: str, copy: TemplateCopy | None = None):
    """卡片按钮 —— **只有一个**,指向 HR 配的 HRIS(SF)请假入口。

    此前是两个:「提交请假申请」(HRIS)+「查看我的假期」(Helix)。HR 定的
    (2026-08-18):余额以 SF 为准、申请也在 SF 提,让员工在两个入口之间选一个
    只是把「该去哪」这个问题丢回给他。所以按钮合成一个,落点是 SF。

    ``hris_url`` 没配时退回 Helix 的假期页 —— 一张没有任何按钮的提醒卡是更坏的
    结局,而这条降级路径和「HR 还没填 HRIS 入口」是同一件事,页面上看得见。

    按钮文字优先用 HR 自己写的 ``card_button_label``。不接这一步的话那三个字段
    就是死的:``_build_interactive_card`` 是 ``if extra_actions: … elif url:``,
    我们无条件返回至少一个 action,所以模板的主按钮分支永不执行 —— HR 改完保存、
    看不到任何变化,而校验、快照、未替换占位符扫描全都照样在管它。
    """
    from app.notifications.service import CardAction, LocalizedText

    def _label(field: str, fallback: LocalizedText) -> LocalizedText:
        if copy is None:
            return fallback
        data = copy.model_dump()
        return LocalizedText(
            zh=str(data.get(field) or "").strip() or fallback.zh,
            en=str(data.get(f"{field}_en") or "").strip() or fallback.en,
            ja=str(data.get(f"{field}_ja") or "").strip() or fallback.ja,
        )

    return [CardAction(
        label=_label("card_button_label", LocalizedText(
            zh="查看我的假期",
            en="View my leave balance",
            ja="休暇残高を見る",
        )),
        url=hris_url or MY_LEAVE_PATH,
        style="primary",
    )]


def render_message(
    *,
    copy: TemplateCopy,
    context: dict[str, Any],
    lang: str,
    display_name: str | None = None,
    person_id: uuid.UUID | None = None,
    accounts: list[AccountSnapshot] | None = None,
    hris_url: str = "",
    portal_url: str = "",
    signoff: str = "",
) -> RenderedMessage:
    """把文案 + context 渲染成 HR 过目的成品。

    刻意复用发送路径的两块原语(``_select_template_field`` 的 ja→en→zh 阶梯 +
    ``_safe_format`` 的缺键容忍),而不是另写一套渲染 —— 预览和真实投递用两套
    渲染,预览就不再是证据。
    """
    from app.notifications.service import (
        _resolve_cta_line,
        _safe_format,
        _select_template_field,
        _strip_inline_markup,
    )

    lang = normalize_lang(lang) or DEFAULT_LANG
    tmpl = copy.model_dump()
    full_context = dict(context)
    full_context.setdefault("portal_url", portal_url)
    # 预览按邮件渠道算 cta_line(飞书卡片有按钮,那一行是空的),这样 HR 看到
    # 的是「话最多」的那个版本,不会漏掉正文里的链接。
    full_context.setdefault("cta_line", _resolve_cta_line(
        channel="email_text", portal_url=portal_url or None, lang=lang,
    ))

    def _render(field: str) -> str:
        raw = _select_template_field(tmpl, field, lang) or ""
        try:
            rendered = _safe_format(raw, full_context)
        except (KeyError, IndexError, ValueError):
            rendered = raw
        return _strip_inline_markup(rendered).strip()

    subject = _render("subject")
    body = _render("body_text")
    card_subtitle = _render("card_subtitle")
    button_label = _render("card_button_label")
    # 「没替换掉」只能是**本该被替换**的那些,也就是白名单里的名字。渲染后文本里
    # 的 ``{curly}`` 也可能是 HR 故意写的 ``{{curly}}`` 转义字面量 —— 把它报成
    # 未替换占位符是假阳性,会让一段完全合法的文案显示成红色告警。
    # 四个字段一起扫:卡片副标题和按钮文字同样会被渲染,只扫正文就是漏报。
    substitutable = set(ALLOWED_PLACEHOLDERS) | set(AUTO_PLACEHOLDERS)
    unresolved = sorted({
        name
        for text in (subject, body, card_subtitle, button_label)
        for name in extract_placeholders(text)
        if name in substitutable
    })
    stats = build_card_stats(accounts or [], lang)
    actions = build_card_actions(hris_url, lang, copy)
    from app.notifications.service import resolve_card_text

    return RenderedMessage(
        person_id=person_id,
        display_name=display_name,
        language=lang,
        subject=subject,
        body_text=body,
        card_subtitle=card_subtitle,
        card_button_label=button_label,
        card_footer=signoff,
        card_stats=[
            {
                "label": s.label,
                "value": _strip_inline_markup(str(s.value)),
                "hint": s.hint,
            }
            for s in stats
        ],
        card_actions=[
            {
                "label": resolve_card_text(a.label, lang),
                "url": a.url,
                "style": a.style,
            }
            for a in actions
        ],
        context=full_context,
        unresolved_placeholders=unresolved,
    )


# ── 发送 ────────────────────────────────────────────────────────────────


async def assert_no_batch_in_flight(db: AsyncSession) -> None:
    """同一时刻只允许一个**真实**批次在投递。

    判据是 ``finished_at IS NULL``,不是「``created_at`` 在半小时内」—— 一次
    700 人 × 2 渠道的合法 drain 本身就会超过半小时(``fanout`` 的注释算过:
    500 人 × 2 渠道 × ~2s SMTP 已经过线),按创建时间判会在第 31 分钟放行第二
    批,尾部收件人拿到两张卡。

    僵尸判据换成 ``updated_at``:``run_batch`` 每投若干行刷一次心跳,所以
    「``updated_at`` 停了 ``_STALE_HEARTBEAT`` 这么久」才是真的进程没了 ——
    那时不再挡住 HR(已入队的行由 outbox 重试 worker 兜住,批次本身由
    ``hr.leave_reminder_reap`` 收尸)。

    测试发送被排除在外:一次崩掉的 test send 不该把真实群发挡在门外。
    """
    cutoff = datetime.now(UTC) - _STALE_HEARTBEAT
    rows = (await db.execute(
        select(
            HrLeaveReminderBatch.id,
            HrLeaveReminderBatch.updated_at,
            HrLeaveReminderBatch.created_at,
        ).where(
            HrLeaveReminderBatch.status == BATCH_STATUS_SENDING,
            HrLeaveReminderBatch.finished_at.is_(None),
            HrLeaveReminderBatch.test_mode.is_(False),
        )
    )).all()
    for batch_id, updated_at, created_at in rows:
        beat = updated_at or created_at
        if beat is None or _as_utc(beat) >= cutoff:
            raise ConflictError(
                f"A reminder batch ({batch_id}) is still sending. Wait for it "
                "to finish before starting another one.",
            )


async def create_batch(
    db: AsyncSession,
    *,
    filters: ReminderFilters,
    channels: list[str],
    person_ids: list[uuid.UUID],
    expected_recipient_count: int | None,
    allow_resend: bool,
    test_mode: bool,
    actor_person_id: uuid.UUID,
    trace_id: str,
) -> HrLeaveReminderBatch:
    """解析名单 → 冻结快照 → 落批次行(**不**发送)。

    调用方负责 commit 之后再把投递交给后台任务:发送必须发生在事务之外,
    否则 outbox 的 FK 行锁会被扣在外部 HTTP 调用上。
    """
    hris_url = await get_hris_url(db)
    signoff_config = await get_signoff(db)
    copy, _ = await effective_copy(db)
    validate_copy(copy)  # 存量坏文案不该在群发时才炸
    cohort = await resolve_cohort(db, filters)

    selected = cohort.candidates
    # ``include_expired`` 是**预览可见性**开关,不是发送口径。已过清零日的桶
    # 一旦进了发送,员工收到的是事实错误的信:标题写着「即将到期 · 2025-09-30
    # 清零」、正文写着「距清零还有 0 天 / 最晚可用于 2025-09-30」——一个早就
    # 清零的额度,日期还在过去。同理,没有清零日的桶(窗口条件为空时会入选)
    # 会渲染成「· — 清零」。两类都在这里剔掉,并如实报给 HR。
    sendable = _sendable_rows(selected)
    excluded_undated = len(selected) - len(sendable)
    if test_mode:
        # 测试发送在**任何**名单校验之前就分流:它只发给发起人自己,
        # ``person_ids`` / ``expected_recipient_count`` 与它无关。放在校验之后
        # 会让「带着 person_ids 调 test_mode」的脚本因为「某人已不在名单里」被
        # 422 拒掉 —— 而那些人根本不会收到这封测试信。
        selected = await _test_mode_selection(
            db, cohort, actor_person_id, filters,
            sendable=sendable, person_ids=person_ids,
        )
    else:
        selected = sendable
        if person_ids:
            wanted = {str(p) for p in person_ids}
            selected = [c for c in selected if str(c.person_id) in wanted]
            missing = wanted - {str(c.person_id) for c in selected}
            if missing:
                # 把「本来就发不出去」和「名单真的变了」分开说。合起来说的那
                # 版文案(「their balance, deadline or employment changed.
                # Reload the preview」)对前一种情况是**假话**:什么都没变,
                # HR 照提示 reload 会看到一模一样的名单、再点再报同一个错,
                # 除非他手工取消勾选每一行已过期的,否则永远发不出去。
                undated = {
                    str(c.person_id) for c in cohort.candidates
                    if str(c.person_id) in wanted
                    and (c.expires_in_days is None or c.expires_in_days < 0)
                }
                if undated:
                    raise ValidationError(
                        f"{len(undated)} selected recipient(s) have a balance "
                        "that is already past its deadline (or has no deadline "
                        "at all), so they can never be told it is 'about to "
                        "expire'. Unselect them — the preview marks them — or "
                        "turn off 'include expired'.",
                    )
                raise ValidationError(
                    f"{len(missing)} selected recipient(s) are no longer in the "
                    "cohort (their balance, deadline or employment changed). "
                    "Reload the preview and confirm again.",
                )
            if not allow_resend:
                # 显式勾选**也**要过去重护栏。此前这里是 ``elif``,而前端在正常
                # 路径上**总是**带 person_ids ⇒ ``resend_window_days`` 这条护栏
                # 在真实路径上从不生效,``allow_resend`` 也随之变成 no-op:HR 发完
                # 一轮、页面把刚发过的人重新勾满(那时 outbox 行还是 pending,
                # 去重看不到),再点一次就是同一批人第二张卡,零警告。
                repeats = [c for c in selected if c.reminded_recently]
                if repeats:
                    names = ", ".join(c.display_name for c in repeats[:5])
                    more = f" (+{len(repeats) - 5})" if len(repeats) > 5 else ""
                    raise ConflictError(
                        f"{len(repeats)} selected recipient(s) were already "
                        f"reminded recently: {names}{more}. Turn on 'send "
                        "again' if you mean to chase them, or unselect them.",
                    )
        else:
            # 没有显式选人时,默认口径 = 前端默认勾选的那一批。两条默认不勾的
            # 理由都要在这里成立,否则「不传 person_ids」这条路径会把 HR 明确
            # 排除掉的人也发出去 —— 而页面上他是不勾的。
            selected = [c for c in selected if not c.excluded]
            if not allow_resend:
                selected = [c for c in selected if not c.reminded_recently]
        if not selected:
            raise ValidationError(
                "No recipients can be sent to. Either nobody matches the "
                "filters, everyone in scope was reminded recently (enable "
                "'send again' to include them), or every matched account is "
                "already past its deadline / has no deadline at all "
                f"({excluded_undated} such recipient(s) were excluded — an "
                "expired balance cannot be 'about to expire').",
            )
        if len(selected) > MAX_RECIPIENTS_PER_BATCH:
            raise ValidationError(
                f"{len(selected)} recipients exceeds the per-batch cap of "
                f"{MAX_RECIPIENTS_PER_BATCH}. Narrow the filters.",
            )
        if (
            expected_recipient_count is not None
            and expected_recipient_count != len(selected)
        ):
            raise ConflictError(
                f"The cohort changed since you previewed it: you confirmed "
                f"{expected_recipient_count} recipient(s), it is now "
                f"{len(selected)}. Reload the preview and confirm again.",
            )

    snapshot = [
        {
            "person_id": str(c.person_id),
            "display_name": c.display_name,
            "work_email": c.work_email,
            "department": c.department,
            "accounts": [s.model_dump(mode="json") for s in c.accounts],
            "total_balance_days": c.total_balance_days,
            "earliest_deadline": (
                c.earliest_deadline.isoformat() if c.earliest_deadline else None
            ),
            "expires_in_days": c.expires_in_days,
        }
        for c in selected
    ]
    batch = HrLeaveReminderBatch(
        created_by_person_id=str(actor_person_id),
        trace_id=trace_id or None,
        status=BATCH_STATUS_SENDING,
        test_mode=test_mode,
        channels=list(channels),
        filters=filters.model_dump(mode="json"),
        copy_snapshot=copy.model_dump(),
        recipients_snapshot=snapshot,
        recipient_count=len(snapshot),
        # 从 candidate 对象上数,而不是从刚序列化出来的 dict 里回读 —— 同一个
        # 事实读两遍是让两个数字有机会不一致的最短路径。
        account_count=sum(len(c.accounts) for c in selected),
        at_risk_days=round(sum(c.total_balance_days for c in selected), 2),
        started_at=datetime.now(UTC),
    )
    # ``hris_url`` 与「剔掉了几个无效清零日」也冻结进 filters 快照,事后能解释
    # 卡片按钮指向了哪里、以及为什么发出的人数比预览名单少。
    # ``_signoff`` 和 ``_hris_url`` 一样冻结进批次:选中的那一行寄语要能被
    # 站内通知的重渲染、resend-failed 的克隆行、以及事后的批次详情读到同一句。
    # 种子取批次 id ⇒ 随机模式下同一批次每次读都是同一句。
    batch.filters = {
        **batch.filters,
        "_hris_url": hris_url,
        "_excluded_undated": excluded_undated,
        "_signoff": pick_signoff(signoff_config, seed=str(batch.id)),
    }
    db.add(batch)
    await db.flush()
    return batch


#: 试发收件人的数据来源 —— 前端逐人显示,发送前就能看出会寄出什么。
TEST_SOURCE_COHORT = "cohort"
TEST_SOURCE_OUT_OF_COHORT = "out_of_cohort"
TEST_SOURCE_SAMPLE = "sample"
TEST_SOURCE_BLOCKED = "blocked"

#: ``blocked`` 的原因码(前端翻成三语,不复用后端英文句子)。
BLOCK_NOT_FOUND = "not_found"
BLOCK_INACTIVE = "inactive"
BLOCK_NO_EMAIL = "no_email"
BLOCK_NO_MATCHING_LEAVE = "no_matching_leave"
BLOCK_NO_DEADLINE = "no_deadline"
BLOCK_PAST_DEADLINE = "past_deadline"
#: 有账户、但选中的桶一天都不剩 —— 和「没有清零日」是两回事,分开说。
BLOCK_ZERO_BALANCE = "zero_balance"


class TestRecipient(NamedTuple):
    """一个被挑中的试发对象,以及「他会收到什么」。"""

    person_id: uuid.UUID
    display_name: str
    work_email: str | None
    #: ``cohort`` / ``out_of_cohort`` / ``sample`` / ``blocked``
    source: str
    #: 仅 ``blocked`` 时有值,取上面的原因码
    reason: str | None
    #: ``blocked`` 之外都有,就是真正会被渲染的那一行
    row: CandidateRow | None


def _sendable_rows(rows: Iterable[CandidateRow]) -> list[CandidateRow]:
    """能被告知「即将清零」的行 —— 有清零日、且清零日不在过去。"""
    return [
        c for c in rows
        if c.expires_in_days is not None and c.expires_in_days >= 0
    ]


async def resolve_test_recipients(
    db: AsyncSession,
    *,
    filters: ReminderFilters,
    person_ids: list[uuid.UUID],
    actor_person_id: uuid.UUID,
    cohort: CohortResult | None = None,
    sendable: list[CandidateRow] | None = None,
) -> list[TestRecipient]:
    """逐人判定试发会寄出什么。**预检接口与真正发送共用这个函数**。

    为什么试发不能只允许「当前名单内的人」:HR 想给某个同事(或自己)寄一封样信
    看看长什么样,而那个人是否恰好命中本次口径(余额阈值、到期窗口、财年)是完全
    无关的两件事。旧版把选择器的选项限制在名单里,于是在 500 人的公司里搜一个
    名单外的同事得到的是 ``暂无数据`` —— 看起来就像搜索坏了。现在选择器直接搜
    全员通讯录(``MultiUserSelect``),由这里决定每个人拿到什么:

    * ``cohort`` —— 他在本次名单里:用**和真实群发完全一样**的真数字。
    * ``out_of_cohort`` —— 他不在名单里,但有该类型的假期额度且还没到清零日:
      仍然用他自己的**真**数字(真话,只是这次群发不会包含他)。
    * ``sample`` —— **只有发起人自己**能拿到明确标注的示例数字。给第三方寄一封
      数字是假的「你的年假即将清零」是不能做的事,所以别人走 ``blocked``。
    * ``blocked`` —— 说清楚为什么(没邮箱 / 没有该类型额度 / 额度已过清零日 …),
      前端在发送**之前**就显示出来,而不是点了发送再弹一个 422。

    判定「不在名单里的人有没有真数字」时,把口径放宽成「只保留 HR 选的假期类型」
    —— 余额阈值、到期窗口、财年、是否 closed 都不再限制,因为这些是「要提醒谁」
    的口径,不是「这个人有没有额度」的事实。``include_expired`` 也放开,好把
    「他的桶早就清零了」和「他根本没有这种额度」分成两条不同的提示。
    """
    if not person_ids:
        return []
    if cohort is None:
        cohort = await resolve_cohort(db, filters)
    # 无条件再过一遍 ``_sendable_rows``:调用方传进来的那份**应该**已经过滤过,但
    # 如果它其实是原始 ``candidates``(HR 勾了「含已过期」时里面就有过期桶),
    # 一个早已清零的额度会被当成「在本次名单内」,试发信里写出过去的清零日。
    sendable = _sendable_rows(
        cohort.candidates if sendable is None else sendable,
    )

    wanted: list[str] = []
    for raw_id in person_ids:  # 去重但保持 HR 勾选的顺序
        if str(raw_id) not in wanted:
            wanted.append(str(raw_id))
    in_cohort = {str(c.person_id): c for c in sendable}

    # 只有存在名单外的人时才付第二次解析的代价。
    unresolved = [pid for pid in wanted if pid not in in_cohort]
    relaxed: dict[str, CandidateRow] = {}
    relaxed_all: dict[str, CandidateRow] = {}
    persons: dict[str, IdentityPerson] = {}
    if unresolved:
        relaxed_filters = filters.model_copy(update={
            "fiscal_years": [],
            "min_balance_days": 0.0,
            "expiring_within_days": None,
            "include_expired": True,
            "include_closed_accounts": True,
        })
        relaxed_cohort = await resolve_cohort(db, relaxed_filters)
        relaxed_all = {str(c.person_id): c for c in relaxed_cohort.candidates}
        relaxed = {
            str(c.person_id): c
            for c in _sendable_rows(relaxed_cohort.candidates)
        }
        rows = (await db.execute(
            select(IdentityPerson).where(IdentityPerson.id.in_(unresolved))
        )).scalars().all()
        persons = {str(p.id): p for p in rows}

    out: list[TestRecipient] = []
    for pid in wanted:
        row = in_cohort.get(pid)
        if row is not None:
            out.append(TestRecipient(
                person_id=row.person_id,
                display_name=row.display_name,
                work_email=row.work_email,
                source=TEST_SOURCE_COHORT,
                reason=None,
                row=row,
            ))
            continue
        row = relaxed.get(pid)
        if row is not None:
            out.append(TestRecipient(
                person_id=row.person_id,
                display_name=row.display_name,
                work_email=row.work_email,
                source=TEST_SOURCE_OUT_OF_COHORT,
                reason=None,
                row=row,
            ))
            continue

        person = persons.get(pid)
        name = pid
        email: str | None = None
        if person is not None:
            email = person.work_email
            name = person.display_name or person.work_email or pid
        reason = BLOCK_NO_MATCHING_LEAVE
        if person is None or person.deleted_at is not None:
            reason = BLOCK_NOT_FOUND
        elif person.employment_status != "active":
            reason = BLOCK_INACTIVE
        elif not (person.work_email or "").strip():
            reason = BLOCK_NO_EMAIL
        elif pid in relaxed_all:
            # 有账户,但发不出去 —— 三种情况说法不同:一天都不剩 / 没有清零日 /
            # 清零日已过。合起来说会让 HR 拿着一条不对的原因去问当事人。
            stale = relaxed_all[pid]
            if stale.total_balance_days <= 0:
                reason = BLOCK_ZERO_BALANCE
            else:
                reason = (
                    BLOCK_NO_DEADLINE if stale.expires_in_days is None
                    else BLOCK_PAST_DEADLINE
                )
        if reason != BLOCK_NOT_FOUND and pid == str(actor_person_id):
            # 发起人自己:宁可给他一封标注过的示例信,也不要让他连预览都做不了。
            sample = _sample_row(persons[pid], filters)
            if sample is not None:
                out.append(TestRecipient(
                    person_id=sample.person_id,
                    display_name=sample.display_name,
                    work_email=sample.work_email,
                    source=TEST_SOURCE_SAMPLE,
                    reason=None,
                    row=sample,
                ))
                continue
        out.append(TestRecipient(
            # ``pid`` 来自 ``list[uuid.UUID]``,一定是合法 UUID。
            person_id=uuid.UUID(pid),
            display_name=name,
            work_email=email,
            source=TEST_SOURCE_BLOCKED,
            reason=reason,
            row=None,
        ))
    return out


def _sample_row(
    person: IdentityPerson, filters: ReminderFilters,
) -> CandidateRow | None:
    """一份**明确标注**的示例额度,只用于「发起人自己没有可提醒的额度」。

    没有工作邮箱就投不出去,返回 ``None`` 让调用方走 blocked。
    """
    if not (person.work_email or "").strip():
        return None
    today = today_sh()
    sample_deadline = today + timedelta(days=filters.expiring_within_days or 30)
    return CandidateRow(
        person_id=person.id,
        display_name=person.display_name or person.work_email or str(person.id),
        employee_no=person.employee_no,
        work_email=person.work_email,
        department=None,
        bu=None,
        language=resolve_lang(person),
        accounts=[AccountSnapshot(
            account_id=uuid.uuid4(),
            account_type=filters.account_types[0],
            account_type_label=None,
            fiscal_year=today.year - 1,
            balance_days=5.0,
            entitled_days=5.0,
            used_days=0.0,
            deadline=sample_deadline,
            expires_in_days=(sample_deadline - today).days,
            forfeits=True,
            is_current=False,
            account_closed=False,
        )],
        total_balance_days=5.0,
        earliest_deadline=sample_deadline,
        expires_in_days=(sample_deadline - today).days,
        data_synced_at=None,
        last_reminded_at=None,
        last_reminded_batch_id=None,
        reminded_recently=False,
    )


#: ``blocked`` 原因码 → 给 HR 看的一句话(前端有三语版本,这里是 API 兜底)。
_BLOCK_HINTS = {
    BLOCK_NOT_FOUND: "no such person in Helix",
    BLOCK_INACTIVE: "not an active employee",
    BLOCK_NO_EMAIL: "no work email to deliver to",
    BLOCK_NO_MATCHING_LEAVE: (
        "has no leave account of the selected type(s), so a reminder to them "
        "would contain made-up numbers"
    ),
    BLOCK_ZERO_BALANCE: (
        "has 0 days left in the selected bucket(s) — there is nothing to "
        "remind them about"
    ),
    BLOCK_NO_DEADLINE: "their balance has no forfeit date",
    BLOCK_PAST_DEADLINE: "their balance is already past its forfeit date",
}


async def _test_mode_selection(
    db: AsyncSession,
    cohort: CohortResult,
    actor_person_id: uuid.UUID,
    filters: ReminderFilters,
    *,
    sendable: list[CandidateRow] | None = None,
    person_ids: list[uuid.UUID] | None = None,
) -> list[CandidateRow]:
    """试发:给**指定的人**,或(不指定时)给发起人自己。

    逐人的判定规则见 :func:`resolve_test_recipients` —— 预检接口用的是同一个
    函数,所以「预检说能发」和「发送真的能发」不可能各说各话。

    上限 ``MAX_TEST_RECIPIENTS``:试发**不计入去重**,所以如果不设上限,勾 281 个人
    再打上 ``test_mode`` 就成了一条绕开「近期已提醒」护栏的群发通道。

    不指定时退回原行为:发起人自己在名单里 → 用他真实的数字;不在(比如 HR 是海外
    主体、没有假期账户) → 用一份明确标注的示例数字,而不是拿某个真实员工的余额去
    演示。

    从 ``sendable``(已剔掉过期/无清零日的桶)里挑人,而不是从原始 ``candidates``
    —— 否则 HR 勾了「含已过期」时,试发信里会出现「距清零还有 0 天 · 2025-09-30
    清零」这种真实群发**不会**出现的错数字,反而让 HR 误以为群发也含过期桶。
    """
    pool = sendable if sendable is not None else cohort.candidates
    if person_ids:
        if len(person_ids) > MAX_TEST_RECIPIENTS:
            raise ValidationError(
                f"A test send goes to at most {MAX_TEST_RECIPIENTS} people "
                f"(you picked {len(person_ids)}). Test sends deliberately do NOT "
                "count toward the 'reminded recently' guard, so a large one "
                "would be a way around it — use the real send for the whole "
                "cohort.",
            )
        resolved = await resolve_test_recipients(
            db,
            filters=filters,
            person_ids=person_ids,
            actor_person_id=actor_person_id,
            cohort=cohort,
            sendable=pool,
        )
        blocked = [r for r in resolved if r.source == TEST_SOURCE_BLOCKED]
        if blocked:
            detail = "; ".join(
                f"{r.display_name} ({_BLOCK_HINTS.get(r.reason or '', r.reason)})"
                for r in blocked[:5]
            )
            more = f" (+{len(blocked) - 5} more)" if len(blocked) > 5 else ""
            raise ValidationError(
                f"{len(blocked)} of the people you picked cannot receive a test "
                f"reminder: {detail}{more}. Unpick them and try again.",
            )
        return [r.row for r in resolved if r.row is not None]
    me = next(
        (c for c in pool if str(c.person_id) == str(actor_person_id)),
        None,
    )
    if me is not None:
        return [me]
    person = await db.get(IdentityPerson, str(actor_person_id))
    if person is None:
        raise NotFoundError("Person", str(actor_person_id))
    sample = _sample_row(person, filters)
    if sample is None:
        raise ValidationError(
            "Your account has no work email, so a test send cannot be "
            "delivered to you.",
        )
    return [sample]


class BatchTally(NamedTuple):
    """一个批次在 outbox 里的真实投递画像(按**人**去重)。"""

    #: 这一批一共有多少行 outbox(含 resend 克隆出来的行)
    rows: int
    #: 有多少**人**至少有一行推出去了
    persons: int
    #: 其中多少人算送达
    delivered: int
    #: 其中多少人未送达
    undelivered: int


async def batch_delivery_tally(
    db: AsyncSession, batch_id: uuid.UUID,
) -> BatchTally:
    """从 outbox **真值**数这一批的投递画像,**按人去重、取每人最新一行**。

    不用 ``enqueue_then_drain`` 返回的计数:它把两类「其实还没到人手上」的行算成
    delivered(命中静默时段被 park 成 ``pending`` 的行、被并发 worker 抢到
    ``delivering`` 的行)。批次台账要说的是「员工到底收到没有」,所以只认 outbox
    的 ``status``。

    **按人去重**是 ``resend-failed`` 逼出来的:重发是「克隆一行新的」而不是改旧行
    (``notifications.service._resend_one`` 的契约),克隆行照抄
    ``related_resource_*``,所以同一个人会在同一批次里有两行 —— 一行永久 failed、
    一行 delivered。按行数,重发成功之后台账仍然显示「失败 N」;按人取最新一行,
    数字才跟员工的真实处境一致。
    """
    from app.notifications.models import (
        STATUS_DELIVERED,
        STATUS_PARTIAL,
        NotificationOutbox,
    )

    rows = (await db.execute(
        select(
            NotificationOutbox.recipient_person_id,
            NotificationOutbox.status,
            NotificationOutbox.created_at,
            NotificationOutbox.id,
        ).where(
            NotificationOutbox.related_resource_type == BATCH_RESOURCE_TYPE,
            NotificationOutbox.related_resource_id == str(batch_id),
        )
    )).all()
    def _rank(created_at: datetime | None, status: str, row_id: uuid.UUID):
        """哪一行代表这个人的「当前处境」。

        ``created_at`` 的精度只到秒(schema 实测),而重发是批量克隆 —— 源行与克隆
        行经常落在**同一秒**。此时若只靠 uuid4 字符串兜底,取到哪一行是抛硬币:
        实测 40 个「(失败源行, 成功克隆行)」同秒配对,有 22 个被判成未送达,于是
        重发全成功之后台账还写着「失败 22」。所以平票时**优先已送达** —— 一个人
        只要有一行真的推出去了,他的处境就是「收到了」。
        """
        floor = datetime.min.replace(tzinfo=UTC)
        return (
            _as_utc(created_at) if created_at else floor,
            1 if status in (STATUS_DELIVERED, STATUS_PARTIAL) else 0,
            str(row_id),
        )

    latest: dict[str, tuple[datetime | None, uuid.UUID, str]] = {}
    for pid, status, created_at, row_id in rows:
        key = str(pid)
        prev = latest.get(key)
        if prev is None or _rank(created_at, status, row_id) > _rank(
            prev[0], prev[2], prev[1],
        ):
            latest[key] = (created_at, row_id, status)
    delivered = sum(
        1 for _c, _i, st in latest.values()
        if st in (STATUS_DELIVERED, STATUS_PARTIAL)
    )
    return BatchTally(
        rows=len(rows), persons=len(latest),
        delivered=delivered, undelivered=len(latest) - delivered,
    )


def apply_tally(
    batch: HrLeaveReminderBatch, tally: BatchTally, *, skipped_departed: int = 0,
) -> None:
    """把 outbox 真值写回批次行 —— **一个公式,三个写者共用**。

    此前 ``run_batch`` 和 ``reap_stale_batches`` 各写一份,而只有 reaper 那份带
    ``max(0, recipient_count - 行数)``。于是 ``enqueue_then_drain`` 里单行 enqueue
    失败(它 rollback+log+continue,**不抛**)时,``run_batch`` 算出
    ``delivered=97 / failed=0``、状态判 ``completed`` —— HR 看到「收件人 100 ·
    已送达 97 · 失败 0 · 已完成」,三个人凭空消失且页面上没有任何线索。
    """
    batch.enqueued = tally.persons
    batch.delivered = tally.delivered
    # 没有 outbox 行的人(离职被 skip、enqueue 失败)也算没送到。
    never_enqueued = max(0, batch.recipient_count - tally.persons)
    batch.failed = tally.undelivered + max(never_enqueued, skipped_departed)


def _batch_as_of(batch: HrLeaveReminderBatch) -> date:
    """批次的口径日期 —— 建批次时的 Asia/Shanghai 当天。

    ``created_at`` 是 DB 时钟(UTC),换回运营时区才是 HR 点确认时看到的那个
    ``as_of_date``。取不到就退回今天(只会发生在人为构造的批次行上)。
    """
    created = batch.started_at or batch.created_at
    if created is None:
        return today_sh()
    return _as_utc(created).astimezone(ZoneInfo("Asia/Shanghai")).date()


async def run_batch(db: AsyncSession, batch_id: uuid.UUID) -> dict[str, Any]:
    """投递一个批次(在请求事务之外跑)。

    ``enqueue_then_drain`` 保证:每一行 outbox 先被 commit 成 durable 记录,
    再逐条投递;进程中途挂掉,已发出的记录不会被回滚,未发出的行由
    ``notification.outbox_retry`` worker 兜住。

    分块投递(每 ``_HEARTBEAT_EVERY`` 人一块)只为一件事:块间刷一次
    ``updated_at`` 心跳,让 in-flight 闸能区分「还在正常发」和「进程没了」。
    """
    from app.auth.service import is_effectively_departed
    from app.notifications import fanout
    from app.notifications.service import (
        _SEE_TEMPLATE_SENTINEL,
        LocalizedText,
        NotificationEnvelope,
    )

    batch = await db.get(HrLeaveReminderBatch, batch_id)
    if batch is None:
        raise NotFoundError("HrLeaveReminderBatch", str(batch_id))
    # 幂等:已经收过尾的批次不再重发一整套 outbox 行。重复触发的形状有好几种
    # (重复 spawn、reaper 与 drain 撞车、将来加的管理端重跑入口),都在这里止住。
    if batch.finished_at is not None or batch.status != BATCH_STATUS_SENDING:
        return {
            "batch_id": str(batch_id),
            "status": batch.status,
            "already_finished": True,
        }
    hris_url = str((batch.filters or {}).get("_hris_url") or "")
    # 冻结的寄语。老批次(这个字段出现之前建的)没有 ⇒ 走平台默认落款,不回填。
    frozen_signoff = (batch.filters or {}).get("_signoff")
    if not isinstance(frozen_signoff, dict):
        frozen_signoff = None
    # 口径日期 = **建批次那一刻**的 today,不是 drain 时的 today。23:59 建的批、
    # 00:00 之后 drain,重算会把「还有 0 天」变成「还有 -1 天」(或此前被
    # ``max(0, …)`` 抹成「还有 0 天」而其实已过期)—— 那是唯一一条能让真实员工
    # 收到错数字的路径。快照里已经冻结了每个人的 ``expires_in_days``。
    as_of = _batch_as_of(batch)
    # 按钮文字用**发送时冻结**的那份文案(HR 之后改了文案,这一批的按钮不该跟着变)。
    copy_snapshot: TemplateCopy | None = None
    if isinstance(batch.copy_snapshot, dict):
        try:
            copy_snapshot = TemplateCopy.model_validate(batch.copy_snapshot)
        except Exception:  # noqa: BLE001 — 坏快照不该阻断投递,退回内置标签
            logger.warning(
                "hr_leave.expiry_reminder: batch %s has an unusable "
                "copy_snapshot; falling back to built-in button labels",
                batch_id,
            )
    snapshot = list(batch.recipients_snapshot or [])
    channels = list(batch.channels or ["feishu", "email"])
    test_mode = bool(batch.test_mode)
    trace_id = str(batch.trace_id or "")
    actor_user_id = (
        uuid.UUID(batch.created_by_person_id)
        if batch.created_by_person_id else None
    )
    person_ids = [str(r.get("person_id")) for r in snapshot if r.get("person_id")]
    # 同 ``list_exclusions``:空列表跳过查询,别靠 ``or [""]`` —— ``GUID`` 的 bind
    # 会拿 ``""`` 构造 UUID 然后 ``ValueError``。这里现实中到不了(``create_batch``
    # 拒绝空名单),但同一个写法在别处已经真的 500 过一次。
    persons: dict[str, IdentityPerson] = {}
    if person_ids:
        persons = {
            str(p.id): p
            for p in (await db.execute(
                select(IdentityPerson).where(IdentityPerson.id.in_(person_ids))
            )).scalars().all()
        }

    envelopes: list[NotificationEnvelope] = []
    skipped_departed = 0
    for row in snapshot:
        pid = str(row.get("person_id") or "")
        person = persons.get(pid)
        # 预览到发送之间离职了 —— 不发。判据必须是 ``is_effectively_departed``
        # 而不是 ``deleted_at``:Helix 的离职**不软删**,它翻的是
        # ``employment_status`` / ``hris_record_state``。而 fanout 用的
        # ``deliver`` 自己没有 departed 闸(只有 retry / resend 路径有),所以
        # 这里漏了就真的会给刚离职的人发「快去休假」。
        if person is None or person.deleted_at is not None or is_effectively_departed(person):
            skipped_departed += 1
            continue
        accounts = [
            AccountSnapshot.model_validate(a) for a in (row.get("accounts") or [])
        ]
        lang = resolve_lang(person)
        context = build_context(
            display_name=str(row.get("display_name") or person.display_name or pid),
            accounts=accounts,
            lang=lang,
            hris_url=hris_url,
            today=as_of,
            as_of=as_of,
        )
        subtitle = None
        if test_mode:
            subtitle = LocalizedText(
                zh="测试发送 · 只发给你自己(数字可能是示例值)",
                en="Test send · delivered only to you (numbers may be samples)",
                ja="テスト送信 · あなただけに届いています(数値はサンプルの場合あり)",
            )
        envelopes.append(NotificationEnvelope(
            recipient=person,
            # 标题/正文由注册模板按收件人语言渲染;这里的值只是「模板缺失」时
            # 的兜底。正文写哨兵是刻意的:站内通知读时会按模板重渲染,写真正的
            # 散文进去会让收件箱显示这份中文兜底而不是本地化文案。
            title="你的年假即将到期",
            body_text=_SEE_TEMPLATE_SENTINEL,
            channels=channels,
            template_key=TEMPLATE_KEY,
            context=context,
            url=MY_LEAVE_PATH,
            related_resource_type=BATCH_RESOURCE_TYPE,
            related_resource_id=batch_id,
            card_subtitle=subtitle,
            # 落款换成 HR 的寄语。平台默认是「— Helix IT 平台」,而这封信是催人
            # **去休假**的 —— 一行公司签名会把它读成行政通知。
            card_footer=(
                signoff_text(frozen_signoff, lang) or None
                if frozen_signoff else None
            ),
            card_stats=build_card_stats(accounts, lang),
            card_actions=build_card_actions(hris_url, lang, copy_snapshot),
        ))

    error: str | None = None

    async def _heartbeat(done: int, total: int) -> None:
        """让 in-flight 闸与 reaper 看得出这批还在正常推进。

        每 ``_HEARTBEAT_EVERY`` 行刷一次 ``updated_at``,用 ``func.now()`` 而不是
        ``datetime.now(UTC)``:``created_at`` / ``updated_at`` 的 ``server_default``
        写的是 **DB 时钟**,而两个判据(心跳新鲜度、僵尸阈值)都在比较这两列;混用
        应用时钟会在 DB 时区不是 UTC 时把心跳算成「在未来」⇒ reaper 永远不收尸、
        in-flight 闸永久挡住 HR —— 恰好是 reaper 存在的那个场景被时区废掉。

        走 **Core UPDATE** 而不是「``db.get`` 出实例再赋值」,有两个实测出来的
        理由:① 把 ``func.now()`` 赋给一个 ORM 属性之后,SQLAlchemy 无法知道服务端
        算出来的值,会把该属性标成 expired ⇒ 之后在**异步**代码里读它就是
        ``MissingGreenlet``(实测确认);② 省掉每次心跳的那条 SELECT。
        """
        if done % _HEARTBEAT_EVERY and done != total:
            return
        await db.execute(
            sa_update(HrLeaveReminderBatch)
            .where(HrLeaveReminderBatch.id == batch_id)
            .values(updated_at=func.now()),
        )
        await db.commit()

    # savepoint-exempt: ``enqueue_then_drain`` 自己拥有事务边界(逐行 commit,
    # 失败逐行 rollback),所以这里没有「半条未提交的写」需要 savepoint 去撤;
    # 真正的危险是**吞掉异常之后继续用一个 pending-rollback 的 session**
    # (`reference_swallowed_db_error_poisons_transaction`),所以 except 里
    # 第一件事就是显式 rollback,再去写批次行的结局。
    try:
        # **一次**调用,不分块。``enqueue_then_drain`` 的 phase 1 是全批入队、
        # 逐行 commit —— 那正是「进程被 kill 之后剩下的行仍然 PENDING、由
        # notification.outbox_retry 兜住投递」这条保证的全部来源。分块会把
        # phase 1 挪进循环:第 5 块中途被部署重启,第 126..500 号收件人**根本
        # 没有 outbox 行**,重试 worker 无从兜住,而 resend-failed 只能挑已存在
        # 的行 ⇒ 那些人无法通过任何 UI 补发。心跳改由 on_progress 回调承担。
        if envelopes:
            await fanout.enqueue_then_drain(
                db, envelopes, label="hr_leave.expiry_reminder",
                trace_id=trace_id, actor_user_id=actor_user_id,
                on_progress=_heartbeat,
            )
    except Exception as exc:  # noqa: BLE001 — 必须把结果写回批次行
        error = f"{type(exc).__name__}: {exc}"[:500]
        logger.exception("hr_leave.expiry_reminder: batch %s drain failed", batch_id)
        # 不 rollback 的话,下面的 db.get / db.commit 会以 PendingRollbackError
        # 二次爆掉,批次就永远停在 ``sending`` —— HR 既看不到失败原因,也被
        # in-flight 闸挡住。
        try:
            await db.rollback()
        except Exception:  # noqa: BLE001
            logger.exception(
                "hr_leave.expiry_reminder: rollback after drain failure failed",
            )

    # 重新取一遍:fanout 内部 commit 过多次,ORM 实例可能已过期。
    batch = await db.get(HrLeaveReminderBatch, batch_id, populate_existing=True)
    if batch is None:  # pragma: no cover — 批次被并发删掉
        return {"batch_id": str(batch_id), "missing": True}
    if batch.finished_at is not None:
        # reaper 在我们投递期间把这一批判死了(心跳窗口偏紧,或这台机器很慢)。
        # 它写的数字是按 outbox 真值算的,我们继续按同一口径覆盖一遍即可 ——
        # 但要记一条,因为这说明 _STALE_HEARTBEAT 需要再放宽。
        logger.warning(
            "hr_leave.expiry_reminder: batch %s was reaped while still draining; "
            "recomputing from outbox truth", batch_id,
        )
    stats = await batch_delivery_tally(db, batch_id)
    apply_tally(batch, stats, skipped_departed=skipped_departed)
    delivered = batch.delivered
    if error:
        batch.status = BATCH_STATUS_FAILED
        batch.error_summary = error
    elif not envelopes and skipped_departed:
        # 全员在预览→发送之间离职。说清真实原因,别把人引去查飞书集成。
        batch.status = BATCH_STATUS_FAILED
        batch.error_summary = (
            f"Nothing was sent: all {skipped_departed} recipient(s) left the "
            "company between the preview and the send."
        )
    elif delivered == 0 and batch.recipient_count > 0:
        batch.status = BATCH_STATUS_FAILED
        batch.error_summary = (
            batch.error_summary
            or "No recipient could be delivered — check the Feishu / email "
            "integration status in the outbox."
        )
    elif batch.failed > 0:
        batch.status = BATCH_STATUS_PARTIAL
        if skipped_departed and not batch.error_summary:
            batch.error_summary = (
                f"{skipped_departed} recipient(s) were skipped: they left the "
                "company between the preview and the send."
            )
    else:
        batch.status = BATCH_STATUS_COMPLETED
        # 清掉 reaper 可能留下的那条说明。它写「The send task never started —
        # nobody was notified」时是按当时的 outbox 真值判的,而我们现在**确实**
        # 把人送到了 ⇒ 那句话已经不成立;不清的话页面会显示「已完成」+ 一条红色
        # 「一个人都没收到,请重跑」,HR 很可能据此把整轮重发一遍。
        batch.error_summary = None
    batch.finished_at = datetime.now(UTC)
    await db.commit()
    return {
        "batch_id": str(batch_id),
        "status": batch.status,
        "delivered": batch.delivered,
        "failed": batch.failed,
        "skipped": skipped_departed,
    }


async def reap_stale_batches(
    db: AsyncSession, *, now: datetime | None = None,
) -> dict[str, Any]:
    """把僵死在 ``sending`` 的批次按 outbox 真值收尾。

    ``_spawn_background`` 是 API 进程里的 ``asyncio.create_task``,所以进程被
    kill(部署 / OOM)时批次会永久停在 ``sending``。两种形态:

    * **零 outbox 行** —— 最坏的一种:``/send`` 已经 commit 了批次行,进程在
      spawn 之前就没了,于是**一个人都没收到**,而 HR 看到的是「正在发送」,
      永远。没有这个 reaper 就没有任何信号。
    * **有 outbox 行** —— 已入队的行由 ``notification.outbox_retry`` worker
      兜住投递,但批次的 delivered/failed 永远是 0。按 outbox 回填真值。

    幂等:只动 ``status='sending' AND finished_at IS NULL`` 且心跳已停的行。
    """
    now = now or datetime.now(UTC)
    cutoff = now - _STALE_HEARTBEAT
    rows = (await db.execute(
        select(HrLeaveReminderBatch).where(
            HrLeaveReminderBatch.status == BATCH_STATUS_SENDING,
            HrLeaveReminderBatch.finished_at.is_(None),
        )
    )).scalars().all()
    reaped: list[dict[str, Any]] = []
    for batch in rows:
        beat = batch.updated_at or batch.created_at
        if beat is not None and _as_utc(beat) >= cutoff:
            continue  # 还在正常推进
        tally = await batch_delivery_tally(db, batch.id)
        apply_tally(batch, tally)
        total, delivered = tally.persons, tally.delivered
        if total == 0:
            batch.status = BATCH_STATUS_FAILED
            batch.error_summary = (
                "The send task never started (process restart) — nobody was "
                "notified. Re-run the cohort and send again."
            )
        elif delivered == 0:
            batch.status = BATCH_STATUS_FAILED
            batch.error_summary = (
                batch.error_summary
                or "The send task stopped and nothing was delivered; queued "
                "rows are still retried by the notification outbox worker."
            )
        elif batch.failed > 0:
            batch.status = BATCH_STATUS_PARTIAL
            batch.error_summary = (
                batch.error_summary
                or "The send task stopped partway; queued rows are still "
                "retried by the notification outbox worker."
            )
        else:
            batch.status = BATCH_STATUS_COMPLETED
        batch.finished_at = now
        reaped.append({
            "batch_id": str(batch.id),
            "status": batch.status,
            "outbox_rows": total,
            "delivered": delivered,
        })
    if reaped:
        await db.commit()
        logger.warning(
            "hr_leave.expiry_reminder: reaped %d stale batch(es): %s",
            len(reaped), reaped,
        )
    return {"reaped": len(reaped), "batches": reaped}


async def run_batch_in_background(batch_id: uuid.UUID) -> None:
    """后台入口:自己开 session,绝不复用请求的那一个。"""
    from app.core.database import async_session_factory

    async with async_session_factory() as db:
        # savepoint-exempt: 这个 try 拥有**自己的** session(上面刚开的),吞掉
        # 异常不会污染任何调用方的事务;而且 except 分支第一件事就是 rollback,
        # 再把批次标成 failed —— 这条补偿路径本身就是这个 try 存在的理由
        # (不标就永远停在 ``sending``,in-flight 闸会挡住下一次发送半小时)。
        try:
            await run_batch(db, batch_id)
        except Exception:  # noqa: BLE001
            logger.exception(
                "hr_leave.expiry_reminder: background drain for %s failed",
                batch_id,
            )
            # 最后一道补偿。这里再失败就真的无事可做了 —— 只能记日志:批次会停在
            # ``sending``,但 outbox 重试 worker 仍会把已入队的行投出去,而
            # in-flight 闸有 30 分钟上限,不会永久卡住 HR。
            # savepoint-exempt: 跑在本函数自己开的 session 上(不碰任何调用方的
            # 事务),且第一句就是 rollback,进来时事务已经是干净的。
            try:
                await db.rollback()
                batch = await db.get(
                    HrLeaveReminderBatch, batch_id, populate_existing=True,
                )
                if batch is not None and batch.status == BATCH_STATUS_SENDING:
                    batch.status = BATCH_STATUS_FAILED
                    batch.error_summary = (
                        "The send task crashed; queued rows (if any) are still "
                        "retried by the notification outbox worker."
                    )
                    batch.finished_at = datetime.now(UTC)
                    await db.commit()
            except Exception:  # noqa: BLE001
                logger.exception(
                    "hr_leave.expiry_reminder: could not mark batch %s failed",
                    batch_id,
                )


# ── 码表 / 批次读取 ─────────────────────────────────────────────────────


async def list_account_type_options(
    db: AsyncSession,
) -> tuple[list[AccountTypeOption], list[int]]:
    """从真实数据里长出来的假期类型码表 + 可选财年。

    硬编码一份选项就等于要求「租户换了 SF 码表就改代码」;而 HR 需要看到的
    恰恰是「我们库里到底有哪些类型、各有多少人」。
    """
    from app.domains.hr.leave_insights_service import (
        classify_account_type,
        forfeits_at_period_end,
        load_type_overrides,
    )

    overrides = await load_type_overrides(db)
    rows = (await db.execute(
        select(
            HrLeaveAccount.account_type,
            HrLeaveAccount.account_type_label,
            HrLeaveAccount.fiscal_year,
            func.count(HrLeaveAccount.id),
            func.count(func.distinct(HrLeaveAccount.person_id)),
        ).where(
            HrLeaveAccount.deleted_at.is_(None),
        ).group_by(
            HrLeaveAccount.account_type,
            HrLeaveAccount.account_type_label,
            HrLeaveAccount.fiscal_year,
        )
    )).all()

    merged: dict[str, dict[str, Any]] = {}
    years: set[int] = set()
    for code, label, fy, acct_count, person_count in rows:
        entry = merged.setdefault(code, {
            "label": None, "accounts": 0, "persons": 0, "years": set(),
        })
        if label and not entry["label"]:
            entry["label"] = label
        entry["accounts"] += int(acct_count or 0)
        # 不同财年的 distinct person 数不能直接相加(同一个人跨年会重复计),
        # 取最大值作为「这个类型覆盖多少人」的下界,并在字段名上如实说明。
        entry["persons"] = max(entry["persons"], int(person_count or 0))
        if fy:
            entry["years"].add(int(fy))
            years.add(int(fy))

    options = [
        AccountTypeOption(
            code=code,
            label=data["label"],
            classification=classify_account_type(code, overrides),
            forfeits=forfeits_at_period_end(code, overrides),
            account_count=data["accounts"],
            person_count=data["persons"],
            fiscal_years=sorted(data["years"], reverse=True),
        )
        for code, data in merged.items()
    ]
    # 会清零的排前面 —— 那才是这个页面存在的理由。
    options.sort(key=lambda o: (not o.forfeits, o.code))
    return options, sorted(years, reverse=True)


def _active_person_ids_subquery():
    """在职人员 id 的子查询。

    ``hr_leave_accounts.person_id`` 是 ``String(36)``(无 FK,TiDB 约定),
    ``identity_persons.id`` 是 GUID —— 两者在 MySQL/TiDB 上都是 CHAR(36),
    所以 ``IN (subquery)`` 是直接可比的,不需要把 500 个 id 拉到 Python 再拼
    进一条巨大的 IN 列表。
    """
    return select(IdentityPerson.id).where(
        IdentityPerson.employment_status == "active",
        IdentityPerson.deleted_at.is_(None),
    )


async def count_active_and_covered(db: AsyncSession) -> tuple[int, int]:
    active = int((await db.execute(
        select(func.count()).select_from(IdentityPerson).where(
            IdentityPerson.employment_status == "active",
            IdentityPerson.deleted_at.is_(None),
        )
    )).scalar() or 0)
    covered = int((await db.execute(
        select(func.count(func.distinct(HrLeaveAccount.person_id))).where(
            HrLeaveAccount.deleted_at.is_(None),
            HrLeaveAccount.person_id.in_(_active_person_ids_subquery()),
        )
    )).scalar() or 0)
    return active, covered


async def preview(db: AsyncSession, filters: ReminderFilters) -> PreviewResponse:
    cohort = await resolve_cohort(db, filters)
    return PreviewResponse(summary=cohort.summary, candidates=cohort.candidates)


__all__ = [
    "ALLOWED_PLACEHOLDERS",
    "AUTO_PLACEHOLDERS",
    "DEFAULT_HRIS_URL",
    "EDITABLE_COPY_FIELDS",
    "MAX_RECIPIENTS_PER_BATCH",
    "MAX_TEST_RECIPIENTS",
    "MY_LEAVE_PATH",
    "SETTING_DEFAULTS",
    "SETTING_DEFAULT_CHANNELS",
    "SETTING_HRIS_URL",
    "SETTING_RESEND_WINDOW",
    "SETTING_SIGNOFF",
    "TEMPLATE_KEY",
    "TEMPLATE_SETTING_KEY",
    "BatchTally",
    "apply_tally",
    "assert_no_batch_in_flight",
    "batch_delivery_tally",
    "build_card_actions",
    "build_card_stats",
    "build_context",
    "builtin_copy",
    "count_active_and_covered",
    "create_batch",
    "default_filters",
    "default_signoff",
    "effective_copy",
    "extract_placeholders",
    "fiscal_label",
    "fmt_days",
    "get_default_channels",
    "get_default_filters",
    "get_hris_url",
    "get_resend_window_days",
    "get_signoff",
    "list_exclusions",
    "load_exclusions",
    "list_account_type_options",
    "pick_signoff",
    "preview",
    "reap_stale_batches",
    "remove_exclusion",
    "render_message",
    "resolve_cohort",
    "run_batch",
    "run_batch_in_background",
    "set_exclusion",
    "signoff_text",
    "today_sh",
    "validate_copy",
    "validate_hris_url",
    "validate_signoff",
]