feat: add tree utils

src/bk-user/bkuser/apps/sync/converters.py

@@ -28,11 +28,7 @@ class DataSourceUserConverter:
     def __init__(self, data_source: DataSource):
         self.data_source = data_source
         self.field_mapping = self._get_field_mapping()
-        self.custom_field_names = set(
-            TenantUserCustomField.objects.filter(
-                tenant_id=self.data_source.owner_tenant_id,
-            ).values_list("name", flat=True)
-        )
+        self.custom_fields = TenantUserCustomField.objects.filter(tenant_id=self.data_source.owner_tenant_id)
 
     def _get_field_mapping(self) -> List[DataSourceUserFieldMapping]:
         """获取字段映射配置"""
@@ -49,10 +45,7 @@ def _get_field_mapping(self) -> List[DataSourceUserFieldMapping]:
         # 2. 若数据源配置中不存在，或者格式异常，则根据字段配置中生成，字段映射方式为直接映射
         logger.warning("data source (id: %s) has no field mapping, generate from field settings", self.data_source.id)
 
-        for fields in [
-            UserBuiltinField.objects.all(),
-            TenantUserCustomField.objects.filter(tenant_id=self.data_source.owner_tenant_id),
-        ]:
+        for fields in [UserBuiltinField.objects.all(), self.custom_fields]:
             for f in fields:
                 field_mapping.append(  # noqa: PERF401
                     DataSourceUserFieldMapping(
@@ -67,13 +60,14 @@ def _get_field_mapping(self) -> List[DataSourceUserFieldMapping]:
     def convert(self, user: RawDataSourceUser) -> DataSourceUser:
         # FIXME (su) 重构，支持复杂字段映射类型，如表达式，目前都当作直接映射处理（本地数据源只有直接映射）
         mapping = {m.source_field: m.target_field for m in self.field_mapping}
+        props = user.properties
         return DataSourceUser(
             data_source=self.data_source,
             code=user.code,
-            username=user.properties[mapping["username"]],
-            full_name=user.properties[mapping["full_name"]],
-            email=user.properties[mapping["email"]],
-            phone=user.properties[mapping["phone"]],
-            phone_country_code=user.properties[mapping["phone_country_code"]],
-            extras={k: v for k, v in user.properties.items() if k in self.custom_field_names},
+            username=props[mapping["username"]],
+            full_name=props[mapping["full_name"]],
+            email=props[mapping["email"]],
+            phone=props[mapping["phone"]],
+            phone_country_code=props[mapping["phone_country_code"]],
+            extras={f.name: props.get(f.name, f.default) for f in self.custom_fields},
         )

src/bk-user/bkuser/apps/sync/runners.py

@@ -32,7 +32,7 @@ class DataSourceSyncTaskRunner:
     """
     数据源同步任务执行器
 
-    FIXME (su) 后续支持软删除后，需要重构同步逻辑
+    FIXME (su) 1. 同步异常处理，2. Task 状态更新，3. 后续支持软删除后，需要重构同步逻辑
     """
 
     def __init__(self, task: DataSourceSyncTask, context: Dict[str, Any]):
@@ -77,7 +77,7 @@ class TenantSyncTaskRunner:
     """
     租户数据同步任务执行器
 
-    FIXME (su) 后续支持软删除后，需要重构同步逻辑
+    FIXME (su) 1. 同步异常处理，2. Task 状态更新，3. 后续支持软删除后，需要重构同步逻辑
     """
 
     def __init__(self, task: TenantSyncTask):

src/bk-user/bkuser/apps/sync/syncers.py

@@ -14,6 +14,7 @@
 from bkuser.apps.data_source.models import (
     DataSource,
     DataSourceDepartment,
+    DataSourceDepartmentRelation,
     DataSourceDepartmentUserRelation,
     DataSourceUser,
     DataSourceUserLeaderRelation,
@@ -38,7 +39,69 @@ def __init__(
         self.raw_departments = raw_departments
 
     def sync(self):
-        ...
+        self._sync_departments()
+        self._sync_department_relations()
+
+    def _sync_departments(self):
+        """数据源部门同步"""
+        dept_codes = set(
+            DataSourceDepartment.objects.filter(
+                data_source=self.data_source,
+            ).values_list("code", flat=True)
+        )
+        raw_dept_codes = {dept.code for dept in self.raw_departments}
+
+        waiting_create_dept_codes = raw_dept_codes - dept_codes
+        waiting_delete_dept_codes = dept_codes - raw_dept_codes
+        waiting_update_dept_codes = dept_codes & raw_dept_codes
+
+        if waiting_create_dept_codes:
+            self._create_departments([u for u in self.raw_departments if u.code in waiting_create_dept_codes])
+
+        if waiting_delete_dept_codes:
+            self._delete_departments([u for u in self.raw_departments if u.code in waiting_delete_dept_codes])
+
+        if waiting_update_dept_codes:
+            self._update_departments([u for u in self.raw_departments if u.code in waiting_update_dept_codes])
+
+    def _create_departments(self, raw_departments: List[RawDataSourceDepartment]):
+        # FIXME (su) 记录创建的日志
+        departments = [
+            DataSourceDepartment(data_source=self.data_source, code=dept.code, name=dept.name)
+            for dept in raw_departments
+        ]
+        DataSourceDepartment.objects.bulk_create(departments, batch_size=DATA_SOURCE_SYNC_BATCH_SIZE)
+
+    def _delete_departments(self, raw_departments: List[RawDataSourceDepartment]):
+        # FIXME (su) 记录删除的日志
+        DataSourceDepartment.objects.filter(
+            data_source=self.data_source, code__in=[u.code for u in raw_departments]
+        ).delete()
+
+    def _update_departments(self, raw_departments: List[RawDataSourceDepartment]):
+        # FIXME (su) 记录更新日志
+        dept_map = {
+            dept.code: DataSourceDepartment(data_source=self.data_source, code=dept.code, name=dept.name)
+            for dept in raw_departments
+        }
+
+        waiting_update_departments = DataSourceDepartment.objects.filter(
+            data_source=self.data_source, code__in=[u.code for u in raw_departments]
+        )
+        for u in waiting_update_departments:
+            target_dept = dept_map[u.code]
+            u.name = target_dept.deptname
+            # TODO (su) 确认 bulk_update 是否对 auto_now 的 updated_at 生效？
+
+        DataSourceDepartment.objects.bulk_update(
+            waiting_update_departments, fields=["name", "updated_at"], batch_size=DATA_SOURCE_SYNC_BATCH_SIZE
+        )
+
+    def _sync_department_relations(self):
+        """数据源部门关系同步，目前采用全部删除，再重建的方式"""
+        with DataSourceDepartmentRelation.objects.disable_mptt_updates():
+            DataSourceDepartmentRelation.objects.filter(data_source=self.data_source).delete()
+            # TODO (su) 重新建树
 
 
 class DataSourceUserSyncer:

src/bk-user/bkuser/plugins/local/parser.py

@@ -67,7 +67,6 @@ class LocalDataSourceDataParser:
         "full_name",
         "email",
         "phone_number",
-        "organizations",
     ]
 
     def __init__(self, workbook: Workbook):

src/bk-user/bkuser/plugins/local/plugin.py

@@ -29,6 +29,7 @@ class LocalDataSourcePlugin(BaseDataSourcePlugin):
     def __init__(self, plugin_config: LocalDataSourcePluginConfig, workbook: Workbook):
         self.plugin_config = plugin_config
         self.workbook = workbook
+        # FIXME (su) 不要在这里初始化，在 fetch 之前才加载数据 & 检查
         self.parser = LocalDataSourceDataParser(workbook)
         self.parser.parse()
 

src/bk-user/bkuser/utils/tree.py

@@ -0,0 +1,42 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-用户管理(Bk-User) available.
+Copyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from typing import Generator, List, Tuple
+
+from pydantic import BaseModel
+
+
+class TreeNode(BaseModel):
+    id: str
+    children: List["TreeNode"] = []
+
+
+def build_forest_with_parent_relations(relations: List[Tuple[str, str]]) -> List[TreeNode]:
+    """根据提供的父子关系构建树/森林，父子关系结构：(node_id, parent_id)"""
+    node_map = {node_id: TreeNode(id=node_id) for node_id, _ in relations}
+    roots = []
+    for node_id, parent_id in relations:
+        node = node_map[node_id]
+        if not (parent_id and parent_id in node_map):
+            roots.append(node)
+            continue
+
+        node_map[parent_id].children.append(node)
+
+    return roots
+
+
+def bfs_traversal_forest(roots: List[TreeNode]) -> Generator[TreeNode, None, None]:
+    """广度优先遍历森林，确保父节点都在子节点之前"""
+    queue = list(roots)
+    while queue:
+        node = queue.pop(0)
+        yield node
+        queue.extend(node.children)

src/bk-user/tests/utils/test_tree.py

@@ -0,0 +1,79 @@
+# -*- coding: utf-8 -*-
+"""
+TencentBlueKing is pleased to support the open source community by making 蓝鲸智云-用户管理(Bk-User) available.
+Copyright (C) 2017-2021 THL A29 Limited, a Tencent company. All rights reserved.
+Licensed under the MIT License (the "License"); you may not use this file except in compliance with the License.
+You may obtain a copy of the License at http://opensource.org/licenses/MIT
+Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
+an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
+specific language governing permissions and limitations under the License.
+"""
+from typing import List, Tuple
+
+from bkuser.utils.tree import TreeNode, bfs_traversal_forest, build_forest_with_parent_relations
+
+
+def test_build_forest_with_tree_parent_relations():
+    """理想情况，只有一棵树"""
+    relations = [("A", ""), ("B", "A"), ("C", "A"), ("D", "B"), ("E", "B")]
+    roots = build_forest_with_parent_relations(relations)
+    assert roots == [
+        TreeNode(
+            id="A",
+            children=[
+                TreeNode(
+                    id="B",
+                    children=[TreeNode(id="D"), TreeNode(id="E")],
+                ),
+                TreeNode(id="C"),
+            ],
+        )
+    ]
+
+
+def test_build_forest_with_forest_parent_relations():
+    """森林关系测试"""
+    relations = [("A", ""), ("C", "B"), ("D", "B"), ("B", "")]
+    roots = build_forest_with_parent_relations(relations)
+    assert roots == [
+        TreeNode(id="A"),
+        TreeNode(id="B", children=[TreeNode(id="C"), TreeNode(id="D")]),
+    ]
+
+
+def test_build_forest_with_invalid_parent_relations():
+    """森林关系测试，但是某父节点丢失"""
+    relations = [("A", ""), ("C", "B"), ("D", "B")]
+    roots = build_forest_with_parent_relations(relations)
+    assert roots == [TreeNode(id="A"), TreeNode(id="C"), TreeNode(id="D")]
+
+
+def test_build_forest_with_empty_parent_relations():
+    """空关系测试"""
+    relations: List[Tuple[str, str]] = []
+    roots = build_forest_with_parent_relations(relations)
+    assert len(roots) == 0
+
+
+def test_bfs_traversal_forest():
+    """正常情况测试"""
+    roots = [
+        TreeNode(id="A", children=[TreeNode(id="B"), TreeNode(id="C")]),
+        TreeNode(id="D", children=[TreeNode(id="E")]),
+    ]
+    nodes = list(bfs_traversal_forest(roots))
+    assert [n.id for n in nodes] == ["A", "D", "B", "C", "E"]
+
+
+def test_bfs_traversal_forest_empty():
+    """空森林测试"""
+    roots: List[TreeNode] = []
+    nodes = list(bfs_traversal_forest(roots))
+    assert nodes == []
+
+
+def test_bfs_traversal_forest_single():
+    """单个节点测试"""
+    roots = [TreeNode(id="A")]
+    nodes = list(bfs_traversal_forest(roots))
+    assert nodes == roots