focus/ipfcst-console/src/main/java/com/jiayue/ipfcst/console/service/FileAnalysisService.java

package com.jiayue.ipfcst.console.service;

import com.jiayue.ipfcst.common.core.util.DateTimeUtil;
import com.jiayue.ipfcst.common.core.util.NumberUtils;
import com.jiayue.ipfcst.common.data.constant.enums.HolidayTypeEnum;
import com.jiayue.ipfcst.common.data.constant.enums.PredictionModelEnum;
import com.jiayue.ipfcst.common.data.entity.*;
import com.jiayue.ipfcst.common.data.repository.ElectricFieldRepository;
import com.jiayue.ipfcst.common.data.repository.HolidayCalendarRepository;
import com.jiayue.ipfcst.common.data.service.BaseService;
import com.jiayue.ipfcst.fileupload.util.FileUtil;
import lombok.AllArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.time.DateFormatUtils;
import org.springframework.stereotype.Service;

import java.io.*;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.List;

/**
 * 解析电科院NWP和DQ文件
 *
 * @author cuil
 * @version 2.0
 * @since 2018/12/03 11:24
 */

@AllArgsConstructor
@Service
@Slf4j
public class FileAnalysisService extends BaseService {

  ElectricFieldService electricFieldService;

  ForecastPowerShortTermService forecastPowerShortTermService;

  HolidayCalendarRepository holidayCalendarRepository;

  NwpService nwpService;

  FileAnalysisRecordService fileAnalysisRecordService;

  OverHaulPlanService overHaulPlanService;

  SysParameterService sysParameterService;

  ElectricFieldRepository electricFieldRepository;

  HolidayCalendarService holidayCalendarService;

  private final String deleteFilelog = "文件删除异常";

  private final String format = "yyyy-MM-dd HH:mm:ss";

  private final String systemErrorLog = "系统错误: ";

  private final String errorLog = "error";

  private final String analysisFaild = "文件解析失败";
  public static void mvFile(File file, String fdPath) {
    try {
      FileUtils.copyFile(file, new File(fdPath), true);
    } catch (IOException e) {
      log.error("文件移动错误", e);
    }
  }

  /**
   * 创建备份文件夹
   */
  public static String mkDirForTime(String targetRoot, String format) {
    String path;
    File file;
    if (StringUtils.isNotEmpty(format)) {
      long current = System.currentTimeMillis();
      path = DateFormatUtils.format(current, format);
      file = new File(targetRoot + File.separator + path + File.separator);

    } else {
      file = new File(targetRoot + File.separator);
    }

    if (!file.exists() && !file.isFile()) {

      if (file.mkdir()) {
        log.info("已创建文件夹");
      } else {
        log.info("创建文件夹失败,路径：" + file.getPath());
      }
    }
    return file.getPath() + File.separator;
  }

  @SneakyThrows
  public void analysisJob() {
    log.info("-----------------开始执行文件解析任务----------------------");
    long timeD = 15 * 60 * 1000L;
    Long currentDate = DateTimeUtil.getMillisecondsSubDay();//今日凌晨
    boolean flag;
    List<ElectricField> electricFieldList = electricFieldService.getAll();
    List<String> eList = new ArrayList<>();
    for (ElectricField electricField : electricFieldList) {
      eList.add(electricField.getStationCode());
    }
    String path = FileUtil.getDownloadFilePath();
    //downloadFile
    File dirFile = new File(path);
    //判断该目录是否存在，不存在时创建
    if (!dirFile.exists()) {
      dirFile.mkdirs();
    }
    log.info("系统扫描路径【" + dirFile.getPath() + "】");
    for (String stationCode : eList) {
      // downLoadFile/场站编号/new
      String dirFiles = dirFile.getPath() + File.separator + stationCode + File.separator + "new";
      File filePath = new File(dirFiles);
      if (!filePath.exists()) {
        filePath.mkdirs();
      }
      Collection<File> files = FileUtils.listFiles(filePath, new String[]{"RB"}, false);
      String dayStr = new SimpleDateFormat("yyyyMMdd").format(new Date());//当前时间格式化为年月日
      if (files.isEmpty()) {
        for (File file : files) {
          flag = false;
          String fileName = file.getName();
          if (!fileName.contains(dayStr)) {
            try {
              Files.delete(Paths.get(file.getPath()));
              log.info(fileName + "不是当天的文件，删除！");
              break;
            }catch (IOException e2){
              log.info(deleteFilelog,e2);
            }
          }
          if (fileName.length() < 30) {
            //假期文件
            if (file.getName().startsWith("JH")) {
              List<HolidayCalendar> listJq = fileAnalysisJqTerm(file);
              if (listJq.size() > 0) {
                holidayCalendarRepository.deleteAll();
                holidayCalendarRepository.saveAll(listJq);
                flag = true;
              } else {
                flag = false;
              }
            }
            if (file.getName().startsWith("DQ")) {
              try {
                List<ForecastPowerShortTerm> listDq = fileAnalysisShortTerm(file, currentDate, stationCode);
                ForecastPowerShortTerm forecastPowerShortTerm;
                if (listDq.size() > 0) {
                  //如果数据不全，进行补入
                  while (listDq.get(listDq.size() - 1).getForecastTime() < DateTimeUtil.getMillisecondsSubDay() + 4 * 24 * 60 * 60 * 1000 - timeD) {
                    forecastPowerShortTerm = new ForecastPowerShortTerm();
                    forecastPowerShortTerm.setFpValue(listDq.get(listDq.size() - 96).getFpValue());//修正前值
                    forecastPowerShortTerm.setGenDate(new Date()); //装机容量
                    forecastPowerShortTerm.setForecastTime(currentDate);
                    forecastPowerShortTerm.setPredictionModelEnum(PredictionModelEnum.E1);
                    forecastPowerShortTerm.setStationCode(stationCode);
                    listDq.add(forecastPowerShortTerm);
                  }
                  try {
                    //短期数据修正
                    Long startTime = listDq.get(0).getForecastTime();
                    Long endTime = listDq.get(listDq.size() - 1).getForecastTime();//删除相同时间数据
                    forecastPowerShortTermService.deleteBetweenAndGenTime(startTime, endTime, listDq, stationCode);
                    flag = true;
                  } catch (Exception e) {
                    log.error("保存短期数据报错", e);
                    flag = false;
                  }
                } else {
                  log.info(file.getName() + "文件数据内容为空、不能正常解析 、移除该文件、执行数据修正功能");
                  flag = false;
                }
              } catch (Exception e) {
                flag = false;
                log.error("解析DQ文件失败", e);
              }
            }
            if (file.getName().startsWith("NWP")) {
              try {
                List<Nwp> listNwp = fileAnalysisNwp(file, stationCode);
                Nwp nwpData;
                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");
                if (listNwp.size() > 0) {
                  while (listNwp.get(listNwp.size() - 1).getPreTime() < DateTimeUtil.getMillisecondsSubDay() + 4 * 24 * 60 * 60 * 1000 - timeD) {
                    nwpData = new Nwp();
                    long time = 0;
                    try {
                      time = sdf.parse(listNwp.get(listNwp.size() - 96).getPreDate()).getTime() + 24 * 60 * 60 * 1000;
                    } catch (ParseException e) {
                      e.printStackTrace();
                    }
                    nwpData.setFarmId(listNwp.get(listNwp.size() - 96).getFarmId());
                    nwpData.setScDate(listNwp.get(listNwp.size() - 96).getScDate());
                    nwpData.setScTime(listNwp.get(listNwp.size() - 96).getScTime());
                    nwpData.setPreDate(sdf.format(time));
                    nwpData.setPreTime(listNwp.get(listNwp.size() - 1).getPreTime() + timeD);
                    nwpData.setT(listNwp.get(listNwp.size() - 96).getT());//温度
                    nwpData.setRh(listNwp.get(listNwp.size() - 96).getRh());//湿度
                    nwpData.setPressure(listNwp.get(listNwp.size() - 96).getPressure());//气压
                    nwpData.setSwr(listNwp.get(listNwp.size() - 96).getSwr());//辐射
                    nwpData.setLwr(listNwp.get(listNwp.size() - 96).getLwr());//辐射
                    nwpData.setDiffuseRadiation(listNwp.get(listNwp.size() - 96).getDiffuseRadiation());//散接辐射
                    nwpData.setDirectRadiation(listNwp.get(listNwp.size() - 96).getDirectRadiation());//直接辐射

                    nwpData.setSenf(listNwp.get(listNwp.size() - 96).getSenf());//热感通量

                    nwpData.setWs10(listNwp.get(listNwp.size() - 96).getWs10());//10 m 风速
                    nwpData.setWs30(listNwp.get(listNwp.size() - 96).getWs30());//30 m 风速
                    nwpData.setWs50(listNwp.get(listNwp.size() - 96).getWs50());//50 m 风速
                    nwpData.setWs70(listNwp.get(listNwp.size() - 96).getWs70());//70 m 风速
                    nwpData.setWs80(listNwp.get(listNwp.size() - 96).getWs80());//80 m 风速
                    nwpData.setWs90(listNwp.get(listNwp.size() - 96).getWs90());//90 m 风速
                    nwpData.setWs100(listNwp.get(listNwp.size() - 96).getWs100());//100 m 风速
                    nwpData.setWs170(listNwp.get(listNwp.size() - 96).getWs170());//170 m 风速

                    nwpData.setWd10(listNwp.get(listNwp.size() - 96).getWd10());//10 m 风向
                    nwpData.setWd30(listNwp.get(listNwp.size() - 96).getWd30());//30 m 风向
                    nwpData.setWd50(listNwp.get(listNwp.size() - 96).getWd50());//50 m 风向
                    nwpData.setWd70(listNwp.get(listNwp.size() - 96).getWd70());//70 m 风向
                    nwpData.setWd80(listNwp.get(listNwp.size() - 96).getWd80());//80 m 风向
                    nwpData.setWd90(listNwp.get(listNwp.size() - 96).getWd90());//90 m 风向
                    nwpData.setWd100(listNwp.get(listNwp.size() - 96).getWd100());//100 m 风向
                    nwpData.setWd170(listNwp.get(listNwp.size() - 96).getWd170());//170 m 风向
                    nwpData.setStationCode(stationCode);
                    listNwp.add(nwpData);
                  }
                } else {
                  log.info(file.getName() + "文件数据内容为空、不能正常解析 、移除该文件、执行数据修正功能");
                }
                //保存NWP实时数据
                Long startTime = listNwp.get(0).getPreTime();
                Long endTime = listNwp.get(listNwp.size() - 1).getPreTime();//删除相同时间数据
                nwpService.deleteBetweenAndPreTime(startTime, endTime, listNwp, stationCode);
                flag = true;
              } catch (Exception e) {
                log.error("解析NWP文件失败", e);
                flag = false;
              }
            }
            if (flag) {
              //文件解析成功之后，保存记录
              saveFileParsingRecord(file, "1", stationCode);
              //移除文件备份到临时文件下
              moveFile(file);
            } else {
              //文件解析失败之后，保存记录
              saveFileParsingRecord(file, "0", stationCode);
              //移除文件备份到error文件下
              moveFileError(file);
            }
          }
        }
      }
      log.info("-----------------执行文件解析任务完成----------------------");
    }
  }

  /**
   * NWP解析
   *
   * @param file 文件路径
   * @return 样例集合
   */
  private List<Nwp> fileAnalysisNwp(File file, String stationCode) {
    List<Nwp> listNwp = new ArrayList<>();
    if (file.renameTo(file)) {
      InputStreamReader readNwp = null;
      BufferedReader bufferedReaderNwp = null;
      try {
        readNwp = new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8.name());//考虑到编码格式
        bufferedReaderNwp = new BufferedReader(readNwp);
        String lineTxt;
        Nwp nwpData;
        BigDecimal nwpDirectRadiation = new BigDecimal("0.7"); //直接辐射
        BigDecimal nwpDiffuseRadiation = new BigDecimal("0.3"); //散接辐射

        while ((lineTxt = bufferedReaderNwp.readLine()) != null) {
          //NWP文件按照Tab方式截取
          String[] datas = lineTxt.split("\t");
          if (datas.length == 35 && datas[0].startsWith("#")) {
            SimpleDateFormat sdf = new SimpleDateFormat(format);
            nwpData = new Nwp();

            //将截取的文件放入nwpData中
            nwpData.setFarmId(datas[1]);
            nwpData.setScDate(datas[2]);
            nwpData.setScTime(datas[3]);
            nwpData.setPreDate(datas[4]);
            nwpData.setPreTime(sdf.parse(datas[5]).getTime());//采集时间 与 短期预测时间关联
            nwpData.setT(NumberUtils.subtract(new BigDecimal(datas[6]), new BigDecimal("273.15")));//温度

            nwpData.setSenf(new BigDecimal(datas[11]).setScale(2, RoundingMode.HALF_UP));//感热
            nwpData.setSwr(new BigDecimal(datas[9]).setScale(2, RoundingMode.HALF_UP));//短波辐射(相当于总辐射)
            nwpData.setLwr(new BigDecimal(datas[10]).setScale(2, RoundingMode.HALF_UP));//短波辐射(相当于总辐射)
            nwpData.setPressure(new BigDecimal(datas[8]).setScale(2, RoundingMode.HALF_UP));//地表气压
            nwpData.setRh(new BigDecimal(datas[7]).setScale(2, RoundingMode.HALF_UP));//2m相对湿度
            nwpData.setDiffuseRadiation(new BigDecimal(datas[9]).multiply(nwpDiffuseRadiation).setScale(2, RoundingMode.HALF_UP));//散接辐射
            nwpData.setDirectRadiation(new BigDecimal(datas[9]).multiply(nwpDirectRadiation).setScale(2, RoundingMode.HALF_UP));//直接辐射

            nwpData.setWs10(new BigDecimal(datas[19]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs30(new BigDecimal(datas[20]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs50(new BigDecimal(datas[21]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs70(new BigDecimal(datas[22]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs80(new BigDecimal(datas[23]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs90(new BigDecimal(datas[24]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs100(new BigDecimal(datas[25]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWs170(new BigDecimal(datas[26]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd10(new BigDecimal(datas[27]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd30(new BigDecimal(datas[28]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd50(new BigDecimal(datas[29]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd70(new BigDecimal(datas[30]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd80(new BigDecimal(datas[31]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd90(new BigDecimal(datas[32]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd100(new BigDecimal(datas[33]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setWd170(new BigDecimal(datas[34]).setScale(2, RoundingMode.HALF_UP));
            nwpData.setStationCode(stationCode);
            listNwp.add(nwpData);
          }
        }
      } catch (IOException | ParseException | RuntimeException e) {
        log.error(systemErrorLog, e);
        // 进行告警
        File destFile = new File(file.getPath().replaceFirst("new", systemErrorLog));
        if (destFile.exists()) {
          try {
            Files.delete(Paths.get(destFile.getPath()));
          }catch (IOException e2){
            log.info(deleteFilelog,e2);
          }
        }
        try {
          FileUtils.moveFile(file, destFile);
        } catch (IOException e1) {
          log.error(file.getName() + analysisFaild, e);
        }
      } finally {
        close(bufferedReaderNwp, readNwp);
      }
    }
    return listNwp;
  }

  /**
   * 假期解析
   *
   * @param file 文件路径
   * @return 样例集合
   */
  private List<HolidayCalendar> fileAnalysisJqTerm(File file) {
    SimpleDateFormat sdf = new SimpleDateFormat(format);
    List<HolidayCalendar> holidayCalendars = new ArrayList<>();
    // 当文件未被使用时，进行解析上报
    if (file.renameTo(file)) {
      InputStreamReader read = null;
      BufferedReader bufferedReader = null;
      String stringLine;
      HolidayCalendar holiday;
      try {
        read = new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8);
        bufferedReader = new BufferedReader(read);
        while ((stringLine = bufferedReader.readLine()) != null) {
          String[] string_arr = stringLine.split("\t");
          if (string_arr.length == 7 && string_arr[0].startsWith("#")) {
            holiday = new HolidayCalendar();
            holiday.setId(Integer.parseInt(string_arr[1] + ""));
            holiday.setName(string_arr[2] + "");
            holiday.setStartTime(sdf.parse(string_arr[3]).getTime());
            holiday.setEndTime(sdf.parse(string_arr[4]).getTime());
            holiday.setDays(Integer.parseInt(string_arr[5] + ""));
            if (string_arr[6].equals("1")) {
              holiday.setHolidayTypeEnum(HolidayTypeEnum.valueOf(string_arr[6]));
            }
            holidayCalendars.add(holiday);
          }
        }
      } catch (IOException | ParseException | RuntimeException e) {
        log.error(systemErrorLog, e);
        // 进行告警


        File destFile = new File(file.getPath().replaceFirst("new", systemErrorLog));
        if (destFile.exists()) {
          try {
            Files.delete(Paths.get(destFile.getPath()));
          }catch (IOException e2){
            log.info(deleteFilelog,e2);
          }
        }
        try {
          FileUtils.moveFile(file, destFile);
        } catch (IOException e1) {
          log.error(file.getName() + analysisFaild, e);
        }
      } finally {
        close(bufferedReader, read);
      }
    }
    return holidayCalendars;
  }

  /**
   * 短期解析
   *
   * @param file        文件路径
   * @param currentDate 当前时间
   * @return 样例集合
   */
  private List<ForecastPowerShortTerm> fileAnalysisShortTerm(File file, Long currentDate, String stationCode) {
    SimpleDateFormat sdf = new SimpleDateFormat(format);
    List<ForecastPowerShortTerm> forecastPowerShortTerm = new ArrayList<>();
    // 当文件未被使用时，进行解析上报
    if (file.renameTo(file)) {
      InputStreamReader read = null;
      BufferedReader bufferedReader = null;
      String stringLine;
      ForecastPowerShortTerm stf;
      try {
        read = new InputStreamReader(new FileInputStream(file), StandardCharsets.UTF_8);
        bufferedReader = new BufferedReader(read);
        while ((stringLine = bufferedReader.readLine()) != null) {
          String[] string_arr = stringLine.split("\t");

          if (string_arr.length == 4 && string_arr[0].startsWith("#") && StringUtils.isNotEmpty(string_arr[2])) {
              stf = new ForecastPowerShortTerm();
              stf.setFpValue(new BigDecimal(string_arr[3] + ""));
              stf.setForecastTime(sdf.parse(string_arr[2]).getTime());
              stf.setGenDate(new Date(currentDate));
              stf.setPredictionModelEnum(PredictionModelEnum.E1);
              stf.setStationCode(stationCode);
              forecastPowerShortTerm.add(stf);
          }
        }
      } catch (IOException | ParseException | RuntimeException e) {
        log.error(systemErrorLog, e);
        // 进行告警
        File destFile = new File(file.getPath().replaceFirst("new", systemErrorLog));
        if (destFile.exists()) {
          try {
            Files.delete(Paths.get(destFile.getPath()));
          }catch (IOException e2){
            log.info(deleteFilelog,e2);
          }
        }
        try {
          FileUtils.moveFile(file, destFile);
        } catch (IOException e1) {
          log.error(file.getName() +analysisFaild, e);
        }
      } finally {
        close(bufferedReader, read);
      }
    }
    return forecastPowerShortTerm;
  }

  /**
   * 移动文件到临时目录下
   *
   * @param file 文件
   */
  private void moveFile(File file) {
    // 移动文件到处理目录
    File destFile = new File(file.getPath().replaceFirst("new", "backupsTemp"));
    log.info("move file :{}, dest file:{}", file, destFile);
    if (destFile.exists()) {
      try {
        Files.delete(Paths.get(destFile.getPath()));
      }catch (IOException e2){
        log.info(deleteFilelog,e2);
      }
    }
    try {
      FileUtils.moveFile(file, destFile);
    } catch (IOException e) {
      log.error("系统移除文件错误：", e);
      // 进行告警
    }
    moveFileBackups(destFile.getParent());
  }

  /**
   * 移动文件到处理目录
   *
   * @param filePath 文件路径
   */
  private void moveFileBackups(String filePath) {
    String targetRoot = filePath.replaceFirst("backupsTemp", "backups");
    String path = mkDirForTime(targetRoot, null);
    path = mkDirForTime(path, "yyyyMMdd");
    // 移动文件夹内容
    File sourceFile = new File(filePath);
    if (sourceFile.exists()) {
      try {
        File[] files = sourceFile.listFiles();
        if (files != null && files.length > 0) {
          for (File f : files) {
            if (f.renameTo(new File(path + f.getName()))) {
              log.info("move file :{}, dest file:{}", path, f.getName());
            }
          }
        }
      } catch (Exception e) {
        // 进行告警
      }
    }
  }

  /**
   * 移动文件到处理错误目录下
   *
   * @param file 文件
   */
  private void moveFileError(File file) {
    File destFile = new File(file.getPath().replaceFirst("new", systemErrorLog));
    if (destFile.exists()) {
      try {
        Files.delete(Paths.get(destFile.getPath()));
      }catch (IOException e2){
        log.info(deleteFilelog,e2);
      }
    }
    try {
      FileUtils.moveFile(file, destFile);
    } catch (IOException e) {
      log.error(file.getName() + analysisFaild, e);
    }
  }

  /**
   * 保存文件解析记录
   *
   * @param file       文件信息
   * @param fileStatus 文件解析状态 1:表示解析成功 0：解析失败
   */
  private void saveFileParsingRecord(File file, String fileStatus, String staticCode) {
    String fileType;
    if (file.getName().startsWith("JH")) {
      fileType = "JH";//假期文件
    } else if (file.getName().startsWith("DQ")) {
      fileType = "DQ";//短期文件
    } else {
      fileType = "NWP";//NWP文件
    }
    FileAnalysisRecord fileParsingRecord = new FileAnalysisRecord();
    fileParsingRecord.setFileTitle(file.getName());
    fileParsingRecord.setFileType(fileType);
    fileParsingRecord.setFileStatus(fileStatus);
    fileParsingRecord.setFilePath(file.getPath());
    fileParsingRecord.setFileDescription(DateTimeUtil.getStringDate() + " 正常解析文件");
    fileParsingRecord.setStationCode(staticCode);
    fileAnalysisRecordService.save(fileParsingRecord);
  }


  /**
   * 关闭文件流
   *
   * @param bufferedReader 字符数据
   * @param read           字节流
   */
  private void close(BufferedReader bufferedReader, InputStreamReader read) {
    try {
      if (bufferedReader != null) {
        bufferedReader.close();
      }
      if (read != null) {
        read.close();
      }
    } catch (IOException e) {
      log.error("关闭文件流失败:", e);
    }
  }
}