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."""

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}]})