package com.ycl.task; import com.alibaba.fastjson2.JSONObject; import com.mongodb.client.result.DeleteResult; import com.ycl.feign.UYClient; import com.ycl.platform.domain.entity.TMonitor; import com.ycl.platform.domain.entity.WorkOrder; import com.ycl.platform.domain.param.UY.ImageDetectionParam; import com.ycl.platform.domain.param.UY.MonitorQualifyParam; import com.ycl.platform.domain.param.UY.RecordMetaDSumParam; import com.ycl.platform.domain.param.UY.VideoOnlineParam; import com.ycl.platform.domain.result.SYS.TMonitorResult; import com.ycl.platform.domain.result.UY.ImageDetectionResult; import com.ycl.platform.domain.result.UY.MonitorQualifyResult; import com.ycl.platform.domain.result.UY.RecordMetaDSumResult; import com.ycl.platform.domain.result.UY.VideoOnlineResult; import com.ycl.platform.domain.vo.UpdateOnlineVO; import com.ycl.platform.mapper.TMonitorMapper; import com.ycl.platform.mapper.WorkOrderMapper; import com.ycl.platform.service.UYErrorTypeCheckService; import com.ycl.platform.service.WorkOrderService; import com.ycl.platform.service.YwPointService; import com.ycl.system.domain.SysConfig; import com.ycl.system.mapper.SysConfigMapper; import com.ycl.thread.OnlineCheckThread; import com.ycl.utils.CheckPointUtil; import com.ycl.utils.DateUtils; import constant.ApiConstants; import constant.CheckConstants; import constant.RedisConstant; import enumeration.ErrorType; import enumeration.general.WorkOrderStatusEnum; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; import org.apache.commons.lang3.StringUtils; import org.springframework.beans.BeanUtils; import org.springframework.beans.factory.annotation.Value; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.data.mongodb.core.query.Criteria; import org.springframework.data.mongodb.core.query.Query; import org.springframework.data.redis.core.RedisTemplate; import org.springframework.stereotype.Component; import org.springframework.util.CollectionUtils; import java.text.SimpleDateFormat; import java.util.*; import java.util.concurrent.*; import java.util.function.Function; import java.util.stream.Collectors; //优云对接数据任务 @Slf4j @Component("UYTask") @RequiredArgsConstructor public class UYTask { private final RedisTemplate redisTemplate; private final MongoTemplate mongoTemplate; private final UYClient uyClient; private final UYErrorTypeCheckService uyErrorTypeCheckService; private final YwPointService pointService; private final TMonitorMapper monitorMapper; private final WorkOrderService workOrderService; private final WorkOrderMapper workOrderMapper; private final SysConfigMapper sysConfigMapper; private final CheckPointUtil checkPointUtil; @Value("${youYun.tenantId}") private String tenantId; @Value("${youYun.apikey}") private String apikey; @Value("${youYun.accesskey}") private String accesskey; @Value("${youYun.email}") private String email; @Value("${youYun.passwd}") private String passwd; private static final ExecutorService executorService = new ThreadPoolExecutor(16, 64, 5000, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadPoolExecutor.CallerRunsPolicy() ); // 图像检测 public void imageDetection() { log.info("开始执行图像检测数据同步"); ImageDetectionParam param = new ImageDetectionParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); param.setArealayerno(ApiConstants.AreaNo); JSONObject jsonObject = uyClient.imageDetection(param); if (jsonObject != null) { log.info("数据格式" + jsonObject); Integer statusCode = jsonObject.getInteger("statusCode"); if (ApiConstants.UYSuccessCode.equals(statusCode)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", ImageDetectionResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, ImageDetectionResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); pointService.setDeviceTagByGB(records, CheckConstants.Rule_Category_Video); //存放在mongo中 mongoTemplate.insertAll(records); // 工单生成 uyErrorTypeCheckService.imageDetectionCheck(records); } else { log.error("图像监测数据为空{}", data); } } else { log.error("图像监测数据为空{}", jsonObject); } } else { log.error("图像监测请求失败{}", jsonObject); } } else { log.error("图像监测数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.imageDetectionCheck(mongoTemplate.find(query, ImageDetectionResult.class)); log.info("结束执行图像监测数据同步"); } //一机一档 public void monitorQualifyTask() { //一机一档合格率 // 一机一档注册率 // 档案考核比 log.info("开始执行一机一档合格率数据同步"); MonitorQualifyParam param = new MonitorQualifyParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); JSONObject jsonObject = uyClient.monitorQualify(param); if (jsonObject != null) { String code = jsonObject.getString("code"); if (ApiConstants.UYSuccessCodeStr.equals(code)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", MonitorQualifyResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, MonitorQualifyResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getSerialNumber())) { item.setNo(item.getSerialNumber().getShowValue()); } }); pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Or); pointService.setNew(records); //存放在mongo中 mongoTemplate.insertAll(records); // 工单生成 uyErrorTypeCheckService.monitorQualifyCheck(records); } else { log.error("一机一档合格率数据为空{}", data); } } } else { log.error("一机一档合格率数据为空{}", jsonObject); } } else { log.error("一机一档合格率数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.monitorQualifyCheck(mongoTemplate.find(query, MonitorQualifyResult.class)); log.info("结束一机一档合格率数据同步"); } /** * 点位在线PING检测 * 任务会先执行一次优云同步,然后执行ping检测 * online字段来自于优云,pingOnline为主动ping检测的。存入mongo给数据中心查阅 */ public void pointOnline() throws ExecutionException, InterruptedException { log.info("开始检测点位在线"); Integer times = 2; SysConfig config = new SysConfig(); config.setConfigKey("DAY_OF_POINT_OUTLINE_TIMES"); SysConfig sysConfig = sysConfigMapper.selectConfig(config); if (Objects.nonNull(sysConfig)) { times = Integer.valueOf(sysConfig.getConfigValue()); } else { log.error("请配置离线次数,此次设置为默认值2"); } // 先查出设备IP集合,剔除掉在线情况是未知的,并且只检测正在考核的设备避免多余工单 List monitorList = monitorMapper.getDistinctIP(); //补充错误时间点 Query onlineQuery = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); Map mongoMap = mongoTemplate.find(onlineQuery, TMonitorResult.class) .stream() .collect(Collectors.toMap(TMonitorResult::getNo, Function.identity(), (existing, replacement) -> replacement)); for (TMonitorResult result : monitorList) { TMonitorResult mongoData = mongoMap.get(result.getNo()); if(mongoData!=null){ result.setOffLineTimeStr(mongoData.getOffLineTimeStr()); } } List dataList = new ArrayList<>(48); Integer time = times; List> futureList = monitorList.stream() .map(monitor -> CompletableFuture.supplyAsync(() -> { OnlineCheckThread thread = new OnlineCheckThread(monitor, checkPointUtil, time); return thread.call(); // 假设 OnlineCheckThread 实现了 Callable 接口 }, executorService) .orTimeout(180, TimeUnit.SECONDS) .exceptionally(ex -> { if (ex instanceof TimeoutException) { log.error("任务执行超时:"+monitor.getIp()); } else { log.error("任务执行异常:"+monitor.getIp() +ex); ex.printStackTrace(); } return null; })) .collect(Collectors.toList()); // 等待所有任务完成,但不会无限期等待 CompletableFuture allOf = CompletableFuture.allOf( futureList.toArray(new CompletableFuture[0]) ); try { allOf.get(60, TimeUnit.SECONDS); // 给予额外的5秒来收集结果 } catch (TimeoutException e) { log.error("部分任务未在指定时间内完成"); } catch (Exception e2){ log.error("数据收集异常"+e2); } dataList = futureList.stream() .map(CompletableFuture::join) .filter(Objects::nonNull) .collect(Collectors.toList()); Date now = new Date(); List offLineList = new ArrayList<>(); List onLineList = new ArrayList<>(); //查出数据库纯车辆或纯人脸设备 // List serialNumbers = monitorMapper.selectCarOrFace().stream().map(TMonitor::getSerialNumber).collect(Collectors.toList()); dataList.forEach(item->{ if(item.getPingOnline()) { onLineList.add(item.getIp()); } else if(!item.getPingOnline()) { //筛选出ping离线的设备,更改数据库为离线 offLineList.add(item.getIp()); } }); if(!CollectionUtils.isEmpty(offLineList)) { monitorMapper.batchUpdateOnline(offLineList, now, ApiConstants.UY_OnlineSite_Offline); } if(!CollectionUtils.isEmpty(onLineList)) { monitorMapper.batchUpdateOnline(onLineList, now, ApiConstants.UY_OnlineSite_Online); } //存放到mongo if (!CollectionUtils.isEmpty(dataList)) { List mongoList = new ArrayList<>(); dataList.forEach(item->{ String monitorType = item.getMonitorType(); //同一个设备多个类型每个类型存一条数据,以此在保留数据的情况下区分省厅标签 if(StringUtils.isNotEmpty(monitorType)){ String[] monitors = monitorType.split("/"); for (String type : monitors) { TMonitorResult mongoData = new TMonitorResult(); BeanUtils.copyProperties(item,mongoData); mongoData.setMonitorType(type); if("1".equals(type)){ mongoData.setProvinceTag(mongoData.getProvinceTagVideo()); }else if("2".equals(type)){ mongoData.setProvinceTag(mongoData.getProvinceTagCar()); }else if("3".equals(type)){ mongoData.setProvinceTag(mongoData.getProvinceTagFace()); } mongoList.add(mongoData); } } }); //如果存在之前的数据先删除 Query query = new Query(Criteria.where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, TMonitorResult.class); //存放在mongo中 mongoTemplate.insertAll(mongoList); } //工单(同一IP只生成一个工单) //查询数据库已存在的离线工单获取ip集合,剔除 List ips = workOrderMapper.getOfflineWorkOrder(); List workOrderList = dataList.stream() .filter(item -> Objects.nonNull(item.getWorkOrder()) && (CollectionUtils.isEmpty(ips) || !ips.contains(item.getIp()))) .collect(Collectors.toMap( TMonitorResult::getIp, Function.identity(), (existing, replacement) -> existing // 如果遇到相同的 IP,保留第一个 TMonitorResult 对象 )) .values() .stream() .map(TMonitorResult::getWorkOrder) .collect(Collectors.toList()); if (!CollectionUtils.isEmpty(workOrderList)) { workOrderService.innerAddWorkOrder(workOrderList); } log.info("点位在线监测完成"); } //点位在线率(优云) public void videoOnlineTask() { //视频图像质量 log.info("开始执行点位在线数据同步"); VideoOnlineParam param = new VideoOnlineParam(); param.setPageNum(ApiConstants.PageNo); param.setPageSize(ApiConstants.PageSize); param.setArealayerno(ApiConstants.AreaNo); param.setStatus(ApiConstants.UY_OnlineStatus_All); param.setIcmpStatus(ApiConstants.UY_OnlineStatus_All); JSONObject jsonObject = uyClient.videoOnline(param); if (jsonObject != null) { Integer statusCode = jsonObject.getInteger("statusCode"); if (ApiConstants.UYSuccessCode.equals(statusCode)) { JSONObject data = jsonObject.getJSONObject("data"); if (data != null) { List records = data.getList("records", VideoOnlineResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, VideoOnlineResult.class); //打标签 records.forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video); //存放在mongo中 mongoTemplate.insertAll(records); //更新point表在线状态 Date now = new Date(); List willUpdateList = records.stream().map(item -> { UpdateOnlineVO vo = new UpdateOnlineVO(); vo.setOnline(item.getStatus()); vo.setIp(item.getIpAddr()); vo.setUpdateTime(now); return vo; }).collect(Collectors.toList()); monitorMapper.updateOnlineFromUyOrHk(willUpdateList); //离线生成工单,一个ip只生成一个工单 List workOrders = new ArrayList<>(records.stream() .filter(item -> ApiConstants.UY_OnlineSite_Offline.equals(item.getStatus())) .collect(Collectors.toMap( VideoOnlineResult::getIpAddr, Function.identity(), (existing, replacement) -> existing // 如果遇到相同的 IP,保留第一个(existing) )).values()); uyErrorTypeCheckService.videoOnlineCheck(workOrders); } else { log.error("点位在线结果数据为空{}", data); } } else { log.error("点位在线结果数据为空{}", jsonObject); } } else { log.error("点位在线结果请求失败{}", jsonObject); } } else { log.error("点位在线结果数据为空"); } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.videoOnlineCheck(mongoTemplate.find(query, VideoOnlineResult.class)); log.info("结束执行点位在线数据同步"); } //录像可用 public void recordMetaDSumTask() { //录像可用率 log.info("开始执行录像可用数据同步"); RecordMetaDSumParam param = new RecordMetaDSumParam(); param.setTenantId(tenantId); Calendar instance = Calendar.getInstance(); instance.setTime(new Date()); instance.add(Calendar.DAY_OF_MONTH, -1); SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd"); String yesterday = format.format(instance.getTime()); param.setStatTime(yesterday); JSONObject jsonObject = uyClient.recordMetaDSumList(param); if (jsonObject != null) { if (ApiConstants.UYSuccessCodeStr.equals(jsonObject.getString("code"))) { List records = jsonObject.getList("data", RecordMetaDSumResult.class); if (!CollectionUtils.isEmpty(records)) { //如果今天存在之前的数据先删除 Query query = new Query(Criteria .where("mongoCreateTime").gte(DateUtils.getDayStart(new Date())).lt(DateUtils.getDayEnd(new Date()))); DeleteResult result = mongoTemplate.remove(query, RecordMetaDSumResult.class); records.stream().forEach(item -> { if (Objects.nonNull(item.getDeviceId())) { item.setNo(item.getDeviceId()); } }); //打标签 pointService.setDeviceTagByGB(records,CheckConstants.Rule_Category_Video); //存放在mongo中 mongoTemplate.insertAll(records); // // 工单生成 // uyErrorTypeCheckService.recordMetaDSumCheck(records); } } else { log.error("录像可用数据为空{}", jsonObject); } } // 本地测试 // Query query = new Query(Criteria.where("mongoCreateTime").lt(DateUtils.getDayEnd(new Date()))); // uyErrorTypeCheckService.recordMetaDSumCheck(mongoTemplate.find(query, RecordMetaDSumResult.class)); log.info("结束执行录像可用数据同步"); } }