"""邮箱名对账 —— HR 在 SF 指定的邮箱名 vs 员工实际的企业邮箱。

这是 ``hr.email_name_reconcile`` 定时任务的实现,也是「以后不准再有」那道网。

**它兜的是什么**

`identity_persons.hris_email_name` 是 HR 在 SF `PerPersonal.customString4`
指定的邮箱名(migration 582),由 `HrSyncService._upsert_person` 每次 apply 都写。
主修复在 `account_provisioner`:建 Google 号那一刻优先按这一列取 local-part,
所以只要 HR 在建号前填了,地址就一定对。

但主修复只在**建号那一条路径**上成立。这个对账不假设那条路径永远被走到:

* **还没建号 + 地址不符** —— 自动重跑 `reserve_work_email_for`。既纠正 Helix 侧
  的预告地址,也顺带把 SF PerEmail 回写成新值(reserve 自己就会写回)。这是
  P10026 那个形状——HR 在 create 事件之后才填——在 Day-1 之前的自愈路径。
* **已经建号 + 地址不符** —— 建号之后 Helix 没有任何自动改名能力(Google 改名要
  `Directory API users.update(primaryEmail)`,还牵动飞书 / SF / 各处 identity 行),
  所以只能推卡给 global_admin,把「谁、现在是什么、HR 想要什么」摆在人面前。
  **但只有「真的是另一个名字」才推**(P01574,2026-09-07):标点、词序、去重后缀
  这三类差异都是同一个名字的不同写法,活邮箱不能为它改名,落在 run summary 的
  `spelling_variants` 桶里(看得见、不进任何人的收件箱)。判据见 `_spelling_variant`。

**上线以来这个告警在生产上只产出过一条,而且是误报**:HR 把姓名 `"Koji Betsuyaku"`
填进了邮箱名字段,而员工的 `koji.betsuyaku@` 本来就是对的 —— 老代码把空格和其它
非字母一起删掉,读成 `kojibetsuyaku`,于是每天叫 IT 去把一个正确地址「改成」
一个没人想要的地址。三层一起修的:① 分隔符语义(`_local_part_from_hris_email`
把空格/下划线折成点,3 段以上的显示名直接弃用回落到姓名派生);② 这里的
`_spelling_variant`;③ 卡片打印真正会铸出来的地址而不是 HR 的原始字符串。
SF 里那个脏值本身由 `hris_email_name_malformed` 数据质量发现交给 HR 清,
不再由 IT 顶着。

**为什么必须有这一层**:P10026 之前,`custom07` 的 `$select`(#2020)和改读
`customString4`(#2042)都修过「读不到邮箱名」,每次都只补了当时那一条读取路径。
真正让员工在 Day-1 收到错地址的是「读到了但没人用」,而**没有任何东西在检查
最终结果对不对**。一个每天跑的、只看「HR 想要的 vs 实际是什么」的对账,不依赖
任何一条具体路径正确,所以下一个我们还没想到的形状也会在当天被看见。

对账**只读 Helix 自己的两列**,不打 SF、不打 Google —— 因此便宜、无外部依赖,
可以每天跑而不用担心配额或租户不可达。
"""
from __future__ import annotations

import logging
import re
from datetime import UTC, datetime, timedelta
from typing import Any

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

from app.auth.models import AuthRole, AuthUserRole
from app.identity.person_models import IdentityExternalIdentity, IdentityPerson
from app.notifications.service import (
    NotificationEnvelope,
    NotificationService,
    deliver_enqueued,
)

logger = logging.getLogger(__name__)

# Only look at people hired within this window. A long-tenured employee whose
# ``hris_email_name`` disagrees with a mailbox they have used for years is NOT
# something to rename — HR filling the field late on an old record must not
# page IT about an address the person已经用了几年. 180 days comfortably covers
# "HR filled it after the create event" (P10026 took 9) plus any late Day-1.
_DEFAULT_LOOKBACK_DAYS = 180

_ALERT_TEMPLATE = "hr.email_name_mismatch"
_MAX_LISTED = 20


def _local_part(email: str | None) -> str:
    return (email or "").strip().lower().partition("@")[0]


def _domain_part(email: str | None) -> str:
    return (email or "").strip().lower().partition("@")[2]


def _squash(local: str | None) -> str:
    """Letters + digits only — the identity of a NAME, free of spelling."""
    return re.sub(r"[^a-z0-9]", "", (local or "").lower())


def _spelling_variant(wanted: str, actual: str) -> str | None:
    """Reason ``actual`` is the same *name* as ``wanted``, spelled otherwise.

    Returns ``None`` only when the two are genuinely **different names** —
    which is the one shape worth a human's time (P10026: HR wants
    ``nick.muniz``, the mailbox says ``nicholas.muniz``).

    Two shapes are suppressed, and deliberately only two (P01574,
    2026-09-07). The single production finding this alert ever produced was
    HR typing ``"Koji Betsuyaku"`` against a correct ``koji.betsuyaku@``
    mailbox, and the card asked IT to rename it to ``kojibetsuyaku@``:

    * **same letters, different punctuation** — the strings differ only in
      dots/separators, so a rename buys nothing and risks a working address.
    * **collision-dedup suffix** — the plain address belongs to somebody
      else, so HR's value can never match and no rename is even possible.

    Both land in the run summary's ``spelling_variants`` bucket instead —
    visible in the job's run record for anyone who looks, and nobody's inbox.

    **Word order is NOT suppressed on purpose.** ``chen.wei`` against HR's
    ``wei.chen`` is a genuinely different address for the same person, and
    in the 名.姓 convention one of the two is simply wrong — that is a
    decision for a human, not something to hide. Suppressing it would have
    been a widening beyond the one shape production actually showed.
    """
    if _squash(wanted) == _squash(actual):
        return "same letters, different punctuation"
    squashed_actual = _squash(actual)
    if (
        len(squashed_actual) > 2
        and squashed_actual[-2:].isdigit()
        and squashed_actual[:-2] == _squash(wanted)
    ):
        # ``wei.chen01`` — the plain address was already taken by somebody
        # else, so the dedup loop suffixed it. HR's value can never match
        # and no rename is possible; the suffix IS the correct address.
        return f"collision-dedup suffix {squashed_actual[-2:]}"
    return None


class EmailNameReconcileService:
    """Reconcile ``hris_email_name`` against the live ``work_email``."""

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

    async def run(self, config: dict[str, Any] | None = None) -> dict[str, Any]:
        cfg = config or {}
        dry_run = bool(cfg.get("dry_run", False))
        lookback = int(cfg.get("lookback_days") or _DEFAULT_LOOKBACK_DAYS)
        cutoff = (datetime.now(UTC) - timedelta(days=lookback)).date()

        # Compare against what the provisioner would actually MINT, not
        # against the raw string HR typed. The two differ whenever the name
        # needs sanitising (a space, an apostrophe, over-length), and a
        # predicate that disagrees with the thing it is checking never
        # converges: "nick muniz" would mint ``nickmuniz@`` and then read as
        # a mismatch again tomorrow, re-reserving the same address daily
        # forever. Same helper, same settings, same answer.
        from app.core.corporate_domains import (  # noqa: PLC0415
            get_corporate_email_domains,
        )
        from app.domains.google_workspace.account_provisioner import (  # noqa: PLC0415
            _PROVISIONING_SETTING,
            _local_part_from_hris_email,
        )
        from app.integrations.service import get_platform_setting  # noqa: PLC0415

        settings = await get_platform_setting(
            _PROVISIONING_SETTING, default={}, session=self.db,
        ) or {}
        max_local = int(
            (settings if isinstance(settings, dict) else {}).get(
                "local_part_max_length",
            ) or 30
        )
        corporate_domains = await get_corporate_email_domains(self.db)

        def _wanted_local(value: str | None) -> str:
            return _local_part_from_hris_email(
                value, max_length=max_local, corporate_domains=corporate_domains,
            )

        rows = (await self.db.execute(
            select(IdentityPerson).where(
                IdentityPerson.deleted_at.is_(None),
                IdentityPerson.employment_status == "active",
                IdentityPerson.hris_email_name.isnot(None),
                IdentityPerson.hris_email_name != "",
                IdentityPerson.hire_date.isnot(None),
                IdentityPerson.hire_date >= cutoff,
            )
        )).scalars().all()

        summary: dict[str, Any] = {
            "scanned": len(rows),
            "matched": 0,
            "re_reserved": [],
            "needs_manual_rename": [],
            "spelling_variants": [],
            "skipped": [],
            "errors": [],
            "dry_run": dry_run,
        }
        needs_alert: list[tuple[IdentityPerson, str]] = []

        for person in rows:
            wanted = _wanted_local(person.hris_email_name)
            actual = _local_part(person.work_email)
            if not wanted or wanted == actual:
                summary["matched"] += 1
                continue

            # A live Google identity means the mailbox physically exists;
            # only a human-driven rename can fix it from here.
            has_account = await self.db.scalar(
                select(IdentityExternalIdentity.id).where(
                    IdentityExternalIdentity.person_id == person.id,
                    IdentityExternalIdentity.provider == "google",
                    IdentityExternalIdentity.identity_status == "active",
                    IdentityExternalIdentity.deleted_at.is_(None),
                ).limit(1)
            )
            entry = {
                "person_id": str(person.id),
                "hris_person_id": person.hris_person_id,
                "display_name": person.display_name,
                "work_email": person.work_email,
                "hris_email_name": person.hris_email_name,
            }
            if has_account is not None:
                # The mailbox is live. Only a genuinely DIFFERENT name is
                # worth a rename — see ``_spelling_variant``.
                variant = _spelling_variant(wanted, actual)
                if variant is not None:
                    summary["spelling_variants"].append({
                        **entry, "would_mint": wanted, "reason": variant,
                    })
                    logger.info(
                        "hr.email_name_reconcile: %s keeps %s — HR's %r is the "
                        "same name spelled differently (%s); no rename",
                        person.hris_person_id, person.work_email,
                        person.hris_email_name, variant,
                    )
                    continue
                domain = _domain_part(person.work_email)
                summary["needs_manual_rename"].append(entry)
                needs_alert.append(
                    (person, f"{wanted}@{domain}" if domain else wanted),
                )
                continue

            if dry_run:
                summary["re_reserved"].append({**entry, "status": "dry_run"})
                continue
            # ``best_effort`` (a SAVEPOINT), not a bare try/except: the step
            # below WRITES (person.work_email + an SF write-back audit row).
            # A failed flush deactivates the whole transaction, so merely
            # swallowing the error would leave the session unusable — the
            # next person's re-reserve, and the handler's final commit, would
            # both raise anyway and take the sweep down with them. This is
            # the 2026-08-06 Zoom-seat bug class (PR #2571).
            from app.core.txn import best_effort  # noqa: PLC0415

            # Imported inside the function, not at module scope: hr_lifecycle
            # must not take a hard import edge on the google_workspace domain.
            from app.domains.google_workspace.account_provisioner import (  # noqa: PLC0415
                reserve_work_email_for,
            )
            async with best_effort(
                self.db,
                label=f"hr.email_name_reconcile: re-reserve {person.hris_person_id}",
                log=logger,
            ) as step:
                outcome = await reserve_work_email_for(
                    self.db, person.id,
                    trace_id=f"cron:hr.email_name_reconcile:{person.id}",
                    email_name=person.hris_email_name,
                )
                await self.db.refresh(person)
            if not step.ok:
                summary["errors"].append({
                    "person_id": str(person.id), "error": step.reason(),
                })
                continue
            if outcome.status != "ok":
                # ``skipped`` is NOT a fix, and it must not be filed as one.
                # ``reserve_work_email_for`` skips for several benign, PERMANENT
                # reasons (HR's P99XX test band, a purged tombstone,
                # ``auto_create_on_hire`` off, no domain for the person's role),
                # so the old code's unconditional ``re_reserved`` entry claimed
                # a repair that had not happened — every night, forever, in a
                # bucket named as if it had. Not ``errors`` either: those raise
                # the run red (see the handler), and a nightly red for HR's own
                # test record is its own kind of noise.
                summary["skipped"].append({
                    **entry,
                    "would_mint": wanted,
                    "status": outcome.status,
                    "reason": outcome.reason,
                })
                logger.info(
                    "hr.email_name_reconcile: %s NOT re-reserved (%s) — %s",
                    person.hris_person_id, outcome.status, outcome.reason,
                )
                continue
            summary["re_reserved"].append({
                **entry,
                "status": outcome.status,
                "reason": outcome.reason,
                "work_email_after": person.work_email,
            })
            logger.info(
                "hr.email_name_reconcile: re-reserved %s (%s) → %s [%s]",
                person.hris_person_id, entry["work_email"],
                person.work_email, outcome.status,
            )

        if needs_alert and not dry_run:
            await self._alert_admins(needs_alert)
        return summary

    async def _alert_admins(
        self, persons: list[tuple[IdentityPerson, str]],
    ) -> None:
        """Push the manual-rename list to every active global_admin.

        Takes ``(person, target_address)`` pairs — the target being the
        address the provisioner would actually MINT, not the raw string HR
        typed. Printing the raw value (the 2026-09-06 card) put
        ``→ Koji Betsuyaku@`` in front of IT: not even a legal address, so
        the recipient could not tell what the platform wanted done.

        Deliberately IT-only and deliberately NOT sent to the employee: the
        employee can do nothing about it, and the whole point of the alert is
        that a human has to drive a Google rename.
        """
        admin_ids = list((await self.db.execute(
            select(AuthUserRole.user_id)
            .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
            .where(AuthRole.name.in_(("global_admin", "it_admin")))
            .distinct()
        )).scalars().all())
        admins = []
        for aid in admin_ids:
            admin = await self.db.get(IdentityPerson, aid)
            if (
                admin is not None
                and admin.deleted_at is None
                and (admin.employment_status or "").lower() == "active"
                and (admin.work_email or "").strip()
            ):
                admins.append(admin)
        if not admins:
            logger.warning(
                "hr.email_name_reconcile: %d mailbox(es) need a manual rename "
                "but no active global_admin could be resolved to tell",
                len(persons),
            )
            return

        # Language-neutral detail block — the framing prose lives in the
        # three-language template, so the same block drops into a Japanese
        # admin's card without leaving a Chinese fragment behind.
        listed = "\n".join(
            f"  · {p.display_name or p.hris_person_id} — {p.work_email} "
            f"→ {target}   (SF: {p.hris_email_name!r})"
            for p, target in persons[:_MAX_LISTED]
        )
        if len(persons) > _MAX_LISTED:
            listed += f"\n  · ... (+{len(persons) - _MAX_LISTED} more)"

        notify = NotificationService(self.db)
        queued = []
        for admin in admins:
            env = NotificationEnvelope(
                recipient=admin,
                title=f"[邮箱名不一致] {len(persons)} 位员工的邮箱与 HR 指定的不符",
                body_text="(see template)",
                channels=["feishu", "email"],
                template_key=_ALERT_TEMPLATE,
                context={"count": len(persons), "person_list": listed},
                # 人员花名册（可搜索 + 带邮箱列）。刻意不是
                # ``/admin/hr/sync``（那是同步运行视图，查不了单个人），
                # 也不是 ``/admin/reports/people``（它只是重定向到
                # ``/admin/insights?tab=people`` 这张分析看板）。卡片正文
                # 已经把每个人列全了，这个按钮只是「过去把人翻出来」的顺手入口。
                url="/admin/users",
                related_resource_type="identity_person",
                related_resource_id=persons[0][0].id,
            )
            queued.append((await notify.enqueue(env), env))
        await deliver_enqueued(
            self.db, queued,
            trace_id="system:cron:hr.email_name_reconcile", log=logger,
        )