"""Unified request-detail builder — one rich detail payload for EVERY source the "我的申请" tracker (``build_my_requests``) aggregates. The list rows are source-agnostic; so is this. ``ApprovalDetailModal`` already gives ``employee_request`` rows a polished Feishu-flow detail, but the other sources (cloud scope / DNS / VPN / AI-tool / announcement / domain / api-key release / permission / SSO) only ever had the thin 6-field drawer. This builder returns a single normalised shape the frontend ``RequestDetailModal`` renders consistently for all of them: * summary + requester block (concept-identical to the admin modal), * a domain-specific ``content_fields`` list (the meaningful columns of the backing row, not the raw blob), * a synthesised lifecycle ``timeline`` (submitted → in-flight → terminal) so non-Feishu sources still show *where they are*, * the raw payload for the power-user "原始数据" tab, * ``has_feishu_flow`` so the frontend knows to additionally fetch the rich ``/approval/my/requests/{id}`` enrichment (审批流 + 申请表单 tabs) for the one source that has it. Owner-scoped: the builder resolves the row only when its requester is in ``person_ids`` (``[caller]`` for the self view). A stranger's id raises ``NotFoundError`` — never a 403 — so the endpoint can't be used as an existence oracle for other people's request ids (mirrors ``/approval/my/requests/{id}``). """ from __future__ import annotations from datetime import datetime from typing import Any from uuid import UUID from pydantic import BaseModel, Field from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.models import ApprovalRequest from app.approval.schemas import AdminApprovalTimelineEntry, ApprovalProgress from app.core.exceptions import NotFoundError from app.domains.ai_tool_application.models import AiToolApplication from app.domains.announcement_request.models import AnnouncementRequest from app.domains.apikey.models import ApiKeyReleaseRequest from app.domains.asset.models import EmployeeRequest from app.domains.cmdb.models import ( CmdbCloudAccessGrant, CmdbCloudGrantRenewalRequest, CmdbCloudScope, CmdbCloudScopeRequest, CmdbDnsRecordRequest, ) from app.domains.data_access.models import DataAccessRequest from app.domains.domain_registration.models import DomainRegistration from app.domains.me_universe.my_requests import ( _EMPLOYEE_TYPE_CATEGORY, _FAILURE_STATUSES, _OPEN_STATUSES, CAT_AI, CAT_CLOUD, CAT_EMAIL, CAT_GENERAL, CAT_NETWORK, CAT_SOFTWARE, SRC_AI_TOOL, SRC_ANNOUNCEMENT, SRC_APIKEY_RELEASE, SRC_APPROVAL_FLOW, SRC_CLOUD_GRANT, SRC_CLOUD_SCOPE, SRC_DNS, SRC_DOMAIN, SRC_EMAIL_GROUP_JOIN, SRC_EMPLOYEE, SRC_GITHUB_ID, SRC_IDENTITY_CONSUMER, SRC_PERMISSION, SRC_SSO, SRC_SUPPORT, SRC_VPN, ST_CANCELLED, ST_COMPLETED, ST_FAILED, ST_IN_PROGRESS, ST_PENDING, ST_REJECTED, ST_REVOKED, _normalize_status, cloud_grant_lifecycle_status, cloud_grant_outcome_reason, resolve_apikey_release_approver, resolve_normalized_status, resolve_recourse, ) from app.domains.permission.models import SoftwarePermission, UserPermission from app.domains.sso.models import SsoApp, SsoGrant from app.domains.vpn_application.models import VpnApplication from app.identity.person_models import IdentityPerson class RequestTimelineEvent(BaseModel): """One synthesised lifecycle event. ``key`` is an i18n key the frontend localises (``requestDetail.timeline.``); ``state`` drives the dot colour (done / current / error / wait).""" key: str ts: str | None = None state: str # done | current | error | wait note: str | None = None class RequestContentField(BaseModel): """One human-meaningful field of the backing row. ``key`` is a raw identifier the frontend localises where it knows it (``requestDetail.field.``) and otherwise shows verbatim.""" key: str value: str class UnifiedRequestDetail(BaseModel): id: str # f"{source}:{pk}" source: str category: str request_type: str title: str subtitle: str | None = None status: str # normalised lifecycle raw_status: str created_at: str updated_at: str | None = None deep_link: str feishu_instance_code: str | None = None requester_id: str requester_display: str | None = None requester_email: str | None = None # 工号 + 在职状态:让前端 PersonTag 把离职申请人标灰显示。 requester_employee_no: str | None = None requester_employment_status: str | None = None requester_department: str | None = None content_fields: list[RequestContentField] timeline: list[RequestTimelineEvent] raw: dict[str, Any] # ── Outcome (the unhappy / terminal path) — mirrors MyRequestItem ─ # WHY a terminal request ended this way. For Feishu-routed sources the # reject comment is recovered from the live approval flow when # ``with_live_approval`` runs (``_attach_live_approval_flow``); for the # rest it comes from the row's own columns. None on open / completed. outcome_reason: str | None = None outcome_reason_code: str | None = None needs_attention: bool = False recourse: list[str] = Field(default_factory=list) # employee_request carries a real Feishu approval instance — the # frontend fetches the rich enrichment to render 审批流 + 申请表单. has_feishu_flow: bool = False # Live Feishu approval flow for the OTHER sources that still run a # Feishu approval instance (DNS / cloud-scope / VPN / AI-tool / # announcement / domain / SSO). Populated only by the detail endpoint # (``with_live_approval=True``); the frontend renders the same # ``ApprovalFlowTimeline`` employee_request uses, so a requester can # see WHO is currently approving instead of a bare "Awaiting approval". # All None/empty when the row has no instance or the live fetch failed # — the frontend then falls back to the synthesised ``timeline``. approval_progress: ApprovalProgress | None = None approval_timeline: list[AdminApprovalTimelineEntry] = Field(default_factory=list) approval_timeline_error: str | None = None def _iso(dt: datetime | None) -> str | None: return dt.isoformat() if dt else None def _fields(*pairs: tuple[str, Any]) -> list[RequestContentField]: """Build a content-field list, dropping empty values and stringifying the rest. Order is preserved.""" out: list[RequestContentField] = [] for key, value in pairs: if value is None or value == "": continue out.append(RequestContentField(key=key, value=str(value))) return out def _timeline( *, normalized: str, created_at: datetime, updated_at: datetime | None, outcome_reason: str | None = None, ) -> list[RequestTimelineEvent]: """Synthesise a generic lifecycle timeline from the normalised status. Every request starts with ``submitted`` (done). Terminal states add a final event; still-open states add a ``current`` marker. This is the non-Feishu fallback — employee_request rows overlay the real Feishu flow on top of (well, instead of) this in the frontend. ``outcome_reason`` (when present) is attached as the terminal event's ``note`` so a rejected / failed / revoked request shows WHY inline on the timeline, not just a bare red dot. """ events: list[RequestTimelineEvent] = [ RequestTimelineEvent(key="submitted", ts=_iso(created_at), state="done"), ] end_ts = _iso(updated_at) note = (outcome_reason or "").strip() or None if normalized == ST_PENDING: events.append(RequestTimelineEvent(key="pending", state="current")) elif normalized == ST_IN_PROGRESS: events.append(RequestTimelineEvent(key="inProgress", ts=end_ts, state="current")) elif normalized == ST_COMPLETED: events.append(RequestTimelineEvent(key="completed", ts=end_ts, state="done")) elif normalized == ST_REJECTED: events.append(RequestTimelineEvent(key="rejected", ts=end_ts, state="error", note=note)) elif normalized == ST_CANCELLED: events.append(RequestTimelineEvent(key="cancelled", ts=end_ts, state="wait", note=note)) elif normalized == ST_REVOKED: events.append(RequestTimelineEvent(key="revoked", ts=end_ts, state="error", note=note)) elif normalized == ST_FAILED: events.append(RequestTimelineEvent(key="failed", ts=end_ts, state="error", note=note)) return events async def _resolve_requester( db: AsyncSession, requester_id: str, ) -> tuple[str | None, str | None, str | None, str | None, str | None]: """(display_name, work_email, employee_no, employment_status, department) for the requester, or all-None if the id doesn't resolve.""" try: pid = UUID(str(requester_id)) except (TypeError, ValueError): return None, None, None, None, None person = await db.get(IdentityPerson, pid) if person is None: return None, None, None, None, None return ( person.display_name, person.work_email, person.employee_no, person.employment_status, person.department, ) def _owned(requester_id: Any, person_ids: set[str] | None) -> bool: # ``person_ids is None`` = admin/unscoped resolution (e.g. the admin # urge path): any requester matches. The self view always passes the # caller's id set, so a stranger's row still resolves to None there. return person_ids is None or str(requester_id) in person_ids # ── Per-source detail builders ────────────────────────────────────── # Each returns a fully-built UnifiedRequestDetail or None when the row # doesn't exist / isn't owned by the caller. They share ``_finish``. async def _finish( db: AsyncSession, *, source: str, pk: Any, category: str, request_type: str, title: str, subtitle: str | None, raw_status: str, normalized: str, created_at: datetime, updated_at: datetime | None, deep_link: str, requester_id: Any, feishu_instance_code: str | None, content_fields: list[RequestContentField], raw: dict[str, Any], has_feishu_flow: bool = False, outcome_reason: str | None = None, outcome_reason_code: str | None = None, ) -> UnifiedRequestDetail: display, email, emp_no, emp_status, dept = await _resolve_requester( db, str(requester_id), ) # Reason / code only carry on a terminal row (mirrors ``_mk``). is_terminal = normalized not in _OPEN_STATUSES reason = (outcome_reason or "").strip() or None if is_terminal else None reason_code = outcome_reason_code if (is_terminal and reason) else None return UnifiedRequestDetail( id=f"{source}:{pk}", source=source, category=category, request_type=request_type, title=title, subtitle=subtitle, status=normalized, raw_status=raw_status, created_at=_iso(created_at) or "", updated_at=_iso(updated_at), deep_link=deep_link, feishu_instance_code=feishu_instance_code, requester_id=str(requester_id), requester_display=display, requester_email=email, requester_employee_no=emp_no, requester_employment_status=emp_status, requester_department=dept, content_fields=content_fields, timeline=_timeline( normalized=normalized, created_at=created_at, updated_at=updated_at, outcome_reason=reason, ), raw=raw, has_feishu_flow=has_feishu_flow, outcome_reason=reason, outcome_reason_code=reason_code, needs_attention=normalized in _FAILURE_STATUSES, recourse=resolve_recourse(source, normalized), ) async def _detail_employee(db, pk, person_ids): try: row_id = UUID(str(pk)) except (TypeError, ValueError): return None row = await db.get(EmployeeRequest, row_id) if row is None or not _owned(row.requester_id, person_ids): return None normalized = resolve_normalized_status(SRC_EMPLOYEE, row.request_type, row.status) detail = row.detail if isinstance(row.detail, dict) else {} # The live Feishu approval instance is recorded on the request at # submission time under detail.feishu.instance_code (same field the # supplement-upload / Feishu-cancel paths read). Surfacing it here is # what lets 催办 resolve the CURRENT pending approver (not broadcast to # the manager + IT fallback) and link the recipient into Feishu审批 # instead of the requester's own /my/requests page. None for legacy / # non-Feishu rows → 催办 degrades to the manager+IT fallback as before. feishu_instance_code = (detail.get("feishu") or {}).get("instance_code") or None return await _finish( db, source=SRC_EMPLOYEE, pk=row.id, category=_EMPLOYEE_TYPE_CATEGORY.get(row.request_type, CAT_GENERAL), request_type=row.request_type, title=row.request_type, # frontend localises subtitle=row.request_number, raw_status=row.status, normalized=normalized, created_at=row.created_at, updated_at=row.processed_at, deep_link=f"/my/requests?focus={row.id}", requester_id=row.requester_id, feishu_instance_code=feishu_instance_code, content_fields=_fields( ("request_number", row.request_number), ("reason", row.reason), ("notes", row.notes), ), raw=detail, # employee_request is the one source with a real Feishu instance — # the frontend overlays the rich approval-flow + form-snapshot tabs # by additionally fetching /approval/my/requests/{id}. has_feishu_flow # stays True so _attach_live_approval_flow short-circuits (the # detail endpoint's live overlay is unchanged); the instance_code # above is consumed by the 催办 recipient/deep-link resolution only. has_feishu_flow=True, ) async def _detail_cloud_scope(db, pk, person_ids): row = await db.get(CmdbCloudScopeRequest, str(pk)) if row is None or not _owned(row.requester_person_id, person_ids): return None return await _finish( db, source=SRC_CLOUD_SCOPE, pk=row.id, category=CAT_CLOUD, request_type="cloud_scope_request", title=row.name, subtitle=row.provider, raw_status=row.status, normalized=_normalize_status(row.status), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/cloud/scope-requests", requester_id=row.requester_person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("provider", row.provider), ("scope_name", row.name), ("rejection_reason", row.rejection_reason), ("failure_reason", row.failure_reason), ), raw={"name": row.name, "provider": row.provider, "status": row.status}, outcome_reason=row.rejection_reason or row.failure_reason, ) async def _detail_cloud_grant(db, pk, person_ids): row = await db.get(CmdbCloudAccessGrant, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None scope = ( await db.get(CmdbCloudScope, UUID(str(row.scope_id))) if row.scope_id else None ) fields = [ ("role", row.role_display_name or row.role_key), ("scope_name", (scope.display_name or scope.external_id) if scope else "—"), ] # For a still-pending grant surface the approver as a content field so # the detail view answers "who is this waiting on" too. Owner name when # there is one; the cloud-admin team otherwise (FE leaves the value # blank-safe — the list item's ``pending_approver_kind`` carries the # i18n'd team label). if row.status == "pending" and scope is not None: owner = ( await db.get(IdentityPerson, UUID(str(scope.owner_person_id))) if scope.owner_person_id else None ) if ( owner is not None and owner.employment_status not in ("left", "leaver_pending") ): fields.append(("approver", owner.display_name or owner.work_email)) cg_reason, cg_code = cloud_grant_outcome_reason( row.status, row.detail, row.revoke_reason, ) return await _finish( db, source=SRC_CLOUD_GRANT, pk=row.id, category=CAT_CLOUD, request_type="cloud_access_request", title=( (scope.display_name or scope.external_id) if scope else "cloud access" ), subtitle=row.role_display_name or row.role_key, raw_status=row.status, normalized=cloud_grant_lifecycle_status(row.status, row.detail), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/cloud/grants", requester_id=row.person_id, # Internal approval (no Feishu instance) — the live-flow overlay # is a no-op for this source. feishu_instance_code=None, content_fields=_fields(*fields), raw={ "role_key": row.role_key, "scope_id": str(row.scope_id), "status": row.status, }, outcome_reason=cg_reason, outcome_reason_code=cg_code, ) async def _detail_dns(db, pk, person_ids): row = await db.get(CmdbDnsRecordRequest, str(pk)) if row is None or not _owned(row.requester_person_id, person_ids): return None return await _finish( db, source=SRC_DNS, pk=row.id, category=CAT_NETWORK, request_type=f"dns_{row.action}", title=f"{row.name}.{row.zone_name} ({row.record_type})", subtitle=row.action, raw_status=row.status, normalized=_normalize_status(row.status), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/cloud/dns", requester_id=row.requester_person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("dns_action", row.action), ("record_type", row.record_type), ("name", f"{row.name}.{row.zone_name}"), ("value", row.value), ("error_message", row.error_message), ), raw={ "action": row.action, "record_type": row.record_type, "name": row.name, "zone_name": row.zone_name, "value": row.value, "status": row.status, }, outcome_reason=row.error_message, ) async def _detail_vpn(db, pk, person_ids): row = await db.get(VpnApplication, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None return await _finish( db, source=SRC_VPN, pk=row.id, category=CAT_NETWORK, request_type="vpn_application", title=row.vpn_node, subtitle=None, raw_status=row.status, normalized=resolve_normalized_status(SRC_VPN, "vpn_application", row.status), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("vpn_node", row.vpn_node), ("usage_description", row.usage_description), ("resource_checklist", row.resource_checklist), ), raw={ "vpn_node": row.vpn_node, "usage_description": row.usage_description, "status": row.status, }, outcome_reason=row.revoke_reason, ) async def _detail_ai_tool(db, pk, person_ids): row = await db.get(AiToolApplication, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None return await _finish( db, source=SRC_AI_TOOL, pk=row.id, category=CAT_AI, request_type="ai_tool_application", title=row.ai_name, subtitle=None, raw_status=row.status, normalized=resolve_normalized_status( SRC_AI_TOOL, "ai_tool_application", row.status ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("ai_name", row.ai_name), ("purpose", row.purpose), ), raw={"ai_name": row.ai_name, "purpose": row.purpose, "status": row.status}, ) async def _detail_announcement(db, pk, person_ids): row = await db.get(AnnouncementRequest, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None return await _finish( db, source=SRC_ANNOUNCEMENT, pk=row.id, category=CAT_GENERAL, request_type="announcement_request", title=(row.description or "")[:80] or "announcement_request", subtitle=None, raw_status=row.status, normalized=resolve_normalized_status( SRC_ANNOUNCEMENT, "announcement_request", row.status ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields(("description", row.description)), raw={"description": row.description, "status": row.status}, ) async def _detail_approval_flow(db, pk, person_ids): """Detail for a new-flow request. A source that appears in the tracker list but has no builder here gives its own requester a red 「加载失败」 on 查看 — the #2229 bug class. Registered together with the provider, never separately. """ from app.domains.approval_registry.models import ApprovalFlowRequest row = await db.get(ApprovalFlowRequest, str(pk)) if row is None or not _owned(str(row.requester_person_id), person_ids): return None return await _finish( db, source=SRC_APPROVAL_FLOW, pk=row.id, category=CAT_GENERAL, request_type="approval_flow_request", title=row.proposed_name, subtitle=(row.purpose or "")[:80] or None, raw_status=row.status, normalized=resolve_normalized_status( SRC_APPROVAL_FLOW, "approval_flow_request", row.status ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=str(row.requester_person_id), # No Feishu instance, on purpose: this one request is decided by # IT inside Helix — routing it through Feishu would mean creating # an approval template in order to be allowed to ask for one. feishu_instance_code=None, content_fields=_fields( ("proposed_name", row.proposed_name), ("purpose", row.purpose), ("proposed_category", row.proposed_category), ), raw={ "proposed_name": row.proposed_name, "purpose": row.purpose, "status": row.status, "decision_note": row.decision_note, }, ) async def _detail_domain(db, pk, person_ids): row = await db.get(DomainRegistration, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None return await _finish( db, source=SRC_DOMAIN, pk=row.id, category=CAT_NETWORK, request_type="domain_registration", title=row.domain_name, subtitle=None, raw_status=row.status, normalized=resolve_normalized_status( SRC_DOMAIN, "domain_registration", row.status ), created_at=row.created_at, updated_at=row.updated_at, # Re-registration is submitted on /my/requests (EmployeeRequestPage, # Network), not the read-only /my/cloud/domains list — keep the # reapply recourse landing on a page with a submit form. deep_link="/my/requests", requester_id=row.person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("domain_name", row.domain_name), ("purpose", row.purpose), ), raw={"domain_name": row.domain_name, "purpose": row.purpose, "status": row.status}, outcome_reason=row.revoke_reason, ) async def _detail_apikey_release(db, pk, person_ids): try: row_id = UUID(str(pk)) except (TypeError, ValueError): row_id = pk row = await db.get(ApiKeyReleaseRequest, row_id) if row is None or not _owned(row.requester_id, person_ids): return None # Name who a still-pending release is waiting on (manager, else the # IT-admin team) so the 概览 tab answers "等谁审批" — same approver the # list row + the modal's pending hint show. Only a named manager # becomes a content field; the team fallback rides on the list item's # ``pending_approver_kind`` (the FE i18n's the team label). fields = [ ("reason_code", row.reason_code), ("reason_note", row.reason_note), ] if row.status == "pending": appr_kind, appr_label = await resolve_apikey_release_approver( db, row.requester_id, ) if appr_kind == "owner" and appr_label: fields.append(("approver", appr_label)) return await _finish( db, source=SRC_APIKEY_RELEASE, pk=row.id, category=CAT_SOFTWARE, request_type="apikey_release_request", title=row.reason_code, # frontend localises subtitle=row.reason_note, raw_status=row.status, # Use the shared resolver (NOT bare _normalize_status) so an # ``approved`` (= done) release reads as completed, in lockstep # with the list provider. See _APPROVED_IS_TERMINAL_SOURCES. normalized=resolve_normalized_status( SRC_APIKEY_RELEASE, "apikey_release_request", row.status, ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/me/api-keys", requester_id=row.requester_id, feishu_instance_code=None, content_fields=_fields(*fields), raw={"reason_code": row.reason_code, "status": row.status}, ) async def _detail_permission(db, pk, person_ids): try: row_id = UUID(str(pk)) except (TypeError, ValueError): row_id = pk row = await db.get(UserPermission, row_id) if row is None or not _owned(row.user_id, person_ids): return None perm = await db.get(SoftwarePermission, row.permission_id) return await _finish( db, source=SRC_PERMISSION, pk=row.id, category=CAT_SOFTWARE, request_type="software_permission", title=(perm.display_name if perm else None) or "software_permission", subtitle=None, raw_status=row.status, normalized=_normalize_status(row.status), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.user_id, feishu_instance_code=None, content_fields=_fields( ("permission", perm.display_name if perm else None), ), raw={"status": row.status}, ) async def _detail_sso(db, pk, person_ids): row = await db.get(SsoGrant, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None app = await db.get(SsoApp, row.created_app_id) if row.created_app_id else None return await _finish( db, source=SRC_SSO, pk=row.id, category=CAT_SOFTWARE, request_type="sso_grant", title=row.system_name, subtitle=None, raw_status=row.status, normalized=resolve_normalized_status(SRC_SSO, "sso_grant", row.status), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("system_name", row.system_name), ("sso_provider", row.sso_provider), ("system_owner", row.system_owner_label), ("reason", row.reason), ("acs_url", row.acs_url), ("entity_id", row.entity_id), ("metadata_provided", bool(row.saml_metadata_xml)), ("certificate_provided", bool(row.x509_cert)), ("created_app_id", row.created_app_id), ("app_external_id", app.external_id if app else None), ("idp_config_ready", bool( app and app.idp_sso_url and app.idp_entity_id and app.idp_certificate )), ("config_delivered_at", app.config_delivered_at if app else None), ), # Deliberately exclude SAML XML, x509 certificates and IdP # certificate material from the generic/raw admin payload. raw={ "system_name": row.system_name, "sso_provider": row.sso_provider, "status": row.status, "created_app_id": row.created_app_id, "metadata_provided": bool(row.saml_metadata_xml), "certificate_provided": bool(row.x509_cert), "config_delivered_at": _iso(app.config_delivered_at) if app else None, }, outcome_reason=row.revoke_reason, ) async def _detail_data_access(db, pk, person_ids): try: row_id = UUID(str(pk)) except (TypeError, ValueError): row_id = pk row = await db.get(DataAccessRequest, row_id) if row is None or not _owned(row.requester_person_id, person_ids): return None normalized = { "pending_feishu": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "stale_requires_resubmit": ST_FAILED, "cancelled": ST_CANCELLED, }.get(row.status, row.status) return await _finish( db, source="data_access", pk=row.id, category=CAT_GENERAL, request_type=row.approval_flow_key, title=row.principal_email_snapshot, subtitle=row.entitlement_snapshot, raw_status=row.status, normalized=normalized, created_at=row.created_at, updated_at=row.updated_at, deep_link=f"/my/data-access/requests/{row.id}", requester_id=row.requester_person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("principal_email", row.principal_email_snapshot), ("principal_type", row.principal_type_snapshot), ("entitlement", row.entitlement_snapshot), ("business_reason", row.business_reason), ("reconcile_status", row.reconcile_status), ), raw={ "approval_flow_key": row.approval_flow_key, "status": row.status, "reconcile_status": row.reconcile_status, }, outcome_reason=row.last_reconcile_error, ) async def _detail_cloud_renewal(db, pk, person_ids): # The list collapses all rows from one multi-role submission to its # request_group_id. Legacy single-role rows use their own id. rows = (await db.execute( select(CmdbCloudGrantRenewalRequest).where( (CmdbCloudGrantRenewalRequest.request_group_id == str(pk)) | (CmdbCloudGrantRenewalRequest.id == str(pk)) ).order_by(CmdbCloudGrantRenewalRequest.created_at) )).scalars().all() if not rows or not _owned(rows[0].requested_by_person_id, person_ids): return None row = rows[0] scope = await db.get(CmdbCloudScope, row.scope_id) scope_name = scope.display_name or scope.external_id if scope else row.scope_id normalized = { "pending_owner": ST_PENDING, "pending_admin": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "lapsed": ST_CANCELLED, }.get(row.status, row.status) return await _finish( db, source="cloud_renewal", pk=pk, category=CAT_CLOUD, request_type="cloud_grant_renewal", title=scope_name, subtitle=f"{len(rows)} role(s)", raw_status=row.status, normalized=normalized, created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/cloud/grants", requester_id=row.requested_by_person_id, feishu_instance_code=None, content_fields=_fields( ("scope_id", row.scope_id), ("requested_ttl_days", row.requested_ttl_days), ("reason", row.reason), ("role_count", len(rows)), ("is_revival", any(r.is_revival for r in rows)), ("applied_expires_at", row.applied_expires_at), ("decision_note", row.decision_note), ), raw={ "request_group_id": row.request_group_id, "status": row.status, "grant_ids": [r.grant_id for r in rows], }, outcome_reason=row.decision_note, ) async def _detail_soc2_email_group(db, pk, person_ids): try: row_id = UUID(str(pk)) except (TypeError, ValueError): row_id = pk row = await db.get(ApprovalRequest, row_id) if ( row is None or row.approval_type != "soc2_email_group_grant" or not _owned(row.applicant_id, person_ids) ): return None normalized = { "pending": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, }.get(row.status, row.status) payload = row.source_payload if isinstance(row.source_payload, dict) else {} return await _finish( db, source="soc2_email_group", pk=row.id, category=CAT_EMAIL, request_type="soc2_email_group_grant", title="soc2@pingcap.com", subtitle=row.external_id, raw_status=row.status, normalized=normalized, created_at=row.created_at, updated_at=row.updated_at, deep_link="/admin/process-approval", requester_id=row.applicant_id or "", feishu_instance_code=row.external_id, content_fields=_fields( ("target_group", "soc2@pingcap.com"), ("feishu_instance_code", row.external_id), ), # External-origin form payload may contain more personal data than # the approval centre needs. Keep only non-sensitive identifiers. raw={"status": row.status, "approval_type": row.approval_type, "external_id": row.external_id, "payload_present": bool(payload)}, ) async def _detail_support(db, pk, person_ids): """Support ticket. Without this the tracker's 「查看」 was a dead end. ``_PROVIDERS`` lists support / email_group_join / identity_consumer, ``_DETAIL_BUILDERS`` did not — and the frontend routes everything except ``employee_request`` through the generic detail modal, so an employee who had ever filed an IT ticket saw their own row in ``/my/requests``, clicked 查看, and got a red "load failed" toast over an empty dialog (#2144). The urge path already degrades gracefully for builder-less sources; the detail path did not. """ from app.domains.support.models import SupportTicket row = await db.get(SupportTicket, str(pk)) if row is None or not _owned(row.requester_id, person_ids): return None return await _finish( db, source=SRC_SUPPORT, pk=row.id, category=CAT_GENERAL, request_type="support_ticket", title=row.title, subtitle=row.ticket_number, raw_status=row.status, normalized=resolve_normalized_status( SRC_SUPPORT, "support_ticket", row.status, ), created_at=row.created_at, updated_at=row.updated_at, # The ticket has a real page of its own — send the reader there # rather than keeping them in a modal that can only summarise. deep_link=f"/me/support/{row.id}", requester_id=row.requester_id, feishu_instance_code=None, content_fields=_fields( ("category", row.category), ("priority", row.priority), ("body", row.body), ), raw={"status": row.status, "ticket_number": row.ticket_number}, # #2245: the column is ``resolution`` — ``resolution_note`` never # existed, so this was a constant None and a resolved ticket's # write-up never reached the detail modal. The repo bans # ``getattr`` defaults on ORM columns for exactly this reason: a # renamed / mistyped column must fail LOUDLY, not silently blank # the field it feeds. outcome_reason=row.resolution, ) async def _detail_email_group_join(db, pk, person_ids): """Mail-group join request (#2144).""" from app.domains.email_group.models import EmailGroup, EmailGroupJoinRequest row = await db.get(EmailGroupJoinRequest, str(pk)) if row is None or not _owned(row.requester_id, person_ids): return None group = await db.get(EmailGroup, str(row.email_group_id)) return await _finish( db, source=SRC_EMAIL_GROUP_JOIN, pk=row.id, category=CAT_EMAIL, request_type="email_group_join", title=(group.email if group else "") or "", subtitle=(group.name if group else None), raw_status=row.status, normalized=resolve_normalized_status( SRC_EMAIL_GROUP_JOIN, "email_group_join", row.status, ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/company/mail-groups", requester_id=row.requester_id, feishu_instance_code=None, content_fields=_fields( ("group_email", group.email if group else None), ("group_name", group.name if group else None), ("reason", row.reason), ), raw={"status": row.status}, # Real column names — direct access so a rename breaks the build # instead of blanking the field (#2245). outcome_reason=row.decision_note, ) async def _detail_github_id(db, pk, person_ids): """GitHub ID 修改单。""" from app.domains.github_org.models import GithubIdChangeRequest row = await db.get(GithubIdChangeRequest, str(pk)) if row is None or not _owned(row.person_id, person_ids): return None # 三态如实呈现:``None`` 是「没问到 GitHub」,不是「查无此人」。把它 # 渲染成后者,会让看详情的人以为自己填错了。 verified = { True: "yes", False: "no", None: "unknown", }[row.github_exists] return await _finish( db, source=SRC_GITHUB_ID, pk=row.id, category=CAT_GENERAL, request_type="github_id_change", title=row.new_github_id, subtitle=row.old_github_id, raw_status=row.status, normalized=resolve_normalized_status( SRC_GITHUB_ID, "github_id_change", row.status, ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/me/settings?tab=profile", requester_id=row.person_id, feishu_instance_code=None, content_fields=_fields( ("old_github_id", row.old_github_id), ("new_github_id", row.new_github_id), ("github_exists", verified), ("reason", row.reason), # 回写 HRIS 失败是申请人看得见的事实:值在 Helix/飞书已经生效, # 但 SF 还是旧的。藏起来会让人以为哪里都同步好了。 ("hris_write_status", row.hris_write_status), ), raw={"status": row.status}, outcome_reason=row.decision_note, ) async def _detail_identity_consumer(db, pk, person_ids): """Identity-consumer request (#2144).""" from app.identity.consumers.models import IdentityConsumerRequest row = await db.get(IdentityConsumerRequest, str(pk)) if ( row is None or row.deleted_at is not None or not _owned(row.requester_person_id, person_ids) ): return None return await _finish( db, source=SRC_IDENTITY_CONSUMER, pk=row.id, category=CAT_GENERAL, request_type="identity_consumer_request", title=row.name, subtitle=row.slug, raw_status=row.status, normalized=resolve_normalized_status( SRC_IDENTITY_CONSUMER, "identity_consumer_request", row.status, ), created_at=row.created_at, updated_at=row.updated_at, deep_link="/my/requests", requester_id=row.requester_person_id, feishu_instance_code=row.feishu_instance_code, content_fields=_fields( ("slug", row.slug), # #2245: the columns are ``description`` and ``reject_reason`` # — ``purpose`` / ``decision_note`` do not exist on this model, # so both were constant None. Worst of the three: a REJECTED # identity-consumer request showed a red terminal state with no # reason, while ``reject_reason`` held one, and ``_finish`` # gates ``outcome_reason_code`` + the timeline note on that # value being present — the entire "why" layer went dark for # this source. ("purpose", row.description), ), raw={"status": row.status, "slug": row.slug}, outcome_reason=row.reject_reason, ) _DETAIL_BUILDERS = { SRC_EMPLOYEE: _detail_employee, SRC_CLOUD_SCOPE: _detail_cloud_scope, SRC_CLOUD_GRANT: _detail_cloud_grant, SRC_DNS: _detail_dns, SRC_VPN: _detail_vpn, SRC_AI_TOOL: _detail_ai_tool, SRC_ANNOUNCEMENT: _detail_announcement, SRC_APPROVAL_FLOW: _detail_approval_flow, SRC_DOMAIN: _detail_domain, SRC_APIKEY_RELEASE: _detail_apikey_release, SRC_PERMISSION: _detail_permission, SRC_SSO: _detail_sso, SRC_SUPPORT: _detail_support, SRC_EMAIL_GROUP_JOIN: _detail_email_group_join, SRC_GITHUB_ID: _detail_github_id, SRC_IDENTITY_CONSUMER: _detail_identity_consumer, "data_access": _detail_data_access, "cloud_renewal": _detail_cloud_renewal, "soc2_email_group": _detail_soc2_email_group, } def _parse_item_id(item_id: str) -> tuple[str, str]: """Split a tracker item id ``"{source}:{pk}"``. ``pk`` may itself contain no colon (UUIDs / ints don't), so split on the FIRST colon.""" source, _, pk = item_id.partition(":") if not source or not pk: raise NotFoundError("request", item_id) return source, pk def _reject_comment_from_timeline( timeline: list[AdminApprovalTimelineEntry], ) -> str | None: """Recover the approver's rejection note from a live Feishu flow. Feishu carries the rejection reason as the ``comment`` on the task that was REJECTED (and sometimes as a standalone comment entry). Scan newest- first for a rejecting task with a comment, else fall back to the last non-empty comment. Returns trimmed text or None. """ # Bind the trimmed comment to a local rather than re-deriving it after the # guard: `(entry.comment or "").strip()` being truthy DOES imply # `entry.comment` is a non-empty str, but that is only provable by reading # both halves together — a type checker cannot narrow through it, and # neither can a reader skimming for None-safety. for entry in reversed(timeline): comment = (entry.comment or "").strip() if ( entry.kind == "task" and (entry.status or "").upper() in ("REJECTED", "REJECT") and comment ): return comment for entry in reversed(timeline): comment = (entry.comment or "").strip() if comment: return comment return None def _set_terminal_timeline_note( detail: UnifiedRequestDetail, note: str, ) -> None: """Attach ``note`` to the synthesised timeline's terminal error event (rejected / failed / revoked) so the reason shows inline even on the fallback timeline. No-op if there is no error event.""" for event in detail.timeline: if event.state == "error" and not event.note: event.note = note return async def _attach_live_approval_flow( db: AsyncSession, detail: UnifiedRequestDetail, ) -> None: """Overlay the live Feishu approval flow onto ``detail`` in place. Only for the non-``employee_request`` sources that still run a Feishu instance (``employee_request`` has its own ApprovalDetailModal + ``/approval/my/requests/{id}`` enrichment and carries ``feishu_instance_code=None`` here). One best-effort HTTP round-trip; on any failure the synthesised ``timeline`` stays as the fallback. Side benefit for the outcome layer: when the row ended in a failure and has no local reason column (the Feishu-routed sources whose reject note lives only in the approval instance), recover the approver's rejection comment from the flow and surface it as ``outcome_reason`` + the terminal timeline note. """ if not detail.feishu_instance_code or detail.has_feishu_flow: return # Lazy import: keep the approval domain off the import path of the # list/urge builders (and dodge any import cycle through the aggregator). from app.approval.service import ApprovalService progress, timeline, error = await ApprovalService(db).build_live_feishu_flow( detail.feishu_instance_code, ) detail.approval_progress = progress detail.approval_timeline = timeline detail.approval_timeline_error = error if ( detail.needs_attention and not detail.outcome_reason and timeline ): comment = _reject_comment_from_timeline(timeline) if comment: detail.outcome_reason = comment _set_terminal_timeline_note(detail, comment) async def build_my_request_detail( db: AsyncSession, *, person_id: UUID | None, item_id: str, enforce_owner: bool = True, with_live_approval: bool = False, ) -> UnifiedRequestDetail: """Unified detail for one tracker row. ``item_id`` is the ``"{source}:{pk}"`` id from ``build_my_requests``. When ``enforce_owner`` is True (the self view) the row must belong to ``person_id`` — a stranger's id raises ``NotFoundError`` (404, never 403, so the endpoint can't enumerate other people's request ids). When False (admin paths such as 催办) ownership is not checked and ``person_id`` may be None. Also raises ``NotFoundError`` for an unknown source or a missing row. ``with_live_approval`` (detail endpoint only): when True, sources that carry a Feishu ``instance_code`` get the live approval flow + current approver(s) overlaid via :func:`_attach_live_approval_flow`. The 催办 path leaves it False so 催办 stays a pure DB read with no Feishu round-trip. """ source, pk = _parse_item_id(item_id) builder = _DETAIL_BUILDERS.get(source) if builder is None: raise NotFoundError("request", item_id) # enforce_owner=True → scope to the caller. A None person_id under # enforce_owner becomes an EMPTY set (matches nothing → NotFound), NOT # None — None means admin/unscoped and would silently bypass the owner # check. enforce_owner=False (admin path) → None (unscoped) is intended. if enforce_owner: person_ids: set[str] | None = {str(person_id)} if person_id else set() else: person_ids = None detail = await builder(db, pk, person_ids) if detail is None: raise NotFoundError("request", item_id) if not isinstance(detail, UnifiedRequestDetail): raise TypeError(f"invalid request detail builder result for {source}") if with_live_approval: await _attach_live_approval_flow(db, detail) return detail