"""离职 HR 告知信:`identity_persons` 两列幂等台账 + `hr.pending_leaver_notice` 任务行。

**要解决的事**:HR 的《离职流程指引》要求「Global HRIS 离职审批批完、离职交接触发
的那一刻」就把交接指引推给本人。今天这件事在 Helix 里**根本没有发生过** ——
`OffboardingProcess` 是 SF 把 `User.status` 翻成 `'f'` 之后才建的(生产 87 条流程里
80 条的 `created_at >= leaver_date`),而 `dispatch_post_commit_actions` 的
future-leaver 闸把「流程已启动」卡片一并推迟到最后工作日;#1739 只补了会话切断的
重放,通知那一半从来没被重放过。结果:通知期内几乎无触达 —— 87 条流程里只有 6 位
离职者在 LWD 之前收到过任何离职域通知,其中只有 2 位真的 delivered(2026-08-31
复核更正:本段首版写的「零触达」过头了)。

**新信号**(2026-08-31 在两个生产 SF 租户实测):离职审批一批完,SF 就把
`EmpEmployment.endDate` 写成最后工作日、并写一条带离职原因的
`EmpEmploymentTermination` 行,而 `User.status` 要到最后工作日过了才翻 `'f'`。
所以「`status='t'` + `endDate` 在未来 + 有离职原因行」正好等价于「审批已完成、
尚未到最后工作日」。实测当天 CN 租户 6 人命中,其中 5 位是中国正式员工,Helix 侧
`employment_status` 全是 `active`、`termination_date` 全是 NULL —— 一个都不知道。
在职实习生 9 人全部 `endDate=None` 且无离职原因行,所以「实习期结束日」不会被
误当成离职(这条是把实习生纳入推送范围的前提)。

两列做幂等台账,挂在人身上(与 `it_info_sent_at` / `day90_survey_sent_at` 同构):

- `offboarding_notice_sent_at` —— 推送成功(至少一个渠道真的投递)的时刻;
- `offboarding_notice_last_day` —— 那次推送里写的最后工作日。HR 改期后这一列与
  SF 不一致,于是重推一次带新日期的信;日期没变则永不重复打扰。

任务排 `0 9-19 * * *`(Asia/Shanghai):审批都发生在工作时间,按小时轮询足够贴近
「立即」,同时保证不会在凌晨给人发飞书 DM。

Revision ID: 581
Revises: 580
"""
from __future__ import annotations

import uuid

import sqlalchemy as sa

from alembic import op

revision = "581"
down_revision = "580"
branch_labels = None
depends_on = None

TABLE = "identity_persons"

COLUMNS = (
    ("offboarding_notice_sent_at", "DATETIME NULL"),
    ("offboarding_notice_last_day", "DATE NULL"),
)

_KEY = "hr.pending_leaver_notice"
_DISPLAY = "离职告知信推送(HRIS 审批完成即推)"
_DESCRIPTION = (
    "SF 离职审批一批完(EmpEmployment.endDate 落到未来 + 有 "
    "EmpEmploymentTermination 离职原因行,而 User.status 仍是 't'),就把 HR 的"
    "《离职流程指引》推给本人(飞书卡片 + 邮件,三语)。\n"
    "默认范围:中国员工(is_china_employee)且雇佣类型为正式员工(1070)/ 实习生"
    "(1072);其余类型暂不推送。命中 HRIS 连接器"
    "`excluded_termination_reason_codes`(人员分类转换 / 跨 DC 调动 / 兼岗结束)"
    "的不推 —— 那些人并没有离职。\n"
    "幂等键 identity_persons.offboarding_notice_sent_at + "
    "offboarding_notice_last_day:最后工作日被 HR 改期才会重推一次。\n"
    'config 可覆写:{"employee_type_codes":["1070","1072"],"china_only":true,'
    '"lookahead_days":365,"dry_run":false}。'
)
# 只在 09:00–19:00(Asia/Shanghai)整点跑:审批发生在工作时间,按小时轮询
# 足够贴近「立即」,又不会半夜给人推离职卡片。
_CRON = "0 9-19 * * *"


def _has_column(conn, table: str, column: str) -> bool:
    return conn.execute(
        sa.text(
            "SELECT 1 FROM information_schema.columns "
            "WHERE table_schema = DATABASE() AND table_name = :t "
            "AND column_name = :c"
        ),
        {"t": table, "c": column},
    ).scalar() is not None


def upgrade() -> None:
    conn = op.get_bind()
    if sa.inspect(conn).has_table(TABLE):
        for name, ddl in COLUMNS:
            if not _has_column(conn, TABLE, name):
                op.execute(f"ALTER TABLE `{TABLE}` ADD COLUMN `{name}` {ddl}")
    conn.execute(
        sa.text(
            "INSERT INTO scheduled_jobs "
            "(id, name, handler_key, display_name, description, "
            " trigger_type, cron_expr, timezone, enabled, config, "
            " created_at, updated_at) "
            "SELECT :id, :name, :k, :dn, :desc, 'cron', :cron, "
            " 'Asia/Shanghai', 1, '{}', NOW(), NOW() "
            "FROM DUAL WHERE NOT EXISTS ("
            "  SELECT 1 FROM scheduled_jobs WHERE handler_key = :k)"
        ),
        {
            "id": str(uuid.uuid4()),
            "name": _KEY,
            "k": _KEY,
            "dn": _DISPLAY,
            "desc": _DESCRIPTION,
            "cron": _CRON,
        },
    )


def downgrade() -> None:
    conn = op.get_bind()
    conn.execute(
        sa.text("DELETE FROM scheduled_jobs WHERE handler_key = :k"),
        {"k": _KEY},
    )
    if sa.inspect(conn).has_table(TABLE):
        for name, _ddl in COLUMNS:
            if _has_column(conn, TABLE, name):
                op.execute(f"ALTER TABLE `{TABLE}` DROP COLUMN `{name}`")