"""T+1 post-leaver verification sweep. Runs daily at 02:00. For every person transitioned to ``employment_status='left'`` in the last ``recent_days`` days, rescan every resource type that could point at them. Anything found is a violation of the platform's data-freshness guarantee → write a critical ``compliance_incident``. This is the safety net for :class:`ProcessAdvanceService` — if the force-transfer cascade failed silently (e.g. a connector error we didn't catch, a new resource type added without force-transfer support), this sweep surfaces it so the admin can manually resolve. """ from __future__ import annotations import logging from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.compliance.models import SEVERITY_CRITICAL, TYPE_LEAVER_ORPHAN from app.compliance.service import ComplianceIncidentService from app.identity.person_models import IdentityPerson from app.notifications.service import NotificationEnvelope, NotificationService logger = logging.getLogger(__name__) class PostLeaverVerifyService: def __init__(self, db: AsyncSession) -> None: self.db = db self.incidents = ComplianceIncidentService(db) self.notify = NotificationService(db) #: Resource types whose predicate raised during the current run. self._failed_types: set[str] = set() async def run(self, *, recent_days: int = 7) -> dict[str, Any]: """Verify no resource points at any recently-left person.""" cutoff = datetime.now(UTC) - timedelta(days=recent_days) left_persons = list(( await self.db.execute( select(IdentityPerson).where( IdentityPerson.employment_status == "left", IdentityPerson.deleted_at.is_(None), # Pre-launch P2 fix: OR in the NULL case so that a # leaver whose last_synced_at was never updated (e.g. # created directly via HRIS import with no subsequent # sync tick) is not silently skipped forever. # Without this, a person who went "left" more than # recent_days ago and whose last_synced_at predates # the window would never be re-scanned, letting # residual resources accumulate with no compliance # incident raised. or_( IdentityPerson.last_synced_at >= cutoff, IdentityPerson.last_synced_at.is_(None), ), ) ) ).scalars().all()) summary: dict[str, Any] = { "persons_scanned": len(left_persons), "residual_resources_found": 0, "incidents_created": 0, "incidents_refreshed": 0, } # Reset per run: a type that failed yesterday and reads fine today # must not keep claiming the sweep is partial. self._failed_types = set() if not left_persons: return summary incidents_detail: list[dict[str, Any]] = [] for person in left_persons: residuals = await self._scan_person(person) summary["residual_resources_found"] += len(residuals) for r in residuals: _, created = await self.incidents.create_or_upsert_incident( incident_type=TYPE_LEAVER_ORPHAN, severity=SEVERITY_CRITICAL, person_id=person.id, resource_type=r["resource_type"], resource_id=r["resource_id"], resource_label=r["label"], title=( f"{person.display_name or person.work_email} 已离职 " f"但仍持有 {r['resource_type']} [{r['label']}]" ), detail={ "leaver_email": person.work_email, "termination_date": ( person.termination_date.isoformat() if person.termination_date else None ), }, ) if created: summary["incidents_created"] += 1 incidents_detail.append({ "person": person.display_name or person.work_email, "resource": f"{r['resource_type']}: {r['label']}", }) else: summary["incidents_refreshed"] += 1 if self._failed_types: summary["failed_types"] = sorted(self._failed_types) if summary["incidents_created"] > 0: await self._alert_it_admins(incidents_detail, summary) return summary # ── Scan ──────────────────────────────────────────────── async def _scan_person( self, person: IdentityPerson, ) -> list[dict[str, Any]]: """Everything still pointing at ``person`` — registry-driven (PR3). This used to be a second, hand-maintained enumeration that knew about 14 of the registry's 29 resource types and named four of them differently from every other consumer. It is now one line: the safety net asks the SAME question the offboarding scanner and the mover packet ask, through :mod:`app.hr_lifecycle.leaver_residuals`, whose registry fails the build if a new resource type arrives without a decision. ``failed_types`` is surfaced in the run summary rather than swallowed — a sweep that could not read a domain must not report that domain as clean. """ from app.hr_lifecycle.leaver_residuals import scan_person_residuals scan = await scan_person_residuals(self.db, person) if scan.failed_types: self._failed_types.update(scan.failed_types) return scan.residuals # ── Alerting ──────────────────────────────────────────── async def _alert_it_admins( self, incidents_detail: list[dict[str, Any]], summary: dict[str, Any], ) -> None: from app.auth.models import AuthRole, AuthUserRole 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"))) # An admin holding BOTH global_admin and the legacy # it_admin alias yields two user_id rows → double DM. .distinct() ) ).scalars().all()) if not admin_ids: logger.warning( "post_leaver_verify: %d new orphan incidents but no IT admin to notify", summary["incidents_created"], ) return # The incident list is kept separate from the surrounding prose: the # 三语 template (``notifications/templates/lifecycle.py``) supplies the # lead-in and the CTA in the recipient's language and interpolates only # this block, so an EN/JA admin doesn't get a Chinese preamble embedded # mid-body. ``body`` stays as the no-template fallback. incident_list = ( "\n".join( f" · [{i['resource']}] ← {i['person']}" for i in incidents_detail[:20] ) + ( f"\n\n... 另有 {len(incidents_detail) - 20} 条未列出" if len(incidents_detail) > 20 else "" ) ) body = ( f"T+1 复查发现 {summary['incidents_created']} 项" "新的已离职人员残留资源,请尽快处理:\n\n" + incident_list + "\n\n详情: /admin/compliance-incidents" ) for admin_id in admin_ids: admin = await self.db.get(IdentityPerson, admin_id) if admin is None or not admin.work_email: continue # Don't leak leaver-naming compliance alerts to a # deactivated / departed admin who still holds an admin role # row (same guard as digest_service CR-411; this path was the # outlier the digest comment wrongly assumed already filtered). if ( admin.deleted_at is not None or (admin.employment_status or "").lower() != "active" ): continue try: await self.notify.dispatch( NotificationEnvelope( recipient=admin, title=f"[合规告警] 发现 {summary['incidents_created']} 项已离职人员残留资源", body_text=body, channels=["feishu", "email"], template_key="compliance_incident_alert", # Land the click on the incident triage page — both # the in-app inbox row and the Feishu card CTA button # resolve from this. Without it the inbox row had no # ``url`` and clicking did nothing. url="/admin/compliance-incidents", related_resource_type="compliance_incident", context={ "count": summary["incidents_created"], "incident_list": incident_list, }, ), trace_id="post_leaver_verify", ) except Exception as exc: # noqa: BLE001 logger.warning( "post_leaver_verify: failed to notify admin %s: %s", admin_id, exc, ) __all__ = ["PostLeaverVerifyService"]