from datetime import date, datetime from decimal import Decimal from typing import Literal from uuid import UUID from pydantic import BaseModel, Field # Valid employment_status values — must match the ORM-side enum in # ``IdentityPerson._ALLOWED_EMPLOYMENT_STATUSES``. Pre-launch P2 fix: # the prior literal included ``"inactive"`` which the ORM validator # rejects, so a PATCH that passed pydantic crashed with ValueError at # flush. The only legal terminal-ish value the admin can set is # ``leaver_pending``; ``left`` MUST go through the offboarding pipeline # (see ``AdminAdjustableEmploymentStatus`` below). EmploymentStatus = Literal["active", "leaver_pending", "left", "pre_hire"] # Subset that ``PATCH /users/{id}`` is allowed to set. ``"left"`` is # REMOVED on purpose: that terminal state must only be reached via the # offboarding pipeline (``POST /api/v1/offboarding/processes/{id}/complete``) # which runs the transfer scan, SSO suspend, Feishu notice, blocker check, # and audit fan-out. A direct PATCH to ``status="left"`` skips ALL of that # and silently strands the user's resources. ``pre_hire`` stays # admin-settable for future-dated joiners. AdminAdjustableEmploymentStatus = Literal["active", "leaver_pending", "pre_hire"] class GoogleVerifyRequest(BaseModel): # Google ID tokens are JWTs ≤ ~2 KB. Upper-bound the size to defend # against multi-megabyte garbage payloads; the lower bound is a # nominal min_length=1 so non-empty fake tokens still reach # ``verify_google_id_token`` (which raises 401 in tests). id_token: str = Field(..., min_length=1, max_length=4096) class PasswordLoginRequest(BaseModel): # We don't pull pydantic[email] just for this — the user lookup # below will fail closed if the value isn't a real email. Cap # length to bound DB query and audit-log size. email: str = Field(..., min_length=3, max_length=320) # Bcrypt truncates at 72 bytes anyway; we cap at 256 so an attacker # can't waste CPU verifying multi-megabyte "passwords". password: str = Field(..., min_length=1, max_length=256) class FeishuAuthorizeUrlRequest(BaseModel): # Frontend supplies its own origin + ``/auth/feishu/callback`` # path so dev and prod each carry their own redirect_uri without # a separate env var. Backend enforces it must match one of # CORS_ORIGINS (defence-in-depth on top of Feishu's own whitelist). redirect_uri: str = Field(..., min_length=1, max_length=2048) class FeishuAuthorizeUrlResponse(BaseModel): authorize_url: str class FeishuVerifyRequest(BaseModel): # Feishu auth codes are short opaque tokens. Cap to defend against # multi-megabyte garbage bodies. code: str = Field(..., min_length=1, max_length=512) # ``state`` is the value we issued in /authorize-url # (``secrets.token_urlsafe(32)`` → 43 chars); cap a little above # that to allow a couple of future-revision bytes without # accepting clearly bogus payloads. state: str = Field(..., min_length=1, max_length=256) redirect_uri: str = Field(..., min_length=1, max_length=2048) class FeishuH5LoginRequest(BaseModel): # H5 SDK ``tt.requestAccess`` flow: the JS SDK ran inside the # Feishu app's own webview returns a short-lived ``code`` directly, # without an OAuth authorize redirect. There is no ``state`` and no # ``redirect_uri`` — CSRF is impossible because the SDK only # produces a code when the page is loaded inside Feishu holding # the user's logged-in session. code: str = Field(..., min_length=1, max_length=512) class TokenResponse(BaseModel): access_token: str token_type: str = "bearer" class UserResponse(BaseModel): id: UUID email: str display_name: str | None avatar_url: str | None department: str | None position: str | None status: str roles: list[str] is_manager: bool is_attendance_subject: bool = False last_login_at: datetime | None created_at: datetime model_config = {"from_attributes": True} class MeResponse(BaseModel): id: UUID email: str display_name: str | None avatar_url: str | None department: str | None position: str | None status: str roles: list[str] is_manager: bool is_attendance_subject: bool = False # User-selected UI / notification locale. ``None`` means "fall back # to the platform default" (``app.default_locale`` platform_setting). # Frontend reads this on boot to seed i18n.changeLanguage; PATCH # /auth/me/profile keeps it in sync when the user toggles language # from /me/settings. preferred_language: str | None = None # Managed asset-location IDs (the asset_admin region scope). Empty # for global_admin (they manage everything regardless), non-empty # for asset_admin holders. The frontend reads this to hide / # disable asset write buttons on rows outside the caller's region. managed_location_ids: list[UUID] = [] class UserListResponse(BaseModel): id: UUID email: str display_name: str | None department: str | None position: str | None status: str roles: list[str] created_at: datetime model_config = {"from_attributes": True} class ManagerBrief(BaseModel): id: UUID email: str display_name: str | None department: str | None model_config = {"from_attributes": True} class SubordinateBrief(BaseModel): id: UUID email: str display_name: str | None department: str | None position: str | None status: str model_config = {"from_attributes": True} class UserDetailResponse(BaseModel): id: UUID email: str display_name: str | None avatar_url: str | None employee_id: str | None department: str | None position: str | None status: str roles: list[str] manager: ManagerBrief | None # HR work-relationships sourced from SF ``EmpJobRelationships`` # (see IdentityPerson: 2448=HRBP, 2450=HR Ops, 2447=调薪负责人). # ``hrbp`` prefers the SF-authoritative ``hris_hrbp_person_id`` and # falls back to the Feishu-cached ``hrbp_person_id`` — matching the # precedence in ``resolve_hrbp_for_person`` so this card reflects the # HRBP the platform actually dispatches leaver/attendance notices to. hrbp: ManagerBrief | None = None hr_ops: ManagerBrief | None = None comp_owner: ManagerBrief | None = None subordinates: list[SubordinateBrief] last_login_at: datetime | None created_at: datetime # HRIS-sourced contact + locale fields (migration 048). ``city``, # ``country`` and ``preferred_language`` are non-sensitive — sent # in the clear. ``mobile_phone`` and ``personal_email`` are PII; # the backend always returns a masked form, and the caller must # hit POST /users/{id}/reveal-contact (audit-logged) to unmask. country: str | None = None city: str | None = None preferred_language: str | None = None # 用工类型 — SF ``EmpJob.customString25`` picklist code, stored raw # (``app.identity.employee_types`` owns the meaning). ``employee_type`` # is its i18n slug and is ``None`` for a code we have not mapped; the # UI then renders the raw code rather than an empty cell. Whether # somebody is a regular employee, a consultant or an outsourced # contractor had no home on any Helix surface before this pair — # ``roles`` next to it is the RBAC role, a different question. employee_type_code: str | None = None employee_type: str | None = None # GitHub handle — SF ``cust_Other.cust_id`` keyed by ``employee_no`` # (migration 369). Non-sensitive, sent in the clear like country/city; # sparse (only carried for staff who have one on file in SF). github_id: str | None = None mobile_phone_masked: str | None = None # ISO-2 of the mobile's own country (migration 391) — the dialing-code # source for downstream phone formatting; non-sensitive, sent in clear so # the admin can see why a number formats to a given country (e.g. an # international assignee's host-country phone). mobile_country_code: str | None = None personal_email_masked: str | None = None has_mobile_phone: bool = False has_personal_email: bool = False # International assignment / 外派 (migration 392) — supplementary HRIS # context. ``is_on_international_assignment`` = host set AND host != home. # Countries are ISO-3; dates are the SF-configured assignment window. assignment_home_country: str | None = None assignment_host_country: str | None = None assignment_start_date: date | None = None assignment_end_date: date | None = None is_on_international_assignment: bool = False # HR Tools / 年假计算器 prefill fields (migration 163). NULL means # HRIS did not carry the value; the calculator falls back to manual # HR input. ``work_start_date`` = 参加工作日期 (社会工龄 N 起点); # ``hire_date`` already on the person row drives 司龄 S. hire_date: date | None = None termination_date: date | None = None work_start_date: date | None = None annual_leave_carry_legal_days: Decimal | None = None annual_leave_carry_welfare_days: Decimal | None = None annual_leave_used_legal_days: Decimal | None = None annual_leave_used_welfare_days: Decimal | None = None # HRIS tombstone (migration 222). See ``UserListItemResponse`` for # rationale. ``hris_purged_reason`` is detail useful on the per-person # admin page (e.g. ``"absent_from_full_sync"``) — irrelevant to the # list view so it's omitted there. hris_record_state: str = "present" hris_purged_at: datetime | None = None hris_purged_reason: str | None = None model_config = {"from_attributes": True} class RevealContactResponse(BaseModel): """Unmasked PII returned by POST /users/{id}/reveal-contact. Kept in a dedicated response so the masked-by-default flow is the default code path and reveal is an explicit opt-in.""" mobile_phone: str | None personal_email: str | None class MyProfilePatchRequest(BaseModel): """Mutable fields on the calling user's own profile. Today only ``preferred_language`` (the user-facing locale used by the web app + email/Feishu notifications). Other PII (mobile phone, personal email) stays read-only on the self-service path; HR / HRIS owns those. Allowed values for ``preferred_language``: ``zh_CN`` / ``en_US`` / ``ja_JP``. Anything else is rejected with 422 — both because we only ship those three locale bundles, and because letting an arbitrary string through would silently degrade the email-template selection (anything not starting with "en" lands in Chinese — surprising if the user typed "fr" expecting French). """ preferred_language: str | None = None """``zh_CN`` / ``en_US`` / ``ja_JP``. Pass ``None`` (i.e. omit from JSON body) to leave unchanged. Pass empty string to unset back to the platform default. Anything else is a validation error.""" class UserListItemResponse(BaseModel): id: UUID email: str display_name: str | None avatar_url: str | None employee_id: str | None department: str | None position: str | None status: str roles: list[str] manager_name: str | None manager_email: str | None manager_employee_no: str | None = None manager_employment_status: str | None = None # 用工类型 (see :class:`UserDetailResponse`). On the list too, not # only the detail page: "which of our people are consultants" is a # roster question, and answering it one profile at a time is the # reason it went unanswered. employee_type_code: str | None = None employee_type: str | None = None last_login_at: datetime | None created_at: datetime # HRIS tombstone (migration 222). ``hris_record_state`` is ``"present"`` # for every normal row; flips to ``"purged"`` only when the daily # full-roster reconciliation finds the person missing from the upstream # HRIS. The UI renders a "已从 HRIS 抹除" / "Purged from HRIS" badge # in that case so the row's continued existence is visibly intentional # rather than a stale ghost. The list endpoint hides purged rows by # default — pass ``include_purged=true`` to surface them. hris_record_state: str = "present" hris_purged_at: datetime | None = None model_config = {"from_attributes": True} class DirectoryPersonResponse(BaseModel): """Minimal-field people-picker shape returned by ``GET /api/v1/directory/people``. Distinct from :class:`UserListItemResponse` — the directory endpoint is used by admin-tier pickers (asset / finance / cloud / HR) and intentionally omits HR governance fields (manager hierarchy, last_login_at, hris_record_state, roles[]) that should remain HR-gated on ``GET /api/v1/users``. Mirrors the field set actually consumed by ``UserSelect`` / ``MultiUserSelect`` so the response payload is small and rate-limit cost stays low. """ id: UUID display_name: str | None work_email: str employee_no: str | None department: str | None employment_status: str model_config = {"from_attributes": True} class DepartmentBreakdown(BaseModel): department: str count: int class RoleBreakdown(BaseModel): role: str count: int class StatusBreakdown(BaseModel): status: str count: int class UserStatsResponse(BaseModel): total_users: int active_users: int departments_count: int with_manager: int by_department: list[DepartmentBreakdown] by_role: list[RoleBreakdown] by_status: list[StatusBreakdown] class OrgTreeNode(BaseModel): id: str node_type: str name: str email: str | None department: str | None position: str | None status: str member_count: int | None = None children: list["OrgTreeNode"] class PositionOrgIncumbent(BaseModel): user_id: str helix_person_id: str | None = None display_name: str email: str | None = None class PositionOrgNode(BaseModel): code: str name: str parent_code: str | None = None department: str | None = None company: str | None = None location: str | None = None status: str target_fte: float | None = None current_fte: int direct_reports_count: int = 0 incumbents: list[PositionOrgIncumbent] children: list["PositionOrgNode"] class DepartmentResponse(BaseModel): department: str count: int class UserUpdateRequest(BaseModel): display_name: str | None = None department: str | None = None position: str | None = None # ``status`` does NOT include ``"left"`` — see comment on # ``AdminAdjustableEmploymentStatus``. To put a user into ``left`` # state, callers must drive the offboarding flow. status: AdminAdjustableEmploymentStatus | None = None class RoleAssignRequest(BaseModel): role_name: str class RoleBatchRequest(BaseModel): """Shared body for POST /auth/roles/batch-assign and batch-revoke.""" user_ids: list[UUID] = Field(min_length=1, max_length=500) role_name: str class RoleResponse(BaseModel): id: UUID name: str description: str | None is_system: bool model_config = {"from_attributes": True}