中级服务同步相关操作,新增库存预警提醒次数

src/main/java/com/glxp/api/dao/idc/DbDao.java

@@ -55,7 +55,7 @@ public interface DbDao {
 	@Select("select column_name columnName, data_type dataType, column_comment columnComment, column_key columnKey, extra as extra,table_name tableName, "
 			+ "concat('(',if(NUMERIC_PRECISION>0,CONVERT(NUMERIC_PRECISION,CHAR),''),if(NUMERIC_SCALE>0,concat(',',CONVERT(NUMERIC_SCALE,CHAR)),''),if(CHARACTER_MAXIMUM_LENGTH>0,CONVERT(CHARACTER_MAXIMUM_LENGTH,CHAR),''),')') as width "
 			+ " from information_schema.columns\r\n"
-			+ " where lower(table_name) = lower(#{tableName}) column_key='PRI' and table_schema = (select database()) order by ordinal_position")
+			+ " where lower(table_name) = lower(#{tableName}) and column_key='PRI' and table_schema = (select database()) order by ordinal_position")
 	List<Map<String, String>> listKeyMysql(String tableName);
 	@Select("select cname as \"columnName\",coltype as \"dataType\",cname \"columnComment\",'' as \"columnKey\",'' as  \"extra\",tname \"tableName\", "
 			+" '('||decode(sign(PRECISION),1,to_char(PRECISION),'')||decode(sign(SCALE),1,','||to_char(SCALE),'')||decode(sign(PRECISION),1,'',decode(sign(SCALE),1,'',to_char(width)))||')' as width "

src/main/java/com/glxp/api/entity/inv/InvRemindMsgEntity.java

@@ -137,4 +137,8 @@ public class InvRemindMsgEntity {
      */
     @TableField(value = "handleMsg")
     private String handleMsg;
-}
+
+    @TableField(value = "remindCount")
+    private int remindCount;
+
+}

src/main/java/com/glxp/api/idc/controller/IdcController.java

@@ -16,55 +16,53 @@ import java.util.Map;
  */
 @RestController
 public class IdcController {
-    @Resource
-    IdcService idcService;
-    @Resource
-    DbDao dbDao;
+	@Resource
+	IdcService idcService;
+	@Resource
+	DbDao dbDao;
+	/*通用上传接口*/
+	//@Log("中继服务数据接收,下级往上级上传")
+	@RequestMapping(value = "/spssync/common/upload")
+	public BaseResponse upload(HttpServletRequest request,
+							   @RequestParam("content") String content,
+							   @RequestParam(value = "files", required = false) MultipartFile[] files) {
+		//
+		return idcService.receive("U",request, content, files);
+	}
+	/*通用下发接口*/
+	//@Log("中继服务数据接收,上级往下级下发")
+	@RequestMapping(value = "/spssync/common/issued")
+	public BaseResponse issued(HttpServletRequest request,
+			@RequestParam("content") String content,
+			@RequestParam(value = "files", required = false) MultipartFile[] files) {
+		//
+		return idcService.receive("I",request, content, files);
+	}
 
-    /*通用上传接口*/
-//    @Log("中继服务数据接收,下级往上级上传")
-    @RequestMapping(value = "/spssync/common/upload")
-    public BaseResponse upload(HttpServletRequest request,
-                               @RequestParam("content") String content,
-                               @RequestParam(value = "files", required = false) MultipartFile[] files) {
-        //
-        return idcService.receive("U", request, content, files);
-    }
+	//@Log("数据同步测试")
+	@RequestMapping(value = "/spssync/common/test")
+	public BaseResponse test(HttpServletRequest request,@RequestBody Map<String, Object> params) {
+		//
+		return idcService.send(params);
+	}
 
-    /*通用下发接口*/
-//    @Log("中继服务数据接收,上级往下级下发")
-    @RequestMapping(value = "/spssync/common/issued")
-    public BaseResponse issued(HttpServletRequest request,
-                               @RequestParam("content") String content,
-                               @RequestParam(value = "files", required = false) MultipartFile[] files) {
-        //
-        return idcService.receive("I", request, content, files);
-    }
+	//@Log("数据同步任务列表拉取")
+	@RequestMapping(value = "/spssync/common/list")
+	public BaseResponse list(HttpServletRequest request,@RequestBody(required = false) Map<String, Object> params) {
+		return idcService.taskList(request,params);
+	}
 
-    //    @Log("数据同步测试")
-    @RequestMapping(value = "/spssync/common/test")
-    public BaseResponse test(HttpServletRequest request, @RequestBody Map<String, Object> params) {
-        //
-        return idcService.send(params);
-    }
+	//@Log("数据同步任务根据任务ID下载数据")
+	@RequestMapping(value = "/spssync/common/download")
+	@ResponseBody
+	public BaseResponse download(HttpServletRequest request,@RequestBody Map<String, Object> params) {
+		return idcService.download(request,params);
+	}
 
-    //    @Log("数据同步任务列表拉取")
-    @RequestMapping(value = "/spssync/common/list")
-    public BaseResponse list(HttpServletRequest request, @RequestBody(required = false) Map<String, Object> params) {
-        return idcService.taskList(request, params);
-    }
-
-    //    @Log("数据同步任务根据任务ID下载数据")
-    @RequestMapping(value = "/spssync/common/download")
-    @ResponseBody
-    public BaseResponse download(HttpServletRequest request, @RequestBody Map<String, Object> params) {
-        return idcService.download(request, params);
-    }
-
-    //    @Log("数据同步任务根据任务ID,返回下载成功标记")
-    @RequestMapping(value = "/spssync/common/success")
-    @ResponseBody
-    public BaseResponse success(HttpServletRequest request, @RequestBody Map<String, Object> params) {
-        return idcService.downlaodSuccess(request, params);
-    }
+	//@Log("数据同步任务根据任务ID,返回下载成功标记")
+	@RequestMapping(value = "/spssync/common/success")
+	@ResponseBody
+	public BaseResponse success(HttpServletRequest request,@RequestBody Map<String, Object> params) {
+		return idcService.downlaodSuccess(request,params);
+	}
 }

src/main/java/com/glxp/api/idc/service/IdcService.java

@@ -22,9 +22,10 @@ public interface IdcService {
 
     BaseResponse download(HttpServletRequest request, Map<String, Object> params);
 
-    void pull();
+    public void asyncFetchTask();
 
     BaseResponse downlaodSuccess(HttpServletRequest request, Map<String, Object> params);
 
     public void asyncIdcTask();
+
 }

src/main/java/com/glxp/api/idc/service/impl/IdcServiceImpl.java

@@ -4,10 +4,10 @@ import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import com.glxp.api.common.res.BaseResponse;
 import com.glxp.api.common.util.ResultVOUtils;
+import com.glxp.api.dao.idc.DbDao;
 import com.glxp.api.dao.schedule.ScheduledDao;
 import com.glxp.api.entity.system.ScheduledEntity;
 import com.glxp.api.entity.system.SystemParamConfigEntity;
-import com.glxp.api.dao.idc.DbDao;
 import com.glxp.api.idc.service.IdcService;
 import com.glxp.api.idc.utils.DBAUtils;
 import com.glxp.api.idc.utils.IDCUtils;
@@ -109,18 +109,27 @@ public class IdcServiceImpl implements IdcService {
     }
 
     /*拉取前一级中继服务数据*/
+    @Async
     @Override
-    public void pull() {
-        Map<String, Object> query = new HashMap<String, Object>();
-        String result = post(getNextHost("U") + "/spssync/common/list", query);
-        JSONObject json = JSON.parseObject(result);
-        if (json.getInteger("code") == 20000 && json.getString("data") != null) {
-            List<Map> list = JSON.parseArray(json.getString("data"), Map.class);
-            if (list != null) {
-                for (Map map : list) {
-                    Map<String, Object> params = new HashMap<String, Object>();
-                    params.put("taskId", map.get("taskId"));
-                    pullData(getNextHost("U") + "/spssync/common/list", params);
+    public void asyncFetchTask() {
+        String[] directions = {"I", "U"};
+        for (String dir : directions) {
+            Map<String, Object> query = new HashMap<String, Object>();
+            String host = getNextHost(dir);
+            if (!StringUtils.isEmpty(host)) {
+                String result = post(host + "/spssync/common/list", query);
+                if (IDCUtils.isJson(result)) {
+                    JSONObject json = JSON.parseObject(result);
+                    if (json != null && json.getInteger("code") == 20000 && json.getString("data") != null) {
+                        List<Map> list = JSON.parseArray(json.getString("data"), Map.class);
+                        if (list != null) {
+                            for (Map map : list) {
+                                Map<String, Object> params = new HashMap<String, Object>();
+                                params.put("taskId", map.get("taskId"));
+                                fetchData(host + "/spssync/common/list", params);
+                            }
+                        }
+                    }
                 }
             }
         }
@@ -135,14 +144,14 @@ public class IdcServiceImpl implements IdcService {
 
         if (direction.equals("I")) {
             tNames = SPMS_TO_UDI_TABLES;
-        } else {
+        } else if (direction.equals("U")) {
             tNames = UDI_TO_SPMS_TABLES;
         }
         /*为顶级或末级，以及下游或上游连通，可执行*/
         Boolean isLastLevel = isLastLevel(direction);
         Boolean isRelay = isRelay(direction);
-        System.out.print("-----数据传输任务开始----" + direction + "\n");
-        System.out.print("-----是否顶级或末级服务：" + isLastLevel + ",是否转发数据:" + isRelay + "----\n");
+        logger.info("-----数据传输任务开始----" + direction + "\n");
+        logger.info("-----是否需要上传或下发数据：" + isLastLevel + ",是否已配置接收地址:" + isRelay + "----\n");
         if (isLastLevel && isRelay) {
             for (String t : tNames) {
                 uploadData(t);
@@ -151,17 +160,22 @@ public class IdcServiceImpl implements IdcService {
 
     }
 
+
     private void uploadData(String t) {
         String[] tn = t.split("/");
         String lastUpdateTime = getUpdateTime(tn[0]);
         Date nowUpdateTime = new Date();
+        List<Map<String, String>> keyList = dbDao.listKeyMysql(tn[0]);
+        String keyColumn = keyList != null && keyList.size() > 0 ? keyList.get(0).get("columnName").toString() : "id";
         Map<String, Object> map = new HashMap<String, Object>();
-        String sqlWhere = "not exists (select fkId from idc_record where fkId=" + tn[0] + ".id)";
+        String sqlWhere = "not exists (select fkId from idc_record where type='" + tn[0] + "' and fkId=" + tn[0] + "." + keyColumn + ")";
         if (!StringUtils.isEmpty(tn[1])) {
-            sqlWhere += " and ifnull(" + tn[1] + ",now())>=cast('" + lastUpdateTime + "' as datetime)";
+            sqlWhere += " and " + tn[1] + ">=cast('" + lastUpdateTime + "' as datetime)";
         } else {
             map.put("isEnd", "1");
         }
+        if (tn[0].equals("thr_order"))
+            logger.info("-------------------" + sqlWhere);
         map.put("sqlWhere", sqlWhere);
         map.put("tableName", tn[0]);
         map.put("filePathColumn", tn[2]);
@@ -177,15 +191,15 @@ public class IdcServiceImpl implements IdcService {
             Date nextTimePoint = cronSequenceGenerator.next(DateUtil.parseDate(lastUpdateTime));
             send = nextTimePoint.before(nowUpdateTime);
         }
-        //if(send) {
-        BaseResponse result = send(map);
-        if (result.getCode() == 20000) {
-            setUpdateTime(tn[0], DateUtil.formatDate(nowUpdateTime));
+        if (send) {
+            BaseResponse result = send(map);
+            if (result.getCode() == 20000) {
+                setUpdateTime(tn[0], DateUtil.formatDate(nowUpdateTime, "yyyy-MM-dd HH:mm:ss"));
+            }
         }
-        //}
     }
 
-    private String pullData(String url, Map<String, Object> params) {
+    private String fetchData(String url, Map<String, Object> params) {
         OkHttpClient client = new OkHttpClient().newBuilder()
                 .build();
         MediaType mediaType = MediaType.parse("application/json");
@@ -292,13 +306,20 @@ public class IdcServiceImpl implements IdcService {
     /*数据同步，从数据库获取数据下发或上传下级中继服务*/
     @Override
     public BaseResponse send(Map<String, Object> params) {
+
         return send(params.get("messageType").toString(), params.get("tableName").toString(), params);
+
+
     }
 
     @Override
     public BaseResponse send(String messageType, String tableName, Map<String, Object> params) {
-        if (sendOnPage(messageType, tableName, params))
-            return ResultVOUtils.success();
+        try {
+            if (sendOnPage(messageType, tableName, params))
+                return ResultVOUtils.success();
+        } catch (Exception ex) {
+            logger.error(ex.getMessage());
+        }
         return ResultVOUtils.error(9999, "");
     }
 
@@ -348,7 +369,7 @@ public class IdcServiceImpl implements IdcService {
     }
 
     private boolean sendOnPage(String messageType, String tableName, Map<String, Object> params) {
-        boolean success = true;
+        boolean success = false;
         Map<String, String> table = dbDao.getMysql(tableName);
         if (!(table != null && table.get("tableName") != null))
             return false;
@@ -368,9 +389,14 @@ public class IdcServiceImpl implements IdcService {
         String filePathColumn = params.get("filePathColumn") != null ? params.get("filePathColumn").toString() : "";
         orderNum++;
         saveIdcLog(messageType, "", tableName + ">" + where, 0, total);
+        if (tableName.equals("thr_order"))
+            logger.info("tableName-->" + tableName + "-->row-->" + total + "-->sqlWhere-->" + where);
         if (total > 0) {
+            success = true;
             params.put("page", 0);
             params.put("limit", limit);
+            List<Map<String, String>> keyList = dbDao.listKeyMysql(tableName);
+            String keyColumn = keyList != null && keyList.size() > 0 ? keyList.get(0).get("columnName").toString() : "id";
             for (int i = 0; i < Math.ceil(total / limit) + 1; i++) {
                 params.replace("page", i * limit);
                 params.replace("limit", limit);
@@ -387,7 +413,7 @@ public class IdcServiceImpl implements IdcService {
                                 files[m] = list.get(m).get(filePathColumn).toString();
                             }
                             recordSql += !StringUtils.isEmpty(recordSql) ? "," : "";
-                            recordSql += "('" + UUID.randomUUID().toString().replaceAll("-", "") + "','" + tableName + "','" + list.get(m).get("id") + "','" + isEnd + "',now())";
+                            recordSql += "('" + UUID.randomUUID().toString().replaceAll("-", "") + "','" + tableName + "','" + list.get(m).get(keyColumn) + "','" + isEnd + "',now())";
                         }
 
                         orderNum++;
@@ -409,6 +435,7 @@ public class IdcServiceImpl implements IdcService {
                     }
                 }
             }
+
         }
         return success;
     }
@@ -477,6 +504,7 @@ public class IdcServiceImpl implements IdcService {
                         String extra = columnList.get(k).get("extra") != null && columnList.get(k).get("extra").toLowerCase().contains("auto") ? "A" : "N";
                         String attrName = columnList.get(k).get("attrName");
                         String value = list.get(i).get(attrName) != null ? list.get(i).get(attrName).toString() : "";
+                        value = DBAUtils.escape(value);
                         String dataType = columnList.get(k).get("dataType").toLowerCase().contains("char") || columnList.get(k).get("dataType").toLowerCase().contains("text") ?
                                 "C" : columnList.get(k).get("dataType").toLowerCase().contains("date") ? "D" : "N";
                         if (extra.equals("N")) {
@@ -575,44 +603,30 @@ public class IdcServiceImpl implements IdcService {
 
     /*获取转发服务地址，当前值允许单向，只使用参数upper_server_ip*/
     private String getNextHost(String direction) {
-        SystemParamConfigEntity systemParamConfigEntity = getSystemParamConfig("upper_server_ip", "中继上传服务地址", "", "");
-        //direction.equals("U") ? getSystemParamConfig("upper_server_ip","中继上传服务地址" , "",""):
-        //getSystemParamConfig("sync_idc_lower_host","下级(下发)中继服务地址" , "","")	;
-        return systemParamConfigEntity.getParamValue();
+        SystemParamConfigEntity systemParamConfigEntity =
+                direction.equals("U") ? getSystemParamConfig("upper_server_host", "自助平台数据接收服务地址", "", "接收由UDI系统上传的数据") :
+                        direction.equals("I") ? getSystemParamConfig("lower_server_host", "UDI系统数据接收服务地址", "", "接收由自助平台下发的数据") : null;
+        String host = systemParamConfigEntity != null ? systemParamConfigEntity.getParamValue() : "";
+        host = !StringUtils.isEmpty(host) && host.substring(host.length() - 1).equals("/") ? host.substring(0, host.length() - 1) : host;
+        return host;
     }
 
-    /*判断是否最后一级*/
+    /*判断是否上传或下发数据*/
     private boolean isLastLevel(String direction) {
-        SystemParamConfigEntity systemParamConfigEntity =
-                direction.equals("I") ? getSystemParamConfig("sync_idc_top", "是否顶级中继服务(连接自助平台)", "0", "0：否；1：是(是,接收下级上传数据后解析入库)") :
-                        getSystemParamConfig("sync_idc_final", "是否末级中继服务(连接UDI管理系统)", "0", "0：否；1：是(是,接收上级下发数据后解析入库)");
-        return systemParamConfigEntity.getParamValue().equals("0") ? false : true;
-
+        return direction.equals("I") || direction.equals("U") ? true : false;
     }
 
     /*检查当前系统为自助平台（下发）还是UDI系统(上传)，返回传输方向,如果都未设置，默认返回下发*/
     private String getDirection() {
-        SystemParamConfigEntity systemParamConfigEntity = getSystemParamConfig("sync_idc_top", "是否顶级中继服务(连接自助平台)", "0", "0：否；1：是(是,接收下级上传数据后解析入库)");
-        System.out.print(">>>>>>>>>>" + JSON.toJSONString(systemParamConfigEntity) + "\n");
-        if (systemParamConfigEntity.getParamValue().equals("0")) {
-            systemParamConfigEntity = getSystemParamConfig("sync_idc_final", "是否末级中继服务(连接UDI管理系统)", "0", "0：否；1：是(是,接收上级下发数据后解析入库)");
-            if (systemParamConfigEntity.getParamValue().equals("1"))
-                return "U";
-        }
-        return "I";
-
+        SystemParamConfigEntity systemParamConfigEntity = getSystemParamConfig("sync_system_type", "系统类型", "IDC", "UDI(UDI管理系统，由UDI系统往自助平台上传数据),SPS(自助平台，由自助平台往UDI系统下发数据),IDC(中继服务,只接收，转发或暂存数据供下一级服务拉取数据)");
+        return systemParamConfigEntity.getParamValue().equals("UDI") ? "U" : systemParamConfigEntity.getParamValue().equals("SPS") ? "I" : "N";
     }
 
     /*是否需要转发*/
     private boolean isRelay(String direction) {
-        String relayStr = direction.equals("U") ? "sync_upstream_enable" : "sync_downstream_enable";
+        String relayHost = getNextHost(direction);
 
-        SystemParamConfigEntity systemParamConfigEntity = systemParamConfigService.selectByParamKey(relayStr);
-        if (systemParamConfigEntity != null && systemParamConfigEntity.getParamValue() != null &&
-                systemParamConfigEntity.getParamValue().equals("0")) {
-            return false;
-        }
-        return true;
+        return !StringUtils.isEmpty(relayHost);
     }
 
 
@@ -707,7 +721,7 @@ public class IdcServiceImpl implements IdcService {
         String updateTime = "";
         Map<String, Object> map = dbDao.get("select * from idc_status where id='" + id + "'");
         if (map != null && map.get("id") != null) {
-            updateTime = DateUtil.formatDate(IDCUtils.parseDate(map.get("statusTime").toString()));
+            updateTime = DateUtil.formatDate(IDCUtils.parseDate(map.get("statusTime").toString()), "yyyy-MM-dd HH:mm:ss");
         } else {
             updateTime = "2000-01-01 00:00:00";
             String sql = "insert into idc_status (id,statusTime) values ('" + id + "',cast('" + updateTime + "' as datetime))";

src/main/java/com/glxp/api/idc/thread/AsyncFetchTask.java

@@ -0,0 +1,53 @@
+package com.glxp.api.idc.thread;
+
+import com.glxp.api.dao.schedule.ScheduledDao;
+import com.glxp.api.entity.system.ScheduledEntity;
+import com.glxp.api.idc.service.IdcService;
+import com.glxp.api.req.system.ScheduledRequest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.scheduling.annotation.EnableScheduling;
+import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+import org.springframework.scheduling.support.CronTrigger;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+
+@Component
+@EnableScheduling
+public class AsyncFetchTask implements SchedulingConfigurer {
+
+    final Logger logger = LoggerFactory.getLogger(AsyncIdcTask.class);
+
+    @Resource
+    private ScheduledDao scheduledDao;
+
+    @Resource
+    private IdcService idcService;
+
+    @Override
+    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
+        scheduledTaskRegistrar.addTriggerTask(() -> process(),
+                triggerContext -> {
+                    ScheduledRequest scheduledRequest = new ScheduledRequest();
+                    scheduledRequest.setCronName("syncFetch");
+                    logger.info("syncFetch----------------");
+                    ScheduledEntity scheduledEntity = scheduledDao.findScheduled(scheduledRequest);
+                    String cron = scheduledEntity != null ? scheduledEntity.getCron() : "15 * * * * ?";
+
+                    if (cron.isEmpty()) {
+                        logger.error("cron is null");
+                    }
+                    logger.info("syncFetch----------------");
+                    return new CronTrigger(cron).nextExecutionTime(triggerContext);
+                });
+    }
+
+    private void process() {
+        logger.info("syncFetch----process------------");
+
+        idcService.asyncFetchTask();
+    }
+
+}

src/main/java/com/glxp/api/idc/utils/DBAUtils.java

@@ -390,8 +390,8 @@ public class DBAUtils {
     public static String escape(String str) {
         String value = str;
         if (value != null && value.contains("'")) {
-            value = value.replaceAll("'", "AAA12345ZZZ");
-            value = value.replaceAll("AAA12345ZZZ", "''");
+            value = value.replaceAll("'", "CCCBBBAAA12345ZZZYYYXXX");
+            value = value.replaceAll("CCCBBBAAA12345ZZZYYYXXX", "''");
         }
         return value;
     }

src/main/java/com/glxp/api/res/inv/InvRemindMsgResponse.java

@@ -132,4 +132,6 @@ public class InvRemindMsgResponse {
      */
     private String zczbhhzbapzbh;
 
+    private Integer remindCount;
+
 }

src/main/java/com/glxp/api/service/inv/impl/InvRemindMsgServiceImpl.java

@@ -205,6 +205,7 @@ public class InvRemindMsgServiceImpl implements InvRemindMsgService {
         msgEntity.setIgnoreStatus(0); //消息忽略状态修改为不忽略
         msgEntity.setHandleMsg(null); //处理方式重置为空
         msgEntity.setUpdateTime(new Date());
+        msgEntity.setRemindCount(msgEntity.getRemindCount() + 1);
         msgEntity.setNextRemindTime(new Date());
 
         if (null == msgEntity.getId()) {

src/main/resources/mybatis/mapper/basic/UdiRlSupDao.xml

@@ -211,7 +211,7 @@
                 AND (basic_udirel.isDisable is null or basic_udirel.isDisable = false)
             </if>
         </where>
-        ORDER BY basic_udirel.updateTime DESC
+        ORDER BY company_product_relevance.updateTime DESC
     </select>
 
 

src/main/resources/mybatis/mapper/inv/InvRemindMsgDao.xml

@@ -19,6 +19,7 @@
                irm.msg,
                irm.handleMsg,
                irm.remark,
+               irm.remindCount,
                bp.ggxh,
                bp.manufactory,
                bp.zczbhhzbapzbh,

src/main/resources/schemas/schema_v2.1.sql

@@ -103,4 +103,7 @@ CALL Pro_Temp_ColumnWork('sup_product', 'allowNoExpire', 'tinyint', 1);
 CALL Pro_Temp_ColumnWork('sup_product', 'allowNoProduct', 'tinyint', 1);
 CALL Pro_Temp_ColumnWork('sup_product', 'allowNoSerial', 'tinyint', 1);
 
+CALL Pro_Temp_ColumnWork('inv_remind_msg', 'remindCount', 'int', 1);
+
+