TrackingFlowTask.java

package track.task;

import common.model.CalculationFlow;
import common.model.Contract;
import common.model.TkioFlow;
import common.repository.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.util.CollectionUtils;
import tkio.model.Account;
import tkio.repository.AccountRepository;
import tkio.repository.AppRepository;
import tkio.service.AccountFlowRestrictService;
import tkio.service.FlowService;
import util.Constant;
import util.DateUtil;
import util.StringUtil;

import java.math.BigDecimal;
import java.math.BigInteger;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;

/**
 * @author liyin
 * @description
 * @date
 */
public class TrackingFlowTask {

    private final Logger logger = LoggerFactory.getLogger(TrackingFlowTask.class);

    @Autowired
    private ContractRepository contractRepository;
    @Autowired
    private AccountRepository accountRepository;
    @Autowired
    private AppRepository appRepository;
    /*@Autowired
    private AccountFlowRestrictService accountFlowRestrictService;*/
    @Autowired
    private FlowService flowService;
    @Autowired
    private TkioFlowRepository tkioFlowRepository;
    @Autowired
    private ContractChangeRepository contractChangeRepository;
    @Autowired
    private CalculationFlowRepository calculationFlowRepository;
    @Autowired
    private TkioFlowHistoryRepository tkioFlowHistoryRepository;

    public void syncFlow() {
        List<CalculationFlow> calculationFlows = calculationFlowRepository.findByStatus(0);
        for (CalculationFlow calculationFlow : calculationFlows) {
            List<TkioFlow> tkioFlowList = new ArrayList<>();
            calculationFlow.setStatus(1);
            calculationFlowRepository.save(calculationFlow);
            tkioFlowRepository.deleteByEmail(calculationFlow.getEmail());
            String email = calculationFlow.getEmail();

            //查询用户下所有appkey
            try {
                Account account = accountRepository.findByEmail(email);
                List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
                List<Long> idList = new ArrayList<>();
                for (Account ac : accountList) {
                    idList.add(ac.getId());
                }
                List<String> appkeys = appRepository.findAppkeysNotDebug(idList);
                String appkeyStr = String.join("','", appkeys);
                appkeyStr = "'" + appkeyStr + "'";

                String ago = DateUtil.format(account.getCreateTime(), DateUtil.C_DATE_PATTON_DEFAULT);//查找最早一天的流量
                if (ago == null) {
                    calculationFlow.setStatus(2);
                    calculationFlowRepository.save(calculationFlow);
                    continue;
                }

                int between = 0;
                try {
                    between = DateUtil.daysBetween(ago, DateUtil.getBeforeDays(1)) + 1;
                } catch (ParseException e) {
                    logger.error("强转错误：", e);
                }
                int startInt = 0;
                if (Integer.valueOf(DateUtil.getHH()) > 10) {
                    startInt = -1;
                }
                for (int ii = between; ii > startInt; ii--) {
                    String yesterday = DateUtil.getBeforeDays(ii);//昨日
                    //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
                    BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
                    if (clickNum != null && clickNum.longValue() > 0) {
                        List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
                        if (contracts.size() == 1) {//只有一个合同
                            TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contracts.get(0));
                            if (tkioFlow != null) {
                                tkioFlowList.add(tkioFlow);
                                tkioFlowRepository.save(tkioFlow);
                            }
                        } else {
                            //多个合同
                            //看昨日被哪几个合同包含了
                            List<Contract> correlationContract = new ArrayList<>();
                            for (Contract contract : contracts) {
                                /*if (ContractStatusEnum.CANCEL.getKey().equals(contract.getStatus()) || ContractStatusEnum.SUSPEND.getKey().equals(contract.getStatus())) {
                                    //中止或作废合同处理结束时间，以方便昨日流量的归属计算
                                    ContractChange contractChange = contractChangeRepository.findByContentCode(ContractStatusEnum.CANCEL.getValue(), contract.getContractCode());
                                    if (contractChange != null) {
                                        contract.setEndDate(new DateTime(contractChange.getCreateTime()).toString("yyyy-MM-dd"));
                                    }
                                }*/
                                String startDate = contract.getStartDate();
                                String endDate = contract.getEndDate();
                                if (DateUtil.getDate(yesterday).getTime() >= DateUtil.getDate(startDate).getTime() && DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(endDate).getTime()) {
                                    correlationContract.add(contract);
                                }
                            }
                            //多个合同时，进行排序，如果第一个合同流量超出就要看第二个合同，以此往下
                            if (correlationContract.size() > 1) {
                                //冒泡
                                //第一优先级：合同开始日期，第二优先级，合同编号大小
                                Contract[] contractsArray = new Contract[correlationContract.size()];
                                contractsArray = correlationContract.toArray(contractsArray);

                                contractsArray = orderByContract(contractsArray);

                                for (int i = 0; i < contractsArray.length; i++) {
                                    TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[i]);
                                    if (tkioFlow != null && i < contractsArray.length - 1 && tkioFlow.getCostFlow() != null && tkioFlow.getCostFlow() > 0L) {//处理成本流量，如果超出了，依次算在下一个合同上
                                        clickNum = BigInteger.valueOf(tkioFlow.getCostFlow());
                                        tkioFlow.setCostFlow(null);
                                        if (tkioFlow.getFlow() > 0) {
                                            tkioFlowList.add(tkioFlow);
                                            tkioFlowRepository.save(tkioFlow);
                                        }
                                    } else {
                                        if (tkioFlow != null) {
                                            tkioFlowList.add(tkioFlow);
                                            tkioFlowRepository.save(tkioFlow);
                                        }
                                        break;
                                    }

                                }
                            } else {
                                if (correlationContract.size() == 0) {//昨日不包含在所有合同中
                                    //排序
                                    Contract[] contractsArray = new Contract[contracts.size()];
                                    contractsArray = contracts.toArray(contractsArray);
                                    contractsArray = orderByContract(contractsArray);
                                    //如果昨日日期在第一个合同之前，则归属到第一个合同的成本，其余区间都归属到前一个合同的成本上
                                    for (int i = 0; i < contractsArray.length; i++) {
                                        if (DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(contractsArray[i].getStartDate()).getTime() || i == contractsArray.length - 1) {
                                            int j = 0;

                                            if (i != 0) {
                                                j = i - 1;
                                            }

                                            if (i == contractsArray.length - 1) {
                                                j = i;
                                            }

                                            TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[j]);
                                            if (tkioFlow != null) {
                                                tkioFlowList.add(tkioFlow);
                                                tkioFlowRepository.save(tkioFlow);
                                            }
                                            break;
                                        }
                                    }
                                } else {
                                    TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, correlationContract.get(0));
                                    if (tkioFlow != null) {
                                        tkioFlowList.add(tkioFlow);
                                        tkioFlowRepository.save(tkioFlow);
                                    }
                                }

                            }
                        }
                    }
                }
                if (tkioFlowList.size() > 0) {
                    //tkioFlowRepository.save(tkioFlowList);
                }
                calculationFlow.setStatus(2);
                calculationFlowRepository.save(calculationFlow);
            } catch (Exception e) {
                logger.error("CalculationFlow:Id::" + calculationFlow.getId() + ":全流量同步失败", e);
                calculationFlow.setStatus(3);
                calculationFlowRepository.save(calculationFlow);
            }
        }
    }

    public void task() {
        List<String> emails = contractRepository.findDistinctEmailByPlatform("tkio");

        List<TkioFlow> tkioFlowList = new ArrayList<>();
        String yesterday = DateUtil.getBeforeDays(1);//昨日
        logger.info("昨日流量同步：" + yesterday);
        for (String email : emails) {

            //查询用户下所有appkey
            Account account = accountRepository.findByEmail(email);
            if(account==null||account.getRootParent()==null){
                logger.warn("【单日流量同步】用户不存在：{}",email);
                continue;
            }
            List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
            List<Long> idList = new ArrayList<>();
            for (Account ac : accountList) {
                idList.add(ac.getId());
            }
            List<String> appkeys = appRepository.findAppkeysNotDebug(idList);
            String appkeyStr = String.join("','", appkeys);
            appkeyStr = "'" + appkeyStr + "'";
            if(CollectionUtils.isEmpty(appkeys)){
                logger.warn("【单日流量同步】该用户没有appkey：{}",email);
                continue;
            }
            //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
            BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
            if (clickNum != null && clickNum.longValue() > 0) {
                List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
                if (contracts.size() == 1) {//只有一个合同
                    TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contracts.get(0));
                    if (tkioFlow != null) {
                        tkioFlowList.add(tkioFlow);
                        tkioFlowRepository.save(tkioFlow);
                    }
                } else {//多个合同
                    //看昨日被哪几个合同包含了
                    List<Contract> correlationContract = new ArrayList<>();
                    for (Contract contract : contracts) {
                        /*if (ContractStatusEnum.CANCEL.getKey().equals(contract.getStatus()) || ContractStatusEnum.SUSPEND.getKey().equals(contract.getStatus())) {
                            //中止或作废合同处理结束时间，以方便昨日流量的归属计算
                            ContractChange contractChange = contractChangeRepository.findByContentCode(ContractStatusEnum.CANCEL.getValue(), contract.getContractCode());
                            if(contractChange!=null){
                                contract.setEndDate(new DateTime(contractChange.getCreateTime()).toString("yyyy-MM-dd"));
                            }
                        }*/
                        String startDate = contract.getStartDate();
                        String endDate = contract.getEndDate();
                        if (DateUtil.getDate(yesterday).getTime() >= DateUtil.getDate(startDate).getTime() && DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(endDate).getTime()) {
                            correlationContract.add(contract);
                        }
                    }
                    //多个合同时，进行排序，如果第一个合同流量超出就要看第二个合同，以此往下
                    if (correlationContract.size() > 1) {
                        //冒泡
                        //第一优先级：合同开始日期，第二优先级，合同编号大小
                        Contract[] contractsArray = new Contract[correlationContract.size()];
                        contractsArray = correlationContract.toArray(contractsArray);

                        contractsArray = orderByContract(contractsArray);

                        for (int i = 0; i < contractsArray.length; i++) {
                            TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[i]);
                            if (tkioFlow != null && i < contractsArray.length - 1 && tkioFlow.getCostFlow() != null && tkioFlow.getCostFlow() > 0L) {//处理成本流量，如果超出了，依次算在下一个合同上
                                clickNum = BigInteger.valueOf(tkioFlow.getCostFlow());
                                tkioFlow.setCostFlow(null);
                                if (tkioFlow.getFlow() > 0) {
                                    tkioFlowList.add(tkioFlow);
                                    tkioFlowRepository.save(tkioFlow);
                                }
                            } else {
                                if (tkioFlow != null) {
                                    tkioFlowList.add(tkioFlow);
                                    tkioFlowRepository.save(tkioFlow);
                                }
                                break;
                            }

                        }
                    } else {
                        if (correlationContract.size() == 0) {//昨日不包含在所有合同中
                            //排序
                            Contract[] contractsArray = new Contract[contracts.size()];
                            contractsArray = contracts.toArray(contractsArray);
                            contractsArray = orderByContract(contractsArray);
                            //如果昨日日期在第一个合同之前，则归属到第一个合同的成本，其余区间都归属到前一个合同的成本上
                            for (int i = 0; i < contractsArray.length; i++) {
                                if (DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(contractsArray[i].getStartDate()).getTime() || i == contractsArray.length - 1) {
                                    int j = 0;

                                    if (i != 0) {
                                        j = i - 1;
                                    }

                                    if (i == contractsArray.length - 1) {
                                        j = i;
                                    }

                                    TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[j]);
                                    if (tkioFlow != null) {
                                        tkioFlowList.add(tkioFlow);
                                        tkioFlowRepository.save(tkioFlow);
                                    }
                                    break;
                                }
                            }


                        } else {
                            TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, correlationContract.get(0));
                            if (tkioFlow != null) {
                                tkioFlowList.add(tkioFlow);
                                tkioFlowRepository.save(tkioFlow);
                            }
                        }

                    }
                }
            }


        }
        if (tkioFlowList.size() > 0) {
            //tkioFlowRepository.save(tkioFlowList);
        }

    }

    //排序
    public Contract[] orderByContract(Contract[] contractsArray) {
        for (int i = 0; i < contractsArray.length - 1; i++) {
            for (int j = 0; j < contractsArray.length - 1 - i; j++) {
                String startDate1 = contractsArray[j].getSignedDate() == null ? contractsArray[j].getStartDate() : contractsArray[j].getSignedDate();
                String startDate2 = contractsArray[j + 1].getSignedDate() == null ? contractsArray[j + 1].getStartDate() : contractsArray[j + 1].getSignedDate();
                if (DateUtil.getDate(startDate1).getTime() > DateUtil.getDate(startDate2).getTime()) {
                    Contract temp = contractsArray[j];
                    contractsArray[j] = contractsArray[j + 1];
                    contractsArray[j + 1] = temp;
                } else if (DateUtil.getDate(startDate1).getTime() == DateUtil.getDate(startDate2).getTime()) {

                    String contractCode = StringUtil.matchNumber(contractsArray[j].getContractCode());
                    String contractCodeMin = StringUtil.matchNumber(contractsArray[j + 1].getContractCode());
                    if (Long.valueOf(contractCode) > Long.valueOf(contractCodeMin)) {
                        Contract temp = contractsArray[j];
                        contractsArray[j] = contractsArray[j + 1];
                        contractsArray[j + 1] = temp;
                    }
                }
            }
        }
        return contractsArray;
    }

    public TkioFlow getTkioFlow(BigInteger clickNum, String yesterday, Contract contract) {
        TkioFlow tkioFlow = new TkioFlow();
        tkioFlow.setDs(yesterday);
        tkioFlow.setEmail(contract.getEmail());
        tkioFlow.setContractCode(contract.getContractCode());
        try {
            if (DateUtil.daysBetween(contract.getStartDate(), yesterday) < 0 || DateUtil.daysBetween(contract.getEndDate(), yesterday) > 0) {//昨日日期早于合同开始日期
                //设置为成本流量
                tkioFlow.setCostFlow(clickNum.longValue());
            } else {
                //查看历史总消耗流量是否超出
                BigDecimal totalFlow = tkioFlowRepository.sumFlowByEmailAndContractCode(contract.getEmail(), contract.getContractCode());
                totalFlow = totalFlow == null ? new BigDecimal(0) : totalFlow;
                Double contractTrackFlow = contract.getTrackFlow() * 10000;
                if (contractTrackFlow.longValue() - totalFlow.longValue() - clickNum.longValue() >= 0 || contract.getPriceLevel() == Constant.tkioPriceLevelNotLimit) {
                    tkioFlow.setFlow(clickNum.longValue());
                } else {
                    tkioFlow.setFlow(contractTrackFlow.longValue() - totalFlow.longValue());
                    tkioFlow.setCostFlow(clickNum.longValue() - tkioFlow.getFlow());
                }
            }
            return tkioFlow;
        } catch (ParseException e) {
            logger.error("合同编号-" + contract.getContractCode() + "-同步昨日流量错误：", e);
        }
        return null;
    }


    public void reset(){
        List<String> emails = contractRepository.findDistinctEmailByPlatform("tkio");

        List<TkioFlow> tkioFlowList = new ArrayList<>();
        //String yesterday = DateUtil.getBeforeDays(1);//昨日
        List<String> dateInterval = DateUtil.getDateInterval("2021-05-25", "2021-05-25");
        for (String yesterday : dateInterval) {
            logger.info("昨日流量同步：" + yesterday);
            for (String email : emails) {

                //查询用户下所有appkey
                Account account = accountRepository.findByEmail(email);
                if(account==null||account.getRootParent()==null){
                    logger.warn("【单日流量同步】用户不存在：{}",email);
                    continue;
                }
                List<Account> accountList = accountRepository.findByRootParent(account.getRootParent());
                List<Long> idList = new ArrayList<>();
                for (Account ac : accountList) {
                    idList.add(ac.getId());
                }
                List<String> appkeys = appRepository.findAppkeysNotDebug(idList);
                String appkeyStr = String.join("','", appkeys);
                appkeyStr = "'" + appkeyStr + "'";
                if(CollectionUtils.isEmpty(appkeys)){
                    logger.warn("【单日流量同步】该用户没有appkey：{}",email);
                    continue;
                }
                //BigInteger clickNum = accountFlowRestrictService.getTotalNum(yesterday, yesterday, appkeyStr, "account_track_flow_restrict", "click_sum");
                BigInteger clickNum = flowService.getFlowByAccount(yesterday,yesterday,appkeys);
                if (clickNum != null && clickNum.longValue() > 0) {
                    List<Contract> contracts = contractRepository.findByPlatformAndEmail("tkio", email);
                    if (contracts.size() == 1) {//只有一个合同
                        TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contracts.get(0));
                        if (tkioFlow != null) {
                            tkioFlowList.add(tkioFlow);
                            tkioFlowRepository.save(tkioFlow);
                        }
                    } else {//多个合同
                        //看昨日被哪几个合同包含了
                        List<Contract> correlationContract = new ArrayList<>();
                        for (Contract contract : contracts) {
                        /*if (ContractStatusEnum.CANCEL.getKey().equals(contract.getStatus()) || ContractStatusEnum.SUSPEND.getKey().equals(contract.getStatus())) {
                            //中止或作废合同处理结束时间，以方便昨日流量的归属计算
                            ContractChange contractChange = contractChangeRepository.findByContentCode(ContractStatusEnum.CANCEL.getValue(), contract.getContractCode());
                            if(contractChange!=null){
                                contract.setEndDate(new DateTime(contractChange.getCreateTime()).toString("yyyy-MM-dd"));
                            }
                        }*/
                            String startDate = contract.getStartDate();
                            String endDate = contract.getEndDate();
                            if (DateUtil.getDate(yesterday).getTime() >= DateUtil.getDate(startDate).getTime() && DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(endDate).getTime()) {
                                correlationContract.add(contract);
                            }
                        }
                        //多个合同时，进行排序，如果第一个合同流量超出就要看第二个合同，以此往下
                        if (correlationContract.size() > 1) {
                            //冒泡
                            //第一优先级：合同开始日期，第二优先级，合同编号大小
                            Contract[] contractsArray = new Contract[correlationContract.size()];
                            contractsArray = correlationContract.toArray(contractsArray);

                            contractsArray = orderByContract(contractsArray);

                            for (int i = 0; i < contractsArray.length; i++) {
                                TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[i]);
                                if (tkioFlow != null && i < contractsArray.length - 1 && tkioFlow.getCostFlow() != null && tkioFlow.getCostFlow() > 0L) {//处理成本流量，如果超出了，依次算在下一个合同上
                                    clickNum = BigInteger.valueOf(tkioFlow.getCostFlow());
                                    tkioFlow.setCostFlow(null);
                                    if (tkioFlow.getFlow() > 0) {
                                        tkioFlowList.add(tkioFlow);
                                        tkioFlowRepository.save(tkioFlow);
                                    }
                                } else {
                                    if (tkioFlow != null) {
                                        tkioFlowList.add(tkioFlow);
                                        tkioFlowRepository.save(tkioFlow);
                                    }
                                    break;
                                }

                            }
                        } else {
                            if (correlationContract.size() == 0) {//昨日不包含在所有合同中
                                //排序
                                Contract[] contractsArray = new Contract[contracts.size()];
                                contractsArray = contracts.toArray(contractsArray);
                                contractsArray = orderByContract(contractsArray);
                                //如果昨日日期在第一个合同之前，则归属到第一个合同的成本，其余区间都归属到前一个合同的成本上
                                for (int i = 0; i < contractsArray.length; i++) {
                                    if (DateUtil.getDate(yesterday).getTime() <= DateUtil.getDate(contractsArray[i].getStartDate()).getTime() || i == contractsArray.length - 1) {
                                        int j = 0;

                                        if (i != 0) {
                                            j = i - 1;
                                        }

                                        if (i == contractsArray.length - 1) {
                                            j = i;
                                        }

                                        TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, contractsArray[j]);
                                        if (tkioFlow != null) {
                                            tkioFlowList.add(tkioFlow);
                                            tkioFlowRepository.save(tkioFlow);
                                        }
                                        break;
                                    }
                                }


                            } else {
                                TkioFlow tkioFlow = getTkioFlow(clickNum, yesterday, correlationContract.get(0));
                                if (tkioFlow != null) {
                                    tkioFlowList.add(tkioFlow);
                                    tkioFlowRepository.save(tkioFlow);
                                }
                            }

                        }
                    }
                }


            }
        }


    }
    public static void main(String[] args) throws ParseException {
        String ago = "2020-10-01";
        int between = DateUtil.daysBetween(ago, DateUtil.getBeforeDays(1)) + 1;
        System.out.println(between);
        for (int i = between; i > 0; i--) {
            System.out.println(DateUtil.getBeforeDays(i));
        }
    }
}