PlanServiceImpl.java

package com.luca.pilot.service.impl;

import cn.hutool.core.map.MapUtil;
import cn.hutool.json.JSONUtil;
import com.luca.pilot.bizEnum.TableEnum;
import com.luca.pilot.entity.*;
import com.luca.pilot.service.PlanService;
import com.luca.pilot.util.CrfUtil;
import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateOneModel;
import com.mongodb.client.model.UpdateOptions;
import com.mongodb.client.model.WriteModel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.mongodb.core.BulkOperations;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.mapping.Document;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import javax.annotation.Resource;
import java.time.*;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static org.springframework.data.mongodb.core.query.Criteria.where;

@Slf4j
@Service
public class PlanServiceImpl implements PlanService {

    @Resource(name = "mongoTemplate")
    private MongoTemplate mongoTemplate;

    @Resource(name = "mongoArchiveTemplate")
    private MongoTemplate mongoArchiveTemplate;

    @Override
    @Scheduled(cron ="0 0/5 * * * ?")
    public void synchronizePlanTask() {
        log.info("--------------synchronizePlanTask start---------");
        String nowTime = parseTimeFromDateTime (LocalDateTime.now());

        SynchronizeTime synchronizeTime = findSynchronizeTime();

        String lastSynchronizeTime = "2024-08-28 09:53:24";
        if (Objects.nonNull(synchronizeTime)) {
            lastSynchronizeTime =  synchronizeTime.getPlan();
        }

        Query countQuery = new Query();
        if (Objects.nonNull(lastSynchronizeTime)) {
            countQuery.addCriteria(where("last_modified_at").gt(lastSynchronizeTime));
        }
        long totalPlanTaskDetailsCount = mongoTemplate.count(countQuery, PlanTaskDetails.class);
        log.info("PlanTaskDetails 在大于 {} 时间内未查询到数据 {} 条", lastSynchronizeTime, totalPlanTaskDetailsCount);

        if (totalPlanTaskDetailsCount == 0) {
            return;
        }

        saveSynchronizeTime(synchronizeTime, nowTime);