"""合作方通讯录的 HTTP 面 —— 全部 admin-gated。 GET /api/v1/partner-directory/overview GET /api/v1/partner-directory/members GET /api/v1/partner-directory/members/export GET /api/v1/partner-directory/links POST /api/v1/partner-directory/links/{id}/review POST /api/v1/partner-directory/sync 名册里是**别家公司员工的个人信息**,所以这一整个路由前缀没有任何员工 可达的入口 —— 连"我是不是被标成合作方了"这种自查都不开。谁是合作方 员工是 IT 的台账问题,不是一个应该对全员广播的事实。 """ from __future__ import annotations import csv import io 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.audit.service import AuditService from app.auth.models import AuthUser from app.core.database import get_db from app.core.dependencies import ( check_sensitive_endpoint_rate_limit, require_admin, require_admin_read, require_browser_session, ) from app.core.response import success_response from app.domains.partner_directory import service from app.domains.partner_directory.models import LINK_STATES, LINK_SURFACES router = APIRouter( prefix="/api/v1/partner-directory", tags=["partner-directory"], ) def _trace_id(request: Request) -> str: return getattr(request.state, "trace_id", None) or "partner-directory" # CSV-formula 注入防护。与 ``google_workspace/router.py`` 里那份是同一 # 段代码 —— 域模块之间不互相 import(CLAUDE.md 规则 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 class ReviewBody(BaseModel): decision: str = Field(pattern="^(confirm|dismiss)$") note: str | None = Field(default=None, max_length=2000) class SyncBody(BaseModel): integration_id: str | None = None #: ``False`` = 只算不写,用来预览「这一轮会标出什么」。 apply_tags: bool = True @router.get("/overview") async def get_overview( db: AsyncSession = Depends(get_db), _user: AuthUser = Depends(require_admin_read()), ) -> dict[str, Any]: return success_response(await service.overview(db)) @router.get("/members") async def get_members( integration_id: str | None = None, q: str | None = None, only_active: bool | None = None, has_github: bool | None = None, include_deleted: bool = False, page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), db: AsyncSession = Depends(get_db), _user: AuthUser = Depends(require_admin_read()), ) -> dict[str, Any]: result = await service.list_members( db, integration_id=integration_id, q=q, only_active=only_active, has_github=has_github, include_deleted=include_deleted, page=page, page_size=page_size, ) return success_response( result["items"], meta={ "total": result["total"], "page": result["page"], "page_size": result["page_size"], }, ) @router.get("/members/export") async def export_members( integration_id: str | None = None, db: AsyncSession = Depends(get_db), _user: AuthUser = Depends(require_admin_read()), ) -> StreamingResponse: result = await service.list_members( db, integration_id=integration_id, include_deleted=True, page=1, page_size=10000, ) buffer = io.StringIO() writer = csv.writer(buffer) writer.writerow([ "name", "en_name", "email", "email_name", "employee_no", "job_title", "department_path", "github_login", "is_active", "is_suspended", "last_seen_at", "deleted_at", ]) for item in result["items"]: writer.writerow([_csv_safe(v) for v in [ item["name"], item["en_name"], item["email"], item["email_name"], item["employee_no"], item["job_title"], " / ".join(item["department_path"] or []), item["github_login"], item["is_active"], item["is_suspended"], item["last_seen_at"], item["deleted_at"], ]]) payload = "" + buffer.getvalue() return StreamingResponse( iter([payload]), media_type="text/csv; charset=utf-8", headers={ "Content-Disposition": 'attachment; filename="partner-directory-members.csv"', }, ) @router.get("/links") async def get_links( integration_id: str | None = None, state: str | None = Query(default=None), surface: str | None = Query(default=None), page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=200), db: AsyncSession = Depends(get_db), _user: AuthUser = Depends(require_admin_read()), ) -> dict[str, Any]: if state and state not in LINK_STATES: state = None if surface and surface not in LINK_SURFACES: surface = None result = await service.list_links( db, integration_id=integration_id, state=state, surface=surface, page=page, page_size=page_size, ) return success_response( result["items"], meta={ "total": result["total"], "page": result["page"], "page_size": result["page_size"], }, ) @router.post("/links/{link_id}/review") async def review_link( link_id: str, body: ReviewBody, request: Request, db: AsyncSession = Depends(get_db), user: AuthUser = Depends(require_admin()), ) -> dict[str, Any]: result = await service.review_link( db, link_id=link_id, decision=body.decision, actor_person_id=str(user.id) if user.id else None, trace_id=_trace_id(request), note=body.note, ) await db.commit() return success_response(result) @router.post("/sync") async def trigger_sync( body: SyncBody, request: Request, _rate: None = Depends(check_sensitive_endpoint_rate_limit), user: AuthUser = Depends(require_admin()), # 与定时任务的手动触发端点同一道闸:立即执行一个会写生产的重任务, # 必须来自浏览器会话,不能用自助 PAT 打。声明在 ``require_admin`` # **之后**,这样 ``request.state.auth_via`` 已经被填好。 _gate: None = Depends(require_browser_session), db: AsyncSession = Depends(get_db), ) -> dict[str, Any]: """同步跑完再返回。CBG 那份名册(75 个部门 / 381 人)实测约 2 分钟。 刻意做成同步的:管理员点这个按钮就是想当场看到「拉到多少人、标了 多少条、多少条被拒」,而把它做成「已提交」再去别处看结果,等于把 一次本来就要盯着看的操作拆成两步。每天的自动同步走的是 cron (``partner_directory.sync``),不经过这里。 """ integrations = await service.list_partner_integrations(db) if body.integration_id: integrations = [ i for i in integrations if str(i.id) == body.integration_id ] results = [] for integration in integrations: results.append(await service.sync_one( db, integration=integration, trace_id=_trace_id(request), apply_tags=body.apply_tags, )) await AuditService(db).log( trace_id=_trace_id(request), actor_type="user", actor_user_id=user.id, action="partner_directory.sync_triggered", status="success", resource_type="integration", resource_id=body.integration_id, detail={ "apply_tags": body.apply_tags, "integrations": len(integrations), "results": results, }, ) await db.commit() return success_response({"results": results})