boat-redis/src/main/java/com/shcd/boat/service/impl/ShorePowerServiceImpl.java

package com.shcd.boat.service.impl;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.shcd.boat.config.RedisCache;
import com.shcd.boat.config.SnowConfig;
import com.shcd.boat.entity.AisShoreLine;
import com.shcd.boat.entity.BoatInfoEntity;
import com.shcd.boat.entity.HkShoreLine;
import com.shcd.boat.entity.ObjCargoDeclareVo;
import com.shcd.boat.entity.ShipRecords;
import com.shcd.boat.entity.ShipSealOffPoints;
import com.shcd.boat.entity.ShorePowerFile;
import com.shcd.boat.entity.ShorePowerinfo;
import com.shcd.boat.entity.SurveillancePointinfo;
import com.shcd.boat.entity.SysEnterprise;
import com.shcd.boat.entity.gh.GHAisInfo;
import com.shcd.boat.entity.ObjCargoDeclare;
import com.shcd.boat.mapper.ShipSealOffPointsMapper;
import com.shcd.boat.mapper.SysEnterpriseMapper;
import com.shcd.boat.service.HaiKangService;
import com.shcd.boat.service.IObjCargoDeclareService;
import com.shcd.boat.service.IShoreLinePointsSpaceService;
import com.shcd.boat.service.IShorePowerFileService;
import com.shcd.boat.service.IShorePowerinfoService;
import com.shcd.boat.service.ISurveillancePointinfoService;
import com.shcd.boat.service.ShorePowerService;
import com.shcd.boat.utils.GhjgRpcUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;


import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;

/**
 * 类名:@Name - ShorePowerServiceImpl
 * 功能描述:@Function Description<>
 * 日期:@Date - 2025/2/19 16:08
 * 作者:@Created by zrj
 */
@Service
@Slf4j
@Transactional
public class ShorePowerServiceImpl implements ShorePowerService {

    @Autowired
    private IObjCargoDeclareService objCargoDeclareService;

    @Autowired
    private IShorePowerinfoService shorePowerinfoService;

    @Autowired
    private IShorePowerFileService shorePowerFileService;

    @Autowired
    private IShoreLinePointsSpaceService shoreLinePointsSpaceService;

    @Autowired
    private ISurveillancePointinfoService surveillancePointinfoService;

    @Autowired
    private HaiKangService haiKangService;

    @Autowired
    private ShipSealOffPointsMapper offPointsMapper;

    @Autowired
    private SysEnterpriseMapper sysEnterpriseMapper;

    @Autowired
    private RedisCache redisUtilTemplate;

    @Autowired
    private SnowConfig snowConfig;

    @Value("${transUrl}")
    private String transUrl;

    @Override
    public void analysisData() {
        try {
            ShorePowerinfo shorePowerinfo = new ShorePowerinfo();
//            LambdaQueryWrapper<ObjCargoDeclare> cargoDeclareLambdaQueryWrapper = new LambdaQueryWrapper<>();
//            cargoDeclareLambdaQueryWrapper.eq(ObjCargoDeclare::getDeleteFlag, 0);
//            List<ObjCargoDeclare> findBoats = objCargoDeclareService.list(cargoDeclareLambdaQueryWrapper);

            //查询货物装卸申报未关账的数据
            List<ObjCargoDeclareVo> findBoats = objCargoDeclareService.getObjCargoDeclareList();

            if (!CollectionUtils.isEmpty(findBoats)) {
                //获取货物装卸的船舶信息
                List<String> mmsis = findBoats.stream().map(ObjCargoDeclareVo::getMmsi).collect(Collectors.toList());
                //获取中远ais，查看货物装卸的船是否存在
                Map<String, Object> exit = isShipExit(mmsis);
                //获取点位信息，超过三个小时的旧的点位信息进行清理
                removePoints();
                //获取所有预警信息
                List<ShorePowerinfo> shorePowerinfoAll = shorePowerinfoService.getShorePowerinfoAll();
                List<String> declareIds = null;
                if (!CollectionUtils.isEmpty(shorePowerinfoAll)){
                    declareIds = shorePowerinfoAll.stream().map(ShorePowerinfo::getDeclareId).collect(Collectors.toList());
                }
                //保存中远ais点位信息
                savePoints(exit);
                //开始进行ais点位与岸线坐标分析
                List<AisShoreLine> aisShoreLineList = analysisAisPoint();
                if (CollectionUtils.isEmpty(aisShoreLineList)){
                    log.info("无ais点位与岸线坐标数据");
                }
//                List<String> shoreLineIds = aisShoreLineList.stream().map(AisShoreLine::getShoreLineId).collect(Collectors.toList());
                log.info("一秒前靠泊的船舶与对应岸线：{}", aisShoreLineList);
                //找到离岸线最近的视频点位，然后调用海康视频抓拍
                List<HkShoreLine> hkShoreLineList = analysisHkPoint();
                //获得视频id，通过id获得视频点位code
                if (CollectionUtils.isEmpty(hkShoreLineList)) {
                    log.info("无海康视频与岸线坐标数据");
                }
                Date startTime = new Date();
                Thread.sleep(1000);
                //保存一秒后中远ais点位信息
                List<String> zyFilterKeysNo = savePoints(exit);
                List<AisShoreLine> aisShoreLineListNew = analysisAisPoint();
                if (CollectionUtils.isEmpty(aisShoreLineListNew) ){
                    log.info("无一秒后中远ais点位信息");
                }
                log.info("一秒后靠泊的船舶与对应岸线：{}", aisShoreLineListNew);
                aisShoreLineListNew.forEach(i -> {
                    aisShoreLineList.forEach(j -> {
                        if (j.getDistance() <= 15 && i.getDistance() > 15 && i.getShoreLineId().equals(j.getShoreLineId()) && i.getMmsi().equals(j.getMmsi())) {
                            //说明该船已经驶出岸线范围，截图抓拍，并查询最早刚驶入范围的开始时间，然后查询岸电数据
                            //todo:调用海康视频流，进行抓拍
                            List<String> xyIds = hkShoreLineList.stream().filter(hk -> i.getShoreLineId().equals(hk.getShorelineid())).map(HkShoreLine::getId).collect(Collectors.toList());
                            List<SurveillancePointinfo> surveillancePointinfoList = null;
                            if (!CollectionUtils.isEmpty(xyIds)){
                                surveillancePointinfoList = surveillancePointinfoService.getSurveillancePointinfoByXyids(xyIds);
                            }
                            List<String> cameraIndexCodes = null;
                            if (!CollectionUtils.isEmpty(surveillancePointinfoList) ){
                                cameraIndexCodes = surveillancePointinfoList.stream().map(SurveillancePointinfo::getCameraIndexCode).collect(Collectors.toList());
                            }

                            List<ShorePowerFile> shorePowerFileList = haiKangService.doOutPicture(cameraIndexCodes);
                            ShorePowerinfo shorePowerinfoDb = shorePowerinfoService.getShorePowerInfoByMMsi(i.getMmsi());
                            shorePowerinfoDb.setWharfName("上海港");
                            //获取该船的岸电情况
                            Date finishSubTime = new Date();
                            try {
                                Thread.sleep(1000);
                            } catch (InterruptedException e) {
                                log.error(e.getMessage());
                            }
                            List<ShipRecords> shipRecordsList = this.getShipRecords(shorePowerinfoDb.getShipBerthingTime(), finishSubTime);
                            if (!CollectionUtils.isEmpty(shorePowerFileList)){
                                ShorePowerFile shorePowerFile = shorePowerFileList.get(0);
                                if (!CollectionUtils.isEmpty(shipRecordsList)) {
                                    List<String> shipNames = findBoats.stream().map(ObjCargoDeclareVo::getShipName).collect(Collectors.toList());
                                    for (ShipRecords shipRecords : shipRecordsList) {
                                        if (shipNames.contains(shipRecords.getShipname())) {
                                            shorePowerinfoDb.setShipDepartureTime(finishSubTime);
                                            shorePowerinfoDb.setIsWarning("0");
                                            shorePowerinfoService.saveOrUpdate(shorePowerinfoDb);
                                            log.info("该条船：" + shipRecords.getShipname() + "有岸电数据");
                                        } else {
                                            String fileId = snowConfig.snowflakeStringId();
                                            shorePowerinfoDb.setShipDepartureTime(finishSubTime);
                                            shorePowerinfoDb.setIsWarning("1");
                                            shorePowerinfoDb.setWarningTime(new Date());
                                            shorePowerFile.setShorePowerFileId(fileId);
                                            shorePowerFile.setShorePowerId(shorePowerinfoDb.getId());
                                            shorePowerFile.setFileId(fileId);
                                            shorePowerFile.setCreateTime(new Date());
                                            shorePowerFile.setDelFlag("0");
                                            shorePowerFile.setCameraIndexCode(cameraIndexCodes.get(0));
                                            shorePowerFileService.insertShorePowerFile(shorePowerFile);
                                            shorePowerinfoService.saveOrUpdate(shorePowerinfoDb);
                                            log.info("该条船：" + shipRecords.getShipname() + "无岸电数据，已预警");
                                        }
                                    }
                                }
                                shoreLinePointsSpaceService.removeByMMsi(i.getMmsi());
                            }

                        } else {
                            //说明该船还在岸线范围，保存该数据
                            //todo:调用海康视频流，进行抓拍
                            List<String> xyIds = hkShoreLineList.stream().filter(hk -> j.getShoreLineId().equals(hk.getShorelineid())).map(HkShoreLine::getId).collect(Collectors.toList());
                            List<SurveillancePointinfo> surveillancePointinfoList = null;
                            if (!CollectionUtils.isEmpty(xyIds)){
                                surveillancePointinfoList = surveillancePointinfoService.getSurveillancePointinfoByXyids(xyIds);
                            }
                            List<String> cameraIndexCodes = null;
                            if (!CollectionUtils.isEmpty(surveillancePointinfoList) ){
                                cameraIndexCodes = surveillancePointinfoList.stream().map(SurveillancePointinfo::getCameraIndexCode).collect(Collectors.toList());
                            }

                            List<ShorePowerFile> shorePowerFileList = haiKangService.doInPicture(cameraIndexCodes);
                            if (CollectionUtils.isEmpty(shorePowerFileList)){
                                log.info("进港无截图");
                            }
                            LambdaQueryWrapper<ShorePowerFile> cargoDeclareLambdaQueryWrapper = new LambdaQueryWrapper<>();
                            cargoDeclareLambdaQueryWrapper.eq(ShorePowerFile::getCameraIndexCode, cameraIndexCodes.get(0));
                            List<ShorePowerFile> shorePowerFileDb = shorePowerFileService.list(cargoDeclareLambdaQueryWrapper);
                            //通过mmsi查询货物装卸申报的船舶名称
                            ObjCargoDeclare objCargoDeclare = objCargoDeclareService.getObjCargoDeclare(i.getMmsi());
                            if (CollectionUtils.isEmpty(shorePowerFileDb)){
                                log.info("未有截图.只截图一次");
                                for (ShorePowerFile shorePowerFile : shorePowerFileList) {
                                    if (!ObjectUtils.isEmpty(objCargoDeclare)) {
                                        SysEnterprise sysEnterprise = sysEnterpriseMapper.selectById(objCargoDeclare.getBusinessUnitId());
                                        String id = snowConfig.snowflakeStringId();
                                        String fileId = snowConfig.snowflakeStringId();
                                        shorePowerinfo.setId(id);
                                        shorePowerinfo.setShipBerthingTime(startTime);
                                        shorePowerinfo.setShipName(objCargoDeclare.getShipName());
                                        shorePowerinfo.setBusinessName(sysEnterprise.getEnterpriseName());
                                        shorePowerinfo.setWharfName("上海港");
                                        shorePowerinfo.setDelFlag("0");
                                        shorePowerinfo.setIsWarning("0");
                                        shorePowerFile.setShorePowerFileId(fileId);
                                        shorePowerFile.setShorePowerId(id);
                                        shorePowerFile.setFileId(fileId);
                                        shorePowerFile.setCreateTime(new Date());
                                        shorePowerFile.setDelFlag("0");
                                        shorePowerFile.setCameraIndexCode(cameraIndexCodes.get(0));
                                        shorePowerinfoService.save(shorePowerinfo);
                                        shorePowerFileService.insertShorePowerFile(shorePowerFile);
                                    } else {
                                        log.info(i.getMmsi() + ",该船舶无货物装卸申报数据");
                                    }
                                }
                            }else {
                                for (ShorePowerFile shorePowerFile : shorePowerFileDb) {
                                    if ("1".equals(shorePowerFile.getInImageStatus())){
                                        if (!ObjectUtils.isEmpty(objCargoDeclare)) {
                                            SysEnterprise sysEnterprise = sysEnterpriseMapper.selectById(objCargoDeclare.getBusinessUnitId());
                                            String id = snowConfig.snowflakeStringId();
                                            shorePowerinfo.setId(id);
                                            shorePowerinfo.setShipBerthingTime(startTime);
                                            shorePowerinfo.setShipName(objCargoDeclare.getShipName());
                                            shorePowerinfo.setBusinessName(sysEnterprise.getEnterpriseName());
                                            shorePowerinfo.setWharfName("上海港");
                                            shorePowerinfo.setDelFlag("0");
                                            shorePowerinfo.setIsWarning("0");
                                            shorePowerinfoService.save(shorePowerinfo);
                                        } else {
                                            log.info(i.getMmsi() + ",该船舶无货物装卸申报数据");
                                        }
                                    }
                                }
                            }
                        }
                    });
                });


                List<ObjCargoDeclareVo> declareList = null;
                if (!CollectionUtils.isEmpty(zyFilterKeysNo)){
                    declareList = findBoats.stream()
                            .filter(boat -> zyFilterKeysNo.contains(boat.getMmsi()))
                            .collect(Collectors.toList());
                }
                if (CollectionUtils.isEmpty(declareList)){
                    log.info("无ais的船与货物装卸申报的船未匹配");
                }
                //无ais获取货物装卸申报的开工时间和完工提交时间
                List<String> finalDeclareIds = declareIds;
                declareList.forEach(i -> {
                    //只有第一次预警，数据库里无数据的情况会进这里
                    if (CollectionUtils.isEmpty(finalDeclareIds)){
                        if (1 == i.getIsFinish()) {
                            //查询岸电接口
                            analysisShorePower(i,shorePowerinfo);
                        }
                    }else {
                        if (1 == i.getIsFinish() && !finalDeclareIds.contains(i.getDeclareNo())) {
                            //查询岸电接口
                            analysisShorePower(i,shorePowerinfo);
                        }
                    }
                });

            }
        } catch (Exception e) {
            log.info(e.getMessage());
        }
    }

    private void analysisShorePower(ObjCargoDeclareVo i, ShorePowerinfo shorePowerinfo) {
        try {
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            log.info(e.getMessage());
        }
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime customDateTime = LocalDateTime.parse(i.getStartWorkTime(), formatter);
        Date date = Date.from(customDateTime.atZone(ZoneId.systemDefault()).toInstant());
        List<ShipRecords> shipRecordsList = this.getShipRecords(date, i.getFinishSubTime(), i.getShipName());
        SysEnterprise sysEnterprise = sysEnterpriseMapper.selectById(i.getBusinessUnitCode());
        shorePowerinfo.setBusinessName(sysEnterprise.getEnterpriseName());
        shorePowerinfo.setWharfName("上海港");
        shorePowerinfo.setDeclareId(i.getDeclareNo());
        if (!CollectionUtils.isEmpty(shipRecordsList)) {
            List<String> shipNames = shipRecordsList.stream().map(ShipRecords::getShipname).collect(Collectors.toList());
            for (ShipRecords shipRecords : shipRecordsList) {
                if (shipNames.contains(i.getShipName())) {
                    shorePowerinfo.setId(snowConfig.snowflakeStringId());
                    shorePowerinfo.setShipName(shipRecords.getShipname());
                    shorePowerinfo.setShipBerthingTime(date);
                    shorePowerinfo.setShipDepartureTime(i.getFinishSubTime());
                    shorePowerinfo.setDelFlag("0");
                    shorePowerinfo.setIsWarning("0");
                    shorePowerinfoService.save(shorePowerinfo);
                    log.info("该条船：" + shipRecords.getShipname() + "有岸电数据");
                }
            }
        } else {
            shorePowerinfo.setId(snowConfig.snowflakeStringId());
            shorePowerinfo.setShipName(i.getShipName());
            shorePowerinfo.setShipBerthingTime(date);
            shorePowerinfo.setShipDepartureTime(i.getFinishSubTime());
            shorePowerinfo.setDelFlag("0");
            shorePowerinfo.setIsWarning("1");
            shorePowerinfo.setWarningTime(new Date());
            shorePowerinfoService.save(shorePowerinfo);
            log.info("该条船：" + i.getShipName() + "无岸电数据，已预警");
        }
    }

    private List<HkShoreLine> analysisHkPoint() {
        return shoreLinePointsSpaceService.analysisHkPoint();
    }

    private void removePoints() {
        List<ShipSealOffPoints> sealOffPointsthis = offPointsMapper.getPointsBySource();
        Date now = new Date();
        if (!CollectionUtils.isEmpty(sealOffPointsthis)) {
            sealOffPointsthis.forEach(i -> {
                long diffMillis = now.getTime() - i.getCreateTime().getTime();
                long diffHours = diffMillis / (1000 * 60 * 60);
                if (diffHours >= 3) {
                    shoreLinePointsSpaceService.removeByMMsi(i.getMmsi());
                }
            });
        }
    }

    private List<String> savePoints(Map<String, Object> exit) {
        List<String> zyFilterKeys;
        List<String> zyFilterKeysNo = null;
        if (!ObjectUtils.isEmpty(exit)){
            //有ais的船
            zyFilterKeys = (List<String>) exit.get("zyFilterKeys");
            //无ais的船
            zyFilterKeysNo = (List<String>) exit.get("zyFilterKeysNo");

            if (!CollectionUtils.isEmpty(zyFilterKeys)) {
                List<GHAisInfo> ghAisInfoList = getGhAisInfo(zyFilterKeys);
                log.info("中远ais数据信息：{}", ghAisInfoList);
                List<ShipSealOffPoints> sealOffPoints = new ArrayList<>();
                for (GHAisInfo ghAisInfo : ghAisInfoList) {
                    ShipSealOffPoints sealOffPointsthis = new ShipSealOffPoints();
                    sealOffPointsthis.setId(snowConfig.snowflakeStringId());
                    sealOffPointsthis.setMmsi(ghAisInfo.getDeviCeid());
                    sealOffPointsthis.setSourceType("中远AIS");
                    sealOffPointsthis.setAisTime(ghAisInfo.getGpsTime());
                    sealOffPointsthis.setPointJson(JSON.toJSONString(ghAisInfo));
                    sealOffPointsthis.setPointGeom(ghAisInfo.getMapx().toString() + " " + ghAisInfo.getMapy().toString());
                    sealOffPoints.add(sealOffPointsthis);
                }
                offPointsMapper.saveZyAll(sealOffPoints, 3857);
                return zyFilterKeys;
            }
        }
        return zyFilterKeysNo;
    }

    private List<ShipRecords> getShipRecords(Date startWorkTime, Date finishSubTime) {
        Integer pageSize = 500;
        JSONObject request = new JSONObject();
        request.put("pageIndex", 1);
        request.put("pageSize", pageSize);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime localDateTime1 = startWorkTime.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
        LocalDateTime localDateTime2 = finishSubTime.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
        String start = localDateTime1.format(formatter);
        String end = localDateTime2.format(formatter);
        request.put("startTime", start);
        request.put("endTime", end);
        log.info("调用连船接口，请求参数：{}", request);
        String shipRecords = GhjgRpcUtils.getShipRecords(transUrl, request);
        JSONObject jsonObject = JSONObject.parseObject(shipRecords);
        List<ShipRecords> list = JSON.parseArray(jsonObject.getJSONArray("rows").toJSONString(), ShipRecords.class);
        return list;
    }

    private List<ShipRecords> getShipRecords(Date startWorkTime, Date finishSubTime, String shipName) {
        Integer pageSize = 500;
        JSONObject request = new JSONObject();
        request.put("pageIndex", 1);
        request.put("pageSize", pageSize);
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        LocalDateTime localDateTime1 = startWorkTime.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
        LocalDateTime localDateTime2 = finishSubTime.toInstant()
                .atZone(ZoneId.systemDefault())
                .toLocalDateTime();
        String start = localDateTime1.format(formatter);
        String end = localDateTime2.format(formatter);
        request.put("startTime", start);
        request.put("endTime", end);
        request.put("shipName", shipName);
        log.info("调用连船接口，请求参数：{}", request);
        String shipRecords = GhjgRpcUtils.getShipRecords(transUrl, request);
        JSONObject jsonObject = JSONObject.parseObject(shipRecords);
        List<ShipRecords> list = JSON.parseArray(jsonObject.getJSONArray("rows").toJSONString(), ShipRecords.class);
        return list;
    }

    public List<AisShoreLine> analysisAisPoint() {
        return shoreLinePointsSpaceService.analysisAisPoint();
    }

    public List<GHAisInfo> getGhAisInfo(List<String> zyFilterKeys) {
        List<GHAisInfo> list = new ArrayList<>();
        zyFilterKeys = zyFilterKeys.stream().map(key -> key + "_zy").collect(Collectors.toList());
        zyFilterKeys.forEach(key -> {
            Object result = redisUtilTemplate.getCacheObject(key);
            if (!ObjectUtils.isEmpty(result)) {
                BoatInfoEntity redisData = JSON.parseObject((String) result, BoatInfoEntity.class);
                GHAisInfo ghAisInfo = redisData.getGhAisInfo();
                list.add(ghAisInfo);
            }
        });
        return list;
    }

    public Map<String, Object> isShipExit(List<String> mmsis) {
        Map<String, Object> map = new HashMap<>();
        if (!CollectionUtils.isEmpty(mmsis)) {
            //todo 1.拿中远AiS
            Collection<String> cacheKeys = redisUtilTemplate.keys("*zy");
            List<String> keys = cacheKeys.stream().collect(Collectors.toList());
            if (!CollectionUtils.isEmpty(keys)) {
                //将中远ais后缀截取
                List<String> zyKeysNo = keys.stream().map(key -> key.replace("_zy", ""))
                        .collect(Collectors.toList());
                List<String> zyFilterKeys = mmsis.stream().filter(key -> zyKeysNo.contains(key)).collect(Collectors.toList());
                List<String> zyFilterKeysNo = mmsis.stream().filter(key -> !zyKeysNo.contains(key)).collect(Collectors.toList());
                List<String> zyFilterKeysNoNull = zyFilterKeysNo.stream().filter(key -> key != null).collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(zyFilterKeys)) {
                    map.put("zyFilterKeys", zyFilterKeys);
                    log.info("中远ais船舶与货物装卸船舶匹配成功:{}", zyFilterKeys);
                    return map;
                } else if (!CollectionUtils.isEmpty(zyFilterKeysNoNull)) {
                    map.put("zyFilterKeysNo", zyFilterKeysNoNull);
                    log.info("中远ais船舶与货物装卸船舶未匹配船舶:{}", zyFilterKeysNoNull);
                    return map;
                }
            }
        }
        log.info("中远ais船舶与货物装卸船舶未匹配");
        return map;
    }
}