"""入职第一天仪式感 — a warm "welcome aboard" page that wraps the existing onboarding machinery in a sense of occasion. A new hire's first authenticated landing shouldn't be a cold to-do list. This aggregates the things that make Day 1 feel like an arrival: who you are here (employee no / title / office / your manager), the gear waiting for you (assigned + reserved devices, an "unboxing" moment), where your onboarding stands, and a gentle first-week checklist of things to explore. Read-only, self-scoped. Reuses the onboarding-readiness signals the manager lifecycle cockpit already computes — no parallel logic. """ from __future__ import annotations import logging from datetime import datetime from typing import Any from pydantic import BaseModel from sqlalchemy import func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload from app.core.working_hours import resolve_timezone from app.domains.asset.models import AssetItem, SoftwareSeatAllocation from app.domains.permission.models import UserPermission from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # A hire counts as "new" (full welcome ceremony) for their first month. _NEW_WINDOW_DAYS = 30 class WelcomeResponse(BaseModel): person: dict[str, Any] is_new: bool days_since_hire: int | None manager: dict[str, Any] | None onboarding: dict[str, Any] | None devices: list[dict[str, Any]] starters: dict[str, Any] generated_at: str def _person_card(p: IdentityPerson) -> dict[str, Any]: return { "id": str(p.id), "display_name": p.display_name or p.work_email, "employee_no": p.employee_no, "department": p.primary_department, "job_title": p.job_title, "avatar_url": p.avatar_url, "work_email": p.work_email, } async def build_my_welcome( db: AsyncSession, person: IdentityPerson, ) -> WelcomeResponse: tz = resolve_timezone(person) now = datetime.now(tz) today = now.date() pid = person.id days_since_hire: int | None = None if person.hire_date is not None: days_since_hire = (today - person.hire_date).days is_new = days_since_hire is not None and -90 <= days_since_hire <= _NEW_WINDOW_DAYS # Manager card (skip soft-deleted / missing pointer). Eager-load # org_memberships so ``primary_department`` (a relationship-backed property) # doesn't trigger a lazy load during response serialisation (MissingGreenlet). manager_card: dict[str, Any] | None = None if person.manager_person_id is not None: mgr = await db.scalar( select(IdentityPerson) .where(IdentityPerson.id == person.manager_person_id) .options(selectinload(IdentityPerson.org_memberships)) ) if mgr is not None and mgr.deleted_at is None: manager_card = _person_card(mgr) onboarding = await _onboarding_snapshot(db, pid, today) devices = await _devices(db, pid) starters = await _starters(db, person) return WelcomeResponse( person={ **_person_card(person), "hire_date": person.hire_date.isoformat() if person.hire_date else None, "work_location": person.work_location_name, }, is_new=is_new, days_since_hire=days_since_hire, manager=manager_card, onboarding=onboarding, devices=devices, starters=starters, generated_at=now.isoformat(), ) async def _onboarding_snapshot(db, pid, today) -> dict[str, Any] | None: """The caller's own onboarding process (if any), with the same readiness + pending-blocker signals the manager lifecycle cockpit surfaces. Reuses ``_pending_blocker_labels`` so blocker classification stays single-sourced.""" from app.domains.me_universe.team_service import _pending_blocker_labels from app.domains.onboarding.models import ( PROC_TERMINAL_STATUSES, OnboardingBlocker, OnboardingProcess, ) proc = await db.scalar( select(OnboardingProcess) .where(OnboardingProcess.person_id == pid) .order_by(OnboardingProcess.created_at.desc()) .limit(1) ) if proc is None: return None # 🔴 A TERMINAL process's blocker rows are frozen: ``reconcile_blockers`` # short-circuits on completed/cancelled and the 5-minute reconcile cron # selects open processes only, so nothing will ever resolve them. This # card renders them as 「IT 正在为你准备剩下的 N 项,很快就好」 — a # promise of work that provably will not happen, shown to the employee # themselves for the rest of the 90-day new-joiner window. Production # has 20 completed-with-gap hires (the reason the IT 硬件待核实 queue # includes completed processes at all), and a hire who later received # their laptop through the ordinary claim flow keeps the frozen row # forever. Once the process is closed the onboarding is over: whatever # is still "pending" is bookkeeping, not something anyone is doing. # The manager-facing twin never had this problem — ``team_service`` # selects ``status.notin_(_TERMINAL_PROC)`` before it even reads the # blockers. if proc.status in PROC_TERMINAL_STATUSES: gaps: dict[str, list[str]] = {"blocker": [], "advisory": []} else: gaps = (await _pending_blocker_labels( db, OnboardingBlocker, OnboardingBlocker.process_id, [proc.id], )).get(proc.id, {"blocker": [], "advisory": []}) days_to_hire = (proc.hire_date - today).days if proc.hire_date else None return { "status": proc.status, "hire_date": proc.hire_date.isoformat() if proc.hire_date else None, "days_to_hire": days_to_hire, "ready": proc.ready_at is not None, "hard_ready": proc.hard_ready_at is not None, "pending_blocker_count": len(gaps["blocker"]), "pending_blocker_labels": gaps["blocker"][:12], "pending_advisory_labels": gaps["advisory"][:12], } async def _devices(db, pid) -> list[dict[str, Any]]: """Gear that is yours or earmarked for you — the unboxing moment. Assigned devices + devices reserved for this incoming hire (备机闭环).""" rows = list((await db.execute( select(AssetItem).where( or_( AssetItem.assigned_to == pid, AssetItem.reserved_for_person_id == pid, ), AssetItem.deleted_at.is_(None), ) )).scalars().all()) out: list[dict[str, Any]] = [] seen: set = set() for a in rows: if a.id in seen: continue seen.add(a.id) out.append({ "id": str(a.id), "asset_name": a.asset_name, "category": a.category, "brand": a.brand, "model": a.model, "status": a.status, # Reserved-but-not-yet-assigned reads as "waiting for you". "reserved": a.assigned_to != pid and a.reserved_for_person_id == pid, }) return out async def _starters(db, person: IdentityPerson) -> dict[str, Any]: """Gentle readiness signals that drive the first-week checklist ticks.""" seat_count = await db.scalar( select(func.count()).select_from(SoftwareSeatAllocation).where( SoftwareSeatAllocation.user_id == person.id, SoftwareSeatAllocation.status == "active", ) ) permission_count = await db.scalar( select(func.count()).select_from(UserPermission).where( UserPermission.user_id == person.id, UserPermission.status == "active", ) ) return { "seat_count": int(seat_count or 0), "permission_count": int(permission_count or 0), "has_avatar": bool(person.avatar_url), } __all__ = ["build_my_welcome", "WelcomeResponse"]