"""Unified admin approval surfaces — the platform-wide approval registry. Two surfaces are built from ONE source registry (``_SPECS``): 1. ``build_admin_approval_overview`` — the **pending** triage board (待办 总览). Every approval-type item *waiting for a decision*, across the whole platform, oldest-pending-first. 2. ``build_admin_request_list`` — the **all-status** request list (申请 列表). Every approval-type item in any lifecycle state (pending / approved / rejected / completed / failed / cancelled), newest-first, so an admin can browse and search the *whole* request history across every domain — not just ``employee_requests``. This is the "生命周期闭环": one list where DNS / domain / VPN / AI-tool / cloud- scope / API-key-release / leaver-override / identity-consumer requests are all visible alongside the asset/software self-service requests. Before this registry, an IT admin had to context-switch across ≥4 pages to see what was pending, and the 申请列表 tab only showed one table (``employee_requests``). The registry reads each source's own rows (READ ONLY — never the source of truth) and normalises them into one list, each item tagged with **where the decision is made** (``decided_in``: ``helix`` = a Helix page decides it, ``feishu`` = the approver acts in Feishu and Helix only mirrors status) and a ``deep_link`` to that surface when one exists. Decisions still happen on the owning page, where the decide UI + RBAC already live — this unifies *discovery*, not the decision itself. Each source declares a :class:`_SourceSpec` (model, requester FK, status map, title/subtitle accessors, deep-link, decided-in). A single generic runner :func:`_run_spec` turns a spec into normalised candidates, so the pending board and the full list share the exact same per-source query shape — there is no second place to keep in sync. """ from __future__ import annotations import logging from collections.abc import Callable from dataclasses import dataclass from datetime import UTC, datetime, timedelta from datetime import date as date_type from typing import Any from pydantic import BaseModel from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.approval.models import ApprovalRequest from app.core.exceptions import ValidationError 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.email_group.models import EmailGroup, EmailGroupJoinRequest from app.domains.leaver_blocked.models import ( DUAL_SIGN_PENDING_SECONDARY, LeaverDualSignOverride, ) from app.domains.sso.models import SsoGrant from app.domains.vpn_application.models import VpnApplication from app.identity.consumers.models import IdentityConsumerRequest from app.identity.person_models import IdentityPerson logger = logging.getLogger(__name__) # Sources — stable literals shared with the FE i18n keys # (``approvalOverview.source.``). SRC_EMPLOYEE = "employee_request" SRC_CLOUD_SCOPE = "cloud_scope" SRC_LEAVER_OVERRIDE = "leaver_override" SRC_DNS = "dns" SRC_DOMAIN = "domain" SRC_VPN = "vpn" SRC_AI_TOOL = "ai_tool" SRC_ANNOUNCEMENT = "announcement" SRC_APIKEY_RELEASE = "apikey_release" SRC_IDENTITY_CONSUMER = "identity_consumer" SRC_CLOUD_GRANT = "cloud_grant" SRC_CLOUD_RENEWAL = "cloud_renewal" SRC_EMAIL_GROUP_JOIN = "email_group_join" SRC_SSO = "sso" SRC_DATA_ACCESS = "data_access" SRC_SOC2_EMAIL_GROUP = "soc2_email_group" # Where the decision is actually made. DECIDED_HELIX = "helix" DECIDED_FEISHU = "feishu" # Canonical lifecycle states — every source's raw status maps onto one of # these so the 申请列表 status filter is uniform across domains. Shared # with FE i18n keys ``approvalOverview.status.``. ST_PENDING = "pending" ST_APPROVED = "approved" ST_REJECTED = "rejected" ST_COMPLETED = "completed" ST_FAILED = "failed" ST_CANCELLED = "cancelled" CANONICAL_STATUSES = ( ST_PENDING, ST_APPROVED, ST_REJECTED, ST_COMPLETED, ST_FAILED, ST_CANCELLED, ) # No per-source cap: pagination, export and analytics must never silently # lose older rows. Optional cap support remains in the runner for bounded # internal callers, and logs whenever such a caller truncates. _PER_SOURCE_CAP: int | None = None _LIST_PER_SOURCE_CAP: int | None = None class AdminApprovalOverviewItem(BaseModel): id: str # f"{source}:{pk}" source: str decided_in: str # helix | feishu requester_display: str | None = None requester_id: str | None = None requester_email: str | None = None requester_employee_no: str | None = None title: str subtitle: str | None = None status: str # raw per-source status (forensic) status_normalized: str # one of CANONICAL_STATUSES created_at: str # ISO 8601 processed_at: str | None = None # ISO 8601, when the source records it age_days: int deep_link: str | None = None # decision/review surface, when one exists detail_available: bool = True # Cached Feishu task assignees for employee requests. The list builders # never fetch Feishu per row. current_approvers: list[str] = [] class AdminApprovalOverviewResponse(BaseModel): items: list[AdminApprovalOverviewItem] counts_by_source: dict[str, int] counts_by_decided_in: dict[str, int] counts_by_status: dict[str, int] total: int page: int page_size: int class AdminApprovalSource(BaseModel): source: str decided_in: str detail_available: bool @dataclass class _Cand: item: AdminApprovalOverviewItem sort_dt: datetime def _naive(dt: datetime) -> datetime: return dt.replace(tzinfo=None) if dt.tzinfo is not None else dt def _age_days(created_at: datetime) -> int: now = datetime.now(UTC) delta = _naive(now) - _naive(created_at) return max(0, delta.days) def _cached_employee_current_approvers(row: EmployeeRequest) -> list[str]: """Read active approvers from the callback snapshot without live I/O.""" feishu = (row.detail or {}).get("feishu") if not isinstance(feishu, dict): return [] snapshot = feishu.get("latest_instance_detail") if not isinstance(snapshot, dict): return [] # Import lazily to keep this registry free of a module-level dependency # on the approval service. from app.approval.service import ApprovalService progress = ApprovalService._build_progress_from_instance( snapshot, include_raw_identifiers=False, ) return progress.current_approvers if progress else [] # ── Source registry ───────────────────────────────────────────────── # Each source declares how to read its rows and normalise them. The # generic runner below is the only query path; there is no per-source # query code to drift. @dataclass class _SourceSpec: source: str model: Any #: Column joined to ``IdentityPerson.id`` for the requester name. requester_fk: Any decided_in: str #: raw status string → one of CANONICAL_STATUSES. status_map: dict[str, str] #: ``(row, requester_name) -> human title`` (FE may localise it). title: Callable[[Any, str | None], str] subtitle: Callable[[Any], str | None] #: Best-effort decision/finish timestamp; ``None`` when the source #: does not record one as a clean column. processed_at: Callable[[Any], datetime | None] #: ``row -> decision/review surface`` (None when none exists). deep_link: Callable[[Any], str | None] #: Extra SQL predicates (for example cloud grants only include Helix- #: originated rows; discovered/imported grants are not approvals). where: tuple[Any, ...] = () #: Optional logical id. Renewal rows from one submission share a group id #: and must appear once in the approval centre. logical_id: Callable[[Any], object] | None = None detail_available: bool = True def pending_raw_statuses(self) -> list[str]: return [raw for raw, canon in self.status_map.items() if canon == ST_PENDING] def raw_statuses_for(self, normalized: str) -> list[str]: return [raw for raw, canon in self.status_map.items() if canon == normalized] _SPECS: list[_SourceSpec] = [ _SourceSpec( source=SRC_EMPLOYEE, model=EmployeeRequest, requester_fk=EmployeeRequest.requester_id, decided_in=DECIDED_HELIX, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "completed": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "failed": ST_FAILED, }, title=lambda r, _n: r.request_type, # FE localises asset.request.type.* subtitle=lambda r: r.request_number, processed_at=lambda r: r.processed_at, deep_link=lambda _r: "/admin/approvals", ), _SourceSpec( source=SRC_CLOUD_SCOPE, model=CmdbCloudScopeRequest, requester_fk=CmdbCloudScopeRequest.requester_person_id, decided_in=DECIDED_HELIX, status_map={ "pending": ST_PENDING, "feishu_sent": ST_PENDING, "feishu_approved": ST_APPROVED, "provisioning": ST_APPROVED, "provisioned": ST_COMPLETED, "rejected": ST_REJECTED, "withdrawn": ST_CANCELLED, "failed": ST_FAILED, }, title=lambda r, _n: r.name, subtitle=lambda r: r.provider, processed_at=lambda _r: None, deep_link=lambda _r: "/admin/cloud/scopes?tab=requests", ), _SourceSpec( source=SRC_LEAVER_OVERRIDE, model=LeaverDualSignOverride, requester_fk=LeaverDualSignOverride.person_id, decided_in=DECIDED_HELIX, status_map={ DUAL_SIGN_PENDING_SECONDARY: ST_PENDING, "approved": ST_APPROVED, "executed": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "expired": ST_CANCELLED, }, title=lambda r, n: n or str(r.person_id), subtitle=lambda _r: None, processed_at=lambda _r: None, deep_link=lambda r: f"/admin/leaver-blocked/override/{r.id}", detail_available=False, ), _SourceSpec( source=SRC_DNS, model=CmdbDnsRecordRequest, requester_fk=CmdbDnsRecordRequest.requester_person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "executing": ST_APPROVED, "executed": ST_COMPLETED, "rejected": ST_REJECTED, "failed": ST_FAILED, "cancelled": ST_CANCELLED, }, title=lambda r, _n: f"{r.name}.{r.zone_name} ({r.record_type})", subtitle=lambda r: r.action, processed_at=lambda _r: None, deep_link=lambda _r: "/admin/cloud/dns", ), _SourceSpec( source=SRC_DOMAIN, model=DomainRegistration, requester_fk=DomainRegistration.person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "revoked": ST_CANCELLED, }, title=lambda r, _n: r.domain_name, subtitle=lambda _r: None, processed_at=lambda r: r.granted_at, deep_link=lambda _r: "/admin/cloud/domains", ), _SourceSpec( source=SRC_VPN, model=VpnApplication, requester_fk=VpnApplication.person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "revoked": ST_CANCELLED, }, title=lambda r, _n: r.vpn_node, subtitle=lambda _r: None, processed_at=lambda _r: None, # No dedicated Helix admin review page — approver acts in Feishu. deep_link=lambda _r: None, ), _SourceSpec( source=SRC_AI_TOOL, model=AiToolApplication, requester_fk=AiToolApplication.person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "revoked": ST_CANCELLED, }, title=lambda r, _n: r.ai_name, subtitle=lambda _r: None, processed_at=lambda r: r.granted_at, deep_link=lambda _r: None, ), _SourceSpec( source=SRC_ANNOUNCEMENT, model=AnnouncementRequest, requester_fk=AnnouncementRequest.person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, }, title=lambda r, _n: (r.description or "")[:80] or "announcement", subtitle=lambda _r: None, processed_at=lambda _r: None, deep_link=lambda _r: "/admin/announcements", ), _SourceSpec( source=SRC_APIKEY_RELEASE, model=ApiKeyReleaseRequest, requester_fk=ApiKeyReleaseRequest.requester_id, decided_in=DECIDED_HELIX, status_map={ "pending": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, }, title=lambda r, _n: r.reason_code, # FE localises revokeReasonCode.* subtitle=lambda r: r.reason_note, processed_at=lambda r: r.responded_at, deep_link=lambda _r: "/my/todos", ), _SourceSpec( source=SRC_IDENTITY_CONSUMER, model=IdentityConsumerRequest, requester_fk=IdentityConsumerRequest.requester_person_id, decided_in=DECIDED_FEISHU, status_map={ "pending_approval": ST_PENDING, "approved": ST_APPROVED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "failed": ST_FAILED, }, title=lambda r, _n: r.name, subtitle=lambda r: r.slug, processed_at=lambda r: r.decided_at, deep_link=lambda _r: "/admin/identity/consumers", ), _SourceSpec( source=SRC_CLOUD_GRANT, model=CmdbCloudAccessGrant, requester_fk=CmdbCloudAccessGrant.person_id, decided_in=DECIDED_HELIX, status_map={ "pending": ST_PENDING, "active": ST_COMPLETED, "revoked": ST_REJECTED, "disabled": ST_CANCELLED, "expired": ST_CANCELLED, }, title=lambda r, _n: r.role_display_name or r.role_key, subtitle=lambda r: r.scope_id, processed_at=lambda r: r.revoked_at or r.granted_at, deep_link=lambda _r: "/my/todos", where=(CmdbCloudAccessGrant.source == "helix",), logical_id=lambda r: r.request_group_id or r.id, ), _SourceSpec( source=SRC_CLOUD_RENEWAL, model=CmdbCloudGrantRenewalRequest, requester_fk=CmdbCloudGrantRenewalRequest.requested_by_person_id, decided_in=DECIDED_HELIX, status_map={ "pending_owner": ST_PENDING, "pending_admin": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "lapsed": ST_CANCELLED, }, title=lambda r, _n: r.scope_id, subtitle=lambda r: f"{r.requested_ttl_days}d", processed_at=lambda r: r.admin_decided_at or r.owner_decided_at, deep_link=lambda _r: "/my/todos", logical_id=lambda r: r.request_group_id or r.id, ), _SourceSpec( source=SRC_EMAIL_GROUP_JOIN, model=EmailGroupJoinRequest, requester_fk=EmailGroupJoinRequest.requester_id, decided_in=DECIDED_HELIX, status_map={ "pending": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, }, title=lambda r, _n: str(r.email_group_id), subtitle=lambda r: r.reason, processed_at=lambda r: r.decided_at, deep_link=lambda _r: "/company/mail-groups", ), _SourceSpec( source=SRC_SSO, model=SsoGrant, requester_fk=SsoGrant.person_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, "revoked": ST_CANCELLED, }, title=lambda r, _n: r.system_name, subtitle=lambda r: r.sso_provider, processed_at=lambda r: r.revoked_at or r.granted_at, deep_link=lambda _r: "/admin/sso-apps", ), _SourceSpec( source=SRC_DATA_ACCESS, model=DataAccessRequest, requester_fk=DataAccessRequest.requester_person_id, decided_in=DECIDED_FEISHU, status_map={ "pending_feishu": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "stale_requires_resubmit": ST_FAILED, "cancelled": ST_CANCELLED, }, title=lambda r, _n: r.principal_email_snapshot, subtitle=lambda r: r.approval_flow_key, processed_at=lambda r: r.decided_at, # The employee detail route is requester-scoped; global approval # readers use the safe read-only detail modal instead. deep_link=lambda _r: None, ), _SourceSpec( source=SRC_SOC2_EMAIL_GROUP, model=ApprovalRequest, requester_fk=ApprovalRequest.applicant_id, decided_in=DECIDED_FEISHU, status_map={ "pending": ST_PENDING, "approved": ST_COMPLETED, "rejected": ST_REJECTED, "cancelled": ST_CANCELLED, }, title=lambda _r, _n: "soc2@pingcap.com", subtitle=lambda r: r.external_id, processed_at=lambda r: r.updated_at, deep_link=lambda _r: None, where=(ApprovalRequest.approval_type == "soc2_email_group_grant",), ), ] def list_admin_approval_sources() -> list[AdminApprovalSource]: """Public catalogue consumed by every admin approval filter. Keeping this server-owned removes the two frontend arrays that drifted independently and caused identity-consumer (and later SSO) to disappear. """ return [ AdminApprovalSource( source=spec.source, decided_in=spec.decided_in, detail_available=spec.detail_available, ) for spec in _SPECS ] async def _run_spec( db: AsyncSession, spec: _SourceSpec, *, raw_statuses: list[str] | None, start_dt: datetime | None, end_dt: datetime | None, requester_id: object | None, cap: int | None, ) -> list[_Cand]: """Read one source's rows and normalise them into candidates. ``raw_statuses=None`` means *all statuses* (the full-list path); a list narrows to those raw statuses (the pending-board path passes the source's pending raws). Requester filtering goes through the already-present ``IdentityPerson`` join (``IdentityPerson.id`` is a ``Uuid`` column, so the comparison is type-safe — unlike the string-typed per-source ``person_id`` FKs). """ query = select( spec.model, IdentityPerson.display_name, IdentityPerson.work_email, IdentityPerson.employee_no, ).outerjoin( IdentityPerson, IdentityPerson.id == spec.requester_fk ) if raw_statuses is not None: query = query.where(spec.model.status.in_(raw_statuses)) if start_dt is not None: query = query.where(spec.model.created_at >= start_dt) if end_dt is not None: query = query.where(spec.model.created_at <= end_dt) if requester_id is not None: query = query.where(IdentityPerson.id == requester_id) if spec.where: query = query.where(*spec.where) # Fetch cap+1 to detect (and log) truncation — never a silent cap. ordered = query.order_by(spec.model.created_at.desc()) if cap is not None: ordered = ordered.limit(cap + 1) rows = (await db.execute(ordered)).all() if cap is not None and len(rows) > cap: rows = rows[:cap] logger.warning( "admin request registry: source=%s hit per-source cap=%d; " "older rows omitted from this page", spec.source, cap, ) # Resolve human-facing names in batches. The approval centre must not # make admins decipher internal UUIDs, and it must not introduce an N+1 # query by following ORM relationships row by row. email_groups: dict[str, EmailGroup] = {} if spec.source == SRC_EMAIL_GROUP_JOIN: group_ids = {row.email_group_id for row, *_identity in rows} if group_ids: groups = ( await db.execute(select(EmailGroup).where(EmailGroup.id.in_(group_ids))) ).scalars() email_groups = {str(group.id): group for group in groups} cloud_scopes: dict[str, CmdbCloudScope] = {} if spec.source in {SRC_CLOUD_GRANT, SRC_CLOUD_RENEWAL}: scope_ids = {row.scope_id for row, *_identity in rows} if scope_ids: scopes = ( await db.execute( select(CmdbCloudScope).where(CmdbCloudScope.id.in_(scope_ids)) ) ).scalars() cloud_scopes = {str(scope.id): scope for scope in scopes} cands: list[_Cand] = [] seen_logical_ids: set[str] = set() for row, name, work_email, employee_no in rows: logical_id = spec.logical_id(row) if spec.logical_id else row.id logical_id_s = str(logical_id) if logical_id_s in seen_logical_ids: continue seen_logical_ids.add(logical_id_s) raw = row.status processed = spec.processed_at(row) requester_value = getattr(row, spec.requester_fk.key, None) title = spec.title(row, name) subtitle = spec.subtitle(row) if spec.source == SRC_EMAIL_GROUP_JOIN: group = email_groups.get(str(row.email_group_id)) if group is not None: title = group.email subtitle = group.name or subtitle elif spec.source in {SRC_CLOUD_GRANT, SRC_CLOUD_RENEWAL}: scope = cloud_scopes.get(str(row.scope_id)) if scope is not None: scope_name = scope.display_name or scope.external_id if spec.source == SRC_CLOUD_GRANT: title = f"{scope_name} · {title}" else: title = scope_name cands.append( _Cand( item=AdminApprovalOverviewItem( id=f"{spec.source}:{logical_id_s}", source=spec.source, decided_in=spec.decided_in, requester_display=name, requester_id=( str(requester_value) if requester_value is not None else None ), requester_email=work_email, requester_employee_no=employee_no, title=title, subtitle=subtitle, status=raw, status_normalized=spec.status_map.get(raw, raw), created_at=row.created_at.isoformat(), processed_at=processed.isoformat() if processed else None, age_days=_age_days(row.created_at), deep_link=spec.deep_link(row), detail_available=spec.detail_available, current_approvers=( _cached_employee_current_approvers(row) if spec.source == SRC_EMPLOYEE else [] ), ), sort_dt=_naive(row.created_at), ) ) return cands def _matches_query(item: AdminApprovalOverviewItem, needle: str) -> bool: """Case-insensitive substring match across the human-searchable fields. Specs fold the domain identifiers (request number, DNS name, domain name, scope name, …) into ``title`` / ``subtitle``, so those two plus the requester name and source cover every source without per-source knowledge here. """ hay = ( f"{item.title}\n{item.subtitle or ''}" f"\n{item.requester_display or ''}\n{item.requester_email or ''}" f"\n{item.requester_employee_no or ''}\n{item.source}" ).lower() return needle in hay def _parse_date_bound(value: str | None, *, end: bool) -> datetime | None: if not value: return None try: d = date_type.fromisoformat(value) except ValueError as exc: raise ValidationError( f"date must be ISO format (YYYY-MM-DD), got {value!r}", ) from exc return datetime.combine(d, datetime.max.time() if end else datetime.min.time()) async def build_admin_approval_overview( db: AsyncSession, *, source: str | None = None, decided_in: str | None = None, q: str | None = None, page: int = 1, page_size: int = 20, ) -> AdminApprovalOverviewResponse: """Assemble the platform-wide **pending** approval board (待办总览). Admin-scoped (the HTTP route gates with ``require_admin_read``); no per-source role gate here. ``source`` / ``decided_in`` filter the list; ``q`` is a case-insensitive substring search over title / subtitle / requester / source (powers the global-search "待审批" group); ``counts_*`` are computed off the **unfiltered** candidate set so the FE filter chips stay stable. Sorted oldest-pending-first. """ candidates: list[_Cand] = [] for spec in _SPECS: candidates.extend( await _run_spec( db, spec, raw_statuses=spec.pending_raw_statuses(), start_dt=None, end_dt=None, requester_id=None, cap=_PER_SOURCE_CAP, ) ) counts_by_source: dict[str, int] = {} counts_by_decided_in: dict[str, int] = {} counts_by_status: dict[str, int] = {} for c in candidates: counts_by_source[c.item.source] = counts_by_source.get(c.item.source, 0) + 1 counts_by_decided_in[c.item.decided_in] = ( counts_by_decided_in.get(c.item.decided_in, 0) + 1 ) counts_by_status[c.item.status_normalized] = ( counts_by_status.get(c.item.status_normalized, 0) + 1 ) if source: candidates = [c for c in candidates if c.item.source == source] if decided_in: candidates = [c for c in candidates if c.item.decided_in == decided_in] needle = (q or "").strip().lower() if needle: candidates = [c for c in candidates if _matches_query(c.item, needle)] # Oldest-pending-first — the most overdue decision surfaces at the top. candidates.sort(key=lambda c: c.sort_dt) total = len(candidates) start = (page - 1) * page_size page_items = [c.item for c in candidates[start : start + page_size]] return AdminApprovalOverviewResponse( items=page_items, counts_by_source=counts_by_source, counts_by_decided_in=counts_by_decided_in, counts_by_status=counts_by_status, total=total, page=page, page_size=page_size, ) async def build_admin_analytics_summary( db: AsyncSession, *, window_days: int, ) -> dict[str, Any]: """Cross-domain analytics over the same registry as both admin lists.""" now = datetime.now(UTC) start = now - timedelta(days=window_days - 1) start = start.replace(hour=0, minute=0, second=0, microsecond=0) all_rows = await build_admin_request_list( db, page=1, page_size=10_000_000, ) items = all_rows.items window = [ i for i in items if _naive(datetime.fromisoformat(i.created_at)) >= _naive(start) ] today = _naive(now).date() by_status: dict[str, int] = {} by_type: dict[str, dict[str, int]] = {} daily = { (start.date() + timedelta(days=n)).isoformat(): {"created": 0, "processed": 0} for n in range(window_days) } requester_counts: dict[str, dict[str, Any]] = {} turnaround: dict[str, list[float]] = {} for item in window: status = item.status_normalized by_status[status] = by_status.get(status, 0) + 1 request_type = item.title if item.source == SRC_EMPLOYEE else item.source bucket = by_type.setdefault(request_type, { "total": 0, ST_PENDING: 0, ST_APPROVED: 0, ST_REJECTED: 0, ST_COMPLETED: 0, ST_CANCELLED: 0, }) bucket["total"] += 1 if status in bucket: bucket[status] += 1 created = datetime.fromisoformat(item.created_at) created_key = created.date().isoformat() if created_key in daily: daily[created_key]["created"] += 1 if item.processed_at: processed = datetime.fromisoformat(item.processed_at) processed_key = processed.date().isoformat() if processed_key in daily: daily[processed_key]["processed"] += 1 if status in (ST_APPROVED, ST_COMPLETED, ST_REJECTED): hours = max(0.0, (_naive(processed) - _naive(created)).total_seconds() / 3600) turnaround.setdefault(request_type, []).append(hours) rid = item.requester_id or "" rb = requester_counts.setdefault(rid, { "requester_id": rid or None, "display_name": item.requester_display, "work_email": item.requester_email, "employee_no": item.requester_employee_no, "employment_status": None, "count": 0, "rejected_count": 0, }) rb["count"] = int(rb["count"]) + 1 if status == ST_REJECTED: rb["rejected_count"] = int(rb["rejected_count"]) + 1 approved = by_status.get(ST_APPROVED, 0) + by_status.get(ST_COMPLETED, 0) rejected = by_status.get(ST_REJECTED, 0) decided = approved + rejected breakdown: list[dict[str, Any]] = [] for request_type, bucket in sorted(by_type.items(), key=lambda x: -x[1]["total"]): type_decided = bucket[ST_APPROVED] + bucket[ST_COMPLETED] + bucket[ST_REJECTED] breakdown.append({ "request_type": request_type, **bucket, "approval_rate": round( (bucket[ST_APPROVED] + bucket[ST_COMPLETED]) * 1000 / type_decided, ) / 10 if type_decided else None, }) turnaround_rows = [{ "request_type": key, "avg_hours": round(sum(values) / len(values), 1), "max_hours": round(max(values), 1), "decided_count": len(values), } for key, values in turnaround.items()] all_turn = [value for values in turnaround.values() for value in values] return { "window_days": window_days, "checked_at": now.isoformat(), "totals": { "total": len(items), "total_window": len(window), "total_today": sum(1 for i in items if datetime.fromisoformat(i.created_at).date() == today), "pending_open": sum(1 for i in items if i.status_normalized == ST_PENDING), "approved_window": approved, "rejected_window": rejected, "approval_rate": round(approved * 1000 / decided) / 10 if decided else None, "avg_turnaround_hours": round(sum(all_turn) / len(all_turn), 1) if all_turn else None, "decided_window": decided, "decided_for_turnaround": len(all_turn), }, "by_status": by_status, "request_type_breakdown": breakdown, "daily_series": [ {"date": key, **value} for key, value in daily.items() ], "turnaround_by_type": turnaround_rows, "top_requesters": sorted( requester_counts.values(), key=lambda row: -int(row["count"]), )[:10], } async def build_admin_request_list( db: AsyncSession, *, source: str | None = None, request_type: str | None = None, status: str | None = None, # one of CANONICAL_STATUSES requester_id: object | None = None, q: str | None = None, start_date: str | None = None, end_date: str | None = None, page: int = 1, page_size: int = 20, ) -> AdminApprovalOverviewResponse: """Assemble the platform-wide **all-status** request list (申请列表). The cross-domain sibling of the pending board: every request in any lifecycle state, across all sources, newest-first. ``source`` narrows to one domain; ``status`` filters by canonical lifecycle state; ``request_type`` further narrows the asset/software source by its raw sub-type (laptop_replacement / software_seat / …) — it is ignored for other sources. ``q`` substring-searches the same human fields as the board. ``start_date`` / ``end_date`` bound ``created_at`` (inclusive). ``counts_by_source`` / ``counts_by_status`` are computed off the set filtered by everything *except* ``source`` / ``status`` themselves, so the FE filter chips stay populated while one is selected. Per-source rows are capped (``_LIST_PER_SOURCE_CAP``); truncation is logged. """ start_dt = _parse_date_bound(start_date, end=False) end_dt = _parse_date_bound(end_date, end=True) # Always read every source (status NOT pushed to SQL) so the count # chips reflect the other filters and stay stable when a source/status # is selected — mirrors the pending board's "fetch all, then filter". candidates: list[_Cand] = [] for spec in _SPECS: cands = await _run_spec( db, spec, raw_statuses=None, start_dt=start_dt, end_dt=end_dt, requester_id=requester_id, cap=_LIST_PER_SOURCE_CAP, ) # request_type only narrows the asset/software source; its raw # request_type IS the item title there. if request_type and spec.source == SRC_EMPLOYEE: cands = [c for c in cands if c.item.title == request_type] candidates.extend(cands) needle = (q or "").strip().lower() if needle: candidates = [c for c in candidates if _matches_query(c.item, needle)] counts_by_source: dict[str, int] = {} counts_by_status: dict[str, int] = {} counts_by_decided_in: dict[str, int] = {} for c in candidates: counts_by_source[c.item.source] = counts_by_source.get(c.item.source, 0) + 1 counts_by_status[c.item.status_normalized] = ( counts_by_status.get(c.item.status_normalized, 0) + 1 ) counts_by_decided_in[c.item.decided_in] = ( counts_by_decided_in.get(c.item.decided_in, 0) + 1 ) if source: candidates = [c for c in candidates if c.item.source == source] if status: candidates = [c for c in candidates if c.item.status_normalized == status] # Newest-first — a browsable history reads top-down by recency. candidates.sort(key=lambda c: c.sort_dt, reverse=True) total = len(candidates) start = (page - 1) * page_size page_items = [c.item for c in candidates[start : start + page_size]] return AdminApprovalOverviewResponse( items=page_items, counts_by_source=counts_by_source, counts_by_decided_in=counts_by_decided_in, counts_by_status=counts_by_status, total=total, page=page, page_size=page_size, )