from datetime import date, datetime from decimal import Decimal from typing import Any, Literal from uuid import UUID from pydantic import BaseModel, ConfigDict, Field, field_validator ASSET_APPROVAL_TEMPLATE_KEYS = ( "laptop_replacement", "computer_borrow", "asset_return", "offboarding_buyback", ) # Templates a ``leaver_pending`` (离职中) employee may still submit for # themselves, as a carve-out from the Leaver Blocked mutate lock # (2026-08-13). Both are offboarding *obligations* rather than new # privileges: 归还 hands the hardware back, 回购 buys it out. Everything # else on this endpoint hands a departing employee something new # (replacement laptop, borrow, software seat) and stays locked. # # Enforced in ``ApprovalService.create_asset_approval_request`` so the # HTTP route and the ``submit_approval_request`` MCP tool share one # decision. The router / MCP-runtime gates let the *endpoint* through # (see ``_LEAVER_MUTATE_EXACT_ALLOWLIST`` and the ``leaver_allowed`` # flag); this set is what actually decides. LEAVER_SELF_SERVICE_TEMPLATE_KEYS = frozenset({ "asset_return", "offboarding_buyback", }) class ApprovalTemplateOption(BaseModel): value: str text: str class ApprovalTemplateWidget(BaseModel): id: str name: str type: str required: bool = False visible: bool = True printable: bool = True display_condition: dict[str, Any] | None = None options: list[ApprovalTemplateOption] = Field(default_factory=list) raw_option: dict[str, Any] | None = None # Static description content for ``type=="text"`` widgets — the # multi-line notice / FY-year spec block / delivery-time advisory # Feishu admins author in the form designer. Empty for input # widgets. Frontend renders this read-only above/below input # fields so the applicant sees the same context the Feishu form # provides. description: str | None = None # Default value the Feishu form admin pre-baked into the widget # ("是否回购 默认=否", radio default options, prefilled text, …). # Frontend prefills the form with this value on modal open so the # applicant sees the same starting state Feishu's mobile form # would. For radioV2 / select widgets the parser resolves the raw # default against the option set up front so a stale default # cannot fire APPROVAL_FIELD_INVALID_OPTION at submit time. default_value: Any | None = None class ApprovalTemplateStage(BaseModel): """One node on the Feishu approval flow (审批环节). Parsed from the ``node_list`` / ``process_node_list`` payload on ``GET /approval/v4/approvals/{approval_code}`` so the self-service "create request" UI can show the applicant the full chain of approvers BEFORE they submit (the post-submit timeline already surfaces traversed nodes via ``ApprovalProgress.nodes``). The backend also feeds these into ``_build_progress_from_instance`` so un-traversed future nodes appear greyed-out on the detail modal. """ node_id: str | None = None name: str | None = None # ``AND`` (every approver must approve) / ``OR`` (any one approves) / # ``SEQUENCE`` / ``CC`` etc. — Frontend uses this for the badge text # only; we don't make semantic decisions on it server-side. node_type: str | None = None # Pre-resolved approver display names (empty when Feishu uses a # dynamic resolver — e.g. "applicant's direct manager", which is # only known at submission time). Same shape as # ``ApprovalProgressNode.approvers`` so the renderer can share code. approver_names: list[str] = Field(default_factory=list) # IT-admin-authored note shown to employees on the approval-flow # preview. Sourced from ``platform_settings['approval.stage_notes']`` # keyed by ``(approval_code, node_id)`` and joined in at response # construction time — refreshing the stage list from Feishu doesn't # touch notes; nodes that vanish from the definition simply stop # surfacing their orphan note (the storage row stays, harmless). note: str | None = None class FlowStageBundle(BaseModel): """All stages of one Feishu approval flow + its identity. Used by the admin "流程节点备注" Modal so IT can browse every Feishu-backed flow in one place (asset + software + cloud + security flows). Stages already carry notes joined from ``platform_settings`` so the Modal can edit in place. Public employee-facing endpoint (`GET /flow-stages/{code}`) returns the same shape so non-asset flows (api_key, vpn, dns, ...) can render the same preview Modal as the asset templates. """ flow_key: str label: str approval_code: str approval_name: str | None = None stages: list[ApprovalTemplateStage] = Field(default_factory=list) # When the upstream Feishu fetch failed for THIS flow only (e.g. # the approval_code is invalid / template was archived), the # admin Modal still needs to render the row + show the error # rather than vanishing the entry. Empty / None on success. fetch_error: str | None = None class StageNoteUpdateRequest(BaseModel): """``PUT /approval/admin/stage-notes`` body — upsert one note. Setting ``note=null`` deletes the entry. Setting to empty string is treated the same as null (no point persisting an empty note). """ approval_code: str = Field(min_length=1, max_length=128) node_id: str = Field(min_length=1, max_length=128) note: str | None = Field(default=None, max_length=2000) model_config = ConfigDict(extra="forbid") class AssetApprovalTemplateResponse(BaseModel): template_key: str approval_code: str approval_name: str requires_upload: bool = False widgets: list[ApprovalTemplateWidget] stages: list[ApprovalTemplateStage] = Field(default_factory=list) class AssetApprovalSubmitRequest(BaseModel): template_key: Literal[ "laptop_replacement", "computer_borrow", "asset_return", "offboarding_buyback", "software_seat", "software_addon", ] asset_id: UUID | None = None reason: str | None = Field(None, max_length=4000) notes: str | None = Field(None, max_length=4000) # ``answers`` ships into the Feishu approval form. We cap each # field to a generous 8 KB and the whole dict to 64 keys so a buggy # / malicious client (e.g. an LLM agent) can't inflate one approval # row to multi-megabytes — that would bloat audit_logs.detail and # break the Feishu submission anyway. The stricter per-template # widget validation lives in the approval service. answers: dict[str, Any] = Field(default_factory=dict) # Pre-launch hardening 2026-05-04 (P0 AP1): the previous # ``extra_source_data`` / ``extra_detail`` fields were declared # with ``exclude=True`` — but ``exclude=True`` only blocks # serialisation, not input parsing, so any authenticated employee # could POST a request with ``extra_source_data={"seat_allocation_id": # ""}`` and have the Feishu callback # activate someone else's seat. These two plumbing knobs have been # moved off the public schema and onto kw-only parameters of # ``ApprovalService.create_asset_approval_request`` — only internal # callers (asset/router.request_seat, mcp/tools) can pass them. # Defence-in-depth: refuse unknown fields on the public payload so # a future schema addition doesn't reopen this hole by accident. model_config = ConfigDict(extra="forbid") @field_validator("answers") @classmethod def _bound_answers(cls, v: dict[str, Any]) -> dict[str, Any]: if len(v) > 64: raise ValueError( f"answers may have at most 64 keys; got {len(v)}", ) for k, vv in v.items(): if isinstance(vv, str) and len(vv) > 8000: raise ValueError( f"answers[{k!r}] exceeds 8000 chars " f"({len(vv)} got)", ) if isinstance(vv, (list, tuple)) and len(vv) > 256: raise ValueError( f"answers[{k!r}] is a list of {len(vv)} items " "(max 256)", ) return v class AdminBuybackOnBehalfRequest(BaseModel): """Admin 代提回购 payload — IT starts the offboarding-buyback approval for a leaver-held asset on the leaver's behalf. Only ``asset_id`` is required; the buyback subject (the leaver) and the amount are derived server-side from the asset's current holder + IT policy. The admin optionally supplies a reason and a display currency. """ asset_id: UUID reason: str | None = Field(None, max_length=4000) notes: str | None = Field(None, max_length=4000) currency: str | None = Field(None, max_length=8) base_location: str | None = Field(None, max_length=256) last_working_day: date | None = None model_config = ConfigDict(extra="forbid") class AdminBuybackOnBehalfContextResponse(BaseModel): """Sensitive preflight data for the asset-admin-only buyback modal.""" last_working_day: date | None = None widgets: list[ApprovalTemplateWidget] = Field(default_factory=list) class AssetApprovalUploadResponse(BaseModel): token: str widget_type: str file_name: str class ApprovalBackedEmployeeRequestResponse(BaseModel): id: UUID request_number: str request_type: str status: str requester_id: UUID requester_name: str | None = None requester_email: str | None = None requester_employee_no: str | None = None requester_employment_status: str | None = None asset_id: UUID | None = None asset_tag: str | None = None asset_name: str | None = None reason: str | None = None notes: str | None = None detail: dict[str, Any] | None = None # Canonical buyback figure ``{"amount": "1234.00", "currency": "CNY"}`` # for offboarding_buyback / laptop_replacement-with-buyback rows; # None for every other type (and for no-buyback replacements). # Extracted server-side (``extract_buyback_amount``) so the list # column doesn't have to spelunk widget snapshots client-side. buyback_amount: dict[str, str] | None = None approval_code: str | None = None approval_name: str | None = None approval_instance_code: str | None = None approval_status: str | None = None created_at: datetime processed_at: datetime | None = None processed_by: UUID | None = None class AdminApprovalListItem(BaseModel): """Lean row for the unified admin all-approvals table. Backed by ``employee_requests`` so it covers every request kind (asset / software_seat / api_key / google_*) in one shot. The drawer view fetches a richer payload via :class:`AdminApprovalDetailResponse`. """ id: UUID request_number: str request_type: str status: str requester_id: UUID requester_name: str | None = None requester_email: str | None = None requester_employee_no: str | None = None requester_employment_status: str | None = None requester_department: str | None = None asset_id: UUID | None = None asset_tag: str | None = None asset_name: str | None = None reason: str | None = None approval_name: str | None = None approval_instance_code: str | None = None created_at: datetime processed_at: datetime | None = None class AdminApprovalTimelineEntry(BaseModel): """One row of the Feishu approval flow timeline (审批流程). Merges Feishu ``task_list`` (approver actions) and ``comment_list`` (commenter messages) into a single time-ordered stream so the admin drawer can render "who did what when" without the frontend having to merge two arrays. """ kind: Literal["task", "comment"] status: str | None = None node_id: str | None = None node_name: str | None = None user_id: str | None = None open_id: str | None = None user_name: str | None = None comment: str | None = None occurred_at: datetime | None = None class ApprovalProgressNode(BaseModel): """One node (审批环节) collapsed across all of its approver tasks. Feishu's ``task_list`` ships one entry per (node, approver) pair — a node with three approvers in OR mode shows up as three task rows. The detail drawer's progress indicator wants one Step per node, so the service groups task rows by ``node_id`` (or by arrival order when ``node_id`` is missing on legacy templates) and aggregates approver lists / status. ``node_id`` is the Feishu-side stable identifier (or a synthetic ``__fb_{i}`` fallback when legacy templates omit it). Frontend uses it to match ``AdminApprovalTimelineEntry.node_id`` entries back to their progress node — node *names* aren't unique (legacy templates legitimately have two "部门审批" steps), so name-based matching could route both nodes' timeline entries into the first match. Added 2026-05-15 alongside the approval-detail Modal rewrite to fix that bucketing collision. """ node_id: str | None = None name: str | None = None status: str | None = None # PENDING / APPROVED / REJECTED / DONE / ... approvers: list[str] = Field(default_factory=list) class ApprovalProgress(BaseModel): """Approval-flow progress summary (当前环节 / 共 N 环节). Computed from the live Feishu ``task_list``. ``total_nodes`` is the count of distinct nodes the approval has surfaced so far — Feishu doesn't expose the static template node count separately, so for branching templates this is "nodes that exist along the current path" which is exactly what the requester / admin care about. ``current_node_index`` is 1-based and ``None`` once the flow has terminated (approved / rejected / cancelled). ``sync_lag`` is an explicit signal that no node is PENDING but the flow isn't finished either (e.g. final node has only skipped tasks waiting for Feishu's instance status to propagate). When True, ``current_node_index`` points past the last completed node so the Steps strip renders every node as done. Frontend MUST read this flag directly — inferring sync-lag from ``current_node_name is None`` would false-fire on legitimately-in-progress flows whose current node is anonymous (older Feishu templates omit ``node_name``). """ total_nodes: int completed_nodes: int current_node_index: int | None = None current_node_name: str | None = None current_approvers: list[str] = Field(default_factory=list) is_finished: bool = False sync_lag: bool = False nodes: list[ApprovalProgressNode] = Field(default_factory=list) class AdminApprovalDetailResponse(BaseModel): id: UUID request_number: str request_type: str status: str requester_id: UUID requester_name: str | None = None requester_email: str | None = None requester_employee_no: str | None = None requester_employment_status: str | None = None requester_department: str | None = None asset_id: UUID | None = None asset_tag: str | None = None asset_name: str | None = None reason: str | None = None notes: str | None = None detail: dict[str, Any] | None = None approval_code: str | None = None approval_name: str | None = None approval_instance_code: str | None = None approval_status: str | None = None instance_status_raw: str | None = None timeline: list[AdminApprovalTimelineEntry] = Field(default_factory=list) timeline_fetched_at: datetime | None = None timeline_error: str | None = None progress: ApprovalProgress | None = None created_at: datetime processed_at: datetime | None = None class FeishuUrlVerificationRequest(BaseModel): type: str challenge: str token: str | None = None class FeishuCallbackEnvelope(BaseModel): model_config = ConfigDict(populate_by_name=True) schema_: str | None = Field(default=None, alias="schema") header: dict[str, Any] | None = None event: dict[str, Any] | None = None type: str | None = None challenge: str | None = None token: str | None = None encrypt: str | None = None class FeishuApprovalDefinitionResponse(BaseModel): approval_name: str form: str | list[dict[str, Any]] class FeishuCreateInstanceResponse(BaseModel): instance_code: str instance_status: str | None = None class FeishuResolvedUser(BaseModel): email: str user_id: str | None = None open_id: str | None = None class FeishuAmountValue(BaseModel): amount: Decimal | str currency: str @field_validator("currency") @classmethod def validate_currency(cls, value: str) -> str: return value.upper() class BuybackQuoteResponse(BaseModel): """Quote returned for the offboarding-buyback form's Amount field. ``amount`` is the buyback figure in ``currency``. A registered purchase entity's banking currency is authoritative, keeping the amount aligned with the payment information shown to the employee. ``prices`` remains available for finance and audit comparisons. """ asset_id: UUID amount: Decimal currency: str default_currency: str supported_currencies: list[str] mode: Literal["depreciation", "floor"] tier: int months_used: int prices: dict[str, Decimal]