"""HTTP surface for the Google Workspace orphan-account inventory. Endpoints: GET /api/v1/google-workspace/orphans GET /api/v1/google-workspace/orphans/export POST /api/v1/google-workspace/orphans/{id}/owner POST /api/v1/google-workspace/orphans/{id}/purpose GET /api/v1/google-workspace/orphans/leaver/{person_id} All routes require admin (IT Admin / Asset Admin). Read endpoints do NOT include the listing in any employee-facing response — orphan accounts are infra metadata, not anything an employee should see. """ from __future__ import annotations import csv import io import uuid from typing import Any from fastapi import APIRouter, Depends, Query, Request from fastapi.responses import StreamingResponse from pydantic import BaseModel, Field from sqlalchemy.ext.asyncio import AsyncSession from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import require_admin, require_admin_read from app.core.response import success_response from app.domains.google_workspace.orphan_service import OrphanAccountService router = APIRouter( prefix="/api/v1/google-workspace", tags=["google-workspace"], ) class AssignOwnerBody(BaseModel): new_owner_id: str = Field(min_length=1, max_length=64) purpose: str | None = Field(default=None, max_length=2000) class UpdatePurposeBody(BaseModel): purpose: str | None = Field(default=None, max_length=2000) def _trace_id(request: Request) -> str: return getattr(request.state, "trace_id", None) or "google-workspace" # CSV-formula injection guard + BOM-prefixed CSV response. Same # pattern as ``cmdb/router.py::_csv_response`` — duplicated locally # because domain modules must not import from each other (CLAUDE.md # rule 3). _CSV_INJECTION_PREFIXES = ("=", "+", "-", "@", "\t", "\r") def _csv_safe(value: Any) -> Any: if isinstance(value, str) and value.startswith(_CSV_INJECTION_PREFIXES): return "'" + value return value def _csv_response(rows: list[dict], filename: str) -> StreamingResponse: if not rows: content = "" else: buf = io.StringIO() writer = csv.DictWriter(buf, fieldnames=list(rows[0].keys())) writer.writeheader() for r in rows: writer.writerow({k: _csv_safe(v) for k, v in r.items()}) content = buf.getvalue() # UTF-8 BOM so Excel-on-Windows opens Chinese display names / # purposes without GBK autodetect mangling them. body = ("" + content).encode("utf-8") return StreamingResponse( iter([body]), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{filename}"'}, ) @router.get("/orphans") async def list_orphans( request: Request, unassigned_only: bool = Query( False, description="只看未指派负责人的孤儿账号", ), owner_user_id: str | None = Query( None, description="按负责人 person_id 过滤", ), search: str | None = Query(None, description="按邮箱 / 显示名模糊搜索"), include_suspended: bool = Query( True, description="是否包含已停用的 Google 账号(默认包含)", ), include_deleted: bool = Query( False, description=( "是否包含已删除账号(``deleted_at IS NOT NULL`` —— " "google.seat_sync 检测到 Google 上游已删除而打的标)。默认隐藏。" ), ), user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: svc = OrphanAccountService(db) owner_uuid: uuid.UUID | None = None if owner_user_id: try: owner_uuid = uuid.UUID(owner_user_id) except ValueError: owner_uuid = None rows = await svc.list( unassigned_only=unassigned_only, owner_user_id=owner_uuid, search=search, include_suspended=include_suspended, include_deleted=include_deleted, ) return success_response(data=rows, meta={"total": len(rows)}) @router.get("/orphans/export") async def export_orphans( unassigned_only: bool = Query(False), search: str | None = Query(None), include_suspended: bool = Query(True), include_deleted: bool = Query(False), user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ) -> StreamingResponse: """Export the orphan-account inventory as CSV, honouring the same filters as the listing so what the admin sees is what they get.""" svc = OrphanAccountService(db) rows = await svc.list( unassigned_only=unassigned_only, search=search, include_suspended=include_suspended, include_deleted=include_deleted, limit=None, # compliance export: never truncate the inventory ) out = [] for r in rows: if r["deleted_at"]: status = "deleted" elif r["suspended"]: status = "suspended" else: status = "active" out.append({ "primary_email": r["primary_email"], "display_name": r["display_name"] or "", "status": status, "is_admin": "yes" if r["is_admin"] else "", "former_employee": "yes" if r["former_employee"] else "", "owner": r["owner_display_name"] or r["owner_work_email"] or "", "owner_email": r["owner_work_email"] or "", "purpose": r["purpose"] or "", "org_unit_path": r["org_unit_path"] or "", "creation_time": (r["creation_time"] or "")[:19], "last_login_time": (r["last_login_time"] or "")[:19], "first_seen_at": r["first_seen_at"][:19], "last_seen_at": r["last_seen_at"][:19], "deleted_at": (r["deleted_at"] or "")[:19], }) return _csv_response(out, "google_shared_accounts.csv") @router.get("/orphans/leaver/{person_id}") async def list_orphans_for_leaver( person_id: str, user: AuthUser = Depends(require_admin_read()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """Compact list used by the offboarding panel — just the orphans a specific leaver owns. Saves the panel from filtering the full listing client-side.""" try: person_uuid = uuid.UUID(person_id) except ValueError: return success_response(data=[], meta={"total": 0}) svc = OrphanAccountService(db) rows = await svc.list(owner_user_id=person_uuid) return success_response(data=rows, meta={"total": len(rows)}) @router.post("/orphans/{orphan_id}/owner") async def assign_owner( orphan_id: str, body: AssignOwnerBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: svc = OrphanAccountService(db) row = await svc.assign_owner( uuid.UUID(orphan_id), uuid.UUID(body.new_owner_id), purpose=body.purpose, trace_id=_trace_id(request), actor_user_id=user.id, ) return success_response(data=row) @router.post("/orphans/{orphan_id}/purpose") async def update_purpose( orphan_id: str, body: UpdatePurposeBody, request: Request, user: AuthUser = Depends(require_admin()), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: svc = OrphanAccountService(db) row = await svc.update_purpose( uuid.UUID(orphan_id), body.purpose, trace_id=_trace_id(request), actor_user_id=user.id, ) return success_response(data=row)