"""离职告知信 —— HRIS 审批一批完就推给本人。

**为什么需要一条新链路,而不是在离职流程里加一张卡。**

Helix 的 ``OffboardingProcess`` 建得太晚。SF 只在最后工作日过了之后才把
``User.status`` 翻成 ``'f'``,同步据此才 stage 出 offboard 事件 —— 生产 87 条
流程里 80 条的 ``created_at >= leaver_date``。而 ``dispatch_post_commit_actions``
的 future-leaver 闸又把「流程已启动」卡片推迟到最后工作日;#1739 只补了会话
切断的重放,**通知那一半从来没有被任何 T+0 流程重放过**。两件事叠起来的结果:
通知期(公告到最后工作日之间那几周,恰恰是唯一还来得及交接的窗口)几乎无触达:
生产 87 条流程里**只有 6 位**离职者在自己 LWD 之前收到过任何离职域通知,其中
**只有 2 位真的 delivered**(2026-08-31 复核;首版这里写的「零触达」是过头的,
已按 `notification_outbox` 逐行推翻)。真正为零的是 HR 的《离职流程指引》本身
—— 这封信此前不存在。

HR 想要的时点是「Global HRIS 离职审批批完、离职交接触发」的那一刻,比上面这
条链路早几周,所以它不可能挂在离职流程上 —— 那时流程还不存在。

**判据(2026-08-31 在两个生产 SF 租户实测,不是推断):**

审批一通过,SF 立刻把 ``EmpEmployment.endDate`` 写成最后工作日、并写一条带
离职原因(``EmpEmploymentTermination.customString85``)的行;``User.status``
仍然是 ``'t'``,要到最后工作日过了才翻 ``'f'``。所以

    status == 't'  +  endDate 在未来  +  存在离职原因行

正好等价于「离职审批已完成、尚未到最后工作日」。实测当天 CN 租户命中 6 人,
Helix 侧 ``employment_status`` 全是 ``active``、``termination_date`` 全是 NULL
—— 一个都不知道。

第三个条件不是冗余:在职实习生 9 人全部 ``endDate=None`` 且没有离职原因行,
说明 SF 并**不**把「实习期计划结束日」写进 ``endDate`` —— 这是把实习生纳入
推送范围的前提。反过来,命中 HRIS 连接器 ``excluded_termination_reason_codes``
(D03 调往中国数据中心 / D04 兼岗结束 / D07 人员分类转换 / D08 调往德国数据
中心)的人写了离职原因行却**没有离职**,必须排除 —— 与全量/增量同步用的是
同一份配置,口径不会漂。

**范围**:HR 2026-08-31 的口径是「先只推中国员工的正式员工与实习生」,所以
默认 ``china_only=True`` + ``employee_type_codes={1070, 1072}``。两者都从
scheduled job 的 ``config`` 读,HR 想扩到别的雇佣类型/国家时改配置即可,不必
改代码 —— 但默认值就是 HR 今天要的那一套。

**幂等**:``identity_persons.offboarding_notice_sent_at`` +
``offboarding_notice_last_day``。只有真的投递出去(至少一个渠道)才盖戳 ——
盖在投递失败上会把这封信永久静音;这条与 ``offboarding/escalation.py``
CR-314 的教训同源。最后工作日被 HR 改期时(且仅在改期时)重推一封带新日期
的信。
"""
from __future__ import annotations

import logging
from datetime import UTC, date, datetime, timedelta
from typing import Any
from uuid import UUID

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

from app.core.exceptions import format_exception_for_log
from app.core.working_hours import resolve_timezone
from app.identity.person_models import IdentityPerson
from app.identity.successfactors_client import SuccessFactorsClient
from app.identity.successfactors_sync import _parse_date as parse_sf_date
from app.notifications.service import (
    CardSection,
    CardStat,
    LocalizedText,
    NotificationEnvelope,
    NotificationService,
)

logger = logging.getLogger(__name__)

#: SF ``EmpJob.customString25`` 雇佣类型码 —— HR 这次要推的两类。
#: 1070 正式员工 / 1072 实习生。1071 EOR、1073 劳务顾问、1074 外包、
#: 1075 派遣暂不推送(HR 2026-08-31 明确「其他员工类型暂时不用推送」)。
EMPLOYEE_TYPE_REGULAR = "1070"
EMPLOYEE_TYPE_INTERN = "1072"
DEFAULT_EMPLOYEE_TYPE_CODES: tuple[str, ...] = (
    EMPLOYEE_TYPE_REGULAR,
    EMPLOYEE_TYPE_INTERN,
)

#: 往前看多远。SF 里最长的通知期是三个月量级;一年足够宽,又不至于把
#: 多年后的合同结束日(如果哪天出现)卷进来。
DEFAULT_LOOKAHEAD_DAYS = 365

#: 已经离开的人不再推送 —— 信里写的全是「请在最后工作日前完成」。
_SKIP_EMPLOYMENT_STATUSES = frozenset({"left"})

_TEMPLATE_BY_TYPE = {
    EMPLOYEE_TYPE_REGULAR: "offboarding.hr_notice_regular",
    EMPLOYEE_TYPE_INTERN: "offboarding.hr_notice_intern",
}

#: 「北京地区人事政策」那一段只推北京员工(HR 文档原文:仅北京员工推送)。
#: 判据用**工作城市**,不是 ``office_code`` —— 2651 是 bj-office 邮件组,
#: 生产里它还挂着大连 / 武汉 / 郑州 / Amsterdam / Santa Clara 的人,而档案
#: 调转、北京市工作居住证、党组织关系转接都是按劳动关系所在城市走的。
#: 反向也成立:生产里有一位 city=北京市 而 office_code 为空的在职员工。
_BEIJING_CITY_PREFIXES = ("北京", "Beijing")

_BEIJING_SECTION = LocalizedText(
    zh=(
        "\n8. 北京地区人事政策相关（可联系许增华）\n"
        "· 人事档案：请提供调档材料，人力协助调转；无接收单位可自行托管人才市场。\n"
        "· 北京市工作居住证：离职当日办理解绑注销，请 60 天内完成新单位变更挂靠，"
        "建议 30 天内提交，避免超期证件失效。\n"
        "· 党员员工：及时报备去向，7 个工作日内办理党组织关系转接。\n"
    ),
    en=(
        "\n8. Beijing-specific HR matters (contact 许增华 / Zenghua Xu)\n"
        "· Personnel file: please provide the transfer paperwork and HR will "
        "help move your file; without a receiving employer you may place it "
        "with a talent-service centre yourself.\n"
        "· Beijing work & residence permit: it is unbound and cancelled on "
        "your last working day. Re-register it under your new employer within "
        "60 days — we suggest submitting within 30 — or the permit lapses.\n"
        "· Party members: report your destination promptly and transfer your "
        "Party membership within 7 working days.\n"
    ),
    ja=(
        "\n8. 北京地区の人事関連事項（担当：许增华 / Zenghua Xu）\n"
        "· 人事档案（人事ファイル）：移管書類をご提出いただければ人事が移管を"
        "サポートします。受け入れ先がない場合は人材センターでの保管も可能です。\n"
        "· 北京市工作居住証：最終出社日に解除・抹消されます。60 日以内に新しい"
        "勤務先へ変更手続きを行ってください（30 日以内のご提出を推奨。期限超過で"
        "失効します）。\n"
        "· 党員の方：異動先を速やかに報告し、7 営業日以内に党組織関係の移管手続き"
        "を行ってください。\n"
    ),
)

_EMPTY_SECTION = LocalizedText(zh="", en="", ja="")

_CTA_URL = "/my/requests"


def _is_beijing(person: IdentityPerson) -> bool:
    """工作城市在北京。``city`` 优先,回落到 ``work_location_name``。"""
    for value in (person.city, person.work_location_name):
        text = (value or "").strip()
        if text and any(
            text.startswith(prefix) for prefix in _BEIJING_CITY_PREFIXES
        ):
            return True
    return False


def _local_today(person: IdentityPerson) -> date:
    try:
        tz = resolve_timezone(person)
    except Exception:  # noqa: BLE001 — 时区解析失败退回 UTC,不是致命错误
        tz = None
    now = datetime.now(UTC)
    return (now.astimezone(tz) if tz else now).date()


class PendingLeaverNoticeService:
    """把 HR 的《离职流程指引》推给刚被批准离职、尚未到最后工作日的人。"""

    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.notify = NotificationService(db)

    # ── SF 侧:找出「审批已完成、还没到最后工作日」的人 ───────────

    async def _collect_pending_terminations(
        self, *, lookahead_days: int,
    ) -> tuple[dict[str, date], dict[str, str], list[dict[str, Any]]]:
        """扫描每个已启用的 HRIS 实例。

        返回 ``(userId → 最后工作日, userId → 离职原因码, 每实例诊断)``。
        跨实例重复的人(CN/EU 镜像同一个人)按最后工作日**较早**的那条收敛,
        与 SF 里「谁先到期谁生效」一致;两个租户的 endDate 实测一致,这里只是
        不让顺序决定结果。

        单个实例失败不影响其他实例:诊断里记一条,由 handler 决定是否 raise。
        """
        from app.domains.hr.service import normalize_reason_codes
        from app.integrations.compat import list_hris_instances
        from app.integrations.drivers.successfactors import (
            build_auth_from_config,
        )

        by_user: dict[str, date] = {}
        reasons: dict[str, str] = {}
        diagnostics: list[dict[str, Any]] = []

        instances = await list_hris_instances(session=self.db)
        for row, config in instances:
            base_url = (config.get("base_url") or "").rstrip("/")
            if not base_url:
                diagnostics.append({
                    "instance": row.name, "error": "missing base_url",
                })
                continue
            excluded = normalize_reason_codes(
                (config or {}).get("excluded_termination_reason_codes")
            )
            try:
                client = SuccessFactorsClient(
                    base_url=base_url, auth=build_auth_from_config(config),
                )
                today = datetime.now(UTC).date()
                employments = await client.list_termination_employments(
                    today + timedelta(days=1),
                    today + timedelta(days=lookahead_days),
                )
                candidate_dates: dict[str, date] = {}
                for row_emp in employments:
                    user_id = (row_emp or {}).get("userId")
                    end_date = parse_sf_date(row_emp.get("endDate"))
                    if not user_id or end_date is None:
                        continue
                    prior = candidate_dates.get(user_id)
                    if prior is None or end_date < prior:
                        candidate_dates[user_id] = end_date
                if not candidate_dates:
                    diagnostics.append({
                        "instance": row.name, "future_end_dates": 0,
                    })
                    continue

                # SF 把 ``User.status`` 留在 't' 直到最后工作日过去。已经是
                # 'f' 的人走的是既有离职流程(那条链路会自己发卡片),这里
                # 只负责通知期内的人。
                users = await client.get_users(sorted(candidate_dates))
                still_active = {
                    u.get("userId")
                    for u in users
                    if str((u or {}).get("status") or "").strip().lower() == "t"
                }
                # 有离职原因行 = 离职确实被执行过(不是合同/实习计划结束日)。
                instance_reasons = (
                    await client.get_termination_reasons_for_users(
                        sorted(candidate_dates)
                    )
                )

                kept = 0
                for user_id, end_date in candidate_dates.items():
                    if user_id not in still_active:
                        continue
                    reason = (instance_reasons.get(user_id) or "").strip()
                    if not reason:
                        continue
                    if reason.upper() in excluded:
                        continue
                    prior = by_user.get(user_id)
                    if prior is None or end_date < prior:
                        by_user[user_id] = end_date
                        reasons[user_id] = reason
                    kept += 1
                diagnostics.append({
                    "instance": row.name,
                    "future_end_dates": len(candidate_dates),
                    "approved_pending": kept,
                })
            except Exception as exc:  # noqa: BLE001 — 逐实例降级
                logger.exception(
                    "pending_leaver_notice: instance %s failed", row.name,
                )
                diagnostics.append({
                    "instance": row.name,
                    "error": format_exception_for_log(exc, max_len=300),
                })
        return by_user, reasons, diagnostics

    # ── 主流程 ────────────────────────────────────────────────

    async def run(self, config: dict[str, Any] | None = None) -> dict[str, Any]:
        cfg = config or {}
        raw_types = cfg.get("employee_type_codes")
        type_codes = frozenset(
            str(code).strip()
            for code in (raw_types or DEFAULT_EMPLOYEE_TYPE_CODES)
            if str(code).strip()
        ) or frozenset(DEFAULT_EMPLOYEE_TYPE_CODES)
        china_only = bool(cfg.get("china_only", True))
        lookahead_days = int(cfg.get("lookahead_days") or DEFAULT_LOOKAHEAD_DAYS)
        dry_run = bool(cfg.get("dry_run", False))

        summary: dict[str, Any] = {
            "approved_pending": 0,
            "matched_persons": 0,
            "candidates": 0,
            "notified": 0,
            "skipped_out_of_scope": 0,
            "skipped_already_sent": 0,
            "dry_run": dry_run,
            "instances": [],
            "errors": [],
        }

        by_user, reasons, diagnostics = await self._collect_pending_terminations(
            lookahead_days=lookahead_days,
        )
        summary["instances"] = diagnostics
        summary["approved_pending"] = len(by_user)
        if not by_user:
            return summary

        rows = list((
            await self.db.execute(
                select(IdentityPerson).where(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.source_system == "successfactors",
                    IdentityPerson.hris_person_id.in_(sorted(by_user)),
                )
            )
        ).scalars().all())
        summary["matched_persons"] = len(rows)

        due: list[tuple[UUID, date]] = []
        for person in rows:
            last_day = by_user.get(person.hris_person_id or "")
            if last_day is None:
                continue
            if person.employment_status in _SKIP_EMPLOYMENT_STATUSES:
                summary["skipped_out_of_scope"] += 1
                continue
            if china_only and not person.is_china_employee:
                summary["skipped_out_of_scope"] += 1
                continue
            if (person.employee_type_code or "") not in type_codes:
                summary["skipped_out_of_scope"] += 1
                continue
            if not (person.work_email or "").strip():
                summary["skipped_out_of_scope"] += 1
                continue
            # 最后工作日已经过去(SF 还没翻 status)——这封信全是「请在最后
            # 工作日前完成」,过期发出去只是噪音,交给既有离职流程。
            if last_day < _local_today(person):
                summary["skipped_out_of_scope"] += 1
                continue
            if (
                person.offboarding_notice_sent_at is not None
                and person.offboarding_notice_last_day == last_day
            ):
                summary["skipped_already_sent"] += 1
                continue
            due.append((person.id, last_day))

        summary["candidates"] = len(due)
        if dry_run:
            return summary

        for person_id, last_day in due:
            # 重新读一遍:上面那轮筛选之后可能已经过了几秒到几分钟(SF 三次
            # 往返),而 ``_send_notice`` 会写这行的幂等戳。
            fresh = await self.db.get(IdentityPerson, person_id)
            if fresh is None or fresh.deleted_at is not None:
                continue
            if (
                fresh.offboarding_notice_sent_at is not None
                and fresh.offboarding_notice_last_day == last_day
            ):
                continue
            try:
                sent = await self._send_notice(
                    fresh,
                    last_day=last_day,
                    reason_code=reasons.get(fresh.hris_person_id or ""),
                )
                if sent:
                    await self.db.commit()
                    summary["notified"] += 1
                else:
                    await self.db.rollback()
                    summary["errors"].append({
                        "person_id": str(person_id),
                        "error": "no channel delivered",
                    })
            except Exception as exc:  # noqa: BLE001 — 逐人降级
                await self.db.rollback()
                logger.exception(
                    "pending_leaver_notice: dispatch failed for %s", person_id,
                )
                summary["errors"].append({
                    "person_id": str(person_id),
                    "error": format_exception_for_log(exc, max_len=500),
                })
        return summary

    # ── 单人投递 ──────────────────────────────────────────────

    async def _send_notice(
        self,
        person: IdentityPerson,
        *,
        last_day: date,
        reason_code: str | None,
    ) -> bool:
        """投递一封告知信。真的投出去(至少一个渠道)才盖幂等戳并返回 True。"""
        from app.audit.service import AuditService

        template_key = _TEMPLATE_BY_TYPE.get(
            person.employee_type_code or "", "offboarding.hr_notice_regular",
        )
        is_intern = person.employee_type_code == EMPLOYEE_TYPE_INTERN
        person_name = person.display_name or person.work_email
        last_day_str = last_day.isoformat()
        days_left = (last_day - _local_today(person)).days
        beijing_section = (
            _BEIJING_SECTION
            if (not is_intern and _is_beijing(person))
            else _EMPTY_SECTION
        )
        trace_id = "pending_leaver_notice"

        env = NotificationEnvelope(
            recipient=person,
            title="📄 您的离职审批已完成 — 离职交接指引",
            body_text=(
                f"亲爱的小伙伴，感谢一路相伴！您的离职审批已完成，"
                f"最后工作日为 [b]{last_day_str}[/b]。"
                f"请在最后工作日前完成工作交接、资产归还与账号权限处理。"
            ),
            channels=["feishu", "email"],
            template_key=template_key,
            context={
                "person_name": person_name,
                "last_working_day": last_day_str,
                "beijing_section": beijing_section,
            },
            url=_CTA_URL,
            card_subtitle=LocalizedText(
                zh=f"最后工作日：{last_day_str}（还有 {days_left} 天）",
                en=(
                    f"Last working day: {last_day_str} "
                    f"({days_left} day{'s' if days_left != 1 else ''} left)"
                ),
                ja=f"最終出社日：{last_day_str}（残り {days_left} 日）",
            ),
            card_stats=[
                CardStat(
                    label=LocalizedText(
                        zh="最后工作日", en="Last working day", ja="最終出社日",
                    ),
                    value=last_day_str,
                ),
                CardStat(
                    label=LocalizedText(zh="距今", en="Time left", ja="残り"),
                    value=LocalizedText(
                        zh=f"{days_left} 天",
                        en=(
                            f"{days_left} day"
                            if days_left == 1 else f"{days_left} days"
                        ),
                        ja=f"{days_left} 日",
                    ),
                ),
            ],
            card_sections=[
                CardSection(
                    title=LocalizedText(
                        zh="账号会在什么时候关",
                        en="When your accounts close",
                        ja="アカウント停止のタイミング",
                    ),
                    body=LocalizedText(
                        zh=(
                            "飞书、邮箱、业务系统等账号将于最后工作日 "
                            "[b]24:00[/b] 统一关停，并同时解绑手机号、身份证等"
                            "实名认证信息。请提前退出与公司及业务相关的群组；"
                            "如为群主 / 管理员，请务必提前完成身份转让。"
                        ),
                        en=(
                            "Feishu, email and business-system accounts all "
                            "close at [b]24:00[/b] on your last working day, "
                            "and the real-name details bound to them are "
                            "unbound at the same time. Please leave "
                            "company-related groups beforehand, and transfer "
                            "ownership of any group you own or administer."
                        ),
                        ja=(
                            "Feishu・メール・業務システムのアカウントは最終出社日の "
                            "[b]24:00[/b] に一斉停止され、実名認証情報も同時に"
                            "解除されます。関連グループからは事前にご退出いただき、"
                            "オーナー / 管理者の場合は権限の移譲をお願いします。"
                        ),
                    ),
                    tone="warn",
                ),
                CardSection(
                    title=LocalizedText(
                        zh="有问题找谁",
                        en="Who to ask",
                        ja="お問い合わせ先",
                    ),
                    body=LocalizedText(
                        zh=(
                            "离职流程相关问题请咨询 HR Helpdesk；"
                            "背调需求可发送至 hr@pingcap.com 或联系 HRBP；"
                            "资产归还与回购的操作详情可在 IT Help Desk 查询。"
                        ),
                        en=(
                            "For questions about the offboarding process, "
                            "contact the HR Helpdesk; for background-check "
                            "requests write to hr@pingcap.com or your HRBP; "
                            "the IT Help Desk has the step-by-step guide for "
                            "returning or buying back assets."
                        ),
                        ja=(
                            "退職手続きについては HR Helpdesk へ、"
                            "リファレンスチェックのご依頼は hr@pingcap.com "
                            "または HRBP へ、資産の返却・買い取り手順は "
                            "IT Help Desk でご確認ください。"
                        ),
                    ),
                    tone="info",
                ),
            ],
            related_resource_type="identity_person",
            related_resource_id=person.id,
        )
        result = await self.notify.dispatch(env, trace_id=trace_id)
        if not result.delivered_channels:
            # 一个渠道都没投出去时**不**盖戳 —— 盖了就是把这封信永久静音,
            # 下一轮再也不会重试(与 offboarding/escalation.py CR-314 同源)。
            logger.warning(
                "pending_leaver_notice: no channel delivered for %s — "
                "leaving the idempotency stamp unset so the next tick retries",
                person.work_email,
            )
            return False

        person.offboarding_notice_sent_at = datetime.now(UTC)
        person.offboarding_notice_last_day = last_day
        await self.db.flush()
        await AuditService(self.db).log(
            trace_id=trace_id,
            actor_type="system",
            actor_user_id=None,
            action="offboarding.hr_notice_sent",
            status="success",
            resource_type="identity_person",
            resource_id=str(person.id),
            detail={
                "work_email": person.work_email,
                "hris_person_id": person.hris_person_id,
                "last_working_day": last_day.isoformat(),
                "employee_type_code": person.employee_type_code,
                "termination_reason_code": reason_code,
                "template_key": template_key,
                "beijing_section": beijing_section is _BEIJING_SECTION,
                "delivered": result.delivered_channels,
            },
        )
        return True