From e965aad7248d662e65a3dc92636f2b910ae12b2c Mon Sep 17 00:00:00 2001 From: kx_guo Date: Fri, 18 Sep 2026 13:46:04 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=9D=E7=95=99=E9=92=89=E9=92=89=E7=94=A8?= =?UTF-8?q?=E6=88=B7=E4=BA=BA=E5=B7=A5=E6=8C=87=E5=AE=9A=E8=A7=92=E8=89=B2?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/services/dingtalk_auth.py | 19 ++++++++++++++++--- tests/test_auth.py | 8 ++++++++ 2 files changed, 24 insertions(+), 3 deletions(-) diff --git a/app/services/dingtalk_auth.py b/app/services/dingtalk_auth.py index 5681d23..17cae50 100644 --- a/app/services/dingtalk_auth.py +++ b/app/services/dingtalk_auth.py @@ -192,7 +192,7 @@ class DingTalkAuthService: 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( + mapped_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), @@ -201,7 +201,7 @@ class DingTalkAuthService: user = UserAccount( emp_no=f"dt_{user_id}"[:20], name=self._display_name(detail, user_id), - role=role, + role=mapped_role, status="启用", dingtalk_id=user_id, ) @@ -209,7 +209,7 @@ class DingTalkAuthService: user.name = self._display_name(detail, user.name or user_id) user.dept = (self._department_name(detail) or user.dept or "")[:50] or None user.position = str(detail.get("title") or detail.get("position") or detail.get("job_title") or user.position or "")[:50] or None - user.role = role + user.role = self.resolve_synced_role(user.role, mapped_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 @@ -237,6 +237,19 @@ class DingTalkAuthService: return "admin" if is_admin else "user" + @staticmethod + def resolve_synced_role(current_role: str, mapped_role: str) -> str: + """Apply DingTalk role mapping without downgrading assigned roles. + + Existing roles are treated as administrator-managed data. DingTalk may + promote a regular user when it reports a manager flag, but a missing + flag must never downgrade an existing admin (or another assigned role). + """ + + if current_role == "user" and mapped_role == "admin": + return "admin" + return current_role + @staticmethod def _is_dingtalk_manager(detail: dict) -> bool: """Read administrator/department-leader flags from DingTalk API variants.""" diff --git a/tests/test_auth.py b/tests/test_auth.py index bdcfe63..fab81f3 100644 --- a/tests/test_auth.py +++ b/tests/test_auth.py @@ -10,6 +10,14 @@ def test_dingtalk_role_mapping() -> None: assert DingTalkAuthService.map_role("未知部门", "普通员工") == "user" +def test_dingtalk_sync_only_promotes_user_and_never_downgrades_admin() -> None: + assert DingTalkAuthService.resolve_synced_role("user", "admin") == "admin" + assert DingTalkAuthService.resolve_synced_role("user", "user") == "user" + assert DingTalkAuthService.resolve_synced_role("admin", "user") == "admin" + assert DingTalkAuthService.resolve_synced_role("admin", "admin") == "admin" + assert DingTalkAuthService.resolve_synced_role("reviewer", "admin") == "reviewer" + + def test_dingtalk_manager_flag_variants() -> None: assert DingTalkAuthService._is_dingtalk_manager({"admin": True}) assert DingTalkAuthService._is_dingtalk_manager({"leader_in_dept": [{"leader": True}]})