"""Server-side DingTalk auth-code exchange and local account upsert.""" import json import time from datetime import UTC, datetime import httpx from app.core.config import get_settings from app.core.exceptions import AppException from app.core.security import create_access_token from app.models.entities import OperationAudit, UserAccount from app.repositories.users import UserRepository from app.schemas.auth import DingTalkConfigData, TokenData from app.services.auth import AuthService TOKEN_URL = "https://oapi.dingtalk.com/gettoken" USERINFO_URL = "https://oapi.dingtalk.com/topapi/v2/user/getuserinfo" USERDETAIL_URL = "https://oapi.dingtalk.com/topapi/v2/user/get" class DingTalkServiceError(AppException): def __init__(self, message: str = "钉钉服务暂时不可用，请稍后重试") -> None: super().__init__(message, code=503, status_code=503) class DingTalkAuthService: _token_cache: tuple[str, float] | None = None def __init__(self, session) -> None: # noqa: ANN001 self.session = session self.repo = UserRepository(session) @staticmethod def config() -> DingTalkConfigData: settings = get_settings() if not settings.dingtalk_app_key or not settings.dingtalk_agent_id or not settings.dingtalk_corp_id: raise DingTalkServiceError("钉钉免登未配置，请联系管理员") return DingTalkConfigData( corp_id=settings.dingtalk_corp_id, app_key=settings.dingtalk_app_key, agent_id=settings.dingtalk_agent_id, ) async def login(self, auth_code: str) -> TokenData: settings = get_settings() if not settings.dingtalk_app_secret or not settings.dingtalk_app_secret.get_secret_value().strip(): raise DingTalkServiceError("钉钉免登未配置，请联系管理员") access_token = await self._app_access_token() user_id = await self._exchange_user_id(access_token, auth_code) detail = await self._user_detail(access_token, user_id) user = await self._upsert_user(user_id, detail) if user.status != "启用": raise AppException("用户账号已停用", code=401, status_code=401) permissions = await self.repo.permissions_for_role(user.role) permission_keys = sorted({item.func_key for item in permissions}) self.session.add( OperationAudit( actor_user_id=user.id, actor_emp_no=user.emp_no, action="auth.dingtalk_login", ) ) return TokenData( access_token=create_access_token(user.id, role=user.role), expires_in=settings.jwt_expire_minutes * 60, user=AuthService.to_user_data(user, permission_keys), ) async def _app_access_token(self) -> str: settings = get_settings() if self._token_cache and self._token_cache[1] > time.time() + 60: return self._token_cache[0] try: async with httpx.AsyncClient(timeout=8) as client: response = await client.get( TOKEN_URL, params={ "appkey": settings.dingtalk_app_key, "appsecret": settings.dingtalk_app_secret.get_secret_value(), }, ) response.raise_for_status() data = response.json() except httpx.HTTPError as exc: raise DingTalkServiceError() from exc if data.get("errcode") or not data.get("access_token"): raise DingTalkServiceError() self._token_cache = (data["access_token"], time.time() + int(data.get("expires_in", 7200))) return data["access_token"] async def _exchange_user_id(self, access_token: str, auth_code: str) -> str: try: async with httpx.AsyncClient(timeout=8) as client: response = await client.post( USERINFO_URL, params={"access_token": access_token}, json={"code": auth_code}, ) response.raise_for_status() data = response.json() except httpx.HTTPError as exc: raise AppException("钉钉免登码已失效，请重新进入工作台", code=401, status_code=401) from exc user_id = (data.get("result") or {}).get("userid") if data.get("errcode") or not user_id: raise AppException("钉钉免登码已失效，请重新进入工作台", code=401, status_code=401) return str(user_id) async def _user_detail(self, access_token: str, user_id: str) -> dict: try: async with httpx.AsyncClient(timeout=8) as client: response = await client.post( USERDETAIL_URL, params={"access_token": access_token}, json={"userid": user_id, "language": "zh_CN"}, ) response.raise_for_status() data = response.json() except httpx.HTTPError: return {"userid": user_id} return data.get("result") or {"userid": user_id} async def _upsert_user(self, user_id: str, detail: dict) -> UserAccount: user = await self.repo.get_by_dingtalk_id(user_id) role = self.map_role( str(detail.get("dept_name") or detail.get("deptName") or ""), str(detail.get("title") or detail.get("position") or ""), self._is_dingtalk_manager(detail), ) if user is None: user = UserAccount( emp_no=f"dt_{user_id}"[:20], name=str(detail.get("name") or user_id)[:50], role=role, status="启用", dingtalk_id=user_id, ) self.session.add(user) user.name = str(detail.get("name") or user.name)[:50] user.dept = str(detail.get("dept_name") or user.dept or "")[:50] or None user.position = str(detail.get("title") or user.position or "")[:50] or None user.role = role user.dingtalk_union_id = str(detail.get("unionid") or user.dingtalk_union_id or "")[:64] or None user.dingtalk_avatar = str(detail.get("avatar") or user.dingtalk_avatar or "")[:512] or None user.dingtalk_job_number = str(detail.get("job_number") or user.dingtalk_job_number or "")[:64] or None user.dingtalk_title = str(detail.get("title") or user.dingtalk_title or "")[:128] or None dept_id = detail.get("main_dept_id") or detail.get("dept_id") or user.dingtalk_dept_id if isinstance(dept_id, list): dept_id = dept_id[0] if dept_id else None try: user.dingtalk_dept_id = int(dept_id) if dept_id is not None else None except (TypeError, ValueError): user.dingtalk_dept_id = None user.dingtalk_is_dept_manager = self._is_dingtalk_manager(detail) role_codes = detail.get("role_list") or user.dingtalk_role_codes if isinstance(role_codes, (list, dict)): role_codes = json.dumps(role_codes, ensure_ascii=False) user.dingtalk_role_codes = str(role_codes or "")[:512] or None user.last_login_at = datetime.now(UTC) user.dingtalk_id = user_id user.updated_at = datetime.now(UTC) await self.session.flush() return user @staticmethod def map_role(dept: str, title: str, is_admin: bool = False) -> str: """Map DingTalk users to admin (including review) or regular user.""" return "admin" if is_admin else "user" @staticmethod def _is_dingtalk_manager(detail: dict) -> bool: """Read administrator/department-leader flags from DingTalk API variants.""" direct = ( detail.get("admin"), detail.get("is_admin"), detail.get("isAdmin"), detail.get("is_dept_manager"), detail.get("isDeptManager"), ) if any(value is True or str(value).lower() in {"1", "true"} for value in direct if value is not None): return True leaders = detail.get("leader_in_dept") or detail.get("leaderInDept") or [] if isinstance(leaders, dict): leaders = [leaders] if isinstance(leaders, list): return any( item is True or str(item).lower() in {"1", "true"} or isinstance(item, dict) and any( value is True or str(value).lower() in {"1", "true"} for value in item.values() ) for item in leaders ) return False