"""Read-only service powering ``/admin/offboarding``. Reads: * ``offboarding_processes`` joined with ``identity_persons`` and pending ``offboarding_blockers``. * The most recent ``leaver_block_audit`` rows per person. * Active ``leaver_dual_sign_overrides`` per person. * ``workflow_runs`` of type ``offboarding`` for the timeline. Returns paginated DTOs aimed directly at the dashboard frontend; no mutation surface here. """ from __future__ import annotations from collections import defaultdict from dataclasses import dataclass from datetime import UTC, datetime from typing import Any from uuid import UUID from zoneinfo import ZoneInfo from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.core.workflow_journal import ( WorkflowJournal, WorkflowRun, ) from app.domains.leaver_blocked.models import ( BLOCKING_DUAL_SIGN_STATUSES, DUAL_SIGN_APPROVED, DUAL_SIGN_PENDING_SECONDARY, LeaverBlockAudit, LeaverDualSignOverride, ) from app.domains.offboarding.models import ( PROC_BLOCKED, PROC_CANCELLED, PROC_COMPLETED, RESO_PENDING, SLA_BREACH, SLA_NONE, SLA_URGENT, SLA_WARN, OffboardingBlocker, OffboardingProcess, compute_triage_bucket, ) from app.domains.provisioning.models import ProvisioningTask from app.identity.person_models import IdentityPerson @dataclass class LeaverBlockedSummary: """Top-of-page KPI strip.""" total_blocked: int sla_warn: int sla_urgent: int sla_breach: int pending_overrides: int approved_overrides_ready_to_execute: int today_completed: int @dataclass class LeaverBlockedRow: """One row in the dashboard table.""" process_id: UUID person_id: UUID person_name: str person_email: str department: str | None leaver_date: str | None blocked_for_hours: int | None sla_level: str pending_blocker_count: int pending_blocker_types: list[str] current_step: str | None has_active_override: bool override_status: str | None workflow_run_id: UUID | None last_step_at: str | None class LeaverBlockedDashboardService: def __init__(self, db: AsyncSession) -> None: self.db = db self.journal = WorkflowJournal(db) @staticmethod def _blocker_to_dict( b: OffboardingBlocker, task_meta: dict[UUID, tuple[str, str]], ) -> dict[str, Any]: """Single source of truth for blocker serialization in this service. Mirrors :class:`OffboardingBlockerResponse` for the fields that the leaver-blocked detail panel renders; the dict shape is legacy (the original endpoint pre-dates the OffboardingBlockerResponse schema) and downstream consumers depend on the literal keys. Migration 148 added the structured error columns and the triage_bucket field — included here so the unified offboarding console can render the same 3-bucket UI off this endpoint as off the offboarding-list endpoint. """ task_status, task_mode = ( task_meta.get(b.provisioning_task_id, (None, None)) if b.provisioning_task_id is not None else (None, None) ) return { "id": str(b.id), "resource_type": b.resource_type, "resource_id": ( str(b.resource_id) if b.resource_id else None ), "label": b.label, "severity": b.severity, "resolution": b.resolution, "resolution_ref": b.resolution_ref, "notes": b.notes, "last_error_message": b.last_error_message, "last_error_at": ( b.last_error_at.isoformat() if b.last_error_at else None ), "error_count": b.error_count or 0, "triage_bucket": compute_triage_bucket( resolution=b.resolution, severity=b.severity, error_count=b.error_count or 0, provisioning_task_status=task_status, provisioning_task_execution_mode=task_mode, ), "software_id": ( str(b.software_id) if b.software_id else None ), "provisioning_task_id": ( str(b.provisioning_task_id) if b.provisioning_task_id else None ), "provisioning_task_status": task_status, "provisioning_task_execution_mode": task_mode, "created_at": ( b.created_at.isoformat() if b.created_at else None ), "updated_at": ( b.updated_at.isoformat() if b.updated_at else None ), } # ── KPIs ─────────────────────────────────────────────────── async def get_summary(self) -> LeaverBlockedSummary: rows = list((await self.db.execute( select(OffboardingProcess) .where(OffboardingProcess.status.notin_([ PROC_COMPLETED, PROC_CANCELLED, ])) )).scalars().all()) total_blocked = sum(1 for p in rows if p.status == PROC_BLOCKED) sla_warn = sum(1 for p in rows if p.sla_breach_level == SLA_WARN) sla_urgent = sum(1 for p in rows if p.sla_breach_level == SLA_URGENT) sla_breach = sum(1 for p in rows if p.sla_breach_level == SLA_BREACH) pending_overrides = await self.db.scalar( select(func.count(LeaverDualSignOverride.id)).where( LeaverDualSignOverride.status == DUAL_SIGN_PENDING_SECONDARY, ) ) or 0 approved_overrides = await self.db.scalar( select(func.count(LeaverDualSignOverride.id)).where( LeaverDualSignOverride.status == DUAL_SIGN_APPROVED, ) ) or 0 # Office-local day boundary (Asia/Shanghai, the platform office TZ # — mirrors sla_service's fallback) so "今日完成离职" rolls over at # local midnight rather than 08:00 local (UTC midnight). Converted # to UTC for comparison against the UTC-stored completed_at. office_tz = ZoneInfo("Asia/Shanghai") today_start = datetime.combine( datetime.now(office_tz).date(), datetime.min.time(), tzinfo=office_tz, ).astimezone(UTC) today_completed = await self.db.scalar( select(func.count(OffboardingProcess.id)).where( OffboardingProcess.status == PROC_COMPLETED, OffboardingProcess.completed_at >= today_start, ) ) or 0 return LeaverBlockedSummary( total_blocked=total_blocked, sla_warn=sla_warn, sla_urgent=sla_urgent, sla_breach=sla_breach, pending_overrides=int(pending_overrides), approved_overrides_ready_to_execute=int(approved_overrides), today_completed=int(today_completed), ) # ── List ──────────────────────────────────────────────────── async def list_blocked( self, *, sla_level: str | None = None, keyword: str | None = None, limit: int = 100, offset: int = 0, ) -> tuple[list[LeaverBlockedRow], int]: query = ( select(OffboardingProcess) .where(OffboardingProcess.status.notin_([ PROC_COMPLETED, PROC_CANCELLED, ])) ) if sla_level: query = query.where(OffboardingProcess.sla_breach_level == sla_level) # We must filter / sort joined with identity_persons; do via a # subselect on person_id. all_procs = list((await self.db.execute(query)).scalars().all()) # Resolve person rows person_ids = [p.person_id for p in all_procs] person_rows = list((await self.db.execute( select(IdentityPerson).where(IdentityPerson.id.in_(person_ids)) )).scalars().all()) if person_ids else [] person_by_id = {p.id: p for p in person_rows} # Keyword filter on display_name / work_email — done in Python # because the population is small (typically ≤ a few dozen # blocked leavers at a time) and the join logic is already # cheap. Avoids tearing apart the OffboardingProcess query. if keyword: kw = keyword.strip().lower() filtered_procs = [] for p in all_procs: person = person_by_id.get(p.person_id) if person is None: continue hay = " ".join([ (person.display_name or "").lower(), (person.work_email or "").lower(), ]) if kw in hay: filtered_procs.append(p) all_procs = filtered_procs # Pre-compute pending blocker counts per process blocker_rows = list((await self.db.execute( select(OffboardingBlocker).where( OffboardingBlocker.process_id.in_([p.id for p in all_procs]), OffboardingBlocker.resolution == RESO_PENDING, ) )).scalars().all()) if all_procs else [] blockers_by_process: dict[UUID, list[OffboardingBlocker]] = defaultdict(list) for b in blocker_rows: blockers_by_process[b.process_id].append(b) # Override status per process override_rows = list((await self.db.execute( select(LeaverDualSignOverride).where( LeaverDualSignOverride.process_id.in_([p.id for p in all_procs]), LeaverDualSignOverride.status.in_(BLOCKING_DUAL_SIGN_STATUSES), ) )).scalars().all()) if all_procs else [] override_by_process = {o.process_id: o for o in override_rows} # Sort: SLA breach first, then urgent, warn, none. Within the # same SLA level, longest-blocked first. sla_rank = { SLA_BREACH: 0, SLA_URGENT: 1, SLA_WARN: 2, SLA_NONE: 3, } def _sort_key(p: OffboardingProcess) -> tuple[int, datetime]: # Same tz-naive guard as the row-build loop below: TiDB # returns DATETIME columns tz-naive; the fallback # ``datetime.now(UTC)`` is tz-aware. Mixing the two in a # single sort key raises ``TypeError: can't compare # offset-naive and offset-aware datetimes`` the moment any # row in the result set has ``first_blocked_at IS NULL``. fb = p.first_blocked_at if fb is not None and fb.tzinfo is None: fb = fb.replace(tzinfo=UTC) return ( sla_rank.get(p.sla_breach_level, 9), fb or datetime.now(UTC), ) all_procs.sort(key=_sort_key) total = len(all_procs) windowed = all_procs[offset:offset + limit] rows: list[LeaverBlockedRow] = [] now = datetime.now(UTC) for p in windowed: person = person_by_id.get(p.person_id) if person is None: continue blockers = blockers_by_process.get(p.id, []) override = override_by_process.get(p.id) # TiDB returns DATETIME columns as tz-naive even though the # SQLAlchemy column is ``DateTime(timezone=True)``. Subtracting # from ``datetime.now(UTC)`` (tz-aware) raises ``TypeError: # can't subtract offset-naive and offset-aware datetimes``, # which surfaced as a hard 500 + "加载列表失败" toast on # ``/admin/offboarding`` for any active blocked process # with a populated ``first_blocked_at``. Same fix pattern as # ``successfactors_sync._last_successful_until``. first_blocked_at = p.first_blocked_at if first_blocked_at is not None and first_blocked_at.tzinfo is None: first_blocked_at = first_blocked_at.replace(tzinfo=UTC) rows.append(LeaverBlockedRow( process_id=p.id, person_id=person.id, person_name=person.display_name or person.work_email, person_email=person.work_email, department=person.primary_department, leaver_date=p.leaver_date.isoformat() if p.leaver_date else None, blocked_for_hours=( int((now - first_blocked_at).total_seconds() // 3600) if first_blocked_at else None ), sla_level=p.sla_breach_level, pending_blocker_count=len(blockers), pending_blocker_types=sorted({b.resource_type for b in blockers}), current_step=p.current_step, has_active_override=override is not None, override_status=override.status if override else None, workflow_run_id=p.workflow_run_id, last_step_at=( p.updated_at.isoformat() if p.updated_at else None ), )) return rows, total # ── Detail ───────────────────────────────────────────────── async def get_detail(self, process_id: UUID) -> dict[str, Any]: proc = await self.db.get(OffboardingProcess, process_id) if proc is None: raise ValueError(f"OffboardingProcess {process_id} not found") person = await self.db.get(IdentityPerson, proc.person_id) if person is None: raise ValueError(f"Person {proc.person_id} not found") blockers = list((await self.db.execute( select(OffboardingBlocker).where( OffboardingBlocker.process_id == proc.id, ) )).scalars().all()) # Batch-fetch provisioning task status/execution_mode for any # blocker that has one, so the per-row triage_bucket reflects # task state (failed / notified-manual / in-flight) instead of # naive "is this advisory or blocker". task_ids = { b.provisioning_task_id for b in blockers if b.provisioning_task_id is not None } task_meta: dict[UUID, tuple[str, str]] = {} if task_ids: for r in ( await self.db.execute( select( ProvisioningTask.id, ProvisioningTask.status, ProvisioningTask.execution_mode, ).where(ProvisioningTask.id.in_(task_ids)) ) ).all(): task_meta[r.id] = (r.status, r.execution_mode) # Recent audit denials for this person recent_audit = list((await self.db.execute( select(LeaverBlockAudit) .where(LeaverBlockAudit.person_id == person.id) .order_by(LeaverBlockAudit.attempted_at.desc()) .limit(20) )).scalars().all()) # Active + recent overrides overrides = list((await self.db.execute( select(LeaverDualSignOverride) .where(LeaverDualSignOverride.process_id == proc.id) .order_by(LeaverDualSignOverride.created_at.desc()) .limit(10) )).scalars().all()) # Workflow run + step events for the offboarding flow run: WorkflowRun | None = None if proc.workflow_run_id is not None: run = await self.journal.get_run(proc.workflow_run_id) if run is None: run = await self.journal.get_run_by_resource( "offboarding_process", proc.id, ) step_events = ( await self.journal.list_step_events(run.id) if run else [] ) return { "process": { "id": str(proc.id), "status": proc.status, "leaver_date": proc.leaver_date.isoformat() if proc.leaver_date else None, "scanned_at": proc.scanned_at.isoformat() if proc.scanned_at else None, "ready_at": proc.ready_at.isoformat() if proc.ready_at else None, "completed_at": proc.completed_at.isoformat() if proc.completed_at else None, "sso_suspended_at": proc.sso_suspended_at.isoformat() if proc.sso_suspended_at else None, "first_blocked_at": proc.first_blocked_at.isoformat() if proc.first_blocked_at else None, "sla_breach_level": proc.sla_breach_level, "current_step": proc.current_step, }, "person": { "id": str(person.id), "display_name": person.display_name or person.work_email, "work_email": person.work_email, "employment_status": person.employment_status, "termination_date": person.termination_date.isoformat() if person.termination_date else None, "manager_person_id": str(person.manager_person_id) if person.manager_person_id else None, "department": person.primary_department, "force_logout_at": person.force_logout_at.isoformat() if person.force_logout_at else None, }, "blockers": [ self._blocker_to_dict(b, task_meta) for b in blockers ], "leaver_block_audit": [ { "id": str(a.id), "attempted_at": a.attempted_at.isoformat(), "attempted_via": a.attempted_via, "denied_reason": a.denied_reason, "pending_blocker_count": a.pending_blocker_count, "actor_user_id": ( str(a.attempted_by_user_id) if a.attempted_by_user_id else None ), "client_ip": a.client_ip, } for a in recent_audit ], "overrides": [ { "id": str(o.id), "status": o.status, "primary_signer_user_id": str(o.primary_signer_user_id), "primary_signed_at": o.primary_signed_at.isoformat() if o.primary_signed_at else None, "secondary_signer_user_id": ( str(o.secondary_signer_user_id) if o.secondary_signer_user_id else None ), "secondary_signed_at": o.secondary_signed_at.isoformat() if o.secondary_signed_at else None, "executed_at": o.executed_at.isoformat() if o.executed_at else None, "expires_at": o.expires_at.isoformat() if o.expires_at else None, "overridden_blocker_ids": list(o.overridden_blocker_ids or []), } for o in overrides ], "workflow_run": ( { "id": str(run.id), "workflow_type": run.workflow_type, "status": run.status, "current_step": run.current_step, "started_at": run.started_at.isoformat(), "last_step_at": run.last_step_at.isoformat() if run.last_step_at else None, "step_count": run.step_count, "step_log": run.step_log or [], } if run else None ), "workflow_steps": [ { "id": str(s.id), "sequence": s.sequence, "step_name": s.step_name, "status": s.status, "started_at": s.started_at.isoformat(), "completed_at": s.completed_at.isoformat() if s.completed_at else None, "duration_ms": s.duration_ms, "attempt": s.attempt, "error_class": s.error_class, "error_message": s.error_message, "actor_kind": s.actor_kind, "actor_user_id": ( str(s.actor_user_id) if s.actor_user_id else None ), "payload": s.payload, } for s in step_events ], } __all__ = [ "LeaverBlockedDashboardService", "LeaverBlockedSummary", "LeaverBlockedRow", ]