BiSheng 三方组织同步F00917 个任务的拆解、依赖编排与 Provider Reconciler 架构实现要点【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng本文以 BiSheng 开源仓库中 v2.5.0 特性「009-org-sync」的任务拆解文档tasks.md为主体结合其配套规格 spec.md 与 org_sync 模块源码 展开。读者将掌握三方组织同步功能从 ORM 建模、Provider 实现、差异调和Reconciler到 Celery 异步调度、API 端点的完整任务拆分方式以及任务之间的依赖关系与并行执行策略可作为理解该功能源码结构、验收标准与测试覆盖的直接索引。一、功能背景企业组织架构的自动同步F009 为 BiSheng 引入第三方组织架构同步能力企业可以把飞书、企微、钉钉等平台中的部门树与员工信息同步进 BiSheng自动创建部门、用户与部门成员关系并维护 OpenFGA 权限元组支持手动触发与 Cron 定时同步。其价值在于系统管理员无需再逐个手工创建部门和用户组织变动调岗、离职后权限也会自动调整。该特性采用Provider Reconciler的分层架构Provider负责从第三方平台拉取数据并统一转换为标准 DTOReconciler纯逻辑的差异比较引擎把远程数据与本地数据对比产出差异操作列表OrgSyncService同步编排器把「认证 → 拉取 → 调和 → 执行 → 记录」串成完整流程。从任务拆解的角度看该功能共拆出17 个任务T-01 ~ T-17分布在5 个 Phase中任务间有明确的依赖顺序与可并行的「Wave」划分是理解 DDD 模块化开发与异步同步系统设计的一个完整样例。二、任务总览5 个 Phase、17 个任务tasks.md 将整个特性拆解为以下结构Phase主题任务Phase 1Foundation基础设施T-01 ORM 模型与 DAO、T-02 数据库迁移、T-03 错误码、T-04 User 模型扩展、T-05 Provider 抽象基类与远程 DTOPhase 2Providers数据源T-06 飞书 Provider、T-07 通用 API Provider、T-08 企微/钉钉 stubPhase 3Sync Engine同步引擎T-09 部门差异引擎、T-10 人员差异引擎、T-11 同步编排器Phase 4Infrastructure APIT-12 Celery 任务与 Beat 调度、T-13 请求/响应 DTO、T-14 配置 CRUD API、T-15 执行/测试/历史/远程树 APIPhase 5Testing测试T-16 Reconciler 单元测试、T-17 API 集成 E2E 测试每个任务都包含「文件、内容、覆盖 AC、验证」四要素其中「覆盖 AC」字段把任务与 spec.md 中的验收标准AC-01 ~ AC-34建立了可追溯关系这是该拆解文档最值得借鉴的一点验收标准驱动任务拆解测试通过「覆盖 AC」反向追溯需求。三、Phase 1Foundation——数据模型与领域骨架T-01OrgSyncConfig OrgSyncLog ORM 模型与 DAO新建src/backend/bisheng/org_sync/domain/models/org_sync.py定义两张核心表OrgSyncConfig同步配置表。含tenant_id多租户隔离、providerfeishu/wecom/dingtalk/generic_api、config_name、auth_typeapi_key/passwordoauth 预留、auth_configFernet 加密的 JSON 文本、sync_scope如{root_dept_ids: [id1,id2]}null 表示全量、schedule_typemanual/cron、cron_expression、sync_statusidle/running 运行时互斥、last_sync_at/last_sync_result、statusactive/disabled/deleted等字段并带有唯一约束uk_tenant_provider_name(tenant_id, provider, config_name)。OrgSyncLog同步日志表。记录config_id、trigger_typemanual/scheduled、statusrunning/success/partial/failed以及各统计计数器dept_created/dept_updated/dept_archived、member_created/member_updated/member_disabled/member_reactivated另有error_detailsJSON 字段格式[{entity_type, external_id, error_msg}]记录部分失败明细。DAO 层提供OrgSyncConfigDao.acreate / aget_by_id / aget_list / aupdate / aset_sync_status / aget_active_cron_configs以及OrgSyncLogDao.acreate / aupdate / aget_by_config分页。值得注意的实现细节见 org_sync.py 源码加解密辅助函数encrypt_auth_config/decrypt_auth_config基于 Fernet 实现AD-02复用settings.secret_key与数据库密码加密方式一致decrypt_auth_config对空字符串返回{}避免空配置触发 Fernet InvalidToken 异常。CAS 式状态更新aset_sync_status(config_id, old_status, new_status)使用UPDATE ... WHERE sync_status old_status的原子条件更新并返回rowcount 0这是并发互斥的第一道闸门。T-02数据库迁移脚本新建 Alembic 迁移v2_5_0_f009_org_sync.py包含CREATE TABLE org_sync_config全列 索引 唯一键CREATE TABLE org_sync_log全列 索引ALTER TABLE user ADD COLUMN source VARCHAR(32) NOT NULL DEFAULT localALTER TABLE user ADD COLUMN external_id VARCHAR(128) NULLCREATE UNIQUE INDEX uk_user_source_external_id ON user(source, external_id)。迁移依赖 T-01先有模型定义再写迁移验证方式是在 MySQL 上执行 upgrade/downgrade 无报错。T-03错误码模块 220新建src/backend/bisheng/common/errcode/org_sync.py定义 22000~22009 共 10 个错误码类命名遵循OrgSync{Error}Error。源码中均已实现见 errcode/org_sync.py错误码类名场景22000OrgSyncConfigNotFoundError配置 ID 不存在或不属于当前租户22001OrgSyncConfigDuplicateError同 provider config_name 已存在22002OrgSyncAuthFailedErrorProvider 认证失败22003OrgSyncAlreadyRunningError该配置正在同步中22004OrgSyncProviderErrorProvider API 错误或未实现22005OrgSyncPermissionDeniedError无组织同步操作权限22006OrgSyncInvalidConfigError配置字段缺失或不合法22007OrgSyncFetchError从 Provider 拉取数据失败22008OrgSyncReconcileError调和过程不可恢复错误22009OrgSyncConfigDisabledError配置已禁用模块编码 220 已在版本契约 release-contract.md 中注册。T-04User 模型扩展 UserDao 新方法修改src/backend/bisheng/user/domain/models/user.py为 User 新增source: str Field(defaultlocal, ...) # local/feishu/wecom/dingtalk/generic_api external_id: Optional[str] Field(defaultNone, ...) # 外部员工 ID并增加唯一约束UniqueConstraint(source, external_id, nameuk_user_source_external_id)——与 Department 模型的uk_source_external_id保持一致AD-06单表索引、查询直接、reconcile 匹配性能更好。UserDao 新增两个方法aget_by_source_external_id(source, external_id)按 source external_id 精确匹配用户aget_by_source(source, tenant_id)查询某来源的全部用户供 reconcile 一次性加载。兼容性设计新字段均有默认值sourcelocal、external_idNULL存量用户不受迁移影响。T-05Provider 抽象基类 远程 DTO新建src/backend/bisheng/org_sync/domain/providers/base.py与src/backend/bisheng/org_sync/domain/schemas/remote_dto.py。OrgSyncProvider是抽象基类定义了 4 个抽象方法见 base.py 源码class OrgSyncProvider(ABC): def __init__(self, auth_config: dict): ... abstractmethod async def authenticate(self) - bool: ... # 校验凭证并获取 token abstractmethod async def fetch_departments(self, root_dept_idsNone) - list[RemoteDepartmentDTO]: ... abstractmethod async def fetch_members(self, department_idsNone) - list[RemoteMemberDTO]: ... abstractmethod async def test_connection(self) - dict: ... # 返回 connected/org_name/total_depts/total_members工厂方法get_provider(provider, auth_config)通过懒加载注册表_PROVIDER_REGISTRY实例化对应 Provider对未知 provider 抛出OrgSyncProviderError(msgUnknown provider: ...)。远程 DTO 是两个 dataclassRemoteDepartmentDTOexternal_id, name, parent_external_id(None根), sort_orderRemoteMemberDTOexternal_id, name, email, phone, primary_dept_external_id, secondary_dept_external_ids, status(active/disabled)。Provider 的扩展点是该特性可扩展性的核心新增第三方平台只需实现 4 个方法并注册到工厂无需修改同步引擎逻辑。四、Phase 2Providers——飞书完整实现与通用 APIT-06FeishuProvider 完整实现新建src/backend/bisheng/org_sync/domain/providers/feishu.py对接飞书开放平台通讯录 API v3见 feishu.py 源码方法飞书 API说明authenticatePOST/auth/v3/tenant_access_token/internal用 app_id app_secret 换取 tenant_access_tokenfetch_departmentsGET/contact/v3/departments/{id}/childrenBFS 遍历部门树支持 scope 过滤page_token 分页fetch_membersGET/contact/v3/users?department_idX按部门拉取成员page_token 分页、按 open_id 去重test_connectionauthenticate GET/contact/v3/departments/0验证连接并返回根部门信息关键实现细节使用httpx.AsyncClienttimeout30s发起请求token 缓存 2 小时实例属性_token_expires_at提前 1 分钟刷新并发控制asyncio.Semaphore(5)限制并发请求数避免触发第三方限流429 指数退避重试wait 1 * (2 ** attempt)即 1s/2s/4s最多重试 3 次超过则抛OrgSyncFetchError成员状态识别飞书返回的status.is_frozen/status.is_resigned映射为disabled否则active。T-07GenericAPIProvider 完整实现新建src/backend/bisheng/org_sync/domain/providers/generic_api.py面向无法直接对接飞书/企微/钉钉的长尾数据源。其核心是可配置端点 字段映射{ departments_url: https://api.example.com/departments, members_url: https://api.example.com/members, api_key: sk-xxx, param_location: header, field_mapping: { dept_id: id, dept_name: name, dept_parent_id: parentId, member_id: employeeId, member_name: fullName, member_email: email, member_phone: mobile, member_primary_dept: mainDepartment, member_secondary_depts: otherDepartments, member_status: status } }标准响应格式为{departments: [...]}与{members: [...]}通过field_mapping映射到标准 DTO若返回不符合标准格式Provider 抛OrgSyncFetchError并附带原始响应摘要。认证按auth_type区分api_key 模式把 key 放入 header/query 参数password 模式走基本认证。T-08WeComProvider DingTalkProvider stub新建wecom.py/dingtalk.py两个 Provider 继承OrgSyncProvider4 个方法均抛出OrgSyncProviderError(msgWeChat Work provider not implemented)类文档注释中写明 API 契约留给后续实现者。这是 AD-08 决策的落地飞书覆盖国内主流场景、通用 API 覆盖长尾需求企微/钉钉留 stub 降低首版交付范围。五、Phase 3Sync Engine——差异调和与编排T-09/T-10Reconciler 差异引擎纯逻辑、无 IO新建src/backend/bisheng/org_sync/domain/services/reconciler.py。这是整个同步引擎最核心的部分纯逻辑、无 IO 依赖输入数据输出操作列表因此可以完全脱离数据库做单元测试。部门调和reconcile_departments(remote_depts, local_depts, source) → list[DeptOperation]操作类型为CreateDept | UpdateDept | MoveDept | ArchiveDept判定逻辑创建remote 有而 local 无 → CreateDept更新两边都有且 name 不同 → UpdateDept移动两边都有且 parent_external_id 对应父节点变化 → MoveDept归档local 有source 匹配而 remote 无 → ArchiveDept含子树级联归档本地冲突local Department.sourcelocal 但 external_id 匹配远程 → 强制覆盖 name 并把 source 改为第三方AC-18拓扑排序创建操作按「父先于子」排序Kahn 算法归档按「子先于父」排序按 path 深度倒序若检测到循环引用跳过受影响节点。人员调和reconcile_members(...) → list[MemberOperation]操作类型为CreateMember | UpdateMember | TransferMember | DisableMember | ReactivateMember判定逻辑创建remote 有而 local 无且 statusactive → CreateMember更新name/email/phone 变化 → UpdateMember转岗主部门变化 → TransferMember带old_primary_dept_idtouch_primary区分是否真的动了主部门附属部门增减 → 通过add_secondary_external_ids/remove_secondary_dept_ids表达禁用local 有而 remote 无或 remote.statusdisabled → DisableMember携带待清理的 dept_ids重新激活local.delete1 且 remote.statusactive → ReactivateMember本地冲突User.sourcelocal 但 external_id 匹配远程 → 强制覆盖并改 source。源码中对两种冲突场景的处理reconciler.py部门的 local_by_ext 只收录有 external_id 的部门人员的 local_by_ext 在 source 冲突时优先生效第三方来源记录。T-11OrgSyncService 同步编排器新建src/backend/bisheng/org_sync/domain/services/org_sync_service.pyexecute_sync(config_id, trigger_type, trigger_user)按 spec §7 的 16 步主流程执行1. 加载 OrgSyncConfig → 解密 auth_config 2. 获取互斥锁 a. 检查 config.sync_status idle b. 原子 UPDATE sync_statusrunning WHERE sync_statusidleaffected0 则已被占用 c. 获取 Redis 锁 bisheng:lock:org_sync:{config_id}TTL30min 3. 创建 OrgSyncLog(statusrunning, start_timenow) 4. 实例化 Provider 5. provider.authenticate() 6. 拉取远程部门: fetch_departments(sync_scope) 7. 加载本地部门: Department where sourceprovider and tenant_id 8. Reconciler.reconcile_departments() → dept_ops 9. 执行 dept_ops_apply_dept_ops 10. 拉取远程人员: fetch_members() 11. 加载本地用户: User where sourceprovider and tenant_id经 UserTenant 12. Reconciler.reconcile_members() → member_ops 13. 执行 member_ops_apply_member_ops 14. 更新 OrgSyncLog(status, 统计, end_time) 15. 更新 OrgSyncConfig(last_sync_at, last_sync_result) 16. 释放互斥锁(sync_statusidle, Redis unlock)关键实现决策双重并发保护AD-04DB 的sync_status原子 CAS 保证持久性Redis 分布式锁保证进程崩溃后 30 分钟 TTL 自动释放部分失败处理AD-10逐条 try/except失败项累计进error_details最终日志状态置为partial不中断整体同步新用户密码AD-07secrets.token_hex(32)生成 64 位随机哈希用户无法猜测密码只能通过 SSO 登录绕过权限检查AD-11同步是系统级操作直接操作 DAO DepartmentChangeHandler绕过 DepartmentService 的_check_permission()与source_readonly检查同时通过 ChangeHandler 保证 OpenFGA 元组一致INV-4。_apply_dept_ops对四类操作分别处理CreateDept 创建 Department 后触发on_createdUpdateDept 更新 nameMoveDept 更新 parent_id path 批量替换后触发on_movedArchiveDept 置 archived 并清空成员保留 admin后触发on_archived。_apply_member_ops中 DisableMember 还会清理 Redis 登录态。六、Phase 4Infrastructure API——异步调度与接口层T-12Celery 任务 Beat 定时调度新建src/backend/bisheng/worker/org_sync/tasks.py包含两个任务bisheng_celery.task(acks_lateTrue, time_limit1800, soft_time_limit1500) def execute_org_sync(config_id, trigger_type, trigger_userNone): Async sync execution. Tenant context via INV-8. ... bisheng_celery.task(acks_lateTrue) def check_org_sync_schedules(): Beat task: check cron configs and dispatch if due. Runs every 60s. ...调度设计要点AD-10Beat 检查任务动态调度而非静态配置check_org_sync_schedules作为 Beat 任务每 60 秒执行一次通过 croniter 解析各活跃配置的cron_expression匹配时间则 dispatchexecute_org_sync生成trigger_typescheduled的日志任务路由bisheng.worker.org_sync.*: {queue: knowledge_celery}AD-03同步频率低复用 knowledge_celery 队列无需独立队列time_limit1800, soft_time_limit1500约束单次同步最长 30 分钟多租户上下文传递INV-8发送任务时 tenant_id 通过 Celery headers 注入inject_tenant_headersignalWorker 端before_tasksignal 调用set_current_tenant_id()恢复到 ContextVar被禁用statusdisabled的 cron 配置不会触发AC-32。T-13请求/响应 DTO新建src/backend/bisheng/org_sync/domain/schemas/org_sync_schema.pyOrgSyncConfigCreateprovider, config_name, auth_type, auth_config(dict), sync_scope, schedule_type, cron_expressionOrgSyncConfigUpdateauth_type, auth_config, sync_scope, schedule_type, cron_expression, status全部 Optional支持部分更新OrgSyncConfigRead完整字段auth_config 为脱敏后的 dictOrgSyncLogRead完整字段RemoteTreeNodeexternal_id, name, children递归结构用于远程树预览mask_sensitive_fields(auth_config)把 app_secret/api_key/password 等敏感字段替换为****AC-34。T-14配置 CRUD API5 个端点新建src/backend/bisheng/org_sync/api/endpoints/sync_config.py路由聚合于org_sync/api/router.py并注册进全局src/backend/bisheng/api/router.py端点说明POST /api/v1/org-sync/configs创建auth_config 加密入库GET /api/v1/org-sync/configs配置列表当前租户脱敏GET /api/v1/org-sync/configs/{id}配置详情脱敏PUT /api/v1/org-sync/configs/{id}更新auth_config 合并更新解密 → 合并 → 重新加密DELETE /api/v1/org-sync/configs/{id}软删除statusdeleted同步中不可删返回 22003源码中的关键校验见 sync_config.py_get_config_or_404同时校验tenant_id匹配与status ! deleted跨租户访问返回 22000AC-08创建时的唯一约束冲突Duplicate entry/uk_tenant_provider_name映射为 22001。所有端点通过UserPayload.get_admin_user依赖注入强制管理员权限非管理员返回 22005AC-33响应统一走UnifiedResponseModelresp_200。T-15执行/测试/历史/远程树 API4 个端点新建src/backend/bisheng/org_sync/api/endpoints/sync_exec.py端点说明POST /api/v1/org-sync/configs/{id}/test测试连接返回connected/org_name/total_depts/total_membersPOST /api/v1/org-sync/configs/{id}/execute手动触发检查 sync_status → dispatch Celery 任务 → 返回 log_id异步GET /api/v1/org-sync/configs/{id}/logs?page1limit20同步历史PageData 分页GET /api/v1/org-sync/configs/{id}/remote-tree远程组织树预览嵌套结构约束要点execute 在配置sync_statusrunning时返回 22003、statusdisabled时返回 22009remote-tree 同步调用 Provider 可能耗时 5-10 秒大组织前端建议加 loading 态Celery Worker 未启动时任务进入队列等待API 立即返回 log_id异步语义。七、Phase 5Testing——单元、集成与 E2ET-16Reconciler 单元测试新建src/backend/test/org_sync/test_org_sync_reconciler.py源码中位于 test_org_sync_reconciler.py纯逻辑测试覆盖部门create / 第三方改名 / 本地部门改名 source 变更 / move / archive / 归档级联含本地子部门/ 拓扑序父先于子/ 循环引用跳过人员create / update / transfer主部门变更/ 附属部门增减 / disable离职/ reactivate重新出现/ 本地冲突强制覆盖。测试通过MagicMock构造 Department/User/UserDepartment 伪对象验证 Reconciler 产出的操作类型与字段全程无 IO覆盖 AC-16 ~ AC-28。T-17API 集成测试 E2E 测试API 集成测试test_org_sync_api.pyTestClient配置 CRUD happy path error path包括重复创建 22001、跨租户拒绝 22008、测试连接成功/认证失败、execute 已运行/已禁用、日志分页、远程树、非管理员 40322005等E2E 测试test_e2e_org_sync.pymock Provider 真实 DBtest_full_sync_flow配置 → 触发 → 验证 Department/User/UserDepartment/OrgSyncLog 状态test_incremental_sync首次同步后修改远程数据 → 二次同步 → 验证增量变更test_member_disable_on_departure员工离职 → 验证禁用 清理test_multi_tenant_isolation两个租户各自同步互不影响。八、任务依赖图与并行策略tasks.md 给出了完整的依赖图核心链路是T-01 (ORM) ──┬── T-02 (迁移) ├── T-05 (Provider ABC) ── T-06 (飞书) / T-07 (通用API) / T-08 (stub) T-03 (错误码)─┤ T-04 (User扩展)┤ ├── T-09 (部门Reconciler) ──┐ ├── T-10 (人员Reconciler) ──┤ │ v └── T-11 (OrgSyncService) ─┘ ├── T-12 (Celery) T-13 (DTO)┤ v T-14 (配置CRUD API) ── T-15 (执行API) ── T-17 (APIE2E) ^ T-16 (Reconciler测试) ────────┘据此划分的 8 个并行 WaveWave 1可并行T-01 T-03 T-04 T-05Wave 2T-01 完成后T-02Wave 3T-05 完成后可并行T-06 T-07 T-08Wave 4T-01/T-04/T-09/T-10 完成后T-11Wave 5T-11 完成后可并行T-12 T-13Wave 6T-13 完成后T-14Wave 7T-14 完成后T-15Wave 8T-15 完成后可并行T-16 T-17。这种「先 Foundation 打底、再 Provider 并行、引擎最后收敛、API 串行上垒、测试收尾」的编排最大化并行度的同时保证了每个任务输入就绪是可复用的多任务特性开发节奏。九、边界情况与质量约束从 spec.md 与 tasks.md 中沉淀出的关键边界处理也是实现验收时的重要依据飞书 API 分页自动循环、429 指数退避重试最多 3 次1s/2s/4s部分失败不中断整体同步OrgSyncLog.statuspartialerror_details记录entity_type/external_id/error_msg进程崩溃后 Redis 锁 TTL30min自动释放sync_status可通过手动 API 或下次同步前健康检查重置远程部门树循环引用拓扑排序检测到环后记录错误并跳过受影响子树同名部门在同一父级下冲突捕获DepartmentNameDuplicateError记入 error_details不中断User.external_id唯一约束冲突时走更新而非创建sync_scope为空/null 时同步全部远程部门与人员multi_tenant.enabledfalse时 tenant_id 自动填充默认租户id1不支持双向同步、实时 Webhook、OAuth 回调认证OAuth 字段保留 coming_soon 标记。非功能要求上同步 1000 部门 10000 人员应在 5 分钟内完成Provider 并发 Semaphore5本地数据一次性批量加载到内存 map 避免 N1OpenFGA 双写失败记入 FailedTuple 补偿表INV-4所有端点要求管理员权限tenant_id 自动过滤防止跨租户泄漏INV-1。十、从任务到源码阅读路线图如果希望在仓库中按任务的顺序阅读实现推荐路线数据层org_sync.pyORM DAO Fernet 加解密→ errcode/org_sync.py220xx 错误码Provider 层base.pyABC 工厂→ feishu.py完整实现含 BFS 遍历、Semaphore、429 退避→ generic_api.py / wecom.py / dingtalk.py引擎层reconciler.py纯逻辑差异引擎 拓扑排序API 层sync_config.py配置 CRUD→ sync_exec.py执行/测试/历史/远程树测试层test_org_sync_reconciler.py纯逻辑单测→ test_org_sync_api.pyAPI 集成→ e2e/test_e2e_org_sync.py全链路。F009 以 17 个任务覆盖了从建表、加密、并发控制、差异调和、异步调度到接口暴露的完整闭环其「任务 ↔ AC 验收标准 ↔ 测试用例」的三向可追溯结构以及 Provider 抽象带来的低成本扩展能力是阅读 BiSheng 后端 DDD 模块化实践的良好入口。【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 BiSheng 三方组织同步（F009）：17 个任务的拆解、依赖编排与 Provider + Reconciler 架构实现要点
BiSheng 三方组织同步F00917 个任务的拆解、依赖编排与 Provider Reconciler 架构实现要点【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng本文以 BiSheng 开源仓库中 v2.5.0 特性「009-org-sync」的任务拆解文档tasks.md为主体结合其配套规格 spec.md 与 org_sync 模块源码 展开。读者将掌握三方组织同步功能从 ORM 建模、Provider 实现、差异调和Reconciler到 Celery 异步调度、API 端点的完整任务拆分方式以及任务之间的依赖关系与并行执行策略可作为理解该功能源码结构、验收标准与测试覆盖的直接索引。一、功能背景企业组织架构的自动同步F009 为 BiSheng 引入第三方组织架构同步能力企业可以把飞书、企微、钉钉等平台中的部门树与员工信息同步进 BiSheng自动创建部门、用户与部门成员关系并维护 OpenFGA 权限元组支持手动触发与 Cron 定时同步。其价值在于系统管理员无需再逐个手工创建部门和用户组织变动调岗、离职后权限也会自动调整。该特性采用Provider Reconciler的分层架构Provider负责从第三方平台拉取数据并统一转换为标准 DTOReconciler纯逻辑的差异比较引擎把远程数据与本地数据对比产出差异操作列表OrgSyncService同步编排器把「认证 → 拉取 → 调和 → 执行 → 记录」串成完整流程。从任务拆解的角度看该功能共拆出17 个任务T-01 ~ T-17分布在5 个 Phase中任务间有明确的依赖顺序与可并行的「Wave」划分是理解 DDD 模块化开发与异步同步系统设计的一个完整样例。二、任务总览5 个 Phase、17 个任务tasks.md 将整个特性拆解为以下结构Phase主题任务Phase 1Foundation基础设施T-01 ORM 模型与 DAO、T-02 数据库迁移、T-03 错误码、T-04 User 模型扩展、T-05 Provider 抽象基类与远程 DTOPhase 2Providers数据源T-06 飞书 Provider、T-07 通用 API Provider、T-08 企微/钉钉 stubPhase 3Sync Engine同步引擎T-09 部门差异引擎、T-10 人员差异引擎、T-11 同步编排器Phase 4Infrastructure APIT-12 Celery 任务与 Beat 调度、T-13 请求/响应 DTO、T-14 配置 CRUD API、T-15 执行/测试/历史/远程树 APIPhase 5Testing测试T-16 Reconciler 单元测试、T-17 API 集成 E2E 测试每个任务都包含「文件、内容、覆盖 AC、验证」四要素其中「覆盖 AC」字段把任务与 spec.md 中的验收标准AC-01 ~ AC-34建立了可追溯关系这是该拆解文档最值得借鉴的一点验收标准驱动任务拆解测试通过「覆盖 AC」反向追溯需求。三、Phase 1Foundation——数据模型与领域骨架T-01OrgSyncConfig OrgSyncLog ORM 模型与 DAO新建src/backend/bisheng/org_sync/domain/models/org_sync.py定义两张核心表OrgSyncConfig同步配置表。含tenant_id多租户隔离、providerfeishu/wecom/dingtalk/generic_api、config_name、auth_typeapi_key/passwordoauth 预留、auth_configFernet 加密的 JSON 文本、sync_scope如{root_dept_ids: [id1,id2]}null 表示全量、schedule_typemanual/cron、cron_expression、sync_statusidle/running 运行时互斥、last_sync_at/last_sync_result、statusactive/disabled/deleted等字段并带有唯一约束uk_tenant_provider_name(tenant_id, provider, config_name)。OrgSyncLog同步日志表。记录config_id、trigger_typemanual/scheduled、statusrunning/success/partial/failed以及各统计计数器dept_created/dept_updated/dept_archived、member_created/member_updated/member_disabled/member_reactivated另有error_detailsJSON 字段格式[{entity_type, external_id, error_msg}]记录部分失败明细。DAO 层提供OrgSyncConfigDao.acreate / aget_by_id / aget_list / aupdate / aset_sync_status / aget_active_cron_configs以及OrgSyncLogDao.acreate / aupdate / aget_by_config分页。值得注意的实现细节见 org_sync.py 源码加解密辅助函数encrypt_auth_config/decrypt_auth_config基于 Fernet 实现AD-02复用settings.secret_key与数据库密码加密方式一致decrypt_auth_config对空字符串返回{}避免空配置触发 Fernet InvalidToken 异常。CAS 式状态更新aset_sync_status(config_id, old_status, new_status)使用UPDATE ... WHERE sync_status old_status的原子条件更新并返回rowcount 0这是并发互斥的第一道闸门。T-02数据库迁移脚本新建 Alembic 迁移v2_5_0_f009_org_sync.py包含CREATE TABLE org_sync_config全列 索引 唯一键CREATE TABLE org_sync_log全列 索引ALTER TABLE user ADD COLUMN source VARCHAR(32) NOT NULL DEFAULT localALTER TABLE user ADD COLUMN external_id VARCHAR(128) NULLCREATE UNIQUE INDEX uk_user_source_external_id ON user(source, external_id)。迁移依赖 T-01先有模型定义再写迁移验证方式是在 MySQL 上执行 upgrade/downgrade 无报错。T-03错误码模块 220新建src/backend/bisheng/common/errcode/org_sync.py定义 22000~22009 共 10 个错误码类命名遵循OrgSync{Error}Error。源码中均已实现见 errcode/org_sync.py错误码类名场景22000OrgSyncConfigNotFoundError配置 ID 不存在或不属于当前租户22001OrgSyncConfigDuplicateError同 provider config_name 已存在22002OrgSyncAuthFailedErrorProvider 认证失败22003OrgSyncAlreadyRunningError该配置正在同步中22004OrgSyncProviderErrorProvider API 错误或未实现22005OrgSyncPermissionDeniedError无组织同步操作权限22006OrgSyncInvalidConfigError配置字段缺失或不合法22007OrgSyncFetchError从 Provider 拉取数据失败22008OrgSyncReconcileError调和过程不可恢复错误22009OrgSyncConfigDisabledError配置已禁用模块编码 220 已在版本契约 release-contract.md 中注册。T-04User 模型扩展 UserDao 新方法修改src/backend/bisheng/user/domain/models/user.py为 User 新增source: str Field(defaultlocal, ...) # local/feishu/wecom/dingtalk/generic_api external_id: Optional[str] Field(defaultNone, ...) # 外部员工 ID并增加唯一约束UniqueConstraint(source, external_id, nameuk_user_source_external_id)——与 Department 模型的uk_source_external_id保持一致AD-06单表索引、查询直接、reconcile 匹配性能更好。UserDao 新增两个方法aget_by_source_external_id(source, external_id)按 source external_id 精确匹配用户aget_by_source(source, tenant_id)查询某来源的全部用户供 reconcile 一次性加载。兼容性设计新字段均有默认值sourcelocal、external_idNULL存量用户不受迁移影响。T-05Provider 抽象基类 远程 DTO新建src/backend/bisheng/org_sync/domain/providers/base.py与src/backend/bisheng/org_sync/domain/schemas/remote_dto.py。OrgSyncProvider是抽象基类定义了 4 个抽象方法见 base.py 源码class OrgSyncProvider(ABC): def __init__(self, auth_config: dict): ... abstractmethod async def authenticate(self) - bool: ... # 校验凭证并获取 token abstractmethod async def fetch_departments(self, root_dept_idsNone) - list[RemoteDepartmentDTO]: ... abstractmethod async def fetch_members(self, department_idsNone) - list[RemoteMemberDTO]: ... abstractmethod async def test_connection(self) - dict: ... # 返回 connected/org_name/total_depts/total_members工厂方法get_provider(provider, auth_config)通过懒加载注册表_PROVIDER_REGISTRY实例化对应 Provider对未知 provider 抛出OrgSyncProviderError(msgUnknown provider: ...)。远程 DTO 是两个 dataclassRemoteDepartmentDTOexternal_id, name, parent_external_id(None根), sort_orderRemoteMemberDTOexternal_id, name, email, phone, primary_dept_external_id, secondary_dept_external_ids, status(active/disabled)。Provider 的扩展点是该特性可扩展性的核心新增第三方平台只需实现 4 个方法并注册到工厂无需修改同步引擎逻辑。四、Phase 2Providers——飞书完整实现与通用 APIT-06FeishuProvider 完整实现新建src/backend/bisheng/org_sync/domain/providers/feishu.py对接飞书开放平台通讯录 API v3见 feishu.py 源码方法飞书 API说明authenticatePOST/auth/v3/tenant_access_token/internal用 app_id app_secret 换取 tenant_access_tokenfetch_departmentsGET/contact/v3/departments/{id}/childrenBFS 遍历部门树支持 scope 过滤page_token 分页fetch_membersGET/contact/v3/users?department_idX按部门拉取成员page_token 分页、按 open_id 去重test_connectionauthenticate GET/contact/v3/departments/0验证连接并返回根部门信息关键实现细节使用httpx.AsyncClienttimeout30s发起请求token 缓存 2 小时实例属性_token_expires_at提前 1 分钟刷新并发控制asyncio.Semaphore(5)限制并发请求数避免触发第三方限流429 指数退避重试wait 1 * (2 ** attempt)即 1s/2s/4s最多重试 3 次超过则抛OrgSyncFetchError成员状态识别飞书返回的status.is_frozen/status.is_resigned映射为disabled否则active。T-07GenericAPIProvider 完整实现新建src/backend/bisheng/org_sync/domain/providers/generic_api.py面向无法直接对接飞书/企微/钉钉的长尾数据源。其核心是可配置端点 字段映射{ departments_url: https://api.example.com/departments, members_url: https://api.example.com/members, api_key: sk-xxx, param_location: header, field_mapping: { dept_id: id, dept_name: name, dept_parent_id: parentId, member_id: employeeId, member_name: fullName, member_email: email, member_phone: mobile, member_primary_dept: mainDepartment, member_secondary_depts: otherDepartments, member_status: status } }标准响应格式为{departments: [...]}与{members: [...]}通过field_mapping映射到标准 DTO若返回不符合标准格式Provider 抛OrgSyncFetchError并附带原始响应摘要。认证按auth_type区分api_key 模式把 key 放入 header/query 参数password 模式走基本认证。T-08WeComProvider DingTalkProvider stub新建wecom.py/dingtalk.py两个 Provider 继承OrgSyncProvider4 个方法均抛出OrgSyncProviderError(msgWeChat Work provider not implemented)类文档注释中写明 API 契约留给后续实现者。这是 AD-08 决策的落地飞书覆盖国内主流场景、通用 API 覆盖长尾需求企微/钉钉留 stub 降低首版交付范围。五、Phase 3Sync Engine——差异调和与编排T-09/T-10Reconciler 差异引擎纯逻辑、无 IO新建src/backend/bisheng/org_sync/domain/services/reconciler.py。这是整个同步引擎最核心的部分纯逻辑、无 IO 依赖输入数据输出操作列表因此可以完全脱离数据库做单元测试。部门调和reconcile_departments(remote_depts, local_depts, source) → list[DeptOperation]操作类型为CreateDept | UpdateDept | MoveDept | ArchiveDept判定逻辑创建remote 有而 local 无 → CreateDept更新两边都有且 name 不同 → UpdateDept移动两边都有且 parent_external_id 对应父节点变化 → MoveDept归档local 有source 匹配而 remote 无 → ArchiveDept含子树级联归档本地冲突local Department.sourcelocal 但 external_id 匹配远程 → 强制覆盖 name 并把 source 改为第三方AC-18拓扑排序创建操作按「父先于子」排序Kahn 算法归档按「子先于父」排序按 path 深度倒序若检测到循环引用跳过受影响节点。人员调和reconcile_members(...) → list[MemberOperation]操作类型为CreateMember | UpdateMember | TransferMember | DisableMember | ReactivateMember判定逻辑创建remote 有而 local 无且 statusactive → CreateMember更新name/email/phone 变化 → UpdateMember转岗主部门变化 → TransferMember带old_primary_dept_idtouch_primary区分是否真的动了主部门附属部门增减 → 通过add_secondary_external_ids/remove_secondary_dept_ids表达禁用local 有而 remote 无或 remote.statusdisabled → DisableMember携带待清理的 dept_ids重新激活local.delete1 且 remote.statusactive → ReactivateMember本地冲突User.sourcelocal 但 external_id 匹配远程 → 强制覆盖并改 source。源码中对两种冲突场景的处理reconciler.py部门的 local_by_ext 只收录有 external_id 的部门人员的 local_by_ext 在 source 冲突时优先生效第三方来源记录。T-11OrgSyncService 同步编排器新建src/backend/bisheng/org_sync/domain/services/org_sync_service.pyexecute_sync(config_id, trigger_type, trigger_user)按 spec §7 的 16 步主流程执行1. 加载 OrgSyncConfig → 解密 auth_config 2. 获取互斥锁 a. 检查 config.sync_status idle b. 原子 UPDATE sync_statusrunning WHERE sync_statusidleaffected0 则已被占用 c. 获取 Redis 锁 bisheng:lock:org_sync:{config_id}TTL30min 3. 创建 OrgSyncLog(statusrunning, start_timenow) 4. 实例化 Provider 5. provider.authenticate() 6. 拉取远程部门: fetch_departments(sync_scope) 7. 加载本地部门: Department where sourceprovider and tenant_id 8. Reconciler.reconcile_departments() → dept_ops 9. 执行 dept_ops_apply_dept_ops 10. 拉取远程人员: fetch_members() 11. 加载本地用户: User where sourceprovider and tenant_id经 UserTenant 12. Reconciler.reconcile_members() → member_ops 13. 执行 member_ops_apply_member_ops 14. 更新 OrgSyncLog(status, 统计, end_time) 15. 更新 OrgSyncConfig(last_sync_at, last_sync_result) 16. 释放互斥锁(sync_statusidle, Redis unlock)关键实现决策双重并发保护AD-04DB 的sync_status原子 CAS 保证持久性Redis 分布式锁保证进程崩溃后 30 分钟 TTL 自动释放部分失败处理AD-10逐条 try/except失败项累计进error_details最终日志状态置为partial不中断整体同步新用户密码AD-07secrets.token_hex(32)生成 64 位随机哈希用户无法猜测密码只能通过 SSO 登录绕过权限检查AD-11同步是系统级操作直接操作 DAO DepartmentChangeHandler绕过 DepartmentService 的_check_permission()与source_readonly检查同时通过 ChangeHandler 保证 OpenFGA 元组一致INV-4。_apply_dept_ops对四类操作分别处理CreateDept 创建 Department 后触发on_createdUpdateDept 更新 nameMoveDept 更新 parent_id path 批量替换后触发on_movedArchiveDept 置 archived 并清空成员保留 admin后触发on_archived。_apply_member_ops中 DisableMember 还会清理 Redis 登录态。六、Phase 4Infrastructure API——异步调度与接口层T-12Celery 任务 Beat 定时调度新建src/backend/bisheng/worker/org_sync/tasks.py包含两个任务bisheng_celery.task(acks_lateTrue, time_limit1800, soft_time_limit1500) def execute_org_sync(config_id, trigger_type, trigger_userNone): Async sync execution. Tenant context via INV-8. ... bisheng_celery.task(acks_lateTrue) def check_org_sync_schedules(): Beat task: check cron configs and dispatch if due. Runs every 60s. ...调度设计要点AD-10Beat 检查任务动态调度而非静态配置check_org_sync_schedules作为 Beat 任务每 60 秒执行一次通过 croniter 解析各活跃配置的cron_expression匹配时间则 dispatchexecute_org_sync生成trigger_typescheduled的日志任务路由bisheng.worker.org_sync.*: {queue: knowledge_celery}AD-03同步频率低复用 knowledge_celery 队列无需独立队列time_limit1800, soft_time_limit1500约束单次同步最长 30 分钟多租户上下文传递INV-8发送任务时 tenant_id 通过 Celery headers 注入inject_tenant_headersignalWorker 端before_tasksignal 调用set_current_tenant_id()恢复到 ContextVar被禁用statusdisabled的 cron 配置不会触发AC-32。T-13请求/响应 DTO新建src/backend/bisheng/org_sync/domain/schemas/org_sync_schema.pyOrgSyncConfigCreateprovider, config_name, auth_type, auth_config(dict), sync_scope, schedule_type, cron_expressionOrgSyncConfigUpdateauth_type, auth_config, sync_scope, schedule_type, cron_expression, status全部 Optional支持部分更新OrgSyncConfigRead完整字段auth_config 为脱敏后的 dictOrgSyncLogRead完整字段RemoteTreeNodeexternal_id, name, children递归结构用于远程树预览mask_sensitive_fields(auth_config)把 app_secret/api_key/password 等敏感字段替换为****AC-34。T-14配置 CRUD API5 个端点新建src/backend/bisheng/org_sync/api/endpoints/sync_config.py路由聚合于org_sync/api/router.py并注册进全局src/backend/bisheng/api/router.py端点说明POST /api/v1/org-sync/configs创建auth_config 加密入库GET /api/v1/org-sync/configs配置列表当前租户脱敏GET /api/v1/org-sync/configs/{id}配置详情脱敏PUT /api/v1/org-sync/configs/{id}更新auth_config 合并更新解密 → 合并 → 重新加密DELETE /api/v1/org-sync/configs/{id}软删除statusdeleted同步中不可删返回 22003源码中的关键校验见 sync_config.py_get_config_or_404同时校验tenant_id匹配与status ! deleted跨租户访问返回 22000AC-08创建时的唯一约束冲突Duplicate entry/uk_tenant_provider_name映射为 22001。所有端点通过UserPayload.get_admin_user依赖注入强制管理员权限非管理员返回 22005AC-33响应统一走UnifiedResponseModelresp_200。T-15执行/测试/历史/远程树 API4 个端点新建src/backend/bisheng/org_sync/api/endpoints/sync_exec.py端点说明POST /api/v1/org-sync/configs/{id}/test测试连接返回connected/org_name/total_depts/total_membersPOST /api/v1/org-sync/configs/{id}/execute手动触发检查 sync_status → dispatch Celery 任务 → 返回 log_id异步GET /api/v1/org-sync/configs/{id}/logs?page1limit20同步历史PageData 分页GET /api/v1/org-sync/configs/{id}/remote-tree远程组织树预览嵌套结构约束要点execute 在配置sync_statusrunning时返回 22003、statusdisabled时返回 22009remote-tree 同步调用 Provider 可能耗时 5-10 秒大组织前端建议加 loading 态Celery Worker 未启动时任务进入队列等待API 立即返回 log_id异步语义。七、Phase 5Testing——单元、集成与 E2ET-16Reconciler 单元测试新建src/backend/test/org_sync/test_org_sync_reconciler.py源码中位于 test_org_sync_reconciler.py纯逻辑测试覆盖部门create / 第三方改名 / 本地部门改名 source 变更 / move / archive / 归档级联含本地子部门/ 拓扑序父先于子/ 循环引用跳过人员create / update / transfer主部门变更/ 附属部门增减 / disable离职/ reactivate重新出现/ 本地冲突强制覆盖。测试通过MagicMock构造 Department/User/UserDepartment 伪对象验证 Reconciler 产出的操作类型与字段全程无 IO覆盖 AC-16 ~ AC-28。T-17API 集成测试 E2E 测试API 集成测试test_org_sync_api.pyTestClient配置 CRUD happy path error path包括重复创建 22001、跨租户拒绝 22008、测试连接成功/认证失败、execute 已运行/已禁用、日志分页、远程树、非管理员 40322005等E2E 测试test_e2e_org_sync.pymock Provider 真实 DBtest_full_sync_flow配置 → 触发 → 验证 Department/User/UserDepartment/OrgSyncLog 状态test_incremental_sync首次同步后修改远程数据 → 二次同步 → 验证增量变更test_member_disable_on_departure员工离职 → 验证禁用 清理test_multi_tenant_isolation两个租户各自同步互不影响。八、任务依赖图与并行策略tasks.md 给出了完整的依赖图核心链路是T-01 (ORM) ──┬── T-02 (迁移) ├── T-05 (Provider ABC) ── T-06 (飞书) / T-07 (通用API) / T-08 (stub) T-03 (错误码)─┤ T-04 (User扩展)┤ ├── T-09 (部门Reconciler) ──┐ ├── T-10 (人员Reconciler) ──┤ │ v └── T-11 (OrgSyncService) ─┘ ├── T-12 (Celery) T-13 (DTO)┤ v T-14 (配置CRUD API) ── T-15 (执行API) ── T-17 (APIE2E) ^ T-16 (Reconciler测试) ────────┘据此划分的 8 个并行 WaveWave 1可并行T-01 T-03 T-04 T-05Wave 2T-01 完成后T-02Wave 3T-05 完成后可并行T-06 T-07 T-08Wave 4T-01/T-04/T-09/T-10 完成后T-11Wave 5T-11 完成后可并行T-12 T-13Wave 6T-13 完成后T-14Wave 7T-14 完成后T-15Wave 8T-15 完成后可并行T-16 T-17。这种「先 Foundation 打底、再 Provider 并行、引擎最后收敛、API 串行上垒、测试收尾」的编排最大化并行度的同时保证了每个任务输入就绪是可复用的多任务特性开发节奏。九、边界情况与质量约束从 spec.md 与 tasks.md 中沉淀出的关键边界处理也是实现验收时的重要依据飞书 API 分页自动循环、429 指数退避重试最多 3 次1s/2s/4s部分失败不中断整体同步OrgSyncLog.statuspartialerror_details记录entity_type/external_id/error_msg进程崩溃后 Redis 锁 TTL30min自动释放sync_status可通过手动 API 或下次同步前健康检查重置远程部门树循环引用拓扑排序检测到环后记录错误并跳过受影响子树同名部门在同一父级下冲突捕获DepartmentNameDuplicateError记入 error_details不中断User.external_id唯一约束冲突时走更新而非创建sync_scope为空/null 时同步全部远程部门与人员multi_tenant.enabledfalse时 tenant_id 自动填充默认租户id1不支持双向同步、实时 Webhook、OAuth 回调认证OAuth 字段保留 coming_soon 标记。非功能要求上同步 1000 部门 10000 人员应在 5 分钟内完成Provider 并发 Semaphore5本地数据一次性批量加载到内存 map 避免 N1OpenFGA 双写失败记入 FailedTuple 补偿表INV-4所有端点要求管理员权限tenant_id 自动过滤防止跨租户泄漏INV-1。十、从任务到源码阅读路线图如果希望在仓库中按任务的顺序阅读实现推荐路线数据层org_sync.pyORM DAO Fernet 加解密→ errcode/org_sync.py220xx 错误码Provider 层base.pyABC 工厂→ feishu.py完整实现含 BFS 遍历、Semaphore、429 退避→ generic_api.py / wecom.py / dingtalk.py引擎层reconciler.py纯逻辑差异引擎 拓扑排序API 层sync_config.py配置 CRUD→ sync_exec.py执行/测试/历史/远程树测试层test_org_sync_reconciler.py纯逻辑单测→ test_org_sync_api.pyAPI 集成→ e2e/test_e2e_org_sync.py全链路。F009 以 17 个任务覆盖了从建表、加密、并发控制、差异调和、异步调度到接口暴露的完整闭环其「任务 ↔ AC 验收标准 ↔ 测试用例」的三向可追溯结构以及 Provider 抽象带来的低成本扩展能力是阅读 BiSheng 后端 DDD 模块化实践的良好入口。【免费下载链接】bishengBISHENG is an open LLM devops platform for next generation Enterprise AI applications. Powerful and comprehensive features include: GenAI workflow, RAG, Agent, Unified model management, Evaluation, SFT, Dataset Management, Enterprise-level System Management, Observability and more.项目地址: https://gitcode.com/GitHub_Trending/bi/bisheng创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考