使用 Snail Job 定时任务同步 Welink 部门和用户信息

本文将以同步 Welink 部门和用户为例，介绍如何使用 Snail Job 创建定时任务并适配 RuoYiPlus 项目。

在上文中实现了 Welink 的绑定与扫码登录功能，但是项目一开始是没有用户的，于是就有了本文，通过定时任务来获取 Welink 部门和用户信息构造系统部门和用户，并进行绑定。

如何使用 Snail Job 创建定时任务并启动可参考：snail-job集群定时任务 | 掘金

Snail Job 文档

编写任务类

CreateDeptAndUserFromWelinkJob.java 中使用 @JobExecutor 注解给任务命名，在后台创建任务时就不需要全限定类名，需要编写 jobExecute 方法来完成实际操作：

@Slf4j
@RequiredArgsConstructor
@Component
@JobExecutor(name = "createDeptAndUserFromWelinkJob")
public class CreateDeptAndUserFromWelinkJob {
    private final SocialProperties socialProperties;
    private final ISysDeptService deptService;
    private final ISysUserService userService;
    private final ISysSocialService socialService;
    private final ISysTenantService tenantService;
    private List<SysDeptBo> deptBoList; // 部门列表
    private final List<SysUserBo> userBoList = new ArrayList<>(); // 用户列表
    private final String SOURCE = "welink";
    private static String accessToken;

    public void getAccessToken(){
        SocialLoginConfigProperties obj = socialProperties.getType().get(SOURCE);
        if (ObjectUtil.isNull(obj)) {
            SnailJobLog.REMOTE.error("not find {} application profile", SOURCE);
            return;
        }
        AuthRequest authRequest = SocialUtils.getAuthRequest(SOURCE, socialProperties);
        AuthToken authToken = authRequest.getAccessToken(new AuthCallback());
        accessToken = authToken.getAccessToken();
    }

    private boolean getDeptList(String deptRootCode) {
        JSONArray allDeptJsonArray = new JSONArray();

        final int pageSize = 50; // 最大 100
        int pageNo = 1;
        boolean hasMore = false;

        getAccessToken();

        if( StringUtils.isEmpty(accessToken)) {
            SnailJobLog.REMOTE.error("{} get access token fail", SOURCE);
            return false;
        }
        do {
            Map<String, String> headers = new HashMap<>();
            headers.put("Content-Type", "application/json;charset=UTF-8");
            headers.put("x-wlk-gray", "0");
            headers.put("x-wlk-Authorization", accessToken);

            Map<String, String> params = new HashMap<>();
            params.put("deptCode", deptRootCode);
            params.put("recursiveflag", "1"); // 递归查询子部门
            params.put("offset", String.valueOf(pageNo));
            params.put("limit", String.valueOf(pageSize));

            try {
                String body = OkHttpUtil.get(UrlBuilder.fromBaseUrl(AuthWelinkSource.WELINK.deptList()).build(), headers, params);
                JSONObject object = JSONObject.parseObject(body);
                log.info("get dept list body: {}", object);
                if(this.checkResponse(object)){
                    hasMore = object.getInteger("totalCount") > pageNo * pageSize;
                    pageNo++;
                    JSONArray deptArray = object.getJSONArray("departmentInfo");
                    allDeptJsonArray.addAll(deptArray);
                }else{
                    SnailJobLog.REMOTE.error("{} get dept list fail, pageNo: {}", SOURCE, pageNo);
                    // 部分出错，如分页查询时失败
                    return false;
                }
            } catch (Exception e) {
                SnailJobLog.REMOTE.error("{} get dept list fail, {}", SOURCE, e.getMessage());
                return false;
            }
        } while (hasMore);

        if(!allDeptJsonArray.isEmpty()){
            List<JSONObject> deptListJson = JSONArray.parseArray(allDeptJsonArray.toJSONString(), JSONObject.class);
            // 将 JSONObject 转换为 SysDeptBo
            deptBoList = deptListJson.stream().map(json -> {
                SysDeptBo bo = new SysDeptBo();
                bo.setDeptId(json.getLong("deptCode"));
                bo.setDeptName(json.getString("deptNameCn"));
                bo.setParentId(json.getLong("fatherCode"));
                bo.setDeptCategory(SOURCE);
                bo.setOrderNum(json.getInteger("orderNo"));
                return bo;
            }).collect(Collectors.toList());

            // 手动创建根部门
            SysDeptBo root = new SysDeptBo();
            root.setDeptId(Convert.toLong(deptRootCode));
            root.setOrderNum(0);
            root.setDeptName("xxxxxx有限公司");
            root.setParentId(0L);
            root.setDeptCategory(SOURCE);
            deptBoList.add(root);
        }
        return true;
    }

    private boolean getUserListByDeptId(String deptId){
        if(StringUtils.isEmpty(accessToken)) {
            SnailJobLog.REMOTE.error("{} get access token fail", SOURCE);
            return false;
        }
        JSONArray userJsonArray = new JSONArray();
        final int pageSize = 50; // 最大 50
        int pageNo = 1;
        boolean hasMore = false;
        do {
            Map<String, String> headers = new HashMap<>();
            headers.put("Content-Type", "application/json;charset=UTF-8");
            headers.put("x-wlk-gray", "0");
            headers.put("x-wlk-Authorization", accessToken);

            Map<String, String> params = new HashMap<>();
            params.put("deptCode", deptId);
            params.put("pageNo", String.valueOf(pageNo));
            params.put("pageSize", String.valueOf(pageSize));

            try {
                String body = OkHttpUtil.get(UrlBuilder.fromBaseUrl(AuthWelinkSource.WELINK.userList()).build(), headers, params);
                JSONObject object = JSONObject.parseObject(body);
                log.info("get user list body: {}", object);
                if(this.checkResponse(object)){
                    hasMore = object.getInteger("hasMore") == 0; // 0 表示还有下一页
                    pageNo++;
                    JSONArray userArray = object.getJSONArray("data");
                    userJsonArray.addAll(userArray);
                }else{
                    SnailJobLog.REMOTE.error("{} get user list fail。deptId: {}, pageNo: {}", SOURCE, deptId, pageNo);
                    // 部分出错（如某部门没有用户）或某个分页查询时失败
                    return false;
                }
            } catch (Exception e) {
                SnailJobLog.REMOTE.error("{} get user list fail, {}", SOURCE, e.getMessage());
                return false;
            }

        } while(hasMore);

        if(!userJsonArray.isEmpty()){
            List<JSONObject> userListJson = JSONArray.parseArray(userJsonArray.toJSONString(), JSONObject.class);
            List<SysUserBo> userBoListOneDept = userListJson.stream().map(json -> {
                SysUserBo bo = new SysUserBo();
                Map<String, Object> remark = new HashMap<>();
                remark.put("welinkId", json.getString("userId"));
                remark.put("avatar", json.getString("avatar")); // welink avatar 为链接， 系统 avatar 为 bigint
                bo.setRemark(new JSONObject(remark).toJSONString()); // welink 用户id 为字符串而系统表为 long，保存用于更新 sys_social 表
                bo.setDeptId(Convert.toLong(json.get("mainDeptCode")));
                bo.setUserName(json.getString("employeeId"));
                bo.setNickName(json.getString("userNameCn"));
                bo.setStatus("3".equals(json.getString("userStatus")) ? "0" : "1"); // 0 正常 1 停用
                bo.setPhonenumber(json.getString("mobileNumber").replaceFirst("^\\+\\d+-", "")); // 去除 '+86-' 或 '+1-' 等
                String rawSex = json.getString("sex");
                bo.setSex("M".equalsIgnoreCase(rawSex) ? "0" : ("F".equalsIgnoreCase(rawSex) ? "1" : "2")); // 'M' -> '0' ; 'F' -> '1'; else -> '2'
                bo.setEmail(json.getString("userEmail"));
                bo.setPassword(BCrypt.hashpw("Cd123456$"));
                return bo;
            }).toList();
            // 过滤用户名不为空或空白字符
            userBoListOneDept = userBoListOneDept.stream().filter(bo -> !StringUtils.isBlank(bo.getUserName())).toList();
            userBoList.addAll(userBoListOneDept);
        }
        return true;
    }

    private boolean createSysDept(String deptRootCode) {
        if (StringUtils.isBlank(deptRootCode)) {
            SnailJobLog.REMOTE.error("{} create dept list fail, deptRootCode is null or blank string", SOURCE);
            return false;
        }
        if (deptBoList.isEmpty()) {
            SnailJobLog.REMOTE.error("{} create dept list fail, dept list is empty", SOURCE);
            return false;
        }

        boolean flag;
        try {
            flag = deptService.batchInsertOrUpdateDept(deptBoList);
        } catch (Exception e) {
            SnailJobLog.REMOTE.error("{} create dept list fail, {}", SOURCE, e.getMessage());
            return false;
        }
        return flag;
    }

    private boolean createUser(){
        if (userBoList.isEmpty()) {
            SnailJobLog.REMOTE.error("{} create user list fail, user list is empty", SOURCE);
            return false;
        }
        DataPermissionHelper.enableIgnore(); // 忽略数据权限
        boolean flag;
        try {
            flag = userService.batchInsertOrUpdateUserByUserName(userBoList); // username email phone 一定唯一，插入不用校验
        } catch (Exception e) {
            SnailJobLog.REMOTE.error("{} create user list fail, {}", SOURCE, e.getMessage());
            return false;
        }
        return flag;
    }

    private boolean bindWelink(){
        if(!createUser()){
            return false;
        }
        SnailJobLog.REMOTE.info("create {} user success! userCount: {}", SOURCE, userBoList.size());
        // welink 绑定信息更新
        boolean bindFlag = true;

        for(SysUserBo bo: userBoList){
            try{
                JSONObject object = JSON.parseObject(bo.getRemark());
                if(Objects.nonNull(object) && object.containsKey("welinkId")) {
                    String welinkId = object.getString("welinkId");
                    String authId = SOURCE + welinkId;
                    SysSocialBo socialBo = new SysSocialBo();
                    socialBo.setUserId(bo.getUserId());
                    socialBo.setAuthId(authId);
                    socialBo.setOpenId(welinkId);
                    socialBo.setSource(SOURCE);
                    socialBo.setUserName(bo.getUserName());
                    socialBo.setNickName(bo.getNickName());
                    socialBo.setEmail(bo.getEmail());
                    socialBo.setAccessToken(accessToken);
                    if (object.containsKey("avatar")) {
                        socialBo.setAvatar(object.getString("avatar"));
                    }
                    // 查询是否已经绑定用户
                    SysSocialBo params = new SysSocialBo();
                    params.setUserId(bo.getUserId());
                    params.setSource(SOURCE);
                    List<SysSocialVo> socialList = socialService.queryList(params);
                    boolean success;
                    if (socialList.isEmpty()) {
                        success = socialService.insertByBo(socialBo);
                    } else {
                        socialBo.setId(socialList.getFirst().getId());
                        // 通过 id 更新
                        // bindFlag =  socialService.updateByBo(socialBo);  // 不能这么写，如果新值和旧值一样，返回的是 false
                        socialService.updateByBo(socialBo);
                        List<SysSocialVo> checkList = socialService.selectByAuthId(socialBo.getAuthId());
                        success = CollUtil.isNotEmpty(checkList);
                    }

                    if(!success) {
                        bindFlag = false;
                        break;
                    }
                }
            } catch (JSONException e){
                // json 解析出错，自动进入下一个循环
            }
            catch (Exception e){
                bindFlag = false; // 数据库报错
                break;
            }
        }
        return bindFlag;
    }

    public ExecuteResult jobExecute(JobArgs jobArgs) {
        JSONObject object;
        try {
            object = JSON.parseObject(jobArgs.getJobParams().toString());
        }catch (JSONException e){
            return ExecuteResult.failure("JSON parsing error in JobParams");
        }
        String tenantId = object.getString("tenantId");
        String rootDeptId = object.getString("rootDeptId");
        SnailJobLog.REMOTE.info("租户：{}, 根部门ID：{}", tenantId, rootDeptId);
        if(StringUtils.isBlank(tenantId)){
            return ExecuteResult.failure("tenantId is blank");
        }
        if(StringUtils.isBlank(rootDeptId)){
            return ExecuteResult.failure("rootDeptId is blank");
        }
        SysTenantVo tenant = tenantService.queryByTenantId(tenantId);
        if(ObjectUtil.isNull(tenant)){
            return ExecuteResult.failure("tenantId is not exist");
        }
        // 设置所属租户
        TenantHelper.setDynamic(tenantId);

        //创建部门
        boolean getDeptFlag = getDeptList(rootDeptId);
        if(getDeptFlag){
            SnailJobLog.REMOTE.info("get {} dept list success!", SOURCE);
            boolean insertDeptFlag = createSysDept(rootDeptId);
            if(insertDeptFlag){
                SnailJobLog.REMOTE.info("create {} dept success! deptCount: {}", SOURCE, deptBoList.size());
            }else{
                return ExecuteResult.failure("create {} dept fail", SOURCE);
            }
        }else{
            return ExecuteResult.failure("get {} dept list fail", SOURCE );
        }

        // 创建用户并绑定
        boolean getUserFlag = true;
        for(SysDeptBo deptBo: deptBoList){
            String deptCode = deptBo.getDeptId().toString();
            boolean getUserOneDeptFlag = getUserListByDeptId(deptCode); // 循环获取每个部门的所有用户
            if(!getUserOneDeptFlag){
                getUserFlag = false;
                break; // 某一个部门出错则终止
            }
        }
        if(getUserFlag){
            SnailJobLog.REMOTE.info("get {} user list success!", SOURCE);
            boolean bindFlag = bindWelink();
            if(bindFlag){
                SnailJobLog.REMOTE.info("bind {} user success!", SOURCE);
            }else{
                return ExecuteResult.failure("bind {} user fail", SOURCE);
            }
        }else{
            return ExecuteResult.failure("get {} user list fail", SOURCE);
        }


        return ExecuteResult.success("任务成功" );
    }


    private boolean checkResponse(JSONObject object) {
        if(Objects.isNull(object)) {
            log.error("response body is null, request fail！");
            return false;
        }else if(!Objects.equals(object.getString("code"), "0")){
            log.error("response body message: {}", object.getString("message"));
            return false;
        }
        return true;
    }

}

其中发送 Http 请求的封装代码参考上文附录

在 AuthWelinkSource 中定义接口地址，增加以下内容：

public enum AuthWelinkSource implements AuthSource {
    WELINK;

    // 省略...

    public String deptList() {
        return "https://open.welink.huaweicloud.com/api/contact/v3/departments/list";
    }

    public String userList() {
        return "https://open.welink.huaweicloud.com/api/contact/v2/user/list";
    }
}

Service 类中方法扩充

任务类中调用的 deptService.batchInsertOrUpdateDept 和 userService.batchInsertOrUpdateUserByUserName 方法是为了批量创建或更新部门和用户，但是 RuoYiPlus 中没有，需要自己实现。

SysDeptServiceImpl.java 新增：

/**
 * 批量新增或更新部门信息
 *
 * @param list:  部门信息列表
 * @return boolean 是否成功
 */
@Override
public boolean batchInsertOrUpdateDept(List<SysDeptBo> list) {
    if (list.isEmpty()) {
        return false;
    }
    // 将 bo 转换为 sysDept, 暂不构建 ancestors
    List<SysDept> sysDeptList = MapstructUtils.convert(list, SysDept.class);
    if (Objects.isNull(sysDeptList)){
        return false;
    }
    // 构建 Map 快速查找
    Map<Long, SysDept> deptMap = sysDeptList.stream().collect(Collectors.toMap(SysDept::getDeptId, d -> d));
    sysDeptList.forEach(d -> {
        d.setAncestors(buildAncestors(d, deptMap));
    });

    return baseMapper.insertOrUpdateBatch(sysDeptList);
}

// 构造祖先部门列表
private String buildAncestors(SysDept dept, Map<Long, SysDept> deptMap) {
    List<Long> ancestors = new ArrayList<>();
    Long parentId = dept.getParentId();
    while(ObjectUtil.isNotNull(parentId) && parentId != 0) {
        ancestors.add(parentId);
        SysDept parent = deptMap.get(parentId);
        if(ObjectUtil.isNotNull(parent)) {
            parentId = parent.getParentId();
        } else {
            break; // 没有找到上级
        }
    }
    Collections.reverse(ancestors); // 倒序，从顶级到父级
    return ancestors.stream().map(String::valueOf).collect(Collectors.joining(StringUtils.SEPARATOR));
}

SysUserServiceImpl.java 新增：

/**
 * 批量插入或更新用户
 *
 * @param list:  用户信息列表
 * @return boolean 是否成功
 */
@Override
@Transactional(rollbackFor = Exception.class)
public boolean batchInsertOrUpdateUserByUserName(List<SysUserBo> list) {
    if (list.isEmpty()) {
        return false;
    }

    // 将 bo 转换为 sysUser
    List<SysUser> sysUserList = MapstructUtils.convert(list, SysUser.class);
    if (sysUserList == null){
        return false;
    }

    // 批量插入或更新 （根据 userName 判断是插入还是更新，没有一定是插入）
    // 仅单个租户中 userName 唯一
    // 此处不校验 userName, email, phone 唯一性，由函数调用方自行处理
    for(SysUser sysUser : sysUserList){
        int rows = baseMapper.update(sysUser, new LambdaQueryWrapper<SysUser>().eq(SysUser::getUserName, sysUser.getUserName()));
        if( rows == 0){ // 未更新说明不存在
            // 如果是插入，会自动回填 sysUser.userId
            baseMapper.insert(sysUser);
        }else {
            // 如果是更新，需要手动查询
            SysUser dbUser = baseMapper.selectOne(new LambdaQueryWrapper<SysUser>().eq(SysUser::getUserName, sysUser.getUserName()));
            if(dbUser != null){
                sysUser.setUserId(dbUser.getUserId());
            }
        }
    }
    // 回填生成的 userid 更新其它表
    IntStream.range(0, sysUserList.size()).forEach(i -> {
        SysUser sysUser = sysUserList.get(i);
        list.get(i).setUserId(sysUser.getUserId());
    });
    return true;
}

不要忘了在对应抽象类中定义 batchInsertOrUpdateDept 和 batchInsertOrUpdateUserByUserName 方法

后台创建任务

下图中表示每月 1 日的凌晨 1 点执行一次任务，既支持 Cron 表达式也支持固定时间间隔执行。