MDM同步代码

code/jyyy/nckd-jimin-jyyy-hr/src/main/java/nckd/jimin/jyyy/hr/haos/staff/plugin/task/SyncAdminOrgTask.java

@@ -2,9 +2,13 @@ package nckd.jimin.jyyy.hr.haos.staff.plugin.task;
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import kd.bos.algo.DataSet;
+import kd.bos.algo.Row;
 import kd.bos.context.RequestContext;
 import kd.bos.dataentity.entity.DynamicObject;
 import kd.bos.dataentity.entity.DynamicObjectCollection;
+import kd.bos.db.DB;
+import kd.bos.db.DBRoute;
 import kd.bos.exception.KDException;
 import kd.bos.logging.Log;
 import kd.bos.logging.LogFactory;
@@ -13,6 +17,7 @@ import kd.bos.orm.query.QFilter;
 import kd.bos.orm.util.CollectionUtils;
 import kd.bos.schedule.executor.AbstractTask;
 import kd.bos.util.StringUtils;
+import kd.hr.haos.business.servicehelper.AdminOrgQueryServiceHelper;
 import kd.sdk.plugin.Plugin;
 import kd.hr.hbp.business.servicehelper.HRBaseServiceHelper;
 
@@ -21,6 +26,7 @@ import java.text.ParseException;
 import java.text.SimpleDateFormat;
 import java.util.*;
 import java.util.stream.Collectors;
+import java.util.stream.Stream;
 
 /**
  * 2025-04-16 Tyx
@@ -52,6 +58,8 @@ public class SyncAdminOrgTask extends AbstractTask implements Plugin {
             log.info("-------- SyncAdminOrgTask 开始执行同步行政组织 --------");
             //最后调MDM的jsonObject
             JSONObject ob = new JSONObject();
+            JSONObject posOb = new JSONObject();
+            JSONObject personOb = new JSONObject();
             initDateRange(map);
             HRBaseServiceHelper helper = new HRBaseServiceHelper(orgChgRecord_entity);
             String selectProperties = "adminorg,orgchgentry.changescene.number,orgchgentry.operationtime";
@@ -63,22 +71,6 @@ public class SyncAdminOrgTask extends AbstractTask implements Plugin {
                 SyncUtil.createLog(SyncUtil.v_success, startTime, endTime, "未查询到变化组织", "未调用", SyncUtil.SyncOrg);
             }
             else {
-                //发生变化的组织Map
-                //按照变动场景编码分成不同的List，后续根据场景分开处理
-                //ToDo 如果查询时间内，某个组织先发生组织调整，再修改除名称外其他信息怎么处理？
-                //按照组织分组，只取时间段内最新的一条记录
-                //暂时只处理（10，20，30，40，50）;
-                //ToDo（80，90）目前前台界面做不了该业务，后续考虑;
-                //ToDo 60尚不明确，需确认；
-                //1010_S	    组织新设
-                //1020_S	    上级调整
-                //1030_S	    组织更名
-                //1040_S	    组织停用
-                //1050_S	    初始化新增
-                //1060_S	    初始化禁用
-                //1080_S	    组织合并
-                //1090_S	    组织拆分
-                //CS_1110_SY01	组织修订
 //            Map<String, List<DynamicObject>> changeOperateMap = (Map)changeOperateCol.stream().collect(Collectors.groupingBy(
 //                            dyx -> dyx.getString("orgchgentry.changescene.number")
 //                    ));
@@ -94,17 +86,29 @@ public class SyncAdminOrgTask extends AbstractTask implements Plugin {
                 log.info("-------- 变化组织数（去重）：" + changeOperateMap.keySet().size() + " --------");
                 log.info("-------- 变化组织Id：" + changeOperateMap.keySet() + " --------");
 
-                // ToDo 根据变动操作字段获取发生了组织调整的组织，后续需要根据这些组织同步下级岗位和人员
+                //过滤出上级调整和修改名称的组织
+                List<Long> changeOrgIdList = changeOperateCol.stream().filter(dyx -> dyx.get("orgchgentry.changescene.number").equals("1020_S") || dyx.get("orgchgentry.changescene.number").equals("1030_S"))
+                        .map(dyx -> dyx.getLong("adminOrg")).collect(Collectors.toList());
+                //查询所有下级组织
+                List<Map<String, Object>> a = AdminOrgQueryServiceHelper.batchQueryAllSubOrg(changeOrgIdList, new Date());
+                //转换成Set
+                Set subOrgIdset = a.stream().map(i -> i.get("orgId")).collect(Collectors.toSet());
                 //组织历史查询，如果查询时间内发生多次变化，
                 //haos_adminorgdetail 组织历史查询 所属表t_haos_adminorg Id与t_org_org一样
                 selectProperties = "id,boid,number,name,orglongname,level,nckd_easid,parentorg.id,parentorg.name,parentorg.number,parentorg.nckd_easid,establishmentdate,bsed,enable,adminorgtype.name,disabledate,modifytime,creator.name";
-                QFilter idQFilter = new QFilter("id", "in", changeOperateMap.keySet());
+                //合并Set 变化组织的IdSet+更名/调整上级的下级所有组织的IdSet
+                Set totalOrgSet = (Set) Stream.concat(subOrgIdset.stream(), changeOperateMap.keySet().stream()).collect(Collectors.toSet());
+                QFilter idQFilter = new QFilter("id", "in", totalOrgSet);
                 DynamicObject[] orgDyArr = new HRBaseServiceHelper(adminOrgDetail_entity).query(selectProperties, new QFilter[]{idQFilter});
                 //处理下长编码/长名称，星瀚内是4VTX1ACRU8A9!4VU/45/K1PJZ!4W5LG9NNW9EW这种格式 -- 直接取行政组织结构里面的长编码
                 Map<Long, DynamicObject> bosOrgMap = SyncUtil.initOrgInfo();
                 log.info("-------- 查询完组织详细信息数：" + orgDyArr.length + " --------");
                 //构建入参
                 buildJSON(ob, orgDyArr, changeOperateMap, bosOrgMap);
+                //构建岗位入参
+                buildPosJSON(posOb, totalOrgSet, bosOrgMap);
+                //构建人员入参
+                buildPersonJSON(personOb, totalOrgSet, bosOrgMap);
                 //获取调用接口地址
                 String url = SyncUtil.getUrl("org_url");
                 log.info("-------- 组织同步调用url : " + url + " --------");
@@ -114,6 +118,26 @@ public class SyncAdminOrgTask extends AbstractTask implements Plugin {
                 String status = SyncUtil.dealResponseStatus(response);
                 //记录日志
                 SyncUtil.createLog(status, startTime, endTime, ob.toJSONString(), response.toJSONString(), SyncUtil.SyncOrg);
+                if(posOb.getJSONArray("obj").size() > 0) {
+                    url = SyncUtil.getUrl("position_url");
+                    log.info("-------- 组织同步下级岗位调用url : " + url + " --------");
+                    //调用接口
+                    response = SyncUtil.doPostByHttpClient(url, posOb);
+                    //处理返回结果
+                    status = SyncUtil.dealResponseStatus(response);
+                    //记录日志
+                    SyncUtil.createLog(status, startTime, endTime, posOb.toJSONString(), response.toJSONString(), SyncUtil.SyncOrg_Position);
+                }
+                if(personOb.getJSONArray("obj").size() > 0) {
+                    url = SyncUtil.getUrl("person_url");
+                    log.info("-------- 组织同步下级人员调用url : " + url + " --------");
+                    //调用接口
+                    response = SyncUtil.doPostByHttpClient(url, personOb);
+                    //处理返回结果
+                    status = SyncUtil.dealResponseStatus(response);
+                    //记录日志
+                    SyncUtil.createLog(status, startTime, endTime, personOb.toJSONString(), response.toJSONString(), SyncUtil.SyncOrg_Person);
+                }
             }
         } catch (ParseException e) {
             SyncUtil.createLog(SyncUtil.v_error, startTime, endTime, e.getMessage(), "未调用", SyncUtil.SyncOrg);
@@ -211,6 +235,120 @@ public class SyncAdminOrgTask extends AbstractTask implements Plugin {
         ob.put("obj", obj);
     }
 
+    /**
+     * 构建职位入参
+     * @param ob
+     * @param totalOrgSet
+     * @param bosOrgMap
+     */
+    private void buildPosJSON(JSONObject ob, Set totalOrgSet, Map<Long, DynamicObject> bosOrgMap) {
+        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        String selectFields = "id,adminorg.id,adminorg.name,adminorg.number,adminorg.nckd_easid,parent.name,parent.id,parent.number,number,name,enable,isleader,modifytime,issensitive,nckd_isgxp";
+        QFilter filter = new QFilter("adminorg.id", QCP.in, totalOrgSet);
+        filter.and("iscurrentversion",QCP.equals,"1");
+        DynamicObjectCollection posDyArr = new HRBaseServiceHelper("hbpm_positionhr").queryOriginalCollection(selectFields, new QFilter[]{filter});
+        //构造岗位的树形结构
+        Map<Long, Integer> levelMap = SyncUtil.initPositionTree();
+        JSONArray obj = new JSONArray();
+        for(DynamicObject pos : posDyArr) {
+            JSONObject data = new JSONObject(true);
+            //主体标签
+            data.put("MAIN_BODY_ID", "JY");
+            //级别
+            data.put("MD_LEVEL", levelMap.get(pos.getLong("id")));
+            //所属行政组织名称
+            data.put("POSI_SUB_DEP_NAME",pos.getString("adminorg.name"));
+            //所属行政组织ID 老岗位用EASID 新岗位用星瀚ID
+            String parentId = StringUtils.isBlank(pos.getString("adminorg.nckd_easid")) ? pos.getString("adminorg.id") : pos.getString("adminorg.nckd_easid");
+            data.put("POSI_SUB_DEP_CODE",parentId);
+            //所属行政组织全程 长名称
+            data.put("POSI_SUB_DEP_LNAME",bosOrgMap.get(pos.getLong("adminorg.id")).getString("structure.fullname"));
+            //汇报岗位名称
+            data.put("POSI_REPPOST_NAME",pos.getString("parent.name"));
+            //汇报岗位编码
+            data.put("POSI_REPPOST_CODE",pos.getString("parent.number"));
+            //岗位编码
+            data.put("MD_CODE",pos.getString("number"));
+            //岗位名称
+            data.put("MD_DESCRIPTION",pos.getString("name"));
+            //岗位状态 MDM:1-启用，2-禁用；星瀚：1-启用，0-禁用
+            int enable = pos.getInt("enable");
+            if(enable == 1) {
+                data.put("POSI_STATE","1");
+            }
+            else if(enable == 0){
+                data.put("POSI_STATE","2");
+            }
+            else {
+                data.put("POSI_STATE","0");
+            }
+            //是否负责人职位
+            int isLeader = pos.getInt("isleader");
+            if(isLeader == 1) {
+                data.put("POSI_POSIT_CHARGE","是");
+            }
+            else {
+                data.put("POSI_POSIT_CHARGE","否");
+            }
+            //是否涉密岗位
+            int isSecret = pos.getInt("issensitive");
+            if(isSecret == 1) {
+                data.put("POSI_IS_SECRET","是");
+            }
+            else {
+                data.put("POSI_IS_SECRET","否");
+            }
+            //是否GXP
+            int isGxp = pos.getInt("nckd_isgxp");
+            if(isGxp == 1) {
+                data.put("POSI_IS_GXP","是");
+            }
+            else {
+                data.put("POSI_IS_GXP","否");
+            }
+            //最后更新时间
+            data.put("POSI_LAST_UPTIME",sdf.format(pos.getDate("modifytime")));
+            obj.add(data);
+        }
+        //入参排序
+        SyncUtil.setJSONArraySorted(obj, "MD_LEVEL");
+        ob.put("obj", obj);
+    }
+
+    /**
+     * 构建人员入参
+     * @param personOb
+     * @param totalOrgSet
+     * @param bosOrgMap
+     */
+    private void buildPersonJSON(JSONObject personOb, Set totalOrgSet, Map<Long, DynamicObject> bosOrgMap) {
+        Map<Long, JSONArray> partTimeMap = SyncUtil.getPartTimeInfo(bosOrgMap);
+        StringBuilder sb = SyncUtil.getPersonQuerySql();
+        String ids = (totalOrgSet.stream().map(Object::toString).collect(Collectors.joining(","))).toString();
+        sb.append(" and org.fid in (").append(ids).append(") ");
+        DataSet dataSet = DB.queryDataSet(this.getClass().getName(), DBRoute.of("hr"), sb.toString());
+        JSONArray obj = new JSONArray();
+        String[] fieldNames = dataSet.getRowMeta().getFieldNames();
+        while(dataSet.hasNext()) {
+            JSONObject data = new JSONObject(true);
+            Row rowData = dataSet.next();
+            //循环put数据
+            for (String field : fieldNames) {
+                data.put(field.toUpperCase(), rowData.get(field));
+            }
+            //处理组织长名称
+            data.put("EMP_W_DEPLNAME", bosOrgMap.get(rowData.getLong("orgid")).getString("structure.fullname"));
+            data.put("EMP_W_EMPLOYNUMBER", "test-20250401-0011");
+
+            //处理兼职
+            JSONArray partJob = partTimeMap.get(rowData.getLong("PERSONID"));
+            if(partJob != null) {
+                data.put("partJob", partJob);
+            }
+            obj.add(data);
+        }
+        personOb.put("obj", obj);
+    }
 
     /**
      * 根据调度作业上配置的参数初始化时间范围

code/jyyy/nckd-jimin-jyyy-hr/src/main/java/nckd/jimin/jyyy/hr/haos/staff/plugin/task/SyncPersonTask.java

@@ -51,7 +51,7 @@ public class SyncPersonTask extends AbstractTask implements Plugin {
             String url = SyncUtil.getUrl("person_url");
             log.info("-------- 人员同步调用url : " + url + " --------");
             //处理兼职
-            Map<Long, JSONArray> partTimeMap = getPartTimeInfo(orgDyxMap);
+            Map<Long, JSONArray> partTimeMap = SyncUtil.getPartTimeInfo(orgDyxMap);
             //构建入参
             buildJSON(ob, dataSet, orgDyxMap, partTimeMap);
             if(ob.getJSONArray("obj").size() == 0) {
@@ -123,64 +123,6 @@ public class SyncPersonTask extends AbstractTask implements Plugin {
         return dataSet;
     }
 
-    /**
-     * 获取兼职档案， 人员ID - JSONArray数据
-     *
-     * @return
-     */
-    public Map<Long, JSONArray> getPartTimeInfo (Map<Long, DynamicObject> orgDyxMap) {
-        StringBuilder sb = new StringBuilder();
-        sb.append("/*dialect*/ select a.fpersonId personId, \n");
-        sb.append("        pos.fid EMP_P_PRIMARY_KEY, \n");
-        sb.append("        to_char(a.fstartdate,'yyyy-MM-dd') EMP_P_STARTDATE, \n");
-        sb.append("        to_char(a.fenddate,'yyyy-MM-dd') EMP_P_ENDDATE, \n");
-        sb.append("        '上海济煜' EMP_P_SUBCOMPANY, \n");
-        sb.append("        org.fid ORGID, \n");
-        sb.append("        case when org.fk_nckd_easid = ' ' then org.fid::VARCHAR else org.fk_nckd_easid end EMP_P_DEPCODE, \n");
-        sb.append("        org.fname EMP_P_DEPNAME, \n");
-        sb.append("        pos.fnumber EMP_P_POSICODE, \n");
-        sb.append("        pos.fname EMP_P_POSINAME, \n");
-        sb.append("        pos1.fnumber EMP_P_PARPOSICODE, \n");
-        sb.append("        pos1.fname EMP_P_PARPOSINAME, \n");
-        sb.append("        '2' EMP_P_FORTYPE, \n");
-        sb.append("        '否' EMP_P_DELETLOGO \n");
-        sb.append(" from t_hrpi_ermanfile a \n");
-        sb.append(" left join t_hrpi_erfiletype b on a.ffiletypeid = b.fid \n");
-        sb.append(" left join t_hbss_postype c on b.fpostypeid = c.fid \n");
-        sb.append(" left join t_hrpi_empposorgrel d on d.fid = a.fempposrelid \n");
-        sb.append(" left join t_hbpm_position pos on pos.fid = d.fpositionid \n");
-        sb.append(" left join t_hbpm_position pos1 on pos1.fid = pos.fparentid  \n");
-        sb.append(" left join t_haos_adminorg org on org.fid = d.fadminorgid \n");
-        sb.append(" where c.fnumber = '1020_S' \n");
-        sb.append("    and a.fiscurrentversion = '1' \n");
-        log.info("-------- 查询人员兼职sql :" + sb.toString() + " -------- ");
-        DataSet dataSet = DB.queryDataSet(this.getClass().getName(), DBRoute.of("hr"), sb.toString());
-
-        Map<Long, JSONArray> partTimeMap = new HashMap<Long, JSONArray>();
-        String[] fieldNames = dataSet.getRowMeta().getFieldNames();
-        while(dataSet.hasNext()) {
-            JSONObject data = new JSONObject(true);
-            Row rowData = dataSet.next();
-            //循环put数据
-            for (String field : fieldNames) {
-                data.put(field.toUpperCase(), rowData.get(field));
-            }
-            data.put("EMP_P_DEPLNAME",orgDyxMap.get(rowData.getLong("ORGID")).getString("structure.fullname"));
-            JSONArray partTimeArr = partTimeMap.get(rowData.getLong("PERSONID"));
-            if(partTimeArr != null) {
-                partTimeArr.add(data);
-            }
-            else {
-                partTimeArr = new JSONArray();
-                partTimeArr.add(data);
-            }
-            partTimeMap.put(rowData.getLong("PERSONID"), partTimeArr);
-        }
-        return partTimeMap;
-    }
-
-
-
     /**
      * 根据调度作业上配置的参数初始化时间范围
      * @param map

code/jyyy/nckd-jimin-jyyy-hr/src/main/java/nckd/jimin/jyyy/hr/haos/staff/plugin/task/SyncPositionTask.java

@@ -117,7 +117,7 @@ public class SyncPositionTask extends AbstractTask implements Plugin {
             data.put("MD_LEVEL", levelMap.get(pos.getLong("id")));
             //所属行政组织名称
             data.put("POSI_SUB_DEP_NAME",pos.getString("adminorg.name"));
-            //所属行政组织ID TODO 是EASID还是星瀚ID？
+            //所属行政组织ID 老岗位用EASID 新岗位用星瀚ID
             String parentId = StringUtils.isBlank(pos.getString("adminorg.nckd_easid")) ? pos.getString("adminorg.id") : pos.getString("adminorg.nckd_easid");
             data.put("POSI_SUB_DEP_CODE",parentId);
             //所属行政组织全程 长名称
@@ -176,6 +176,8 @@ public class SyncPositionTask extends AbstractTask implements Plugin {
 
 
 
+
+
     public void initDateRange(Map<String, Object> map) throws ParseException {
         boolean isAuto = Boolean.valueOf(map.get("isAuto").toString());
         if(!isAuto) {

code/jyyy/nckd-jimin-jyyy-hr/src/main/java/nckd/jimin/jyyy/hr/haos/staff/plugin/task/SyncUtil.java

@@ -2,8 +2,12 @@ package nckd.jimin.jyyy.hr.haos.staff.plugin.task;
 
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
+import kd.bos.algo.DataSet;
+import kd.bos.algo.Row;
 import kd.bos.dataentity.entity.DynamicObject;
 import kd.bos.dataentity.entity.DynamicObjectCollection;
+import kd.bos.db.DB;
+import kd.bos.db.DBRoute;
 import kd.bos.exception.KDException;
 import kd.bos.logging.Log;
 import kd.bos.logging.LogFactory;
@@ -271,6 +275,7 @@ public class SyncUtil {
     public static Date getLastSyncEndTime(String type) throws ParseException {
         QFilter filter = new QFilter("nckd_synctype", QCP.equals, type);
         filter.and("nckd_status", QCP.equals, "A");
+        filter.and("enable", QCP.equals, "1");
         DynamicObjectCollection logInfos = QueryServiceHelper.query(KEY_ENTITY_LOG, "nckd_endtime", new QFilter[]{filter}, "nckd_endtime desc", 1);
         if(logInfos.size() > 0) {
             return logInfos.get(0).getDate("nckd_endtime");
@@ -383,6 +388,69 @@ public class SyncUtil {
         return sb;
     }
 
+    /**
+     * 兼职查询sql
+     * @return
+     */
+    public static StringBuilder getPersonPartTimeSql () {
+        StringBuilder sb = new StringBuilder();
+        sb.append("/*dialect*/ select a.fpersonId personId, \n");
+        sb.append("        pos.fid EMP_P_PRIMARY_KEY, \n");
+        sb.append("        to_char(a.fstartdate,'yyyy-MM-dd') EMP_P_STARTDATE, \n");
+        sb.append("        to_char(a.fenddate,'yyyy-MM-dd') EMP_P_ENDDATE, \n");
+        sb.append("        '上海济煜' EMP_P_SUBCOMPANY, \n");
+        sb.append("        org.fid ORGID, \n");
+        sb.append("        case when org.fk_nckd_easid = ' ' then org.fid::VARCHAR else org.fk_nckd_easid end EMP_P_DEPCODE, \n");
+        sb.append("        org.fname EMP_P_DEPNAME, \n");
+        sb.append("        pos.fnumber EMP_P_POSICODE, \n");
+        sb.append("        pos.fname EMP_P_POSINAME, \n");
+        sb.append("        pos1.fnumber EMP_P_PARPOSICODE, \n");
+        sb.append("        pos1.fname EMP_P_PARPOSINAME, \n");
+        sb.append("        '2' EMP_P_FORTYPE, \n");
+        sb.append("        '否' EMP_P_DELETLOGO \n");
+        sb.append(" from t_hrpi_ermanfile a \n");
+        sb.append(" left join t_hrpi_erfiletype b on a.ffiletypeid = b.fid \n");
+        sb.append(" left join t_hbss_postype c on b.fpostypeid = c.fid \n");
+        sb.append(" left join t_hrpi_empposorgrel d on d.fid = a.fempposrelid \n");
+        sb.append(" left join t_hbpm_position pos on pos.fid = d.fpositionid \n");
+        sb.append(" left join t_hbpm_position pos1 on pos1.fid = pos.fparentid  \n");
+        sb.append(" left join t_haos_adminorg org on org.fid = d.fadminorgid \n");
+        sb.append(" where c.fnumber = '1020_S' \n");
+        sb.append("    and a.fiscurrentversion = '1' \n");
+        return sb;
+    }
 
+    /**
+     * 获取兼职档案， 人员ID - JSONArray数据
+     *
+     * @return
+     */
+    public static Map<Long, JSONArray> getPartTimeInfo(Map<Long, DynamicObject> orgDyxMap) {
+        StringBuilder sb = getPersonPartTimeSql();
+        logger.info("-------- 查询人员兼职sql :" + sb.toString() + " -------- ");
+        DataSet dataSet = DB.queryDataSet("SyncUtilGetPartTime", DBRoute.of("hr"), sb.toString());
+
+        Map<Long, JSONArray> partTimeMap = new HashMap<Long, JSONArray>();
+        String[] fieldNames = dataSet.getRowMeta().getFieldNames();
+        while(dataSet.hasNext()) {
+            JSONObject data = new JSONObject(true);
+            Row rowData = dataSet.next();
+            //循环put数据
+            for (String field : fieldNames) {
+                data.put(field.toUpperCase(), rowData.get(field));
+            }
+            data.put("EMP_P_DEPLNAME",orgDyxMap.get(rowData.getLong("ORGID")).getString("structure.fullname"));
+            JSONArray partTimeArr = partTimeMap.get(rowData.getLong("PERSONID"));
+            if(partTimeArr != null) {
+                partTimeArr.add(data);
+            }
+            else {
+                partTimeArr = new JSONArray();
+                partTimeArr.add(data);
+            }
+            partTimeMap.put(rowData.getLong("PERSONID"), partTimeArr);
+        }
+        return partTimeMap;
+    }
 
 }