"""Onboarding domain HTTP endpoints. Two public surfaces: * Cockpit (``GET /api/v1/onboarding/cockpit/queue``) — read-only, backed by ``OnboardingCockpitService``. * Process state-machine (``GET / PUT / POST /api/v1/onboarding/ processes/*``) — admin CRUD on ``onboarding_processes`` rows, mirrors offboarding's router shape. All read endpoints gated by ``require_lifecycle_read`` (cockpit + list/get; global_admin + hr + auditor + management) or ``require_admin`` (mutations). Self-service employee endpoints live in ``/me/onboarding`` and are introduced by PR-4. """ from __future__ import annotations from datetime import date as date_type from decimal import Decimal from typing import Any from uuid import UUID from fastapi import APIRouter, Body, Depends, Query, Request from pydantic import BaseModel, Field, field_validator, model_validator from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import require_admin, require_lifecycle_read from app.core.response import success_response from app.domains.asset.hardware_prep import COMPUTER_CATEGORIES from app.domains.onboarding.ack_service import ( OnboardingAckService, serialize_ack_item, ) from app.domains.onboarding.cockpit_service import OnboardingCockpitService from app.domains.onboarding.hardware_gap_service import ( GAP_LOOKBACK_DAYS, OnboardingHardwareGapService, ) from app.domains.onboarding.models import ( PROC_OPEN_STATUSES, OnboardingBlocker, OnboardingProcess, compute_triage_bucket, ) from app.domains.onboarding.service import OnboardingService router = APIRouter( prefix="/api/v1/onboarding", tags=["onboarding"], ) # ───────────────────────── Cockpit (read) ───────────────────────── @router.get("/cockpit/queue") async def get_cockpit_queue( request: Request, _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """In-flight onboarding queue — every OPEN process (scanning / preparing / ready / live) with hire_date ``<= today+14d``, including already-onboard ``live`` hires still missing a Day-1 dimension, with per-person readiness across the ten provisioning dimensions. Read-only and admin-gated (global_admin / auditor / management). No pagination — the steady-state window is at most a few dozen rows and the cockpit wants the whole set on one screen. Reads from ``onboarding_processes`` + ``onboarding_blockers`` (the state-machine tables introduced by migration 229) rather than recomputing the 7 dimensions per request. """ payload = await OnboardingCockpitService(db).build_queue() return success_response(data=payload, request=request) # ───────────────────────── Process list / detail ───────────────────────── def _serialize_process(proc: OnboardingProcess) -> dict[str, Any]: return { "id": str(proc.id), "person_id": str(proc.person_id), "hire_date": proc.hire_date.isoformat(), "status": proc.status, "scanned_at": _iso(proc.scanned_at), "ready_at": _iso(proc.ready_at), "live_at": _iso(proc.live_at), "completed_at": _iso(proc.completed_at), "cancelled_at": _iso(proc.cancelled_at), "current_step": proc.current_step, "first_ungated_at": _iso(proc.first_ungated_at), "sla_breach_level": proc.sla_breach_level, "hardware_allocation": proc.hardware_allocation, "backfilled": bool(proc.backfilled), "created_at": _iso(proc.created_at), "updated_at": _iso(proc.updated_at), } def _serialize_blocker(blocker: OnboardingBlocker) -> dict[str, Any]: return { "id": str(blocker.id), "process_id": str(blocker.process_id), "resource_type": blocker.resource_type, "resource_id": ( str(blocker.resource_id) if blocker.resource_id else None ), "label": blocker.label, "severity": blocker.severity, "resolution": blocker.resolution, "resolution_ref": blocker.resolution_ref, "provisioning_task_id": ( str(blocker.provisioning_task_id) if blocker.provisioning_task_id else None ), "notes": blocker.notes, "last_error_message": blocker.last_error_message, "last_error_at": _iso(blocker.last_error_at), "error_count": int(blocker.error_count), "triage_bucket": compute_triage_bucket( resolution=blocker.resolution, severity=blocker.severity, error_count=int(blocker.error_count), ), "created_at": _iso(blocker.created_at), "updated_at": _iso(blocker.updated_at), } def _iso(value: Any) -> str | None: if value is None: return None if hasattr(value, "isoformat"): return value.isoformat() return str(value) @router.get("/processes") async def list_processes( request: Request, status: list[str] | None = Query(default=None), open_only: bool = Query(default=False), hire_date_from: date_type | None = Query(default=None), hire_date_to: date_type | None = Query(default=None), limit: int = Query(default=50, le=200), offset: int = Query(default=0, ge=0), _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Paginated list of onboarding processes.""" status_filter: list[str] | None if open_only: status_filter = list(PROC_OPEN_STATUSES) else: status_filter = status or None rows, total = await OnboardingService(db).list_processes( status_in=status_filter, hire_date_from=hire_date_from, hire_date_to=hire_date_to, limit=limit, offset=offset, ) return success_response( data={ "items": [_serialize_process(r) for r in rows], }, meta={"total": total, "limit": limit, "offset": offset}, request=request, ) # NOTE: must stay registered BEFORE ``GET /processes/{process_id}`` — # FastAPI matches in registration order and the path-param route would # otherwise swallow ``export.xlsx`` as a (non-UUID) process id. @router.get("/processes/export.xlsx") async def export_processes( request: Request, status: list[str] | None = Query(default=None), open_only: bool = Query(default=False), hire_date_from: date_type | None = Query(default=None), hire_date_to: date_type | None = Query(default=None), current_user: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Export the filtered onboarding process list as XLSX. Filter signature mirrors :func:`list_processes` minus pagination — same ``open_only`` shorthand, same service query, so any filter fix on the list applies here automatically. Rows are the ``_serialize_process`` shape join-enriched with the hire's display name / email / department / job title (the bare ``person_id`` UUID is useless in a spreadsheet). Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from sqlalchemy import select from sqlalchemy.orm import selectinload from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response from app.identity.person_models import IdentityPerson status_filter: list[str] | None if open_only: status_filter = list(PROC_OPEN_STATUSES) else: status_filter = status or None procs, total = await OnboardingService(db).list_processes( status_in=status_filter, hire_date_from=hire_date_from, hire_date_to=hire_date_to, limit=EXPORT_ROW_LIMIT, offset=0, ) person_ids = {p.person_id for p in procs} people: dict[UUID, IdentityPerson] = {} if person_ids: # ``department`` is a property over org_memberships — eager-load # so the row loop doesn't trip async lazy-loading. result = ( await db.execute( select(IdentityPerson) .options(selectinload(IdentityPerson.org_memberships)) .where(IdentityPerson.id.in_(person_ids)) ) ).scalars().all() people = {p.id: p for p in result} rows: list[dict[str, Any]] = [] for proc in procs: row = _serialize_process(proc) person = people.get(proc.person_id) row["person_name"] = person.display_name if person else None row["person_email"] = person.work_email if person else None row["department"] = person.department if person else None row["job_title"] = person.job_title if person else None rows.append(row) columns = [ ("person_name", "人员"), ("person_email", "邮箱"), ("department", "部门"), ("job_title", "职位"), ("hire_date", "入职日期"), ("status", "状态"), ("sla_breach_level", "SLA等级"), ("current_step", "当前步骤"), ("scanned_at", "扫描时间"), ("ready_at", "就绪时间"), ("live_at", "Live时间"), ("completed_at", "完成时间"), ("cancelled_at", "取消时间"), ("backfilled", "补录"), ("created_at", "创建时间"), ("id", "流程ID"), ] return await xlsx_export_response( db, request, current_user, columns=columns, rows=rows, total_matched=total, filename_stem="onboarding_processes", sheet_title="Onboarding Processes", audit_action="onboarding.processes.export", audit_resource_type="onboarding_process", filters={ "status": status, "open_only": open_only, "hire_date_from": hire_date_from, "hire_date_to": hire_date_to, }, ) @router.get("/processes/{process_id}") async def get_process( process_id: UUID, request: Request, _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): proc, blockers = await OnboardingService(db).get_process(process_id) return success_response( data={ "process": _serialize_process(proc), "blockers": [_serialize_blocker(b) for b in blockers], }, request=request, ) @router.get("/processes/{process_id}/account-timeline") async def get_account_action_timeline( process_id: UUID, request: Request, _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Per-hire account-action timeline from audits and SF input snapshots. Answers "what did Helix do to this person's account and did it succeed?" — the create/link/fail events (with the Google error text), welcome-email outcome, process transitions, and the distinct raw SF identity snapshots received before transformation — surfaced on the onboarding process detail page. Read-only, admin-gated. """ svc = OnboardingService(db) proc, _ = await svc.get_process(process_id) events = await svc.get_account_action_timeline( proc.person_id, process_id=proc.id ) return success_response( data={"person_id": str(proc.person_id), "events": events}, request=request, ) # ───────────────────────── Process mutations ───────────────────────── class HardwareAllocationBody(BaseModel): allocation: dict[str, Any] | None = Field(default=None) @router.put("/processes/{process_id}/hardware-allocation") async def set_hardware_allocation( process_id: UUID, body: HardwareAllocationBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(process_id)) proc = await OnboardingService(db).set_hardware_allocation( process_id, body.allocation, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=_serialize_process(proc), request=request) @router.post("/processes/{process_id}/reconcile") async def reconcile_process( process_id: UUID, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): flipped = await OnboardingService(db).reconcile_blockers(process_id) await db.commit() return success_response( data={"blockers_flipped": flipped}, request=request, ) @router.post("/processes/{process_id}/advance-now") async def advance_now( process_id: UUID, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(process_id)) proc = await OnboardingService(db).advance_now( process_id, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=_serialize_process(proc), request=request) @router.post("/processes/{process_id}/mark-live") async def mark_live( process_id: UUID, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Manual 「直接就位」 override — force an open process to ``live``. Unlike ``advance-now`` (ready → live only), this promotes a scanning / preparing / ready process straight to ``live`` for a hire an admin has verified is set up out-of-band. Idempotent-safe: a 409 on an already-live or terminal process. """ trace_id = getattr(request.state, "trace_id", str(process_id)) proc = await OnboardingService(db).mark_live_manual( process_id, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=_serialize_process(proc), request=request) class CancelProcessBody(BaseModel): reason: str | None = None @router.post("/processes/{process_id}/cancel") async def cancel_process( process_id: UUID, body: CancelProcessBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(process_id)) proc = await OnboardingService(db).cancel_process( process_id, trace_id=trace_id, actor_user_id=admin.id, reason=body.reason, ) await db.commit() return success_response(data=_serialize_process(proc), request=request) # ───────────────────────── Blocker mutations ───────────────────────── class SkipBlockerBody(BaseModel): reason: str = Field(..., min_length=1, max_length=512) @router.post("/blockers/{blocker_id}/skip") async def skip_blocker( blocker_id: UUID, body: SkipBlockerBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(blocker_id)) blocker = await OnboardingService(db).skip_blocker( blocker_id, reason=body.reason, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=_serialize_blocker(blocker), request=request) class MarkResolvedBody(BaseModel): note: str | None = Field(default=None, max_length=512) # ─────────────────── Day-1 hardware gap worklist ─────────────────── class RegisterHireDeviceBody(BaseModel): """Back-register a hire's computer that never reached the ledger. Mirrors the intake form's hardware subset — anything the asset domain can derive (asset tag, canonical entity name, resolved location) is left to it rather than duplicated here. """ category: str = Field(default="laptop", max_length=32) asset_tag: str | None = Field(default=None, max_length=64) brand: str | None = Field(default=None, max_length=64) model: str | None = Field(default=None, max_length=128) brand_id: UUID | None = None model_id: UUID | None = None serial_number: str | None = Field(default=None, max_length=128) purchase_date: date_type | None = None purchase_cost: Decimal | None = Field( default=None, ge=Decimal("0"), decimal_places=2, max_digits=10, ) currency: str | None = Field(default=None, max_length=8) purchase_entity: str | None = Field(default=None, max_length=128) purchase_entity_id: UUID | None = None invoice_number: str | None = Field(default=None, max_length=128) location: str | None = Field(default=None, max_length=256) claim_date: date_type | None = None notes: str | None = Field(default=None, max_length=2000) @field_validator("category") @classmethod def validate_category(cls, v: str) -> str: """Only a computer closes the Day-1 hardware dimension. Registering a monitor here would create a ledger row, leave the blocker pending, and read to the operator as a silent failure. """ if v not in COMPUTER_CATEGORIES: raise ValueError( "category must be one of " + ", ".join(COMPUTER_CATEGORIES) ) return v @model_validator(mode="after") def require_tag_or_entity(self) -> RegisterHireDeviceBody: """Asset-tag generation needs a purchase entity. ``AssetService.intake_asset`` derives an omitted ``asset_tag`` from ``purchase_entity_id`` + ``category`` and raises ``PURCHASE_ENTITY_REQUIRED`` (400) when neither is present. Catching it here turns a mid-transaction 400 into a field-level 422 the form can point at, instead of a failed submit whose message names a parameter this surface doesn't show. """ if not self.asset_tag and self.purchase_entity_id is None: raise ValueError( "either asset_tag or purchase_entity_id is required — the " "asset tag is generated from the purchase entity" ) return self @router.get("/hardware-gaps") async def list_hardware_gaps( request: Request, lookback_days: int = Query( default=GAP_LOOKBACK_DAYS, ge=1, le=1200, ), _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Hires who have started but whose computer is not on the ledger. The closing surface for the Day-1 hardware gap: read-only here, with the two write actions on ``POST .../register-device`` (back-register the real machine) and the existing ``POST /blockers/{id}/skip`` (this hire genuinely needs no company device). """ payload = await OnboardingHardwareGapService(db).list_gaps( lookback_days=lookback_days, ) return success_response(data=payload, request=request) @router.post("/hardware-gaps/{process_id}/register-device") async def register_hire_device( process_id: UUID, body: RegisterHireDeviceBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): """Create the missing asset row, claim it to the hire, close the gap. One transaction. ``global_admin`` only — this writes into the asset ledger on behalf of a region the acting admin may not administer, so it deliberately does not open to the region-scoped ``asset_admin`` tier (which has its own intake + claim surface). """ trace_id = getattr(request.state, "trace_id", str(process_id)) result = await OnboardingHardwareGapService(db).register_device( process_id, payload=body, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=result, request=request) # ───────────────────────── 入职须知 (ack items) ───────────────────────── class AckItemCreateBody(BaseModel): title: str = Field(..., min_length=1, max_length=255) description: str | None = Field(default=None, max_length=4000) doc_url: str | None = Field(default=None, max_length=512) required: bool = True sort_order: int = Field(default=0, ge=0, le=10000) class AckItemUpdateBody(BaseModel): title: str | None = Field(default=None, min_length=1, max_length=255) description: str | None = Field(default=None, max_length=4000) doc_url: str | None = Field(default=None, max_length=512) required: bool | None = None is_active: bool | None = None sort_order: int | None = Field(default=None, ge=0, le=10000) @router.get("/ack-items") async def list_ack_items( request: Request, include_inactive: bool = Query(default=False), _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Admin list of 入职须知 checklist items (employee surface lives under ``/me/onboarding``).""" items = await OnboardingAckService(db).list_items( include_inactive=include_inactive, ) return success_response( data={"items": [serialize_ack_item(i) for i in items]}, request=request, ) @router.post("/ack-items") async def create_ack_item( body: AckItemCreateBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", "onboarding-ack-item") item = await OnboardingAckService(db).create_item( title=body.title, description=body.description, doc_url=body.doc_url, required=body.required, sort_order=body.sort_order, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=serialize_ack_item(item), request=request) @router.put("/ack-items/{item_id}") async def update_ack_item( item_id: UUID, body: AckItemUpdateBody, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(item_id)) item = await OnboardingAckService(db).update_item( item_id, updates=body.model_dump(exclude_unset=True), trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data=serialize_ack_item(item), request=request) @router.get("/analytics/day1-readiness") async def get_day1_readiness_analytics( request: Request, months: int = Query(default=6, ge=1, le=24), _admin: AuthUser = Depends(require_lifecycle_read()), db: AsyncSession = Depends(get_db), ): """Per-month Day-1 readiness rate (hires whose every dimension was in place by their start date) + current past-due hard-gap count. Backfilled and cancelled processes are excluded — both would lie about the rate.""" from app.domains.onboarding.analytics_service import ( OnboardingAnalyticsService, ) payload = await OnboardingAnalyticsService(db).day1_readiness( months=months, ) return success_response(data=payload, request=request) @router.delete("/ack-items/{item_id}") async def delete_ack_item( item_id: UUID, request: Request, admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(item_id)) await OnboardingAckService(db).delete_item( item_id, trace_id=trace_id, actor_user_id=admin.id, ) await db.commit() return success_response(data={"deleted": True}, request=request) @router.post("/blockers/{blocker_id}/mark-resolved") async def mark_blocker_resolved( blocker_id: UUID, request: Request, body: MarkResolvedBody = Body(default_factory=MarkResolvedBody), admin: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ): trace_id = getattr(request.state, "trace_id", str(blocker_id)) blocker = await OnboardingService(db).mark_blocker_resolved( blocker_id, trace_id=trace_id, actor_user_id=admin.id, note=body.note, ) await db.commit() return success_response(data=_serialize_blocker(blocker), request=request)