"""HTTP surface for the unified ``/me`` IT-home aggregation.""" from __future__ import annotations from typing import Annotated from uuid import UUID from fastapi import APIRouter, Depends, Query, Request from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import get_db from app.core.dependencies import get_current_user from app.core.exceptions import NotFoundError from app.core.response import success_response from app.domains.me_universe.cohort import build_my_cohort from app.domains.me_universe.device_gallery import build_my_device_gallery from app.domains.me_universe.footprint import build_my_footprint from app.domains.me_universe.my_requests import ( build_my_requests, build_team_requests, ) from app.domains.me_universe.request_detail import build_my_request_detail from app.domains.me_universe.service import MeUniverseService from app.domains.me_universe.team_service import ( build_team_access_attestation, build_team_api_keys, build_team_cloud, build_team_footprint, build_team_lifecycle, ) from app.domains.me_universe.team_summary import build_my_team_summary from app.domains.me_universe.todo_history import ( build_todo_history, build_todo_history_stats, ) from app.domains.me_universe.urge import urge_request from app.domains.me_universe.welcome import build_my_welcome from app.identity.person_models import IdentityPerson class _UrgeBody(BaseModel): id: str # "{source}:{pk}" tracker item id router = APIRouter( prefix="/api/v1/users", tags=["me-universe"], ) async def _resolve_it_home_target( user_id: str, viewer: IdentityPerson, db: AsyncSession, ) -> IdentityPerson: """Resolve a ``me``/UUID target consistently for IT-home read surfaces.""" if user_id == "me": return viewer try: target_id = UUID(user_id) except (TypeError, ValueError): raise NotFoundError("identity_person", user_id) target = await db.get(IdentityPerson, target_id) if target is None or target.deleted_at is not None: raise NotFoundError("identity_person", user_id) return target # Separate router for /api/v1/me/* paths (the existing one is rooted # at /api/v1/users/* for the historical it-home endpoint). team_router = APIRouter( prefix="/api/v1/me", tags=["me-team"], ) @team_router.get( "/team/footprint", summary="Manager-view aggregate footprint across all direct/transitive reports (PR-D)", ) async def team_footprint( request: Request, include_left: bool = False, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_footprint(db, manager=user, include_left=include_left) return success_response(data=data, request=request) @team_router.get( "/team/cloud", summary="Manager-view per-resource cloud detail (grants/scopes/dns/cdn/domains owned by subordinates) (PR-D)", ) async def team_cloud( request: Request, include_left: bool = False, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_cloud(db, manager=user, include_left=include_left) return success_response(data=data, request=request) @team_router.get( "/team/api-keys", summary="Manager-view AI API keys held by subordinates + per-key month-to-date spend / budget (read-only)", ) async def team_api_keys( request: Request, include_left: bool = False, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_api_keys(db, manager=user, include_left=include_left) return success_response(data=data, request=request) @team_router.get( "/team/lifecycle", summary="Manager-view team people-lifecycle: joiners' onboarding-readiness + leavers' offboarding status/blockers (read-only)", ) async def team_lifecycle( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_lifecycle(db, manager=user) return success_response(data=data, request=request) @team_router.get( "/team/access-attestation", summary="Manager-view team access self-attestation status for the open campaign (read-only)", ) async def team_access_attestation( request: Request, include_left: bool = False, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_access_attestation( db, manager=user, include_left=include_left, ) return success_response(data=data, request=request) @team_router.get( "/requests", summary="Unified '我的申请' tracker — every request the caller submitted across all domains", ) async def my_requests( request: Request, status: str | None = None, category: str | None = None, q: str | None = None, page: Annotated[int, Query(ge=1)] = 1, # 422 on over-cap, not a silent clamp (#2186). Every sibling surface # (``support/router.py``) declares ``Query(ge=1, le=100)``, and the # platform contract is "over the cap is an error". Clamping meant a # client asking for 500 got 100 rows and no signal, then paged from # page 2 with a stride it never agreed to — every subsequent page # off by 400 rows. page_size: Annotated[int, Query(ge=1, le=100)] = 20, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_my_requests( db, person_id=user.id, status=status, category=category, q=(q or "")[:100] or None, page=page, page_size=page_size, ) return success_response(data=data.model_dump(), request=request) @team_router.get( "/requests/detail", summary="Unified rich detail for one '我的申请' row (any source) — owner-scoped", ) async def my_request_detail( request: Request, id: str, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Return the normalised rich detail for one tracker row. ``id`` is the ``"{source}:{pk}"`` id from ``/me/requests``. The builder is hard-scoped to the caller's own submitted requests — a stranger's id surfaces ``NotFound`` (404), never 403, so the endpoint can't enumerate other people's request ids. For the ``employee_request`` source the payload carries ``has_feishu_flow=True`` so the frontend additionally fetches the live Feishu approval flow via ``/approval/my/requests/{pk}``. Every OTHER source that runs a Feishu instance (DNS / cloud-scope / VPN / AI-tool / announcement / domain / SSO) gets the same flow overlaid inline here (``approval_progress`` + ``approval_timeline``) so the requester can see WHO is currently approving — ``with_live_approval``. """ detail = await build_my_request_detail( db, person_id=user.id, item_id=id, with_live_approval=True, ) return success_response(data=detail.model_dump(), request=request) @team_router.post( "/requests/urge", summary="催办 — nudge the approver(s) of one of MY still-open requests", ) async def urge_my_request( request: Request, body: _UrgeBody, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Owner-scoped 催办. The caller must own the request (the builder raises NotFound for a stranger's id). Notifies the requester's manager + IT admins in-app + Feishu; rate-limited to once per request per 24h (Redis). Returns ``urged=False`` with an i18n ``reason`` when the request is terminal or already urged today. """ result = await urge_request(db, item_id=body.id, actor=user, is_admin=False) return success_response(data=result.model_dump(), request=request) @team_router.get( "/requests/export.xlsx", summary="Export the caller's unified '我的申请' tracker as XLSX", ) async def export_my_requests( request: Request, status: str | None = None, category: str | None = None, q: str | None = None, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Export the filtered '我的申请' tracker as XLSX. Filter signature mirrors :func:`my_requests` minus pagination — the tracker page passes the exact query params it is showing, so the download matches the screen. RBAC mirrors the list endpoint too: ``get_current_user`` only; the builder is hard-scoped to the caller's own submitted requests. Reuses ``build_my_requests`` so any filter fix on the list applies here automatically. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response data = await build_my_requests( db, person_id=user.id, status=status, category=category, q=(q or "")[:100] or None, page=1, page_size=EXPORT_ROW_LIMIT, ) columns = [ ("title", "标题"), ("subtitle", "副标题"), ("source", "来源"), ("category", "类别"), ("request_type", "申请类型"), ("status", "状态"), ("raw_status", "原始状态"), ("is_open", "在途"), ("created_at", "提交时间"), ("updated_at", "更新时间"), ("feishu_instance_code", "飞书单号"), ("id", "ID"), ] rows = [item.model_dump() for item in data.items] return await xlsx_export_response( db, request, user, columns=columns, rows=rows, total_matched=data.total, filename_stem="my_requests", sheet_title="My Requests", audit_action="me.requests.export", audit_resource_type="my_requests", filters={"status": status, "category": category, "q": q}, ) @team_router.get( "/todos/done", summary="我的已办 — items the caller has acted on across all todo domains (read-only history)", ) async def my_todos_done( request: Request, type: str | None = None, page: Annotated[int, Query(ge=1)] = 1, # 422 on over-cap, not a silent clamp (#2186). Every sibling surface # (``support/router.py``) declares ``Query(ge=1, le=100)``, and the # platform contract is "over the cap is an error". Clamping meant a # client asking for 500 got 100 rows and no signal, then paged from # page 2 with a stride it never agreed to — every subsequent page # off by 400 rows. page_size: Annotated[int, Query(ge=1, le=100)] = 20, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The read-only 已办 complement to the 待办 queues — every asset transfer / provisioning task / cloud-grant / api-key release / inventory item the CALLER personally handled, newest first. Hard- scoped to the caller's own actions. Optional ``type`` filter (asset_transfer / provisioning / cloud_grant / apikey_release / inventory); bounded pagination (max 100). """ data = await build_todo_history( db, person_id=user.id, type_filter=type, page=page, page_size=page_size, ) return success_response(data=data.model_dump(), request=request) @team_router.get( "/todos/done/stats", summary="我的已办报表 — aggregate analytics over the caller's handled history", ) async def my_todos_done_stats( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The 报表 tab of 我的已办 — total handled, type mix, status mix, and a 12-month activity trend (month buckets in the caller's local timezone), aggregated across all five todo domains. Hard-scoped to the caller's own actions; read-only. Covers the caller's most-recent items per domain (``truncated=true`` flags a heavy history that exceeded the per-type cap). """ data = await build_todo_history_stats(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/team/requests", summary="Manager-view: every request submitted by the caller's direct/transitive reports (read-only)", ) async def team_requests( request: Request, status: str | None = None, category: str | None = None, q: str | None = None, page: Annotated[int, Query(ge=1)] = 1, # 422 on over-cap, not a silent clamp (#2186). Every sibling surface # (``support/router.py``) declares ``Query(ge=1, le=100)``, and the # platform contract is "over the cap is an error". Clamping meant a # client asking for 500 got 100 rows and no signal, then paged from # page 2 with a stride it never agreed to — every subsequent page # off by 400 rows. page_size: Annotated[int, Query(ge=1, le=100)] = 20, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): data = await build_team_requests( db, manager=user, status=status, category=category, q=(q or "")[:100] or None, page=page, page_size=page_size, ) return success_response(data=data.model_dump(), request=request) @team_router.get( "/team/requests/export.xlsx", summary="Export the manager-view team request tracker as XLSX", ) async def export_team_requests( request: Request, status: str | None = None, category: str | None = None, q: str | None = None, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Export the filtered team request tracker as XLSX. Filter signature mirrors :func:`team_requests` minus pagination. RBAC mirrors the list endpoint: ``get_current_user`` only; ``build_team_requests`` scopes rows to the caller's direct / transitive reports and a caller with no reports gets an empty workbook (not 403) — same contract as the list. Capped at :data:`app.core.exporting.EXPORT_ROW_LIMIT` rows. """ from app.core.exporting import EXPORT_ROW_LIMIT, xlsx_export_response data = await build_team_requests( db, manager=user, status=status, category=category, q=(q or "")[:100] or None, page=1, page_size=EXPORT_ROW_LIMIT, ) columns = [ ("requester_display", "申请人"), ("title", "标题"), ("subtitle", "副标题"), ("source", "来源"), ("category", "类别"), ("request_type", "申请类型"), ("status", "状态"), ("raw_status", "原始状态"), ("is_open", "在途"), ("created_at", "提交时间"), ("updated_at", "更新时间"), ("feishu_instance_code", "飞书单号"), ("id", "ID"), ] rows = [item.model_dump() for item in data.items] return await xlsx_export_response( db, request, user, columns=columns, rows=rows, total_matched=data.total, filename_stem="team_requests", sheet_title="Team Requests", audit_action="me.team_requests.export", audit_resource_type="team_requests", filters={"status": status, "category": category, "q": q}, ) @team_router.get( "/footprint", summary="我的 IT 足迹 — the caller's permanent personal IT recap (self-scoped, read-only)", ) async def my_footprint( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The warm, story-shaped counterpart to ``/me`` IT-home: when you joined and your place in line, your first device and longest companion, seats / permissions you've carried, how often IT has had your back, plus a curated milestone timeline. Hard self-scoped — there is no target param, so this surface has no IDOR vector. Copy is emitted language-agnostically (``title_key`` + ``params``) for the frontend to localise.""" data = await build_my_footprint(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/device-gallery", summary="装备图鉴 — every device the caller has ever held, with service tenure (self-scoped, read-only)", ) async def my_device_gallery( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Collection-style view of every device the caller has ever held (current + returned), each with total service days (summed across all ownership stints from the real `asset_operations` trail) and a playful rarity tier. Borrow events are excluded from span math. Hard self-scoped — no target param, no IDOR; never names a previous owner.""" data = await build_my_device_gallery(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/loadout", summary="我的装备墙 — RPG character-sheet view of the caller's gear / attributes / rank (self-scoped, read-only)", ) async def my_loadout( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The game-y face of the personal IT overview — equipped hardware (rarity tiered), a 5-axis attribute spread (assets / accounts / permissions / cloud / security), inventory counts, an achievements preview, and a security rank chip. Composed from the existing self-scoped builders (footprint / device-gallery / security-quiz stats) so it carries no new aggregation logic. Hard self-scoped — no target param, no IDOR.""" from app.domains.me_universe.loadout import build_my_loadout data = await build_my_loadout(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/wrapped-years", summary="年度报告可选年份 — which recaps exist for the caller (self-scoped)", ) async def my_wrapped_years( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Years the caller has a recap for, newest first (hire year → this year). Without this the entry point hardcoded the current year, so opening the recap on January 2nd showed an almost-empty report and there was no way to reach last year's without hand-editing the URL. """ from app.domains.me_universe.wrapped import available_wrapped_years years = await available_wrapped_years(db, person=user) return success_response(data={"years": years}, request=request) @team_router.get( "/wrapped/{year}", summary="年度报告 (Helix Wrapped) — the caller's year-in-review (self-scoped, read-only)", ) async def my_wrapped( request: Request, year: int, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Spotify-Wrapped-style year recap: tenure milestone, security-quiz effort, devices received, tickets opened, badges earned, year-over-year deltas and the derived year keyword. Composed from year-scoped queries in the caller's LOCAL calendar year; hard self-scoped (no target param, no IDOR).""" from app.domains.me_universe.wrapped import build_my_wrapped # Clamp to a sane window — the builder does ``datetime(year + 1, ...)``, # so an out-of-range path value (0, 99999) would raise ValueError → 500. year = max(2000, min(year, 2100)) data = await build_my_wrapped(db, person=user, year=year) return success_response(data=data.model_dump(), request=request) @team_router.get( "/cohort", summary="入职同期 — colleagues who joined on the same day as the caller (self-scoped, read-only)", ) async def my_cohort( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """The people who joined Helix on the caller's exact hire date — a gentle 'you didn't arrive alone'. Resolved from the caller's own hire_date (self-scoped); departed cohort-mates are counted, not named. Lights up (``is_anniversary_today``) on the cohort's shared anniversary.""" data = await build_my_cohort(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/welcome", summary="入职第一天仪式感 — the caller's warm welcome / Day-1 snapshot (self-scoped, read-only)", ) async def my_welcome( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """A new hire's 'welcome aboard' page: who you are here (employee no / title / office / manager), the gear waiting for you (assigned + reserved devices), your onboarding standing, and starter signals for the first-week checklist. Self-scoped; ``is_new`` flags a recent hire so the frontend can dial up the ceremony. Reuses the onboarding-readiness signals the lifecycle cockpit computes.""" data = await build_my_welcome(db, person=user) return success_response(data=data.model_dump(), request=request) @team_router.get( "/team-summary", summary="我的团队 — the caller's manager + immediate-team size for the IT 宇宙 banner (self-scoped, read-only)", ) async def my_team_summary( request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Cheap self-scoped team snapshot for the employee IT-home welcome banner: your manager (skipped if they have left), the number of colleagues who share your manager (``peer_count``), and the number of people who report to you (``direct_report_count``). ≤3 indexed queries — safe on the hot ``/me`` path.""" data = await build_my_team_summary(db, person=user) return success_response(data=data.model_dump(), request=request) @router.get("/{user_id}/it-home") async def get_it_home( user_id: str, request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Aggregated IT footprint for one person. ``user_id`` accepts the literal string ``"me"`` (resolved to the caller's id) or a UUID. The service performs its own RBAC check — self always allowed, admin/auditor allowed for any user, manager allowed for direct/indirect reports up to depth 4. """ target = await _resolve_it_home_target(user_id, user, db) service = MeUniverseService(db) response = await service.get_it_home(viewer=user, target=target) return success_response( data={ "user": response.user, "mode": response.mode, "summary": response.summary, "assets": response.assets, "seats": response.seats, "permissions": response.permissions, "api_keys": response.api_keys, "sso_grants": response.sso_grants, "cloud_servers": response.cloud_servers, "cloud_domains": response.cloud_domains, "cloud_grants": response.cloud_grants, "email_groups": response.email_groups, "registered_systems": response.registered_systems, "workflows": response.workflows, "todos": response.todos, "notifications": response.notifications, "compliance_self_check": response.compliance_self_check, "cost_estimate": response.cost_estimate, "leaver_status": response.leaver_status, }, request=request, ) @router.get("/{user_id}/holdings") async def get_it_holdings( user_id: str, request: Request, user: IdentityPerson = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): """Complete lifecycle-grade holding catalogue for one permitted person. This is on-demand by design: it runs the same exhaustive scan used by mover/offboarding review without adding its cost to the everyday IT-home graph request. Partial category failures are explicit rather than being represented as an empty result. """ target = await _resolve_it_home_target(user_id, user, db) holdings, failed_types = await MeUniverseService(db).get_holdings(user, target) return success_response( data={ "holdings": holdings, "holdings_failed_types": failed_types, }, request=request, )