pollution-warning/src/main/java/com/shanghaichengdi/service/impl/BoatPollutionWarningServiceImpl.java

package com.shanghaichengdi.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.shanghaichengdi.common.enums.*;
import com.shanghaichengdi.common.exception.BizException;
import com.shanghaichengdi.common.properties.GpsStartTimeProperties;
import com.shanghaichengdi.common.properties.WarningProperties;
import com.shanghaichengdi.common.util.LocalDateTimeUtil;
import com.shanghaichengdi.dao.entity.*;
import com.shanghaichengdi.dao.mapper.BoatPollutionWarningMapper;
import com.shanghaichengdi.model.vo.BoatAISVO;
import com.shanghaichengdi.service.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

/**
 * @ClassName BoatPollutionWarningServiceImpl
 * @Author Tevin
 * @DATE 2024/12/27 11:42
 * @Description TODO
 **/
@Slf4j
@Service
public class BoatPollutionWarningServiceImpl extends ServiceImpl<BoatPollutionWarningMapper, BoatPollutionWarning> implements IBoatPollutionWarningService {

    @Resource
    private GpsStartTimeProperties gpsStartTimeProperties;
    @Resource
    private WarningProperties warningProperties;

    @Resource
    private IGpsinfoHistoryService gpsinfoHistoryService;

    @Resource
    private IGpsinfoRealtimeService gpsinfoRealtimeService;

    @Resource
    private IDictDistrictPortService dictDistrictPortService;

    @Resource
    private IBoatPollutionSupervisionService boatPollutionSupervisionService;

    @Resource
    private ISerialService serialService;

    @Resource
    private IBoatWaterPollutantAcceptFService boatWaterPollutantAcceptFService;

    @Resource
    private ICCbShipEepReportRecService ccbShipEepReportRecService;

    /**
     * 处理市内短驳船预警
     */
    public void handleShortBarge() {
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge start==========");
        LocalDateTime handleStartTime = LocalDateTime.now();
        Date maxGpsTime = boatPollutionSupervisionService.getMaxGpsTime(SupervisionTypeEnum.SHORT_BARGE.getStatusCode());
        LocalDateTime startdefaultLocalDateTime = LocalDateTimeUtil.parse(gpsStartTimeProperties.getGpsstartsime());
        Date startTime = Optional.ofNullable(maxGpsTime).orElse(Date.from(startdefaultLocalDateTime.atZone(ZoneId.systemDefault()).toInstant()));
        LocalDateTime startLocalDateTime = startTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
        long diffMillis = Duration.between(startLocalDateTime, LocalDateTime.now()).toMillis();
        if (diffMillis <= 0) {
            return;
        }
        LocalDateTime endLocalDateTime = this.calSuperviseEndTime(startLocalDateTime, diffMillis);
        Date endTime = Date.from(endLocalDateTime.atZone(ZoneId.systemDefault()).toInstant());
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.getShotBargeAis==========");
        //查询开始结束时间段内的市内短驳船AIS历史数据进行分析
//        List<BoatAISVO> boatAISList = gpsinfoHistoryService.getShotBargeAisHistory(startTime, endTime);
        List<BoatAISVO> boatAISList = gpsinfoRealtimeService.getShotBargeAisRealtime(startTime, endTime);
        if (boatAISList.isEmpty()) {
            throw new BizException("未查到船舶AIS数据！");
        }
        // 查询全部有效的监管船舶
        List<BoatPollutionSupervision> boatPollutionSupervisionList = boatPollutionSupervisionService.list();
        if (boatPollutionSupervisionList.isEmpty()) {
            throw new BizException("未查到船舶监管记录！");
        }
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.dictDistrictPortList==========");
        List<DictDistrictPort> dictDistrictPortList = dictDistrictPortService.getDictDistrictPortList();
        Map<String, List<BoatAISVO>> boatAISVOMap = boatAISList.stream().collect(Collectors.groupingBy(BoatAISVO::getDeviceId));
        List<Map<String, List<BoatAISVO>>> boatAISVOMaplist = splitMap(boatAISVOMap, 50);
        boatAISVOMaplist.forEach(subBoatAISVOMap -> CompletableFuture.runAsync(() -> {
            {
                boatAISVOMap.forEach((k, v) -> {
                    //查询船舶最近监管记录
                    List<BoatPollutionSupervision> bpsList = boatPollutionSupervisionList.stream().filter(x -> x.getMmsi().equals(k) && x.getSupervisionState().equals(SupervisionStateEnum.SUPERVISING.getStatusCode())).sorted(Comparator.comparing(BoatPollutionSupervision::getCreatedTime).reversed()).collect(Collectors.toList());
                    LocalDateTime gpsTime = LocalDateTimeUtil.parse(v.get(0).getGpsTime());
                    //进出港申报记录
                    LambdaQueryWrapper<CCbShipEepReportRec> cCbShipEepReportRecLambdaQueryWrapper = new LambdaQueryWrapper<>();
                    if (v.get(0).getShipId() != null) {
                        cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getShipId, v.get(0).getShipId())
                                .le(CCbShipEepReportRec::getReportTime, gpsTime)
                                .ge(CCbShipEepReportRec::getReportTime, gpsTime.minusDays(20))
                                .likeRight(CCbShipEepReportRec::getOrgCode, "29")
                                .orderByDesc(CCbShipEepReportRec::getReportTime);
                    } else if (v.get(0).getDeviceId() != null) {
                        cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getMmsi, v.get(0).getDeviceId())
                                .le(CCbShipEepReportRec::getReportTime, gpsTime)
                                .ge(CCbShipEepReportRec::getReportTime, gpsTime.minusDays(20))
                                .likeRight(CCbShipEepReportRec::getOrgCode, "29")
                                .orderByDesc(CCbShipEepReportRec::getReportTime);
                    } else {
                        cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getShipNameCn, v.get(0).getShipName())
                                .le(CCbShipEepReportRec::getReportTime, gpsTime)
                                .ge(CCbShipEepReportRec::getReportTime, gpsTime.minusDays(20))
                                .likeRight(CCbShipEepReportRec::getOrgCode, "29")
                                .orderByDesc(CCbShipEepReportRec::getReportTime);
                    }
                    List<CCbShipEepReportRec> ccbShipEepReportRecList = ccbShipEepReportRecService.list(cCbShipEepReportRecLambdaQueryWrapper);
                    if (!bpsList.isEmpty()) {
                        BoatPollutionSupervision bps = bpsList.get(0);
                        if (bps.getHasWarning().equals(HasWarningEnum.HAS.getStatusCode())) {
                            //有预警
                            log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.start==========");
                            BoatPollutionWarning boatPollutionWarning = this.getOne(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getSupervisionCode, bps.getCode()));
                            if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.UNDISPOSED.getStatusCode())) {
                                if (gpsTime.isAfter(boatPollutionWarning.getWarningTime().plusDays(5))) {
                                    log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.undisposed.removeWarning==========");
                                    boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                    this.remove(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()));
                                }
                            } else if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.CONFIRM.getStatusCode())) {
                                //查询排放物
                                LambdaQueryWrapper<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFLambdaQueryWrapper = new LambdaQueryWrapper<>();
                                if (v.get(0).getShipName() != null) {
                                    boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                            .eq(BoatWaterPollutantAcceptF::getShipName, v.get(0).getShipName())
                                            .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                            .ge(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                            .le(BoatWaterPollutantAcceptF::getFirstDtm, boatPollutionWarning.getWarningTime().plusDays(5).isAfter(gpsTime) ? gpsTime : boatPollutionWarning.getWarningTime().plusDays(5))
                                            .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                                } else {
                                    boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                            .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                            .ge(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                            .le(BoatWaterPollutantAcceptF::getFirstDtm, boatPollutionWarning.getWarningTime().plusDays(5).isAfter(gpsTime) ? gpsTime : boatPollutionWarning.getWarningTime().plusDays(5))
                                            .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                                }
                                List<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFList = boatWaterPollutantAcceptFService.list(boatWaterPollutantAcceptFLambdaQueryWrapper);
                                if (gpsTime.isAfter(boatPollutionWarning.getWarningTime().plusDays(5))) {
                                    if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.releive.Y5.pollutant_release==========");
                                        boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                .set(BoatPollutionSupervision::getLastOilsewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.OILSEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastSewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.SEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastToxicliquidTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.TOXICLIQUID.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastBadoilTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.BADOIL.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastRubbishTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.RUBBISH.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.POLLUTANT_RELEASE.getStatusCode())
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    } else {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.releive.Y5.warning_exceed==========");
                                        boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.WARNING_EXCEED.getStatusCode())
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    }
                                } else {
                                    if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.releive.N5.pollutant_release==========");
                                        boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                .set(BoatPollutionSupervision::getLastOilsewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.OILSEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastSewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.SEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastToxicliquidTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.TOXICLIQUID.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastBadoilTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.BADOIL.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastRubbishTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.RUBBISH.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.POLLUTANT_RELEASE.getStatusCode())
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    } else {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.releive.N5.updateBoatPollutionSupervision==========");
                                        boatPollutionSupervisionService.lambdaUpdate()
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                    }
                                }
                            } else if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.CANCEL.getStatusCode())) {
                                log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.Warning.overBoatPollutionWarning.cancel.removeWarning==========");
                                this.remove(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()));
                            }
                        } else {
                            //无预警
                            log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.noneWarning.start==========");
                            //查询排放物
                            LambdaQueryWrapper<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFLambdaQueryWrapper = new LambdaQueryWrapper<>();
                            if (v.get(0).getShipName() != null) {
                                boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                        .eq(BoatWaterPollutantAcceptF::getShipName, v.get(0).getShipName())
                                        .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                        .ge(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                        .le(BoatWaterPollutantAcceptF::getFirstDtm, bps.getPredictwarnSwTime().isAfter(gpsTime) ? gpsTime : bps.getPredictwarnSwTime())
                                        .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                            } else {
                                boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                        .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                        .ge(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                        .le(BoatWaterPollutantAcceptF::getFirstDtm, bps.getPredictwarnSwTime().isAfter(gpsTime) ? gpsTime : bps.getPredictwarnSwTime())
                                        .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                            }
                            List<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFList = boatWaterPollutantAcceptFService.list(boatWaterPollutantAcceptFLambdaQueryWrapper);
                            //查询是否到监控周期内的第五天
                            if (gpsTime.isAfter(bps.getStartTime().plusDays(5))) {
                                //是否有排放记录
                                if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                    //结束监管
                                    log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.noneWarning.overBoatPollutionWarning==========");
                                    boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                            .set(BoatPollutionSupervision::getLastOilsewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.OILSEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastSewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.SEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastToxicliquidTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.TOXICLIQUID.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastBadoilTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.BADOIL.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastRubbishTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.RUBBISH.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                } else {
                                    //生成预警
                                    log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.generateBoatPollutionWarning==========");
                                    boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getHasWarning, HasWarningEnum.HAS.getStatusCode())
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getWarningCount, bps.getWarningCount() + 1)
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                    BoatPollutionWarning boatPollutionWarningItem = new BoatPollutionWarning();
                                    boatPollutionWarningItem.setCode(serialService.getCusNumber("bpw"));
                                    boatPollutionWarningItem.setSupervisionCode(bps.getCode());
                                    boatPollutionWarningItem.setBoatName(bps.getBoatName());
                                    boatPollutionWarningItem.setMmsi(bps.getMmsi());
                                    boatPollutionWarningItem.setType(SewageTypeEnum.SEWAGE.getType());
                                    boatPollutionWarningItem.setWarningTime(gpsTime);
                                    boatPollutionWarningItem.setReportCode(ccbShipEepReportRecList.get(0).getReportNo());
                                    boatPollutionWarningItem.setWarningState(WarningStateEnum.WANNING.getStatusCode());
                                    boatPollutionWarningItem.setIsRead(IsReadEnum.UNREAD.getStatusCode());
                                    boatPollutionWarningItem.setDistrict(this.getDistinct(ccbShipEepReportRecList.get(0), dictDistrictPortList));
                                    boatPollutionWarningItem.setBerthName(ccbShipEepReportRecList.get(0).getBerthName());
                                    boatPollutionWarningItem.setRelieveType(RelieveTypeEnum.WARNING.getStatusCode());
                                    boatPollutionWarningItem.setSendmsgFlag(SendMsgFlagEnum.UNSENT.getStatusCode());
                                    boatPollutionWarningItem.setWarnCount(0);
                                    boatPollutionWarningItem.setFilterState(FilterStateEnum.UNDISPOSED.getStatusCode());
                                    this.save(boatPollutionWarningItem);
                                }
                            } else {
                                //是否有排放记录
                                if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                    //结束监管
                                    log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.noneWarning.overBoatPollutionSupervision==========");
                                    boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                            .set(BoatPollutionSupervision::getLastOilsewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.OILSEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastSewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.SEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastToxicliquidTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.TOXICLIQUID.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastBadoilTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.BADOIL.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .set(BoatPollutionSupervision::getLastRubbishTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.RUBBISH.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                } else {
                                    //更新监管
                                    log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge.noneWarning.updateBoatPollutionSupervision==========");
                                    boatPollutionSupervisionService.lambdaUpdate()
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISING.getStatusCode())
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                }
                            }
                        }
                    } else {
                        if (!ccbShipEepReportRecList.isEmpty()) {
                            //新增监管记录
                            log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge saveBoatPollutionSupervision==========");
                            BoatPollutionSupervision boatPollutionSupervision = new BoatPollutionSupervision();
                            boatPollutionSupervision.setCode(serialService.getCusNumber("bps"));
                            boatPollutionSupervision.setMmsi(k);
                            boatPollutionSupervision.setStartTime(gpsTime);
                            boatPollutionSupervision.setGpsTime(gpsTime);
                            boatPollutionSupervision.setIsBreak(IsBreakEnum.VALID_0.getStatusCode());
                            boatPollutionSupervision.setHasWarning(HasWarningEnum.NONE.getStatusCode());
                            boatPollutionSupervision.setReportId(ccbShipEepReportRecList.get(0).getReportId());
                            boatPollutionSupervision.setBerthName(ccbShipEepReportRecList.get(0).getBerthName());
                            boatPollutionSupervision.setBoatName(v.get(0).getShipName());
                            boatPollutionSupervision.setOutCount(0);
                            boatPollutionSupervision.setSupervisionType(SupervisionTypeEnum.SHORT_BARGE.getStatusCode());
                            boatPollutionSupervision.setSupervisionState(SupervisionStateEnum.SUPERVISING.getStatusCode());
                            boatPollutionSupervision.setPredictwarnSwTime(gpsTime.plusDays(Long.parseLong(warningProperties.getPollutionday())));
                            boatPollutionSupervision.setWarningCount(0);
                            boatPollutionSupervisionService.save(boatPollutionSupervision);
                        }

                    }
                });
            }
        }));

        LocalDateTime handleEndTime = LocalDateTime.now();
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge crossTime:[" + Duration.between(handleStartTime, handleEndTime).toMillis() + " milliseconds]==========");
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge end==========");
    }


    /**
     * 处理普通船舶预警
     */
    public void handleOrdinaryShip() {
        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip start==========");
        LocalDateTime handleStartTime = LocalDateTime.now();
        Date maxGpsTime = boatPollutionSupervisionService.getMaxGpsTime(SupervisionTypeEnum.ORDINARY_SHIP_TYPE.getStatusCode());
        LocalDateTime startdefaultLocalDateTime = LocalDateTimeUtil.parse(gpsStartTimeProperties.getGpsstartsime());
        Date startTime = Optional.ofNullable(maxGpsTime).orElse(Date.from(startdefaultLocalDateTime.atZone(ZoneId.systemDefault()).toInstant()));
        LocalDateTime startLocalDateTime = startTime.toInstant().atZone(ZoneId.systemDefault()).toLocalDateTime();
        SimpleDateFormat sdf = new SimpleDateFormat(" yyyy-MM-dd HH:mm:ss ");
        long diffMillis = Duration.between(startLocalDateTime, LocalDateTime.now()).toMillis();
        if (diffMillis <= 0) {
            return;
        }
        LocalDateTime endLocalDateTime = this.calSuperviseEndTime(startLocalDateTime, diffMillis);
        Date endTime = Date.from(endLocalDateTime.atZone(ZoneId.systemDefault()).toInstant());
        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.getShotBargeAis==========");
        //查询开始结束时间段内的市内短驳船AIS历史数据进行分析
        List<BoatAISVO> boatAISList = gpsinfoHistoryService.getOrdinaryShipAisHistory(startTime, endTime);
        if (boatAISList.isEmpty()) {
            throw new BizException("未查到船舶AIS数据！");
        }
        // 查询全部有效的监管船舶
        List<BoatPollutionSupervision> boatPollutionSupervisionList = boatPollutionSupervisionService.list();
        if (boatPollutionSupervisionList.isEmpty()) {
            throw new BizException("未查到船舶监管记录！");
        }
        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.dictDistrictPortList==========");
        List<DictDistrictPort> dictDistrictPortList = dictDistrictPortService.getDictDistrictPortList();
        Map<String, List<BoatAISVO>> boatAISVOMap = boatAISList.stream().collect(Collectors.groupingBy(BoatAISVO::getDeviceId));
        List<Map<String, List<BoatAISVO>>> boatAISVOMaplist = splitMap(boatAISVOMap, 50);
        boatAISVOMaplist.forEach(subBoatAISVOMap -> CompletableFuture.runAsync(() -> {
            {


                boatAISVOMap.forEach((k, v) -> {
                    //查询船舶最近监管记录
                    List<BoatPollutionSupervision> bpsList = boatPollutionSupervisionList.stream().filter(x -> x.getMmsi().equals(k) && x.getSupervisionState().equals(SupervisionStateEnum.SUPERVISING.getStatusCode())).sorted(Comparator.comparing(BoatPollutionSupervision::getCreatedTime).reversed()).collect(Collectors.toList());
                    LocalDateTime gpsTime = LocalDateTimeUtil.parse(v.get(0).getGpsTime());
                    if (!bpsList.isEmpty()) {
                        BoatPollutionSupervision bps = bpsList.get(0);
                        //查询排放物
                        LambdaQueryWrapper<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFLambdaQueryWrapper = new LambdaQueryWrapper<>();
                        if (!v.get(0).getShipName().isEmpty()) {
                            boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getSewageType, v.get(0).getDeviceId())
                                    .eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                    .eq(BoatWaterPollutantAcceptF::getShipName, v.get(0).getShipName())
                                    .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                    .gt(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                    .le(BoatWaterPollutantAcceptF::getFirstDtm, bps.getPredictwarnSwTime().isAfter(gpsTime) ? gpsTime : bps.getPredictwarnSwTime())
                                    .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                        } else {
                            boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getSewageType, v.get(0).getShipName())
                                    .eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                    .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                    .gt(BoatWaterPollutantAcceptF::getFirstDtm, bps.getStartTime())
                                    .le(BoatWaterPollutantAcceptF::getFirstDtm, bps.getPredictwarnSwTime().isAfter(gpsTime) ? gpsTime : bps.getPredictwarnSwTime())
                                    .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                        }
                        List<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFList = boatWaterPollutantAcceptFService.list(boatWaterPollutantAcceptFLambdaQueryWrapper);
                        if (bps.getHasWarning().equals(HasWarningEnum.HAS.getStatusCode())) {
                            //有预警
                            log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.start==========");
                            BoatPollutionWarning boatPollutionWarning = this.getOne(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getSupervisionCode, bps.getCode()));
                            if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.UNDISPOSED.getStatusCode())) {
                                if (gpsTime.isAfter(boatPollutionWarning.getWarningTime().plusDays(5))) {
                                    log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.undisposed.removeWarning==========");
                                    boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                            .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                            .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                            .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                    this.remove(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()));
                                }
                            } else if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.CONFIRM.getStatusCode())) {
                                //判断是否到预警后的第5天
                                if (gpsTime.isAfter(boatPollutionWarning.getWarningTime().plusDays(5))) {
                                    if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                        //有排污
                                        //查询是否出港
                                        boolean isOut = false;
                                        Integer outCount = 0;
                                        LocalDateTime outTime = null;
                                        for (int i = 0; i < v.size(); i++) {
                                            if (v.get(i).getState() == 0) {
                                                outCount++;
                                                if (outCount >= 3) {
                                                    isOut = true;
                                                    outTime = LocalDateTime.parse(v.get(i - 2).getGpsTime());
                                                    break;
                                                }
                                            } else if (v.get(i).getState() == 1) {
                                                outCount = 0;
                                            }
                                        }
                                        if (isOut) {
                                            if (outTime.isBefore(gpsTime)) {
                                                log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.Y5.leave_port==========");
                                                boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                        .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                        .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                        .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                                this.lambdaUpdate()
                                                        .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                        .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                        .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.LEAVE_PORT.getStatusCode())
                                                        .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                            } else {
                                                log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.Y5.pollutant_release2==========");
                                                boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                        .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                        .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                        .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                                this.lambdaUpdate()
                                                        .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                        .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                        .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.POLLUTANT_RELEASE.getStatusCode())
                                                        .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                            }
                                        } else {
                                            log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.Y5.pollutant_release==========");
                                            boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                    .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                    .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                    .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                            this.lambdaUpdate()
                                                    .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                    .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                    .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.POLLUTANT_RELEASE.getStatusCode())
                                                    .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                        }
                                    } else {
                                        //无排污
                                        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.Y5.warning_exceed==========");
                                        boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.WARNING_EXCEED.getStatusCode())
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    }
                                } else {
                                    if (!boatWaterPollutantAcceptFList.isEmpty()) {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.N5.pollutant_release==========");
                                        boatPollutionSupervisionService.lambdaUpdate().set(BoatPollutionSupervision::getEndTime, gpsTime)
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .set(BoatPollutionSupervision::getSupervisionState, SupervisionStateEnum.SUPERVISE_OVER.getStatusCode())
                                                .set(BoatPollutionSupervision::getLastOilsewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.OILSEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastSewageTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.SEWAGE.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastToxicliquidTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.TOXICLIQUID.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastBadoilTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.BADOIL.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .set(BoatPollutionSupervision::getLastRubbishTime, boatWaterPollutantAcceptFList.stream().filter(y -> y.getSewageType().equals(SewageTypeEnum.RUBBISH.getType())).findFirst().map(BoatWaterPollutantAcceptF::getFirstDtm).orElse(null))
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarningState, WarningStateEnum.WANNING_RELIEVE.getStatusCode())
                                                .set(BoatPollutionWarning::getRelieveTime, gpsTime)
                                                .set(BoatPollutionWarning::getRelieveType, RelieveTypeEnum.POLLUTANT_RELEASE.getStatusCode())
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    } else {
                                        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.releive.N5.updateBoatPollutionSupervision==========");
                                        boatPollutionSupervisionService.lambdaUpdate()
                                                .set(BoatPollutionSupervision::getGpsTime, gpsTime)
                                                .eq(BoatPollutionSupervision::getCode, bps.getCode()).update(new BoatPollutionSupervision());
                                        this.lambdaUpdate()
                                                .set(BoatPollutionWarning::getWarnCount, boatPollutionWarning.getWarnCount() + 1)
                                                .eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()).update(new BoatPollutionWarning());
                                    }
                                }
                            } else if (boatPollutionWarning.getFilterState().equals(FilterStateEnum.CANCEL.getStatusCode())) {
                                log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.Warning.overBoatPollutionWarning.removeWarning==========");
                                this.remove(new QueryWrapper<BoatPollutionWarning>().lambda().eq(BoatPollutionWarning::getCode, boatPollutionWarning.getCode()));
                            }
                        }
                    } else {
                        //无预警
                        //查询近三次AIS中是否有一个点在围栏内
                        boolean isIn = false;
                        LocalDateTime inTime = null;
                        for (BoatAISVO boatAISVO : v) {
                            if (boatAISVO.getState() == 1) {
                                inTime = LocalDateTimeUtil.parse(boatAISVO.getGpsTime());
                                isIn = true;
                                break;
                            }
                        }
                        if (isIn) {
                            //有一个点在围栏内
                            //查询进出港申报地是上海的申报记录
                            LambdaQueryWrapper<CCbShipEepReportRec> cCbShipEepReportRecLambdaQueryWrapper = new LambdaQueryWrapper<>();
                            if (v.get(0).getShipId() != null) {
                                cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getShipId, v.get(0).getShipId())
                                        .between(CCbShipEepReportRec::getReportTime, inTime.minusDays(1), inTime)
                                        .eq(CCbShipEepReportRec::getArrivalOrLeave, ArrivalOrLeaveEnum.ARRIVAL.getStatusCode())
                                        .likeRight(CCbShipEepReportRec::getOrgCode, "29").orderByDesc(CCbShipEepReportRec::getReportTime);
                            } else if (v.get(0).getDeviceId() != null) {
                                cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getMmsi, v.get(0).getDeviceId())
                                        .between(CCbShipEepReportRec::getReportTime, inTime.minusDays(1), inTime)
                                        .eq(CCbShipEepReportRec::getArrivalOrLeave, ArrivalOrLeaveEnum.ARRIVAL.getStatusCode())
                                        .likeRight(CCbShipEepReportRec::getOrgCode, "29")
                                        .orderByDesc(CCbShipEepReportRec::getReportTime);
                            } else if (v.get(0).getShipName() != null) {
                                cCbShipEepReportRecLambdaQueryWrapper.eq(CCbShipEepReportRec::getShipNameCn, v.get(0).getShipName())
                                        .between(CCbShipEepReportRec::getReportTime, inTime.minusDays(1), inTime)
                                        .eq(CCbShipEepReportRec::getArrivalOrLeave, ArrivalOrLeaveEnum.ARRIVAL.getStatusCode())
                                        .likeRight(CCbShipEepReportRec::getOrgCode, "29")
                                        .orderByDesc(CCbShipEepReportRec::getReportTime);
                            }
                            List<CCbShipEepReportRec> ccbShipEepReportRecList = ccbShipEepReportRecService.list(cCbShipEepReportRecLambdaQueryWrapper);
                            if (!ccbShipEepReportRecList.isEmpty()) {
                                //查询前5天是否有排放记录
                                LambdaQueryWrapper<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFLambdaQueryWrapper = new LambdaQueryWrapper<>();
                                if (v.get(0).getShipName() != null) {
                                    boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                            .eq(BoatWaterPollutantAcceptF::getShipName, v.get(0).getShipName())
                                            .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                            .gt(BoatWaterPollutantAcceptF::getFirstDtm, gpsTime.minusDays(5))
                                            .le(BoatWaterPollutantAcceptF::getFirstDtm, gpsTime)
                                            .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                                } else {
                                    boatWaterPollutantAcceptFLambdaQueryWrapper.eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getShipName())
                                            .eq(BoatWaterPollutantAcceptF::getMmsi, v.get(0).getDeviceId())
                                            .eq(BoatWaterPollutantAcceptF::getSewageType, SewageTypeEnum.SEWAGE.getType())
                                            .gt(BoatWaterPollutantAcceptF::getFirstDtm, gpsTime.minusDays(5))
                                            .le(BoatWaterPollutantAcceptF::getFirstDtm, gpsTime)
                                            .orderByDesc(BoatWaterPollutantAcceptF::getFirstDtm);
                                }
                                List<BoatWaterPollutantAcceptF> boatWaterPollutantAcceptFList = boatWaterPollutantAcceptFService.list(boatWaterPollutantAcceptFLambdaQueryWrapper);
                                if (boatWaterPollutantAcceptFList.isEmpty()) {
                                    //新增监管记录
                                    log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip saveBoatPollutionSupervision==========");
                                    BoatPollutionSupervision boatPollutionSupervision = new BoatPollutionSupervision();
                                    boatPollutionSupervision.setCode(serialService.getCusNumber("bps"));
                                    boatPollutionSupervision.setMmsi(k);
                                    boatPollutionSupervision.setStartTime(gpsTime);
                                    boatPollutionSupervision.setGpsTime(gpsTime);
                                    boatPollutionSupervision.setIsBreak(IsBreakEnum.VALID_0.getStatusCode());
                                    boatPollutionSupervision.setHasWarning(HasWarningEnum.HAS.getStatusCode());
                                    boatPollutionSupervision.setReportId(ccbShipEepReportRecList.get(0).getReportId());
                                    boatPollutionSupervision.setBerthName(ccbShipEepReportRecList.get(0).getBerthName());
                                    boatPollutionSupervision.setBoatName(v.get(0).getShipName());
                                    boatPollutionSupervision.setOutCount(0);
                                    boatPollutionSupervision.setSupervisionType(SupervisionTypeEnum.ORDINARY_SHIP_TYPE.getStatusCode());
                                    boatPollutionSupervision.setSupervisionState(SupervisionStateEnum.SUPERVISING.getStatusCode());
                                    boatPollutionSupervision.setPredictwarnSwTime(gpsTime.plusDays(Long.parseLong(warningProperties.getPollutionday())));
                                    boatPollutionSupervision.setWarningCount(0);
                                    boatPollutionSupervisionService.save(boatPollutionSupervision);

                                    //生成预警
                                    log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip.generateBoatPollutionWarning==========");
                                    BoatPollutionWarning boatPollutionWarningItem = new BoatPollutionWarning();
                                    boatPollutionWarningItem.setCode(serialService.getCusNumber("obpw"));
                                    boatPollutionWarningItem.setSupervisionCode(boatPollutionSupervision.getCode());
                                    boatPollutionWarningItem.setBoatName(boatPollutionSupervision.getBoatName());
                                    boatPollutionWarningItem.setMmsi(boatPollutionSupervision.getMmsi());
                                    boatPollutionWarningItem.setType(SewageTypeEnum.SEWAGE.getType());
                                    boatPollutionWarningItem.setWarningTime(gpsTime);
                                    boatPollutionWarningItem.setReportCode(ccbShipEepReportRecList.get(0).getReportNo());
                                    boatPollutionWarningItem.setWarningState(WarningStateEnum.WANNING.getStatusCode());
                                    boatPollutionWarningItem.setIsRead(IsReadEnum.UNREAD.getStatusCode());
                                    boatPollutionWarningItem.setDistrict(this.getDistinct(ccbShipEepReportRecList.get(0), dictDistrictPortList));
                                    boatPollutionWarningItem.setBerthName(ccbShipEepReportRecList.get(0).getBerthName());
                                    boatPollutionWarningItem.setRelieveType(RelieveTypeEnum.WARNING.getStatusCode());
                                    boatPollutionWarningItem.setSendmsgFlag(SendMsgFlagEnum.UNSENT.getStatusCode());
                                    boatPollutionWarningItem.setWarnCount(0);
                                    boatPollutionWarningItem.setFilterState(FilterStateEnum.UNDISPOSED.getStatusCode());
                                    this.save(boatPollutionWarningItem);
                                }
                            }

                        }
                    }
                });
            }
        }));
        LocalDateTime handleEndTime = LocalDateTime.now();
        log.info("==========BoatPollutionWarningServiceImpl.handleShortBarge crossTime:[" + Duration.between(handleStartTime, handleEndTime).toMillis() + " milliseconds]==========");
        log.info("==========BoatPollutionWarningServiceImpl.handleOrdinaryShip end==========");

    }

    /**
     * 计算监控时间
     *
     * @param startTime
     * @param diffMillis
     * @return
     */
    private LocalDateTime calSuperviseEndTime(LocalDateTime startTime, long diffMillis) {
        int millsAdd = 0;
        if (diffMillis <= 600000) {
            millsAdd = 600;
        } else if (diffMillis <= 1800000) {
            millsAdd = 1800;
        } else if (diffMillis <= 3600000) {
            millsAdd = 3600;
        } else if (diffMillis <= 5400000) {
            millsAdd = 5400;
        } else if (diffMillis <= 7200000) {
            millsAdd = 7200;
        } else {
            millsAdd = 7200;
        }
        return startTime.plusSeconds(millsAdd);
    }


    /**
     * 获取泊位信息
     *
     * @param cCbShipEepReportRec
     * @param districtPortList
     * @return
     */
    private String getDistinct(CCbShipEepReportRec cCbShipEepReportRec, List<DictDistrictPort> districtPortList) {
//        log.info(report.toString());
        String district = "未知";
        for (int i = 0; i < districtPortList.size(); i++) {
            if (cCbShipEepReportRec.getBerthName().equals(districtPortList.get(i).getBerthname())) {
                district = districtPortList.get(i).getTeam() + "-" + districtPortList.get(i).getLochus();
//                log.info("匹配到内河泊位，所属区县大队为："+district);
            }
        }

        // 进出港类型 0进港 1出港
        if (cCbShipEepReportRec.getArrivalOrLeave() == 0) {
            if (cCbShipEepReportRec.getReportPortCode().indexOf("31") == 0) {
                // 受理机构是01开头代表上海外港
                if (cCbShipEepReportRec.getOrgCode().indexOf("01") == 0) {
                    district = "上海海事局辖区";
//                    log.info("匹配到进港申报到 "+district);
                }
            }
        } else {
            if (cCbShipEepReportRec.getNextPortCode().indexOf("31") == 0) {
                // 受理机构是01开头代表上海外港
                if (cCbShipEepReportRec.getOrgCode().indexOf("01") == 0) {
                    district = "上海海事局辖区";
//                    log.info("匹配到出港申报到 "+district);
                }
            }
        }
        return district;
    }

    private List<Map<String, List<BoatAISVO>>> splitMap(Map<String, List<BoatAISVO>> map, int splitSize) {
        List<Map<String, List<BoatAISVO>>> result = new ArrayList<>();
        int count = 0;
        Map<String, List<BoatAISVO>> subMap = new HashMap<>();

        for (Map.Entry<String, List<BoatAISVO>> entry : map.entrySet()) {
            subMap.put(entry.getKey(), entry.getValue());
            count++;

            if (count == splitSize) {
                result.add(subMap);
                subMap = new HashMap<>();
                count = 0;
            }
        }

        if (!subMap.isEmpty()) {
            result.add(subMap);
        }

        return result;
    }
}