通讯录

haoban-manage3-api/src/main/java/com/gic/haoban/manage/api/service/DealSyncOperationApiService.java

@@ -6,8 +6,10 @@ import com.gic.haoban.manage.api.dto.PreDealLogInfoDTO;
 import com.gic.haoban.manage.api.dto.SyncCheckDTO;
 import com.gic.haoban.manage.api.dto.SyncSingleDealDTO;
 import com.gic.haoban.manage.api.dto.SyncTaskDTO;
+import com.gic.haoban.manage.api.enums.SyncTaskStatusEnum;
 
 import java.util.List;
+import java.util.Set;
 
 /**
  * Created 2020/4/9.
@@ -209,7 +211,11 @@ public interface DealSyncOperationApiService {
      * @param taskId
      */
     void dealTagTask(String wxEnterpriseId, String enterpriseId, List<String> tagIds, String taskId);
-
+    // 同步好友状态定时
     void taskStatusTimer(String params) ;
     void staffTaskStatusTimer(String params) ;
+    // 同步通讯录状态定时
+    void taskSyncStaffTimer(String params) ;
+
+    void sendToMq(String taskId, Set<String> dealList, SyncTaskStatusEnum syncTaskStatusEnum , String mqName) ;
 }

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/dao/mapper/PreDealLogMapper.java

@@ -57,16 +57,6 @@ public interface PreDealLogMapper {
             , @Param("status") int status, @Param("dataType") int dataType);
 
     /**
-     * 获取单条数据
-     *
-     * @param taskId
-     * @param pDataId
-     * @return
-     */
-    List<TabHaobanPreDealLog> listReByPDataId(@Param("taskId") String taskId, @Param("dataId") String pDataId
-            , @Param("dataType") int dataType);
-
-    /**
      * 更新状态
      *
      * @param taskId

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/service/PreDealService.java

@@ -40,15 +40,6 @@ public interface PreDealService {
     public List<TabHaobanPreDealLog> listByPDataId(String taskId, String pDataId, int status);
 
     /**
-     * 根据父数据id 获取子节点
-     *
-     * @param pDataId
-     * @param status
-     * @return
-     */
-    public List<TabHaobanPreDealLog> listReByPDataId(String taskId, String pDataId);
-
-    /**
      * 更新单条数据状态
      *
      * @param dataId
@@ -201,5 +192,6 @@ public interface PreDealService {
 
     public void addTaskStatusCache(String taskId) ;
     public void addTaskStaffStatusCache(String taskId,String staffId , int type) ;
+    public void addTaskDepartStatusCache(String taskId,int type) ;
 
 }

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/service/hm/impl/HmClerkRelationServiceImpl.java

@@ -9,6 +9,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.stream.Collectors;
 
+import com.gic.haoban.manage.api.service.DealSyncOperationApiService;
 import org.apache.commons.collections.CollectionUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
@@ -74,6 +75,8 @@ public class HmClerkRelationServiceImpl implements HmClerkRelationService {
     private SyncTaskService syncTaskService;
     @Autowired
     private PreDealService preDealService;
+    @Autowired
+    private DealSyncOperationApiService dealSyncOperationApiService ;
 
 
     @Override
@@ -284,30 +287,12 @@ public class HmClerkRelationServiceImpl implements HmClerkRelationService {
             return dealLog;
         }).collect(Collectors.toList());
         this.preDealService.insert(dealLogList);
-        this.sendToMq(taskId, dealLogList.stream().map(o->o.getDataId()).collect(Collectors.toSet()), SyncTaskStatusEnum.modify_hm);
+        this.dealSyncOperationApiService.sendToMq(taskId, dealLogList.stream().map(o->o.getDataId()).collect(Collectors.toSet()), SyncTaskStatusEnum.modify_hm,"departmentSyncDealMq2");
         ServiceResponse resp = ServiceResponse.success() ;
         resp.setMessage(taskId);
         return resp ;
     }
 
-    private void sendToMq(String taskId, Set<String> dealList, SyncTaskStatusEnum syncTaskStatusEnum) {
-        //预处理分组任务
-        syncTaskService.updateTaskStatus(taskId, syncTaskStatusEnum.getVal());
-        List<String> listRet = dealList.stream().map(relationId -> {
-            DealParamMqDTO dealParamMqDTO = new DealParamMqDTO();
-            dealParamMqDTO.setTaskId(taskId);
-            dealParamMqDTO.setData(relationId);
-            dealParamMqDTO.setType(syncTaskStatusEnum.getVal());
-            return JSONObject.toJSONString(dealParamMqDTO);
-        }).collect(Collectors.toList());
-
-        GicMQClient clientInstance = GICMQClientUtil.getClientInstance();
-        try {
-            clientInstance.sendBatchMessages("departmentSyncDealMq", listRet, 1);
-        } catch (Exception e) {
-            e.printStackTrace();
-        }
-    }
 
     public String createModifyHmTask(String wxEnterpriseId, String taskName, String staffId, String staffName, int syncTaskType) {
         TabHaobanSyncTask tabHaobanSyncTask = new TabHaobanSyncTask();

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/service/impl/PreDealServiceImpl.java

@@ -33,6 +33,7 @@ public class PreDealServiceImpl implements PreDealService {
 
     public static final String HAOBAN_TASK_ID_SET_CACHE = "haobanTaskSyncIdSetCache" ;
     public static final String HAOBAN_TASK_STAFF_ID_SET_CACHE = "haobanTaskStaffSyncIdSetCache" ;
+    public static final String HAOBAN_TASK_DEPART_SET_CACHE = "haobanTaskDepartSyncIdSetCache" ;
 
     @Autowired
     private PreDealLogMapper preDealLogMapper;
@@ -78,10 +79,6 @@ public class PreDealServiceImpl implements PreDealService {
         return preDealLogMapper.listByPDataId(taskId, pDataId, status, PreDealTypeEnum.dept.getVal());
     }
 
-    @Override
-    public List<TabHaobanPreDealLog> listReByPDataId(String taskId, String pDataId) {
-        return preDealLogMapper.listReByPDataId(taskId, pDataId, PreDealTypeEnum.dept.getVal());
-    }
 
     @Override
     public boolean updateStatusByDataId(String taskId, String dataId, int status, String reason) {
@@ -200,4 +197,11 @@ public class PreDealServiceImpl implements PreDealService {
         RSet<String> set = RedisUtil.getRedisClient().getSet(HAOBAN_TASK_STAFF_ID_SET_CACHE);
         set.add(key) ;
     }
+
+    @Override
+    public void addTaskDepartStatusCache(String taskId, int type) {
+        String key = taskId+"#"+type ;
+        RSet<String> set = RedisUtil.getRedisClient().getSet(HAOBAN_TASK_DEPART_SET_CACHE);
+        set.add(key) ;
+    }
 }

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/service/out/impl/DealSyncOperationApiServiceImpl.java

@@ -709,13 +709,15 @@ public class DealSyncOperationApiServiceImpl implements DealSyncOperationApiServ
             }
         }
     }
-
-    private void sendToMq(String taskId, Set<String> dealList, SyncTaskStatusEnum syncTaskStatusEnum , String mqName) {
+    @Override
+    public void sendToMq(String taskId, Set<String> dealList, SyncTaskStatusEnum syncTaskStatusEnum , String mqName) {
         //预处理分组任务
         if(StringUtils.isEmpty(mqName)) {
             mqName = "departmentSyncDealMq" ;
         }
-        syncTaskService.updateTaskStatus(taskId, syncTaskStatusEnum.getVal());
+        if(!syncTaskStatusEnum.equals(SyncTaskStatusEnum.clerk_sync)) {
+            syncTaskService.updateTaskStatus(taskId, syncTaskStatusEnum.getVal());
+        }
         List<String> listRet = dealList.stream().map(relationId -> {
             DealParamMqDTO dealParamMqDTO = new DealParamMqDTO();
             dealParamMqDTO.setTaskId(taskId);
@@ -726,8 +728,8 @@ public class DealSyncOperationApiServiceImpl implements DealSyncOperationApiServ
 
         GicMQClient clientInstance = GICMQClientUtil.getClientInstance();
         try {
-            logger.info("放入mq={}",JSON.toJSONString(listRet));
-            clientInstance.sendBatchMessages(mqName, listRet, 10);
+            logger.info("放入mq={},{}",mqName,JSON.toJSONString(listRet));
+            clientInstance.sendBatchMessages(mqName, listRet, 1);
         } catch (Exception e) {
             logger.info("发送失败:{},{}", taskId, JSONObject.toJSONString(listRet));
             e.printStackTrace();
@@ -863,6 +865,78 @@ public class DealSyncOperationApiServiceImpl implements DealSyncOperationApiServ
         }
     }
 
+    @Override
+    public void taskSyncStaffTimer(String params) {
+        try {
+            RSet<String> set = RedisUtil.getRedisClient().getSet(PreDealServiceImpl.HAOBAN_TASK_DEPART_SET_CACHE);
+            logger.info("通讯录同步定时开始staff={},{}",set.size(),set);
+            Iterator<String> it = set.iterator() ;
+            while(it.hasNext()) {
+                String v = it.next() ;
+                String[] arr = v.split("#") ;
+                if(arr.length != 2) {
+                    logger.info("异常{},{}",v );
+                    it.remove();
+                    continue;
+                }
+                String taskId = arr[0] ;
+                int type = Integer.valueOf(arr[1]) ;
+                TabHaobanSyncTask task = syncTaskService.getSyncTask(taskId);
+                if(null == task) {
+                    logger.info("任务不存在taskId={}",taskId);
+                    it.remove();
+                    continue;
+                }
+                if(task.getStatusFlag() != 4) {
+                    this.syncTaskService.updateTaskStatus(taskId,4) ;
+                }
+                if(task.getCreateTime().getTime()+1000*60*60*12< System.currentTimeMillis()) {
+                    logger.info("超过一天taskId={},{}",taskId);
+                    it.remove();
+                    continue;
+                }
+                boolean flag = this.syncDepartmentStaffTask(taskId,type) ;
+                if(flag) {
+                    it.remove();
+                }
+            }
+            logger.info("通讯录同步定时完成={}",set);
+        }catch(Exception e) {
+            logger.warn(e.toString(),e);
+        }
+    }
+
+    private boolean syncDepartmentStaffTask(String taskId,int type) {
+        boolean flag = preDealService.checkTask(taskId, type);
+        if (!flag) {
+            logger.info("同步通讯录部门还未完成={}",taskId);
+            return false;
+        }
+        if(type==0) {
+            logger.info("部门成功,开始同步成员:{}", taskId);
+            String key = "haoban_sync_department_task_" + taskId;
+            RedisUtil.lock(key, 3L);
+            // 部门和成员拉取成功后,直接任务完成
+            List<TabHaobanPreDealLog> dealLogs = preDealService.listByTaskId(taskId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.pre.getVal());
+            if (CollectionUtils.isNotEmpty(dealLogs)) {
+                Set<String> dataIds = dealLogs.stream().map(TabHaobanPreDealLog::getDataId).collect(Collectors.toSet());
+                this.sendToMq(taskId, dataIds, SyncTaskStatusEnum.clerk_sync,"departmentSyncDealMq2");
+            }
+            RedisUtil.unlock(key);
+        }else if(type==2) {
+            logger.info("企微通讯录成员成功:{}", taskId);
+            String key = "haoban_sync_clerk_task_" + taskId;
+            RedisUtil.lock(key, 3L);
+            TabHaobanSyncTask syncTask = syncTaskService.getSyncTask(taskId);
+            String wxEnterpriseId = syncTask.getWxEnterpriseId();
+            //同步成功 进入门店处理
+            this.unlockTask(wxEnterpriseId) ;
+            logger.info("同步通讯录完成");
+            RedisUtil.unlock(key);
+        }
+        return true ;
+    }
+
     private boolean updateTaskStatus(String taskId) {
         boolean b = preDealService.checkFriendTask(taskId, PreDealTypeEnum.friend_clerk.getVal());
         if (!b) {

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/task/QywxDeptSyncOperation.java

@@ -7,6 +7,7 @@ import java.util.stream.Collectors;
 import com.gic.haoban.manage.api.service.DealSyncOperationApiService;
 import com.gic.wechat.api.service.qywx.QywxDepartmentApiService;
 import org.apache.commons.collections.CollectionUtils;
+import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -130,14 +131,14 @@ public class QywxDeptSyncOperation implements BaseSyncOperation {
             } else {
                 this.departmentService.edit(dto);
             }
-            if("1".equals(dataId)) {
-                this.syncTaskService.updateTaskStatus(taskId, SyncTaskStatusEnum.compute.getVal());
+            this.preDealService.addTaskDepartStatusCache(taskId,0);
+            reason = this.getStaff(taskId,dataId,enterpriseId,wxEnterpriseId,secretSetting,qwDTO);
+            if(StringUtils.isBlank(reason)) {
+                resultFlag = false ;
             }
-            this.getStaff(taskId,dataId,enterpriseId,wxEnterpriseId,secretSetting,qwDTO);
         } catch (Exception e) {
             resultFlag = false;
-            reason = "处理异常:";
-            e.printStackTrace();
+            reason = "处理异常";
             logger.info("处理异常:{}", e);
         } finally {
             if (!resultFlag) {
@@ -153,10 +154,9 @@ public class QywxDeptSyncOperation implements BaseSyncOperation {
     public void dealException(String taskId, String dataId, String enterpriseId, String reason) {
         logger.info("部门同步异常：{},{}", taskId, dataId);
         preDealService.updateStatusByDataId(taskId, dataId, PreDealTypeEnum.dept.getVal(), PreDealStatusEnum.exception.getVal(), reason);
-        checkDepartmentTask(taskId);
     }
 
-    private void getStaff(String taskId, String dataId, String enterpriseId, String wxEnterpriseId,SecretSettingDTO secretSetting,WxEnterpriseQwDTO qwDTO) {
+    private String getStaff(String taskId, String dataId, String enterpriseId, String wxEnterpriseId,SecretSettingDTO secretSetting,WxEnterpriseQwDTO qwDTO) {
         if("1".equals(dataId)) {
             List<UserDTO> list = null ;
             if(qwDTO.getWxSecurityType()==4) {
@@ -169,7 +169,7 @@ public class QywxDeptSyncOperation implements BaseSyncOperation {
             if (null == list) {
                 logger.info("微信获取部门成员异常：{},{}", taskId, dataId);
                 dealSyncOperationApiService.unlockTask(wxEnterpriseId) ;
-                return;
+                return "获取企微成员异常";
             }
             logger.info("拉取成员数={},taskId={}",list.size(),taskId);
             List<TabHaobanPreDealLog> preDealLogList = list.stream().map(userDTO -> {
@@ -185,7 +185,9 @@ public class QywxDeptSyncOperation implements BaseSyncOperation {
                 return dealLog;
             }).collect(Collectors.toList());
             this.preDealService.insert(preDealLogList);
+            this.syncTaskService.updateTaskStatus(taskId,4) ;
         }
+        return null ;
     }
     @Override
     public void dealSuccess(String taskId, String dataId, String enterpriseId, String wxEnterpriseId) {
@@ -194,39 +196,19 @@ public class QywxDeptSyncOperation implements BaseSyncOperation {
             logger.info("成功修改错误！{}", dataId);
             return;
         }
-        List<TabHaobanPreDealLog> preList = preDealService.listReByPDataId(taskId, dataId);
+        List<TabHaobanPreDealLog> preList = preDealService.listByPDataId(taskId, dataId,0);
         if (CollectionUtils.isNotEmpty(preList)) {
             Set<String> mid = preList.stream().map(tab -> tab.getDataId()).collect(Collectors.toSet());
             dealDepartmentToMq(taskId, mid, SyncTaskStatusEnum.group_sync);
-        } else {
-            checkDepartmentTask(taskId);
         }
     }
 
     @Override
     public void checkDepartmentTask(String taskId) {
-        boolean b = preDealService.checkTask(taskId, PreDealTypeEnum.dept.getVal());
-        if (!b) {
-            return;
-        }
-        logger.info("部门继续成功,开始成员同步:{}", taskId);
-        String key = "haoban_sync_department_task_" + taskId;
-        RedisUtil.lock(key, 3L);
-        // 部门和成员拉取成功后,直接任务完成
-        List<TabHaobanPreDealLog> dealLogs = preDealService.listByTaskId(taskId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.pre.getVal());
-        if (CollectionUtils.isEmpty(dealLogs)) {
-            // 判断成员是否都已完成
-            qywxClerkSyncOperation.checkDepartmentTask(taskId);
-        } else {
-            Set<String> dataIds = dealLogs.stream().map(TabHaobanPreDealLog::getDataId).collect(Collectors.toSet());
-            this.dealDepartmentToMq(taskId, dataIds, SyncTaskStatusEnum.clerk_sync);
-        }
-        RedisUtil.unlock(key);
     }
 
     /**
      * 放入mq处理部门数据
-     *
      * @param taskId
      * @param dealList
      */

haoban-manage3-service/src/main/java/com/gic/haoban/manage/service/task/QywxStaffSyncOperation.java

@@ -59,11 +59,14 @@ public class QywxStaffSyncOperation implements BaseSyncOperation {
         String reason = "";
         //处理clerk
         try {
-            ServiceResponse response = staffApiService.getWxSaveNew(dealParamMqDTO.getData(), dataPre.getWxEnterpriseId());
+            String taskId = dataPre.getTaskId() ;
+            String wxUserId = dealParamMqDTO.getData() ;
+            ServiceResponse response = staffApiService.getWxSaveNew(wxUserId, dataPre.getWxEnterpriseId());
             if (response.getCode() != 1) {
                 dealFlag = false;
                 reason = response.getMessage();
             }
+            this.preDealService.addTaskDepartStatusCache(taskId,2);
         } catch (Exception e) {
             logger.info("企微通讯录成员同步失败：{}", e.getMessage(), e);
             reason = "同步异常:";
@@ -80,41 +83,17 @@ public class QywxStaffSyncOperation implements BaseSyncOperation {
     @Override
     public void dealException(String taskId, String dataId, String enterpriseId, String reason) {
         logger.info("企微通讯录成员处理失败：t:{},d:{}", taskId, dataId);
-        boolean b = preDealService.updateStatusByDataId(taskId, dataId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.exception.getVal(), reason);
-        if (!b) {
-            logger.info("企微通讯录成员处理异常:t:{}, d:{}", taskId, dataId);
-            return;
-        }
-        checkDepartmentTask(taskId);
-
+        this.preDealService.updateStatusByDataId(taskId, dataId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.exception.getVal(), reason);
     }
 
     @Override
     public void dealSuccess(String taskId, String dataId, String enterpriseId, String wxEnterpriseId) {
         logger.info("企微通讯录成员处理成功：t:{},d:{}", taskId, dataId);
-        boolean b = preDealService.updateStatusByDataId(taskId, dataId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.computed.getVal(), "成功");
-        if (!b) {
-            logger.info("企微通讯录成员成功修改错误！{}", dataId);
-            return;
-        }
-        checkDepartmentTask(taskId);
+        this.preDealService.updateStatusByDataId(taskId, dataId, PreDealTypeEnum.clerk.getVal(), PreDealStatusEnum.computed.getVal(), "成功");
     }
 
     @Override
     public void checkDepartmentTask(String taskId) {
-        boolean b = preDealService.checkTask(taskId, PreDealTypeEnum.clerk.getVal());
-        if (!b) {
-            return;
-        }
-        logger.info("企微通讯录成员成功:{}", taskId);
-        String key = "haoban_sync_clerk_task_" + taskId;
-        RedisUtil.lock(key, 3L);
-        TabHaobanSyncTask syncTask = syncTaskService.getSyncTask(taskId);
-        String wxEnterpriseId = syncTask.getWxEnterpriseId();
-        //同步成功 进入门店处理
-        dealSyncOperationApiService.unlockTask(wxEnterpriseId) ;
-        logger.info("同步通讯录完成");
-        RedisUtil.unlock(key);
     }
 
 }

haoban-manage3-service/src/main/resources/mapper/PreDealLogMapper.xml

@@ -195,14 +195,6 @@
         </if>
     </select>
 
-    <select id="listReByPDataId" resultMap="BaseResultMap">
-        select
-        <include refid="Base_Column_List"/>
-        from tab_haoban_pre_deal_log
-        where task_id = #{taskId} and p_data_id=#{dataId} and data_type=#{dataType}
-        and status_flag in(0,1,3)
-    </select>
-
     <update id="updateStatusByDataId">
         update tab_haoban_pre_deal_log
         <set>