bugfix: 修复LDAP同步时code相同但username变更导致用户重建失败

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

Author: justin

src/api/bkuser_core/categories/plugins/ldap/helper.py

@@ -139,6 +139,13 @@ def db_profiles(self) -> Dict[str, Profile]:
         # 由于 bulk_update 需要从数据库查询完整的 Profile 信息, 为提高查询效率, 统一执行查询操作, 减轻数据库负担
         return {profile.username: profile for profile in Profile.objects.filter(category_id=self.category.pk).all()}
 
+    @cached_property
+    def db_profiles_by_code(self) -> Dict[str, Profile]:
+        return {
+            profile.code: profile
+            for profile in Profile.all_objects.filter(category_id=self.category.pk).exclude(code__isnull=True).exclude(code="")
+        }
+
     @cached_property
     def db_departments(self) -> Dict[str, Department]:
         # 由于 bulk_update 需要从数据库查询完整的 Department 信息, 为提高查询效率, 统一执行查询操作, 减轻数据库负担
@@ -182,13 +189,17 @@ def _load_base_info(self):
                 "extras": info.extras,
             }
 
+            self._delete_profile_if_username_changed(info.code, info.username)
+
             # 2. 更新或创建 Profile 对象
             if info.username in self.db_profiles:
                 profile = self.db_profiles[info.username]
                 for key, value in profile_params.items():
                     setattr(profile, key, value)
                 self.db_sync_manager.magic_add(profile, SyncOperation.UPDATE.value)
                 profile_ids.add(profile.id)
+                self.db_profiles[info.username] = profile
+                self.db_profiles_by_code[info.code] = profile
             else:
                 profile = Profile(**profile_params)
                 if self.db_sync_manager.magic_exists(profile):
@@ -202,6 +213,8 @@ def _load_base_info(self):
                     profile.id = self.db_sync_manager.register_id(LdapProfileMeta)
                     self.db_sync_manager.magic_add(profile, SyncOperation.ADD.value)
                     profile_ids.add(profile.id)
+                self.db_profiles[info.username] = profile
+                self.db_profiles_by_code[info.code] = profile
 
             # 3. 维护关联关系
             for full_department_name_list in info.departments:
@@ -235,6 +248,24 @@ def _load_base_info(self):
             self.context.add_record(step=SyncStep.USERS, success=True, username=info.username)
         self.profile_ids = profile_ids
 
+    def _delete_profile_if_username_changed(self, code: str, username: str) -> None:
+        profile = self.db_profiles_by_code.get(code)
+        if not profile or profile.username == username:
+            return
+
+        logger.info(
+            "profile code<%s> exists but username changed from <%s> to <%s>, will hard delete old profile and recreate",
+            code,
+            profile.username,
+            username,
+        )
+
+        old_profile_id = profile.id
+        Profile.all_objects.filter(id=old_profile_id).delete()
+        self.db_profiles.pop(profile.username, None)
+        self.db_profiles_by_code.pop(code, None)
+        return
+
     def _load_leader_info(self):
         # 加载上下级关系
         # WeOps改造,记录原有的上级关系

src/api/bkuser_core/tests/categories/plugins/ldap/test_syncer.py

@@ -172,6 +172,45 @@ def test_sync_users(self, pre_created, test_ldap_syncer, users, expected_adding,
                 in test_ldap_syncer.db_sync_manager._sets[ProfileMeta.target_model].updating_items
             )
 
+    def test_sync_users_recreate_when_code_exists_but_username_changes(self, test_ldap_syncer):
+        make_simple_profile(
+            "old_username",
+            force_create_params={
+                "category_id": test_ldap_syncer.category_id,
+                "code": "same-code",
+            },
+        )
+
+        users = [
+            {
+                "raw_dn": b"CN=dddd aaaaa,OU=shenzhen,DC=center,DC=com",
+                "dn": "CN=dddd aaaaa,OU=shenzhen,DC=center,DC=com",
+                "raw_attributes": {
+                    "displayName": [b"dddd aaaaa"],
+                    "cn": [b"new_username"],
+                    "mail": [b"aaaa@asdf.com"],
+                    "telephonenumber": [b"123412341234"],
+                    "memberOf": [],
+                },
+                "attributes": {
+                    "displayName": "dddd aaaaa",
+                    "sAMAccountName": "new_username",
+                    "mail": "aaaa@asdf.com",
+                    "mobile": "123412341234",
+                    "memberOf": [],
+                },
+                "type": "searchResEntry",
+            }
+        ]
+
+        with mock.patch.object(test_ldap_syncer.fetcher, "fetch") as fetch:
+            fetch.return_value = [], [], users
+            test_ldap_syncer._sync_profile()
+
+        assert not test_ldap_syncer.db_sync_manager.magic_get("old_username", ProfileMeta)
+        new_profile = test_ldap_syncer.db_sync_manager.magic_get("new_username", ProfileMeta)
+        assert new_profile in test_ldap_syncer.db_sync_manager._sets[ProfileMeta.target_model].adding_items
+
     @pytest.mark.parametrize(
         "departments, expected, expected_count",
         [