akds/mjava-akdsbeisen/src/main/java/com/malk/service/personnel/impl/PersonnelSyncServiceImpl.java

package com.malk.service.personnel.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.common.util.concurrent.RateLimiter;
import com.malk.server.aliwork.YDConf;
import com.malk.server.aliwork.YDParam;
import com.malk.server.dingtalk.DDR_New;
import com.malk.server.personnel.PersonnelSyncConf;
import com.malk.service.aliwork.YDClient;
import com.malk.service.dingtalk.DDClient;
import com.malk.service.dingtalk.DDClient_Contacts;
import com.malk.service.personnel.PersonnelSyncService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

import java.time.LocalDate;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Service
public class PersonnelSyncServiceImpl implements PersonnelSyncService {

    @Autowired
    private YDClient ydClient;

    @Autowired
    private DDClient ddClient;

    @Autowired
    private DDClient_Contacts ddClient_contacts;

    @Autowired
    private PersonnelSyncConf conf;

    /** 钉钉 user/get 限速器 (官方 60 QPS, 默认留 10 余量) */
    private RateLimiter ddRateLimiter;
    /** 宜搭写接口限速器 (防止并发线程合计超限) */
    private RateLimiter yidaRateLimiter;

    @PostConstruct
    private void initRateLimiters() {
        ddRateLimiter = RateLimiter.create(DD_API_QPS);
        yidaRateLimiter = RateLimiter.create(YIDA_API_QPS);
    }

    private static final String ACTION_CREATE = "CREATE";
    private static final String ACTION_UPDATE = "UPDATE";
    private static final String ACTION_MARK_OFF = "MARK_OFF";

    // 钉钉官方 QPS 上限 60, 留 10 余量; 宜搭写接口 10 并发需整体压低
    private static final double DD_API_QPS = 50.0;
    private static final double YIDA_API_QPS = 30.0;

    private static final ZoneId CST = ZoneId.of("Asia/Shanghai");

    /** 当天 CST 00:00 epoch 毫秒 (离职时间按天存储, 与应报工时日期格式对齐) */
    private static long todayCstStartMillis() {
        return LocalDate.now(CST).atStartOfDay(CST).toInstant().toEpochMilli();
    }

    @Override
    public Map<String, Object> fullSync(Integer limitOverride) {
        long start = System.currentTimeMillis();
        int limit = effectiveLimit(limitOverride);
        boolean limited = limit > 0;
        log.info("[PersonnelSync] 全量同步开始 limit={}", limited ? limit : "none");

        List<Map> dingUsers = fetchAllDingUsers();
        Map<String, Map> dingUserMap = applyLimit(indexByUserid(dingUsers), limit);
        enrichManagers(dingUserMap);
        log.info("[PersonnelSync] 钉钉拉取 {} 人 (去重后{})", dingUserMap.size(), limited ? ", 已截断前" + limit + "条" : "");

        Map<String, YidaRecord> yidaMap = fetchAllYidaPersonnel();
        log.info("[PersonnelSync] 宜搭人员档案 {} 条", yidaMap.size());

        if (limited) log.info("[PersonnelSync] limit 模式: 本轮跳过 MARK_OFF (拉取非全量)");
        List<Action> actions = diff(dingUserMap, yidaMap, limited);
        Map<String, Long> actionStats = countActions(actions);
        log.info("[PersonnelSync] diff 完成: create={}, update={}, markOff={}, skip={}",
                actionStats.getOrDefault(ACTION_CREATE, 0L),
                actionStats.getOrDefault(ACTION_UPDATE, 0L),
                actionStats.getOrDefault(ACTION_MARK_OFF, 0L),
                (long) dingUserMap.size() + yidaMap.size() - actions.size());

        WriteStats writeStats = concurrentWrite(actions);

        long cost = System.currentTimeMillis() - start;
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("fetched", dingUserMap.size());
        result.put("yidaExisting", yidaMap.size());
        result.put("created", writeStats.created.get());
        result.put("updated", writeStats.updated.get());
        result.put("markedInactive", writeStats.markedInactive.get());
        result.put("failed", writeStats.failed.get());
        result.put("durationMs", cost);
        log.info("[PersonnelSync] 全量同步完成 {}", result);
        return result;
    }

    @Override
    public Map<String, Object> dryRun(Integer limitOverride) {
        long start = System.currentTimeMillis();
        int limit = effectiveLimit(limitOverride);
        boolean limited = limit > 0;
        List<Map> dingUsers = fetchAllDingUsers();
        Map<String, Map> dingUserMap = applyLimit(indexByUserid(dingUsers), limit);
        enrichManagers(dingUserMap);
        Map<String, YidaRecord> yidaMap = fetchAllYidaPersonnel();
        List<Action> actions = diff(dingUserMap, yidaMap, limited);

        Map<String, Long> stats = countActions(actions);
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("fetched", dingUserMap.size());
        result.put("yidaExisting", yidaMap.size());
        Map<String, Object> actionCounts = new LinkedHashMap<>();
        actionCounts.put("create", stats.getOrDefault(ACTION_CREATE, 0L));
        actionCounts.put("update", stats.getOrDefault(ACTION_UPDATE, 0L));
        actionCounts.put("markOff", stats.getOrDefault(ACTION_MARK_OFF, 0L));
        result.put("actions", actionCounts);
        result.put("durationMs", System.currentTimeMillis() - start);
        // 抽样 10 条展示预期动作
        List<Map<String, Object>> sample = new ArrayList<>();
        for (int i = 0; i < Math.min(10, actions.size()); i++) {
            Action a = actions.get(i);
            Map<String, Object> s = new LinkedHashMap<>();
            s.put("action", a.type);
            s.put("userid", a.userid);
            s.put("formData", a.formData);
            sample.add(s);
        }
        result.put("sample", sample);
        return result;
    }

    @Override
    public Map<String, Object> probeDingtalkUsers(int sampleSize) {
        long start = System.currentTimeMillis();
        String token = ddClient.getAccessToken();
        List<Long> deptIds = ddClient_contacts.getDepartmentId_all(token, true);
        List<Map> users = ddClient_contacts.getAllUserDetails(token, true);
        Map<String, Map> byUserid = indexByUserid(users);

        Map<String, Object> result = new LinkedHashMap<>();
        result.put("total", byUserid.size());
        result.put("deptCount", deptIds.size());
        result.put("durationMs", System.currentTimeMillis() - start);

        List<Map> sample = new ArrayList<>();
        int n = Math.min(sampleSize <= 0 ? 3 : sampleSize, users.size());
        for (int i = 0; i < n; i++) sample.add(users.get(i));
        result.put("sample", sample);
        return result;
    }

    @Override
    public List<Map> fetchAllDingUsers() {
        String token = ddClient.getAccessToken();
        List<Map> users = new ArrayList<>(ddClient_contacts.getAllUserDetails(token, true));
        // fixme 外部部门(如 1066052389)不在组织树根部门 1 的递归子树内,getAllUserDetails 拉不到 → 显式补抓
        List<Long> extDepts = conf.getExternalDeptIds();
        if (extDepts != null && !extDepts.isEmpty()) {
            Set<String> seen = new HashSet<>();
            for (Map u : users) {
                Object uid = u.get("userid");
                if (uid != null) seen.add(String.valueOf(uid));
            }
            for (Long dept : extDepts) {
                if (dept == null) continue;
                try {
                    List<Map> ext = ddClient_contacts.listDepartmentUserDetail_all(token, dept);
                    int added = 0;
                    for (Map u : ext) {
                        Object uid = u.get("userid");
                        if (uid != null && seen.add(String.valueOf(uid))) {
                            users.add(u);
                            added++;
                        }
                    }
                    log.info("[PersonnelSync] 补抓外部部门 dept={} 新增 {} 人 (该部门共 {})", dept, added, ext.size());
                } catch (Exception ex) {
                    log.warn("[PersonnelSync] 抓外部部门失败 dept={} err={}", dept, ex.getMessage());
                }
            }
        }
        return users;
    }

    @Override
    public Map probeSingleUser(String userid) {
        String token = ddClient.getAccessToken();
        return ddClient_contacts.getUserInfoById(token, userid);
    }

    @SuppressWarnings("unchecked")
    public Map<String, Object> probeDiff(String userid) {
        Map<String, Object> r = new LinkedHashMap<>();
        // 钉钉
        String token = ddClient.getAccessToken();
        Map ding = ddClient_contacts.getUserInfoById(token, userid);
        r.put("dingUser", ding);
        // 宜搭
        DDR_New result = ydClient.queryData(YDParam.builder()
                .appType(conf.getYidaAppType())
                .systemToken(conf.getYidaSystemToken())
                .formUuid(conf.getFormUuidPersonnel())
                .searchFieldJson("{\"" + conf.getFieldEmployee() + "\":\"" + userid + "\"}")
                .build(), YDConf.FORM_QUERY.retrieve_search_form);
        List<Map> list = (List<Map>) result.getData();
        Map<String, Object> yidaFormData = (list != null && !list.isEmpty()) ? (Map<String, Object>) list.get(0).get("formData") : null;
        r.put("yidaFormData", yidaFormData);
        // 新 formData (在职判定改为存在性, probe 统一用 UPDATE)
        Map<String, Object> newData = toYidaFormData(userid, ding, ACTION_UPDATE);
        r.put("newFormData", newData);
        // 字段对比
        Map<String, Object> diff = new LinkedHashMap<>();
        if (yidaFormData != null) {
            for (Map.Entry<String, Object> e : newData.entrySet()) {
                String f = e.getKey();
                Object nv = e.getValue();
                Object ov = yidaFormData.get(f);
                Object ovId = yidaFormData.get(f + "_id");
                Map<String, Object> d = new LinkedHashMap<>();
                d.put("new", nv);
                d.put("newClass", nv == null ? null : nv.getClass().getSimpleName());
                d.put("old", ov);
                d.put("oldClass", ov == null ? null : ov.getClass().getSimpleName());
                d.put("oldId", ovId);
                d.put("oldIdClass", ovId == null ? null : ovId.getClass().getSimpleName());
                diff.put(f, d);
            }
        }
        r.put("diff", diff);
        return r;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Map<String, Object> probeYidaByUserid(String userid) {
        Map<String, Object> r = new LinkedHashMap<>();
        r.put("userid", userid);
        DDR_New result = ydClient.queryData(YDParam.builder()
                .appType(conf.getYidaAppType())
                .systemToken(conf.getYidaSystemToken())
                .formUuid(conf.getFormUuidPersonnel())
                .searchFieldJson("{\"" + conf.getFieldEmployee() + "\":\"" + userid + "\"}")
                .pageSize(YDConf.PAGE_SIZE_LIMIT)
                .build(), YDConf.FORM_QUERY.retrieve_search_form);
        List<Map> list = (List<Map>) result.getData();
        r.put("count", list == null ? 0 : list.size());
        List<Map<String, Object>> records = new ArrayList<>();
        if (list != null) {
            for (Map item : list) {
                Map<String, Object> rec = new LinkedHashMap<>();
                rec.put("instanceId", item.get("formInstanceId"));
                rec.put("createTime", item.get("gmtCreate"));
                rec.put("modifyTime", item.get("gmtModified"));
                rec.put("creator", item.get("creator"));
                rec.put("formData", item.get("formData"));
                records.add(rec);
            }
        }
        r.put("records", records);
        return r;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Map<String, Object> probeYidaDuplicates() {
        long start = System.currentTimeMillis();
        Map<String, List<Map<String, Object>>> byUserid = new LinkedHashMap<>();
        List<Map<String, Object>> emptyEmployee = new ArrayList<>();
        int currentPage = 1;
        long totalCount;
        int total = 0;
        do {
            DDR_New result = ydClient.queryData(YDParam.builder()
                    .appType(conf.getYidaAppType())
                    .systemToken(conf.getYidaSystemToken())
                    .formUuid(conf.getFormUuidPersonnel())
                    .currentPage(currentPage)
                    .pageSize(YDConf.PAGE_SIZE_LIMIT)
                    .build(), YDConf.FORM_QUERY.retrieve_search_form);
            totalCount = result.getTotalCount();
            List<Map> dataList = (List<Map>) result.getData();
            if (dataList == null || dataList.isEmpty()) break;
            for (Map item : dataList) {
                total++;
                Map<String, Object> formData = (Map<String, Object>) item.get("formData");
                String uid = formData == null ? null : extractEmployeeId(formData, conf.getFieldEmployee());
                Map<String, Object> brief = new LinkedHashMap<>();
                brief.put("instanceId", item.get("formInstanceId"));
                brief.put("createTime", item.get("gmtCreate"));
                brief.put("modifyTime", item.get("gmtModified"));
                if (formData != null) {
                    brief.put("name", formData.get(conf.getFieldName()));
                    brief.put("status", formData.get(conf.getFieldStatus()));
                    brief.put("userType", formData.get(conf.getFieldUserType()));
                }
                if (uid == null || uid.isEmpty()) {
                    emptyEmployee.add(brief);
                } else {
                    byUserid.computeIfAbsent(uid, k -> new ArrayList<>()).add(brief);
                }
            }
            currentPage++;
        } while ((long) (currentPage - 1) * YDConf.PAGE_SIZE_LIMIT < totalCount);

        List<Map<String, Object>> dups = new ArrayList<>();
        for (Map.Entry<String, List<Map<String, Object>>> e : byUserid.entrySet()) {
            if (e.getValue().size() > 1) {
                Map<String, Object> g = new LinkedHashMap<>();
                g.put("userid", e.getKey());
                g.put("count", e.getValue().size());
                g.put("records", e.getValue());
                dups.add(g);
            }
        }
        Map<String, Object> r = new LinkedHashMap<>();
        r.put("totalRecords", total);
        r.put("uniqueUseridCount", byUserid.size());
        r.put("emptyEmployeeCount", emptyEmployee.size());
        r.put("duplicateGroupCount", dups.size());
        r.put("duplicateGroups", dups);
        r.put("emptyEmployeeRecords", emptyEmployee);
        r.put("durationMs", System.currentTimeMillis() - start);
        log.info("[PersonnelSync] probeYidaDuplicates total={} unique={} empty={} duplicateGroups={}",
                total, byUserid.size(), emptyEmployee.size(), dups.size());
        return r;
    }

    /**
     * 一次性清理重复条(本期任务):
     * - jingzhao: 直接删后建条 FHC66571O325JF0LK0EB54NKPOZV3YWZQ98OMOG7
     * - 626967876(吴超): 把同步条 4XC66W81HA43...的钉钉字段 merge 到业务条 XRD66E7127A5... 再删同步条
     * 钉钉字段以同步条为准:属性/员工编号/归属公司/Manager/是否CF/成本中心/入职时间/在职状态
     */
    @Override
    public Map<String, Object> cleanupKnownDuplicatesOnce(boolean dryRun) {
        long start = System.currentTimeMillis();
        Map<String, Object> r = new LinkedHashMap<>();
        r.put("dryRun", dryRun);
        List<Map<String, Object>> actions = new ArrayList<>();

        // ===== Group A: jingzhao =====
        String jingzhaoDeleteId = "FINST-FHC66571O325JF0LK0EB54NKPOZV3YWZQ98OMOG7";
        Map<String, Object> aDel = new LinkedHashMap<>();
        aDel.put("type", "DELETE");
        aDel.put("userid", "jingzhao");
        aDel.put("instanceId", jingzhaoDeleteId);
        aDel.put("reason", "保留早建条 4XC66W81C9Z40Q...; 删除后建空字段重复条");
        if (!dryRun) {
            try {
                ydClient.operateData(YDParam.builder()
                        .appType(conf.getYidaAppType())
                        .systemToken(conf.getYidaSystemToken())
                        .formUuid(conf.getFormUuidPersonnel())
                        .formInstanceId(jingzhaoDeleteId)
                        .build(), YDConf.FORM_OPERATION.delete);
                aDel.put("ok", true);
            } catch (Exception ex) {
                aDel.put("ok", false);
                aDel.put("err", ex.getMessage());
            }
        }
        actions.add(aDel);

        // ===== Group B: 626967876 (吴超) merge then delete =====
        String wuchaoKeepId   = "FINST-XRD66E7127A51G39GBQX1AG83I8525SWNWKOMCZB";
        String wuchaoDeleteId = "FINST-4XC66W81HA43S9G1H3ELL664BSEN23DTEE4NMS703";

        Map<String, Object> mergeFields = new LinkedHashMap<>();
        // 员工编号 textField_mh8xhqc1 = userid 字符串(同步代码 v1.2 后规范)
        mergeFields.put(conf.getFieldJobNumber(), "626967876");
        // 属性: 内部 (钉钉 dept_id=[1057430958] 内部部门, 同步条已写"内部")
        mergeFields.put(conf.getFieldUserType(), conf.getExtAttrValueInternal());
        // 归属公司: 上海 (同步条值)
        mergeFields.put(conf.getFieldCompany(), "上海");
        // Manager: Kevin Xu (640891109) 数组格式
        mergeFields.put(conf.getFieldManager(), Collections.singletonList("640891109"));
        // 是否CF: false (同步条值)
        mergeFields.put(conf.getFieldIsCf(), "false");
        // 成本中心: 35 (同步条值)
        mergeFields.put(conf.getFieldCostCenter(), "35");
        // 入职时间: 1716134400000 (同步条值)
        mergeFields.put(conf.getFieldHiredDate(), 1716134400000L);
        // 在职状态: 在职
        mergeFields.put(conf.getFieldStatus(), conf.getStatusValueActive());
        // 部门: 1057430958 (钉钉同步主部门)
        mergeFields.put(conf.getFieldDepartment(), Collections.singletonList("1057430958"));

        Map<String, Object> bUpd = new LinkedHashMap<>();
        bUpd.put("type", "UPDATE_MERGE");
        bUpd.put("userid", "626967876");
        bUpd.put("instanceId", wuchaoKeepId);
        bUpd.put("mergeFields", mergeFields);
        bUpd.put("note", "钉钉字段贴到业务条,业务字段(客户类型/合同号/numberField/北森编号)保留不动");
        if (!dryRun) {
            try {
                ydClient.operateData(YDParam.builder()
                        .appType(conf.getYidaAppType())
                        .systemToken(conf.getYidaSystemToken())
                        .formUuid(conf.getFormUuidPersonnel())
                        .formInstanceId(wuchaoKeepId)
                        .updateFormDataJson(JSON.toJSONString(mergeFields))
                        .ignoreEmpty(false)
                        .useLatestVersion(true)
                        .build(), YDConf.FORM_OPERATION.update);
                bUpd.put("ok", true);
            } catch (Exception ex) {
                bUpd.put("ok", false);
                bUpd.put("err", ex.getMessage());
            }
        }
        actions.add(bUpd);

        Map<String, Object> bDel = new LinkedHashMap<>();
        bDel.put("type", "DELETE");
        bDel.put("userid", "626967876");
        bDel.put("instanceId", wuchaoDeleteId);
        bDel.put("reason", "同步条信息已 merge 到业务条 XRD66E7127A5...");
        if (!dryRun) {
            try {
                ydClient.operateData(YDParam.builder()
                        .appType(conf.getYidaAppType())
                        .systemToken(conf.getYidaSystemToken())
                        .formUuid(conf.getFormUuidPersonnel())
                        .formInstanceId(wuchaoDeleteId)
                        .build(), YDConf.FORM_OPERATION.delete);
                bDel.put("ok", true);
            } catch (Exception ex) {
                bDel.put("ok", false);
                bDel.put("err", ex.getMessage());
            }
        }
        actions.add(bDel);

        r.put("actions", actions);
        r.put("durationMs", System.currentTimeMillis() - start);
        log.info("[PersonnelSync] cleanupKnownDuplicatesOnce dryRun={} result={}", dryRun, r);
        return r;
    }

    @SuppressWarnings("unchecked")
    @Override
    public Map<String, Object> syncSingle(String userid) {
        long start = System.currentTimeMillis();
        Map<String, Object> result = new LinkedHashMap<>();
        result.put("userid", userid);
        // 1. 钉钉取这个人
        String token = ddClient.getAccessToken();
        Map ding = ddClient_contacts.getUserInfoById(token, userid);
        if (ding == null || ding.get("userid") == null) {
            result.put("action", "NOT_FOUND");
            result.put("durationMs", System.currentTimeMillis() - start);
            log.warn("[PersonnelSync] syncSingle 钉钉查无此人 userid={}", userid);
            return result;
        }
        // 2. 宜搭按 userid 查现有记录
        DDR_New q = ydClient.queryData(YDParam.builder()
                .appType(conf.getYidaAppType())
                .systemToken(conf.getYidaSystemToken())
                .formUuid(conf.getFormUuidPersonnel())
                .searchFieldJson("{\"" + conf.getFieldEmployee() + "\":\"" + userid + "\"}")
                .build(), YDConf.FORM_QUERY.retrieve_search_form);
        List<Map> list = (List<Map>) q.getData();
        String instanceId = null;
        Map<String, Object> existFormData = null;
        if (list != null && !list.isEmpty()) {
            instanceId = String.valueOf(list.get(0).get("formInstanceId"));
            existFormData = (Map<String, Object>) list.get(0).get("formData");
        }
        // 3. 构造 formData 并选 action
        String action = (instanceId == null) ? ACTION_CREATE : ACTION_UPDATE;
        Map<String, Object> formData = toYidaFormData(userid, ding, action);
        if (ACTION_UPDATE.equals(action) && isSameAsYida(formData, existFormData)) {
            result.put("action", "SKIP");
            result.put("instanceId", instanceId);
            result.put("formData", formData);
            result.put("durationMs", System.currentTimeMillis() - start);
            log.info("[PersonnelSync] syncSingle 幂等跳过 userid={}", userid);
            return result;
        }
        // 4. 写入 (复用 executeAction 的指数退避 + 限速)
        WriteStats stats = new WriteStats();
        executeAction(new Action(action, userid, instanceId, formData), stats);
        result.put("action", action);
        result.put("instanceId", instanceId);
        result.put("formData", formData);
        result.put("created", stats.created.get());
        result.put("updated", stats.updated.get());
        result.put("failed", stats.failed.get());
        result.put("durationMs", System.currentTimeMillis() - start);
        log.info("[PersonnelSync] syncSingle 完成 userid={} action={} result={}", userid, action, result);
        return result;
    }

    @Override
    public Map<String, Object> probeStats() {
        long start = System.currentTimeMillis();
        List<Map> users = fetchAllDingUsers();
        int total = users.size();
        int active = 0, inactive = 0, emptyDept = 0, hasExtattr = 0, emptyJobNumber = 0;
        for (Map u : users) {
            if (isActive(u)) active++; else inactive++;
            Object dept = u.get("dept_id_list");
            if (!(dept instanceof List) || ((List) dept).isEmpty()) emptyDept++;
            Object ext = u.get("extattr");
            if (ext instanceof Map && !((Map) ext).isEmpty()) hasExtattr++;
            Object job = u.get("job_number");
            if (job == null || String.valueOf(job).isEmpty()) emptyJobNumber++;
        }
        Map<String, Object> res = new LinkedHashMap<>();
        res.put("total", total);
        res.put("active", active);
        res.put("inactive", inactive);
        res.put("emptyDeptIdList", emptyDept);
        res.put("hasExtattr", hasExtattr);
        res.put("emptyJobNumber", emptyJobNumber);
        res.put("durationMs", System.currentTimeMillis() - start);
        return res;
    }

    // ==================== 内部: 数据抓取 ====================

    @SuppressWarnings("unchecked")
    private Map<String, YidaRecord> fetchAllYidaPersonnel() {
        Map<String, YidaRecord> yidaMap = new LinkedHashMap<>();
        int currentPage = 1;
        long totalCount;
        do {
            DDR_New result = ydClient.queryData(YDParam.builder()
                    .appType(conf.getYidaAppType())
                    .systemToken(conf.getYidaSystemToken())
                    .formUuid(conf.getFormUuidPersonnel())
                    .currentPage(currentPage)
                    .pageSize(YDConf.PAGE_SIZE_LIMIT)
                    .build(), YDConf.FORM_QUERY.retrieve_search_form);
            totalCount = result.getTotalCount();
            List<Map> dataList = (List<Map>) result.getData();
            if (dataList == null || dataList.isEmpty()) break;

            for (Map item : dataList) {
                Map<String, Object> formData = (Map<String, Object>) item.get("formData");
                if (formData == null) continue;
                String userid = extractEmployeeId(formData, conf.getFieldEmployee());
                if (userid == null || userid.isEmpty()) continue;
                YidaRecord rec = new YidaRecord();
                rec.instanceId = String.valueOf(item.get("formInstanceId"));
                rec.formData = formData;
                // fixme 同 userid 重复条防御: 保留早建条 (按 instanceId 字典序较小者) 并 WARN, 避免静默覆盖
                YidaRecord exist = yidaMap.get(userid);
                if (exist != null) {
                    log.warn("[PersonnelSync] 宜搭检测到重复 userid={} 已扫到 instanceId={} 又扫到 instanceId={}",
                            userid, exist.instanceId, rec.instanceId);
                    if (exist.instanceId != null && rec.instanceId != null
                            && rec.instanceId.compareTo(exist.instanceId) > 0) {
                        continue;   // 已有的更早建,保留它
                    }
                }
                yidaMap.put(userid, rec);
            }
            currentPage++;
        } while ((long) (currentPage - 1) * YDConf.PAGE_SIZE_LIMIT < totalCount);
        return yidaMap;
    }

    // ==================== 内部: 差异计算 ====================

    private List<Action> diff(Map<String, Map> dingUserMap, Map<String, YidaRecord> yidaMap, boolean skipMarkOff) {
        List<Action> actions = new ArrayList<>();

        // 钉钉里查到的 -> 一律按"在职"写 (active 布尔不再参与)
        for (Map.Entry<String, Map> e : dingUserMap.entrySet()) {
            String userid = e.getKey();
            Map ding = e.getValue();
            YidaRecord yida = yidaMap.get(userid);

            if (yida == null) {
                actions.add(new Action(ACTION_CREATE, userid, null, toYidaFormData(userid, ding, ACTION_CREATE)));
            } else {
                Map<String, Object> formData = toYidaFormData(userid, ding, ACTION_UPDATE);
                if (isSameAsYida(formData, yida.formData)) continue;   // 幂等跳过
                actions.add(new Action(ACTION_UPDATE, userid, yida.instanceId, formData));
            }
        }

        // 宜搭里有、钉钉里没有 -> 标记离职 (limit 模式下拉取非全量, 跳过此步)
        // ppExt 离职时间: 首次判定离职 status+offlineDate 一起写; 已"离职"但 offlineDate 空 -> 补写(当天);
        //                已有 offlineDate 且状态一致 -> 完全跳过, 保护首次判定日不被覆盖
        if (!skipMarkOff) {
            long todayMs = todayCstStartMillis();
            boolean writeOfflineDate = notBlank(conf.getFieldOfflineDate());
            for (Map.Entry<String, YidaRecord> e : yidaMap.entrySet()) {
                if (dingUserMap.containsKey(e.getKey())) continue;
                YidaRecord yida = e.getValue();
                Object currentStatus = yida.formData.get(conf.getFieldStatus());
                boolean alreadyInactive = conf.getStatusValueInactive().equals(String.valueOf(currentStatus));
                boolean offlineDateBlank = writeOfflineDate && isBlankValue(yida.formData.get(conf.getFieldOfflineDate()));

                if (alreadyInactive && (!writeOfflineDate || !offlineDateBlank)) continue;   // 全部就绪, 跳过

                Map<String, Object> formData = new LinkedHashMap<>();
                if (!alreadyInactive) {
                    formData.put(conf.getFieldStatus(), conf.getStatusValueInactive());
                }
                if (writeOfflineDate && offlineDateBlank) {
                    formData.put(conf.getFieldOfflineDate(), todayMs);
                }
                if (formData.isEmpty()) continue;   // 防御: 无字段变化不写入
                actions.add(new Action(ACTION_MARK_OFF, e.getKey(), yida.instanceId, formData));
            }
        }
        return actions;
    }

    /** dateField 在 formData 里可能是 Long 时间戳或 "yyyy-MM-dd HH:mm:ss" 字符串, 空/空串/字面 "null" 一律视为空 */
    private boolean isBlankValue(Object v) {
        if (v == null) return true;
        String s = String.valueOf(v).trim();
        return s.isEmpty() || "null".equalsIgnoreCase(s);
    }

    // ==================== 内部: 字段映射 ====================

    @SuppressWarnings("unchecked")
    private Map<String, Object> toYidaFormData(String userid, Map ding, String action) {
        Map<String, Object> formData = new LinkedHashMap<>();

        // 人员 (唯一键, 永远写入)
        formData.put(conf.getFieldEmployee(), Collections.singletonList(userid));

        // 在职状态
        String statusValue = ACTION_MARK_OFF.equals(action) ? conf.getStatusValueInactive() : conf.getStatusValueActive();
        formData.put(conf.getFieldStatus(), statusValue);

        // 离职软标记只更新状态字段, 保留其他原值
        if (ACTION_MARK_OFF.equals(action)) return formData;

        // 员工姓名 <- name (目标表该字段 READONLY, 仍按需求强写覆盖)
        Object name = ding.get("name");
        if (notBlank(conf.getFieldName()) && name != null && notBlank(String.valueOf(name))) {
            formData.put(conf.getFieldName(), String.valueOf(name).trim());
        }

        // 员工编号 <- userid (钉钉用户唯一 ID, 与人员 EmployeeField 同源, 但写入 TextField 便于跨模块按字符串引用)
        if (notBlank(conf.getFieldJobNumber())) {
            formData.put(conf.getFieldJobNumber(), userid);
        }
        // 员工工号 <- job_number (目标表 READONLY, 按需求强写覆盖; 空则跳过)
        Object jobNumber = ding.get("job_number");
        if (notBlank(conf.getFieldJobNumber2()) && jobNumber != null && notBlank(String.valueOf(jobNumber))) {
            formData.put(conf.getFieldJobNumber2(), String.valueOf(jobNumber).trim());
        }

        // 员工部门 <- dept_id_list 稳定排序后取首个 (钉钉返回顺序不固定, 避免 diff 抖动)
        Object deptObj = ding.get("dept_id_list");
        if (deptObj instanceof List && !((List) deptObj).isEmpty()) {
            List<Long> sorted = new ArrayList<>();
            for (Object d : (List) deptObj) {
                if (d != null) sorted.add(((Number) d).longValue());
            }
            Collections.sort(sorted);
            if (!sorted.isEmpty()) {
                formData.put(conf.getFieldDepartment(), Collections.singletonList(String.valueOf(sorted.get(0))));
            }
        }

        // 入职时间 <- hired_date (毫秒时间戳, 需要钉钉花名册权限才返回)
        if (notBlank(conf.getFieldHiredDate())) {
            Object hired = ding.get("hired_date");
            if (hired instanceof Number) {
                formData.put(conf.getFieldHiredDate(), ((Number) hired).longValue());
            }
        }

        // Manager <- manager_userid (EmployeeField, 数组格式)
        Object mgr = ding.get("manager_userid");
        if (notBlank(conf.getFieldManager()) && mgr != null && notBlank(String.valueOf(mgr))) {
            formData.put(conf.getFieldManager(), Collections.singletonList(String.valueOf(mgr).trim()));
        }

        // 北森编号 / 归属公司 / 是否CF / 成本中心 <- 钉钉 extattr 自定义字段
        putExtAttr(formData, ding, conf.getFieldBeisenJobNo(), conf.getExtAttrKeyBeisen());
        putExtAttr(formData, ding, conf.getFieldCompany(), conf.getExtAttrKeyCompany());
        putExtAttr(formData, ding, conf.getFieldIsCf(), conf.getExtAttrKeyIsCf());
        putExtAttr(formData, ding, conf.getFieldCostCenter(), conf.getExtAttrKeyCostCenter());

        // 属性 <- 部门含 externalDeptIds ? 外部 : 内部 (extAttrKeyUserType 留空时走部门白名单)
        String userType = resolveUserType(ding);
        if (userType != null) {
            formData.put(conf.getFieldUserType(), userType);
        }

        return formData;
    }

    /** extattr 自定义字段取值并写入 (fieldId 或 extKey 为空 / 取不到值 则不写, 保留宜搭原值) */
    private void putExtAttr(Map<String, Object> formData, Map ding, String fieldId, String extKey) {
        if (!notBlank(fieldId) || !notBlank(extKey)) return;
        String v = readExtAttr(ding, extKey);
        if (notBlank(v)) formData.put(fieldId, v.trim());
    }

    /**
     * 读钉钉自定义字段[key]:
     * - topapi/v2/user/list & user/get 把自定义字段放在 extension (JSON 字符串 {key: value}) 里
     * - 兼容旧式 extattr (Map, 值可能是纯字符串或 {text, value} 枚举)
     */
    @SuppressWarnings("unchecked")
    private String readExtAttr(Map ding, String key) {
        Object ext = ding.get("extension");
        if (ext instanceof String && !((String) ext).trim().isEmpty()) {
            try {
                Object parsed = JSON.parse((String) ext);
                if (parsed instanceof Map) {
                    Object v = ((Map) parsed).get(key);
                    if (v != null && !String.valueOf(v).trim().isEmpty()) return String.valueOf(v);
                }
            } catch (Exception ignored) {}
        }
        Object extattr = ding.get("extattr");
        if (extattr instanceof Map) {
            Object attr = ((Map) extattr).get(key);
            if (attr instanceof Map) {
                Map am = (Map) attr;
                Object v = am.get("value");
                if (v == null) v = am.get("text");
                return v == null ? null : String.valueOf(v);
            } else if (attr != null && !String.valueOf(attr).trim().isEmpty()) {
                return String.valueOf(attr);
            }
        }
        return null;
    }

    private boolean notBlank(String s) {
        return s != null && !s.trim().isEmpty();
    }

    @SuppressWarnings("unchecked")
    private String resolveUserType(Map ding) {
        // 1. 优先读 extattr
        String key = conf.getExtAttrKeyUserType();
        if (key != null && !key.isEmpty()) {
            Object extattrObj = ding.get("extattr");
            if (extattrObj instanceof Map) {
                Map extattr = (Map) extattrObj;
                Object attr = extattr.get(key);
                String raw = null;
                if (attr instanceof Map) {
                    Map am = (Map) attr;
                    Object value = am.get("value");
                    if (value == null) value = am.get("text");
                    if (value != null) raw = String.valueOf(value);
                } else if (attr != null) {
                    raw = String.valueOf(attr);
                }
                if (raw != null && !raw.isEmpty()) {
                    if (conf.getExtAttrValueInternal().equals(raw)) return conf.getExtAttrValueInternal();
                    if (conf.getExtAttrValueExternal().equals(raw)) return conf.getExtAttrValueExternal();
                }
            }
        }
        // 2. 兜底: 部门白名单 (外部部门列表 + 默认规则)
        List<Long> externalDepts = conf.getExternalDeptIds();
        Object deptObj = ding.get("dept_id_list");
        if (deptObj instanceof List && !((List) deptObj).isEmpty()) {
            if (externalDepts != null && !externalDepts.isEmpty()) {
                boolean isExternal = false;
                for (Object d : (List) deptObj) {
                    if (d == null) continue;
                    long deptId = ((Number) d).longValue();
                    if (externalDepts.contains(deptId)) { isExternal = true; break; }
                }
                return isExternal ? conf.getExtAttrValueExternal() : conf.getExtAttrValueInternal();
            }
            if (conf.isFallbackInternalByDefault()) {
                return conf.getExtAttrValueInternal();
            }
        }
        return null;
    }

    private boolean isActive(Map ding) {
        Object active = ding.get("active");
        if (active instanceof Boolean) return (Boolean) active;
        if (active == null) return true;
        return Boolean.parseBoolean(String.valueOf(active));
    }

    // 比较钉钉侧构造的 formData 与宜搭已有 formData 是否所有字段都相等
    @SuppressWarnings("unchecked")
    private boolean isSameAsYida(Map<String, Object> newData, Map<String, Object> yidaData) {
        for (Map.Entry<String, Object> e : newData.entrySet()) {
            String fieldId = e.getKey();
            Object newVal = e.getValue();
            Object oldVal = yidaData.get(fieldId);
            if (newVal instanceof List) {
                // 成员/部门类字段: 宜搭返回时用 _id 后缀取 id 列表
                Object oldIdList = yidaData.get(fieldId + "_id");
                if (oldIdList != null) oldVal = oldIdList;
                if (!listEquals((List) newVal, oldVal)) return false;
            } else {
                if (!Objects.equals(String.valueOf(newVal), String.valueOf(oldVal))) return false;
            }
        }
        return true;
    }

    @SuppressWarnings("unchecked")
    private boolean listEquals(List newList, Object oldObj) {
        List<String> newStr = new ArrayList<>();
        for (Object o : newList) newStr.add(String.valueOf(o));
        List<String> oldStr = new ArrayList<>();
        if (oldObj instanceof List) {
            for (Object o : (List) oldObj) oldStr.add(String.valueOf(o));
        } else if (oldObj != null) {
            oldStr.add(String.valueOf(oldObj));
        }
        if (newStr.size() != oldStr.size()) return false;
        for (String s : newStr) if (!oldStr.contains(s)) return false;
        return true;
    }

    // ==================== 内部: 写入 ====================

    private WriteStats concurrentWrite(List<Action> actions) {
        WriteStats stats = new WriteStats();
        if (actions.isEmpty()) return stats;

        ExecutorService pool = Executors.newFixedThreadPool(Math.max(1, conf.getConcurrency()));
        List<Future<?>> futures = new ArrayList<>();
        for (Action a : actions) {
            futures.add(pool.submit(() -> executeAction(a, stats)));
        }
        for (Future<?> f : futures) {
            try { f.get(); } catch (Exception ex) { log.warn("[PersonnelSync] future 异常", ex); }
        }
        pool.shutdown();
        try { pool.awaitTermination(10, TimeUnit.MINUTES); } catch (InterruptedException ignored) {}
        return stats;
    }

    private void executeAction(Action a, WriteStats stats) {
        int attempt = 0;
        while (true) {
            // 指数退避: 第 1 次不等, 之后 1s/2s/4s... 最长 8s
            if (attempt > 0) {
                long backoffMs = Math.min(1000L * (1L << (attempt - 1)), 8000L);
                try {
                    Thread.sleep(backoffMs);
                } catch (InterruptedException ie) {
                    Thread.currentThread().interrupt();
                    stats.failed.incrementAndGet();
                    return;
                }
            }
            yidaRateLimiter.acquire();
            try {
                if (ACTION_CREATE.equals(a.type)) {
                    ydClient.operateData(YDParam.builder()
                            .appType(conf.getYidaAppType())
                            .systemToken(conf.getYidaSystemToken())
                            .formUuid(conf.getFormUuidPersonnel())
                            .formDataJson(JSON.toJSONString(a.formData))
                            .build(), YDConf.FORM_OPERATION.create);
                    stats.created.incrementAndGet();
                } else {
                    ydClient.operateData(YDParam.builder()
                            .appType(conf.getYidaAppType())
                            .systemToken(conf.getYidaSystemToken())
                            .formUuid(conf.getFormUuidPersonnel())
                            .formInstanceId(a.instanceId)
                            .updateFormDataJson(JSON.toJSONString(a.formData))
                            .ignoreEmpty(false)
                            .useLatestVersion(true)
                            .build(), YDConf.FORM_OPERATION.update);
                    if (ACTION_MARK_OFF.equals(a.type)) {
                        stats.markedInactive.incrementAndGet();
                    } else {
                        stats.updated.incrementAndGet();
                    }
                }
                return;
            } catch (Exception ex) {
                attempt++;
                if (attempt > conf.getMaxRetry()) {
                    log.warn("[PersonnelSync] 写入失败 userid={} action={} err={}", a.userid, a.type, ex.getMessage());
                    stats.failed.incrementAndGet();
                    return;
                }
                log.info("[PersonnelSync] 写入重试 userid={} action={} attempt={} err={}", a.userid, a.type, attempt, ex.getMessage());
            }
        }
    }

    // ==================== 内部: 工具 ====================

    @SuppressWarnings("unchecked")
    private String extractEmployeeId(Map<String, Object> formData, String fieldId) {
        Object raw = formData.get(fieldId + "_id");
        if (raw == null) raw = formData.get(fieldId);
        if (raw == null) return null;
        if (raw instanceof List) {
            List list = (List) raw;
            return list.isEmpty() ? null : String.valueOf(list.get(0));
        }
        String str = String.valueOf(raw).trim();
        if (str.startsWith("[") && str.endsWith("]")) {
            Object parsed = JSON.parse(str);
            if (parsed instanceof List) {
                List pl = (List) parsed;
                return pl.isEmpty() ? null : String.valueOf(pl.get(0));
            }
        }
        return str.isEmpty() ? null : str;
    }

    private Map<String, Map> indexByUserid(List<Map> users) {
        Map<String, Map> map = new LinkedHashMap<>();
        for (Map u : users) {
            Object uid = u.get("userid");
            if (uid != null) map.putIfAbsent(String.valueOf(uid), u);
        }
        return map;
    }

    /** 解析生效的 limit: 入参优先, 否则用配置 limitFirstN; <=0 表示不限 */
    private int effectiveLimit(Integer override) {
        if (override != null) return Math.max(0, override);
        return Math.max(0, conf.getLimitFirstN());
    }

    /** limit>0 时按 userid 升序取前 N 条 (排序保证幂等可复现) */
    private Map<String, Map> applyLimit(Map<String, Map> byUserid, int limit) {
        if (limit <= 0 || byUserid.size() <= limit) return byUserid;
        List<String> keys = new ArrayList<>(byUserid.keySet());
        Collections.sort(keys);
        Map<String, Map> limited = new LinkedHashMap<>();
        for (int i = 0; i < limit; i++) limited.put(keys.get(i), byUserid.get(keys.get(i)));
        return limited;
    }

    /** topapi/v2/user/list 不返回 manager_userid, 按需逐人补 (仅当 fieldManager 已配置; 单人失败不影响整体) */
    @SuppressWarnings("unchecked")
    private void enrichManagers(Map<String, Map> dingUserMap) {
        if (!notBlank(conf.getFieldManager()) || dingUserMap.isEmpty()) return;
        String token = ddClient.getAccessToken();
        int filled = 0;
        for (Map u : dingUserMap.values()) {
            if (u.get("manager_userid") != null) { filled++; continue; }
            Object uid = u.get("userid");
            if (uid == null) continue;
            try {
                ddRateLimiter.acquire();
                Map detail = ddClient_contacts.getUserInfoById(token, String.valueOf(uid));
                Object mgr = detail == null ? null : detail.get("manager_userid");
                if (mgr != null && notBlank(String.valueOf(mgr))) {
                    u.put("manager_userid", String.valueOf(mgr).trim());
                    filled++;
                }
            } catch (Exception ex) {
                log.warn("[PersonnelSync] 取 manager_userid 失败 userid={} err={}", uid, ex.getMessage());
            }
        }
        log.info("[PersonnelSync] manager_userid 已就绪 {}/{} 人", filled, dingUserMap.size());
    }

    private Map<String, Long> countActions(List<Action> actions) {
        Map<String, Long> stats = new HashMap<>();
        for (Action a : actions) {
            stats.merge(a.type, 1L, Long::sum);
        }
        return stats;
    }

    // ==================== 内部: 数据结构 ====================

    private static class YidaRecord {
        String instanceId;
        Map<String, Object> formData;
    }

    private static class Action {
        final String type;
        final String userid;
        final String instanceId;
        final Map<String, Object> formData;

        Action(String type, String userid, String instanceId, Map<String, Object> formData) {
            this.type = type;
            this.userid = userid;
            this.instanceId = instanceId;
            this.formData = formData;
        }
    }

    private static class WriteStats {
        AtomicInteger created = new AtomicInteger();
        AtomicInteger updated = new AtomicInteger();
        AtomicInteger markedInactive = new AtomicInteger();
        AtomicInteger failed = new AtomicInteger();
    }
}