"""DynamicEmailGroupService — HR-facing CRUD for Google Cloud Identity
**dynamic groups** (membership computed from a rule over synced
department / office / country attributes).

Why this is its own service (not folded into EmailGroupService):
dynamic groups have no per-group owner and no Feishu approval — HR edits
take effect immediately (2026-07-22 decision) and are fully audited. The
write path talks to the Cloud Identity API (create / patch query / patch
display) rather than the Directory member API used for static groups.

Every mutation:
  1. Validates + compiles the structured rule to CEL (fail fast).
  2. For rule edits, previews the membership delta and refuses a
     large drop unless the caller explicitly confirms.
  3. Calls Google; on failure raises (no local mutation).
  4. Updates the local ``email_groups`` mirror row, writes an audit row,
     and notifies IT admins (never employees).
"""
from __future__ import annotations

import hashlib
import json
import logging
from uuid import UUID

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload

from app.audit.service import AuditService
from app.core.exceptions import AppException, NotFoundError
from app.core.rbac import can_manage_dynamic_email_groups
from app.domains.email_group import dynamic_cel as cel
from app.domains.email_group.manage_service import real_aliases
from app.domains.email_group.models import EmailGroup
from app.domains.email_group.service import EmailGroupService
from app.identity.person_models import IdentityPerson
from app.notifications.service import NotificationEnvelope, NotificationService

logger = logging.getLogger(__name__)

# Human labels for the field catalog. The frontend maps the field key to
# an i18n key; this is the stable English fallback.
_FIELD_LABELS: dict[str, str] = {
    "Department_Information.L1_Dept": "L1 Department",
    "Department_Information.L2_Dept": "L2 Department",
    "Department_Information.L3_Dept": "L3 Department",
    "Department_Information.L4_Dept": "L4 Department",
    "Employee_Details.Office": "Office",
    "Employee_Details.Country": "Country",
    "Employee_Details.JobFunction": "Job Function",
    "Employee_Details.JobFamily": "Job Family",
    "Employee_Details.Location": "Location",
}

# Applying a rule edit that removes more than this fraction of the
# current members requires explicit confirmation (the g-all incident:
# a silent 402→0 drop). 0.30 = 30%.
_LARGE_DROP_RATIO = 0.30


class DynamicEmailGroupService:
    def __init__(self, db: AsyncSession) -> None:
        self.db = db
        self.audit = AuditService(db)
        self.notify = NotificationService(db)
        self._eg = EmailGroupService(db)
        # Per-request cache of the Google user-attribute scan (list of
        # {email, suspended, fields}). ``False`` = scan attempted and
        # failed; ``None`` = not yet attempted.
        self._attr_cache: list[dict] | None | bool = None

    async def _get_driver(self):
        return await self._eg._get_driver()

    def _require_manage(self, actor_roles: list[str]) -> None:
        if not can_manage_dynamic_email_groups(actor_roles):
            raise AppException(
                code="NOT_AUTHORIZED",
                message="只有 HR / IT Admin 才能管理动态邮件组。",
                status_code=403,
            )

    # ─── Attribute scan (value options + expected-membership eval) ──

    async def _scan_attributes(self, driver) -> list[dict] | None:
        """Return the cached Google user-attribute scan (or None on
        failure). Cached for the life of the service instance so a single
        request that needs both value-counts and rule evaluation makes
        one Google call.
        """
        if self._attr_cache is None:
            scanned = await driver.scan_user_attributes()
            self._attr_cache = scanned if scanned is not None else False
        return None if self._attr_cache is False else self._attr_cache  # type: ignore[return-value]

    @staticmethod
    def _value_counts(users: list[dict]) -> dict[str, dict[str, int]]:
        counts: dict[str, dict[str, int]] = {}
        for u in users:
            for fk, fv in (u.get("fields") or {}).items():
                counts.setdefault(fk, {})[fv] = counts.setdefault(fk, {}).get(fv, 0) + 1
        return counts

    @staticmethod
    def _snapshots(users: list[dict]) -> list[cel.UserSnapshot]:
        return [
            cel.UserSnapshot(
                email=u["email"],
                fields=dict(u.get("fields") or {}),
                suspended=bool(u.get("suspended")),
            )
            for u in users
        ]

    # ─── Rule (de)serialization on the local mirror row ────────────

    @staticmethod
    def _rule_from_row(row: EmailGroup) -> cel.StructuredRule:
        """Reconstruct the structured rule from the stored JSON, falling
        back to parsing the raw ``membership_query`` if the JSON is
        absent (e.g. a row synced before this feature).
        """
        data = row.dynamic_rule
        if isinstance(data, dict):
            return cel.StructuredRule(
                field=data.get("field"),
                match_values=list(data.get("match_values") or []),
                include_emails=list(data.get("include_emails") or []),
                negate=bool(data.get("negate")),
                parseable=bool(data.get("parseable", True)),
                raw_query=data.get("raw_query") or (row.membership_query or ""),
            )
        return cel.parse_membership_query(row.membership_query)

    def _health(
        self,
        rule: cel.StructuredRule,
        value_counts: dict[str, dict[str, int]] | None,
    ) -> tuple[str, list[str]]:
        if not rule.parseable:
            return "google_managed", []
        if value_counts is None:
            return "ok", []
        dead = cel.dead_values(rule, value_counts)
        return ("dead_value" if dead else "ok"), dead

    # ─── Reads ─────────────────────────────────────────────────────

    async def list_groups(self, *, actor_roles: list[str]) -> dict:
        self._require_manage(actor_roles)
        rows = (
            await self.db.execute(
                select(EmailGroup)
                .where(
                    EmailGroup.is_dynamic.is_(True),
                    EmailGroup.status == "active",
                )
                .order_by(EmailGroup.email)
            )
        ).scalars().all()

        value_counts: dict[str, dict[str, int]] | None = None
        attrs_available = True
        try:
            driver = await self._get_driver()
            users = await self._scan_attributes(driver)
            if users is None:
                attrs_available = False
            else:
                value_counts = self._value_counts(users)
        except AppException:
            # Google not configured — still list rows from the mirror.
            attrs_available = False

        items = []
        for row in rows:
            rule = self._rule_from_row(row)
            status, dead = self._health(rule, value_counts)
            items.append(
                {
                    "id": row.id,
                    "email": row.email,
                    "name": row.name,
                    "description": row.description,
                    "direct_members_count": row.direct_members_count,
                    "rule": rule.to_dict() if rule.parseable else None,
                    "parseable": rule.parseable,
                    "raw_query": row.membership_query,
                    "alert_on_missing_values": bool(row.alert_on_missing_values),
                    "health": {"status": status, "dead_values": dead},
                    "aliases": real_aliases(row.aliases),
                    # Tri-state: NULL means the alias list has never
                    # been read, which is NOT "this group has none".
                    "alias_never_synced": row.aliases is None,
                    "last_synced_at": row.last_synced_at,
                }
            )
        return {"groups": items, "attributes_available": attrs_available}

    async def _load_row(self, group_id: UUID) -> EmailGroup:
        row = (
            await self.db.execute(
                select(EmailGroup)
                .where(EmailGroup.id == group_id)
                .options(selectinload(EmailGroup.owners))
            )
        ).scalar_one_or_none()
        if row is None or row.status != "active":
            raise NotFoundError("EmailGroup", str(group_id))
        if row.is_dynamic is not True:
            raise AppException(
                code="NOT_A_DYNAMIC_GROUP",
                message=f"邮件组 {row.email} 不是动态组。",
                status_code=409,
            )
        return row

    async def get_detail(self, *, group_id: UUID, actor_roles: list[str]) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        rule = self._rule_from_row(row)

        value_counts: dict[str, dict[str, int]] | None = None
        attrs_available = True
        try:
            driver = await self._get_driver()
            users = await self._scan_attributes(driver)
            if users is None:
                attrs_available = False
            else:
                value_counts = self._value_counts(users)
        except AppException:
            attrs_available = False

        status, dead = self._health(rule, value_counts)
        field_catalog = [
            {
                "field": fk,
                "label": _FIELD_LABELS.get(fk, fk),
                "options": sorted(
                    (
                        {"value": v, "count": c}
                        for v, c in (value_counts.get(fk, {}) if value_counts else {}).items()
                    ),
                    key=lambda o: (-o["count"], o["value"]),
                ),
            }
            for fk in cel.KNOWN_FIELDS
        ]
        return {
            "id": row.id,
            "email": row.email,
            "name": row.name,
            "description": row.description,
            "direct_members_count": row.direct_members_count,
            "rule": rule.to_dict() if rule.parseable else None,
            "parseable": rule.parseable,
            "raw_query": row.membership_query,
            "alert_on_missing_values": bool(row.alert_on_missing_values),
            "health": {"status": status, "dead_values": dead},
            "aliases": real_aliases(row.aliases),
            "alias_never_synced": row.aliases is None,
            "field_catalog": field_catalog,
            "last_synced_at": row.last_synced_at,
            "attributes_available": attrs_available,
        }

    async def reconcile(self, *, group_id: UUID, actor_roles: list[str]) -> dict:
        """Expected (rule evaluated locally) vs actual (Google members)."""
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        rule = self._rule_from_row(row)
        driver = await self._get_driver()

        users = await self._scan_attributes(driver)
        actual_members = await driver.list_group_members(row.email)
        actual = {
            (m.get("email") or "").strip().lower()
            for m in actual_members
            if m.get("email")
        }

        delta = self._delta(rule, users, actual)
        return {
            "group_id": row.id,
            "group_email": row.email,
            "delta": delta,
        }

    def _delta(
        self,
        rule: cel.StructuredRule,
        users: list[dict] | None,
        actual: set[str],
    ) -> dict:
        if users is None or not rule.parseable:
            return {
                "expected_count": 0,
                "actual_count": len(actual),
                "added": [],
                "removed": [],
                "evaluable": False,
            }
        expected = cel.evaluate_rule(rule, self._snapshots(users)) or set()
        return {
            "expected_count": len(expected),
            "actual_count": len(actual),
            "added": sorted(expected - actual),
            "removed": sorted(actual - expected),
            "evaluable": True,
        }

    # ─── Members (view / export) ───────────────────────────────────

    async def list_members(
        self, *, group_id: UUID, actor_roles: list[str],
    ) -> dict:
        """List a dynamic group's **actual** Google members, enriched
        with Helix person attributes (name / employee_no / job /
        location / country / employment status).

        Read-only. Backs the members Excel export and the
        ``list_dynamic_email_group_members`` MCP tool. Members that don't
        match a known person (nested groups, external addresses) keep
        their email and are returned with blank person fields.
        """
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        driver = await self._get_driver()
        raw = await driver.list_group_members(row.email)

        emails = sorted({
            (m.get("email") or "").strip().lower()
            for m in raw if m.get("email")
        })
        persons: dict[str, IdentityPerson] = {}
        if emails:
            found = (
                await self.db.execute(
                    select(IdentityPerson).where(
                        func.lower(IdentityPerson.work_email).in_(emails)
                    )
                )
            ).scalars().all()
            for p in found:
                key = (p.work_email or "").strip().lower()
                if key:
                    persons[key] = p

        members: list[dict] = []
        for m in raw:
            email = (m.get("email") or "").strip().lower()
            if not email:
                continue
            person = persons.get(email)
            members.append({
                "email": email,
                "display_name": (person.display_name if person else None) or None,
                "employee_no": person.employee_no if person else None,
                "job_title": person.job_title if person else None,
                "job_function": person.job_function if person else None,
                "work_location_name": person.work_location_name if person else None,
                "country": person.country if person else None,
                "employment_status": person.employment_status if person else None,
                "role": m.get("role"),
                "member_type": m.get("type"),
                "google_status": m.get("status"),
            })
        # Owners first, then alphabetical by email — same ordering as the
        # static-group member list.
        members.sort(key=lambda m: (m["role"] != "OWNER", m["email"]))
        return {
            "group_id": str(row.id),
            "group_email": row.email,
            "group_name": row.name,
            "members": members,
            "total": len(members),
        }

    # ─── Preview (before create/update) ────────────────────────────

    async def preview_rule(
        self,
        *,
        group_id: UUID | None,
        rule_in,
        actor_roles: list[str],
    ) -> dict:
        self._require_manage(actor_roles)
        rule = self._build_rule(rule_in)
        compiled = self._compile(rule)
        driver = await self._get_driver()
        users = await self._scan_attributes(driver)

        actual: set[str] = set()
        if group_id is not None:
            row = await self._load_row(group_id)
            members = await driver.list_group_members(row.email)
            actual = {
                (m.get("email") or "").strip().lower()
                for m in members
                if m.get("email")
            }
        delta = self._delta(rule, users, actual)
        large_drop, drop_ratio = self._drop_guard(delta, actual)
        return {
            "compiled_query": compiled,
            "delta": delta,
            "large_drop": large_drop,
            "drop_ratio": drop_ratio,
        }

    @staticmethod
    def _drop_guard(delta: dict, actual: set[str]) -> tuple[bool, float]:
        current = len(actual)
        if current == 0 or not delta.get("evaluable"):
            return False, 0.0
        removed = len(delta.get("removed") or [])
        ratio = removed / current
        return (ratio > _LARGE_DROP_RATIO), round(ratio, 4)

    # ─── Rule build/compile helpers ────────────────────────────────

    @staticmethod
    def _build_rule(rule_in) -> cel.StructuredRule:
        """Coerce an inbound DynamicRuleModel (or dict) into a
        StructuredRule, marking it parseable (it's structured by
        construction).
        """
        field = getattr(rule_in, "field", None)
        match_values = list(getattr(rule_in, "match_values", []) or [])
        include_emails = list(getattr(rule_in, "include_emails", []) or [])
        negate = bool(getattr(rule_in, "negate", False))
        return cel.StructuredRule(
            field=field,
            match_values=match_values,
            include_emails=[e.strip().lower() for e in include_emails],
            negate=negate,
            parseable=True,
        )

    @staticmethod
    def _compile(rule: cel.StructuredRule) -> str:
        try:
            return cel.compile_rule(
                field=rule.field,
                match_values=rule.match_values,
                include_emails=rule.include_emails,
                negate=rule.negate,
            )
        except cel.RuleValidationError as e:
            raise AppException(
                code="INVALID_RULE",
                message=f"规则无效:{e}",
                status_code=422,
            ) from e

    # ─── Writes ────────────────────────────────────────────────────

    async def create_group(
        self,
        *,
        email: str,
        name: str,
        description: str | None,
        rule_in,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        email = email.strip().lower()
        rule = self._build_rule(rule_in)
        compiled = self._compile(rule)

        existing = (
            await self.db.execute(
                select(EmailGroup).where(func.lower(EmailGroup.email) == email)
            )
        ).scalar_one_or_none()
        if existing is not None and existing.status == "active":
            raise AppException(
                code="GROUP_EXISTS",
                message=f"邮件组 {email} 已存在。",
                status_code=409,
            )

        driver = await self._get_driver()
        resp = await driver.create_dynamic_group(
            email=email, name=name, description=description, query=compiled,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_CREATE_DYNAMIC_GROUP_FAILED",
                message=f"Google 创建动态组失败: {resp.error_message}",
                status_code=502,
            )

        rule.raw_query = compiled
        if existing is not None:
            # Revive a soft-deleted row.
            existing.status = "active"
            existing.name = name
            existing.description = description
            existing.is_dynamic = True
            existing.membership_query = compiled
            existing.dynamic_rule = rule.to_dict()
            existing.dynamic_dead_values = None
            existing.drift_signature = None
            row = existing
        else:
            row = EmailGroup(
                email=email,
                name=name,
                description=description,
                status="active",
                created_by=actor.id,
                is_dynamic=True,
                membership_query=compiled,
                dynamic_rule=rule.to_dict(),
            )
            self.db.add(row)
        await self.db.flush()

        await self._audit_and_notify(
            action="email_group.dynamic_created",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={
                "group_email": email,
                "name": name,
                "compiled_query": compiled,
            },
            admin_body=(
                f"HR/{actor.display_name or actor.work_email} 创建了动态邮件组 "
                f"{email}(规则:{compiled})。"
            ),
        )
        await self.db.commit()
        return await self.get_detail(group_id=row.id, actor_roles=actor_roles)

    async def update_rule(
        self,
        *,
        group_id: UUID,
        rule_in,
        confirm_large_drop: bool,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        old_query = row.membership_query
        rule = self._build_rule(rule_in)
        compiled = self._compile(rule)

        driver = await self._get_driver()
        # Drop guard.
        users = await self._scan_attributes(driver)
        members = await driver.list_group_members(row.email)
        actual = {
            (m.get("email") or "").strip().lower()
            for m in members if m.get("email")
        }
        delta = self._delta(rule, users, actual)
        large_drop, drop_ratio = self._drop_guard(delta, actual)
        if large_drop and not confirm_large_drop:
            # The caller has already previewed (which returned the full
            # delta + drop_ratio); this guard just refuses an unconfirmed
            # large drop. The counts are in the message for the audit
            # trail and any non-preview caller (e.g. MCP).
            raise AppException(
                code="LARGE_MEMBERSHIP_DROP",
                message=(
                    f"此改动会移除约 {len(delta['removed'])} 名成员"
                    f"(当前 {len(actual)} 人,降幅 {int(drop_ratio * 100)}%)。"
                    "请先在预览中确认,再以 confirm_large_drop=true 提交。"
                ),
                status_code=409,
            )

        resp = await driver.patch_dynamic_group_query(
            group_email=row.email, query=compiled,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_PATCH_DYNAMIC_GROUP_FAILED",
                message=f"Google 更新动态组规则失败: {resp.error_message}",
                status_code=502,
            )

        rule.raw_query = compiled
        row.membership_query = compiled
        row.dynamic_rule = rule.to_dict()
        # Recompute dead values immediately so the badge is fresh.
        if users is not None:
            row.dynamic_dead_values = cel.dead_values(
                rule, self._value_counts(users),
            )
        await self.db.flush()

        await self._audit_and_notify(
            action="email_group.dynamic_rule_updated",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={
                "group_email": row.email,
                "old_query": old_query,
                "new_query": compiled,
                "expected_delta": delta,
                "confirmed_large_drop": confirm_large_drop,
            },
            admin_body=(
                f"HR/{actor.display_name or actor.work_email} 修改了动态邮件组 "
                f"{row.email} 的成员规则。预计成员 {delta.get('actual_count')}"
                f" → {delta.get('expected_count')}。"
            ),
        )
        await self.db.commit()
        return await self.get_detail(group_id=row.id, actor_roles=actor_roles)

    async def update_display(
        self,
        *,
        group_id: UUID,
        name: str | None,
        description: str | None,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        if name is None and description is None:
            raise AppException(
                code="NOTHING_TO_UPDATE",
                message="没有需要更新的字段。",
                status_code=422,
            )
        driver = await self._get_driver()
        resp = await driver.patch_group_display(
            group_email=row.email,
            display_name=name,
            description=description,
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_PATCH_GROUP_DISPLAY_FAILED",
                message=f"Google 更新组名/描述失败: {resp.error_message}",
                status_code=502,
            )
        old_name, old_desc = row.name, row.description
        if name is not None:
            row.name = name
        if description is not None:
            row.description = description
        await self.db.flush()
        await self._audit_and_notify(
            action="email_group.dynamic_display_updated",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={
                "group_email": row.email,
                "old_name": old_name, "new_name": row.name,
                "old_description": old_desc, "new_description": row.description,
            },
            admin_body=None,  # display-only change — no admin fan-out
        )
        await self.db.commit()
        return await self.get_detail(group_id=row.id, actor_roles=actor_roles)

    async def add_alias(
        self,
        *,
        group_id: UUID,
        alias_email: str,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        alias_email = alias_email.strip().lower()
        driver = await self._get_driver()
        resp = await driver.add_alias(
            target_email=row.email, alias_email=alias_email, target_kind="group",
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_ADD_ALIAS_FAILED",
                message=f"Google 添加别名失败: {resp.error_message}",
                status_code=502,
            )
        # Mirror only when we HAVE a mirror. ``set(row.aliases or [])``
        # on a NULL column writes ``[alias]``, converting "Helix has
        # never read this group's aliases" into the positive claim "this
        # is the complete set" — and the cleanup workbench reads that
        # claim: a confirmed-short list suppresses the ``alias_unknown``
        # indeterminate signal and can put the row back into
        # ``risk: low``. Same reasoning, same shape, as
        # ``EmailGroupAdminService.add_alias``.
        if row.aliases is not None:
            aliases = set(row.aliases)
            aliases.add(alias_email)
            row.aliases = sorted(aliases)
        await self.db.flush()
        await self._audit_and_notify(
            action="email_group.dynamic_alias_added",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={"group_email": row.email, "alias_email": alias_email},
            admin_body=None,
        )
        await self.db.commit()
        return await self.get_detail(group_id=row.id, actor_roles=actor_roles)

    async def remove_alias(
        self,
        *,
        group_id: UUID,
        alias_email: str,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        alias_email = alias_email.strip().lower()
        driver = await self._get_driver()
        resp = await driver.remove_alias(
            target_email=row.email, alias_email=alias_email, target_kind="group",
        )
        if not resp.ok:
            raise AppException(
                code="GOOGLE_REMOVE_ALIAS_FAILED",
                message=f"Google 删除别名失败: {resp.error_message}",
                status_code=502,
            )
        # Same NULL preservation as the add path above, plus: a
        # Google 404 removed nothing, so stripping the address from the
        # mirror would hide an alias that may still be routing mail
        # (Google's non-editable aliases cannot be deleted but ARE
        # merged into this column).
        already = bool((resp.data or {}).get("already_absent"))
        if not already and row.aliases is not None:
            aliases = set(row.aliases)
            aliases.discard(alias_email)
            row.aliases = sorted(aliases)
        await self.db.flush()
        await self._audit_and_notify(
            action="email_group.dynamic_alias_removed",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={
                "group_email": row.email,
                "alias_email": alias_email,
                "already_absent_on_google": already,
            },
            admin_body=None,
            # ``noop``, not ``success``, when Google removed nothing.
            # The static twin (``EmailGroupAdminService.remove_alias``)
            # has recorded it that way since it shipped; logging the same
            # event two ways in the same log means an auditor reading the
            # dynamic row concludes an alias was removed from Google that
            # was never there.
            status="noop" if already else "success",
        )
        await self.db.commit()
        detail = await self.get_detail(group_id=row.id, actor_roles=actor_roles)
        # Surface the no-op to the CALLER, not just to the audit log.
        # Recording ``status="noop"`` while returning a response
        # indistinguishable from a real removal left the UI showing a
        # green 「别名已删除」 with the alias still listed one line below
        # (the mirror deliberately keeps an address Google never
        # removed). The static twin returns ``outcome`` for exactly this.
        detail["alias_outcome"] = "already_absent" if already else "removed"
        return detail

    async def set_missing_alert(
        self,
        *,
        group_id: UUID,
        enabled: bool,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        row.alert_on_missing_values = bool(enabled)
        # Reset the dedup signature so the next drift tick re-evaluates
        # under the new setting.
        row.drift_signature = None
        await self.db.flush()
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action="email_group.dynamic_missing_alert_set",
            status="success",
            resource_type="email_group",
            resource_id=str(row.id),
            detail={"group_email": row.email, "enabled": bool(enabled)},
        )
        await self.db.commit()
        return await self.get_detail(group_id=row.id, actor_roles=actor_roles)

    async def delete_group(
        self,
        *,
        group_id: UUID,
        reason: str,
        actor: IdentityPerson,
        actor_roles: list[str],
        trace_id: str,
    ) -> dict:
        self._require_manage(actor_roles)
        row = await self._load_row(group_id)
        driver = await self._get_driver()
        resp = await driver.delete_group(email=row.email)
        if not resp.ok:
            raise AppException(
                code="GOOGLE_DELETE_GROUP_FAILED",
                message=f"Google 删除动态组失败: {resp.error_message}",
                status_code=502,
            )
        row.status = "deleted"
        await self.db.flush()
        await self._audit_and_notify(
            action="email_group.dynamic_deleted",
            row=row,
            actor=actor,
            trace_id=trace_id,
            detail={"group_email": row.email, "reason": reason},
            admin_body=(
                f"HR/{actor.display_name or actor.work_email} 删除了动态邮件组 "
                f"{row.email}。原因:{reason}"
            ),
        )
        await self.db.commit()
        return {"group_id": str(row.id), "group_email": row.email, "deleted": True}

    # ─── Audit + IT-admin fan-out ──────────────────────────────────

    async def _audit_and_notify(
        self,
        *,
        action: str,
        row: EmailGroup,
        actor: IdentityPerson,
        trace_id: str,
        detail: dict,
        admin_body: str | None,
        status: str = "success",
    ) -> None:
        await self.audit.log(
            trace_id=trace_id,
            actor_type="user",
            actor_user_id=actor.id,
            action=action,
            status=status,
            resource_type="email_group",
            resource_id=str(row.id),
            detail=detail,
        )
        if not admin_body:
            return
        # Notify IT admins only — never employees/managers (IT-only push
        # rule). Best-effort: a notification failure must not roll back
        # the successfully-applied Google + local change.
        try:
            from app.scheduled_jobs.failure_notifier import list_admin_recipients

            admins = await list_admin_recipients(self.db)
            for admin in admins:
                if admin.id == actor.id:
                    continue
                env = NotificationEnvelope(
                    recipient=admin,
                    title=f"动态邮件组变更:{row.email}",
                    body_text=admin_body,
                    template_key="dynamic_email_group_changed",
                    context={
                        "person_name": admin.display_name or admin.work_email or "",
                        "group_email": row.email,
                        "group_name": row.name,
                        "actor_name": actor.display_name or actor.work_email or "",
                        "action": action,
                        # The template frame is localised; this detail line is
                        # not — each caller composes ``admin_body`` in Chinese
                        # with variant-specific facts (compiled rule / member
                        # delta / delete reason) that have no structured form
                        # to translate from. Recipients are IT admins only, so
                        # a Chinese detail inside an EN/JA frame is an accepted
                        # degradation rather than an employee-facing miss.
                        "change_detail": admin_body,
                    },
                    related_resource_type="email_group",
                    related_resource_id=row.id,
                )
                await self.notify.dispatch(
                    env, trace_id=trace_id, actor_user_id=actor.id,
                )
        except Exception:
            logger.exception(
                "dynamic email-group admin notify failed (non-fatal) for %s",
                row.email,
            )


def compute_drift_signature(dead: list[str], missing: list[str]) -> str:
    """Stable dedup signature over the current drift state so the
    notifier only re-alerts IT when the dead/missing value set changes.
    """
    payload = json.dumps(
        {"dead": sorted(dead), "missing": sorted(missing)},
        ensure_ascii=False, sort_keys=True,
    )
    return hashlib.sha256(payload.encode("utf-8")).hexdigest()