/**
 * Copyright 2019 DH
 * All right reserved.
 * 项目名称: 大恒泰山系统
 * 创建日期:2023/5/18
 */
package org.springblade.task;

import cn.hutool.core.io.FileUtil;
import cn.hutool.extra.ftp.Ftp;
import cn.hutool.extra.ftp.FtpMode;
import com.google.common.collect.ImmutableList;

import org.apache.commons.net.ftp.FTPFile;
import org.springblade.core.tool.utils.ConcurrentDateFormat;
import org.springblade.modules.baseinfo.rtu.dto.RtuInfoDTO;
import org.springblade.modules.baseinfo.rtu.entity.RtuInfoEntity;
import org.springframework.beans.factory.annotation.Value;
import lombok.extern.slf4j.Slf4j;
import org.glassfish.jersey.internal.guava.ThreadFactoryBuilder;
import org.rocksdb.RocksDB;
import org.rocksdb.RocksDBException;
import org.springblade.core.tool.utils.Func;
import org.springblade.modules.baseinfo.rtu.service.IRtuBaseInfoService;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ValueOperations;
import org.springframework.scheduling.annotation.EnableScheduling;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
import ucar.ma2.Array;
import ucar.ma2.Index;
import ucar.nc2.NetcdfFile;
import ucar.nc2.NetcdfFiles;
import ucar.nc2.Variable;


import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.*;
import java.nio.charset.Charset;
import java.util.*;
import java.util.concurrent.*;

/***
 * Date:2023/5/18
 * Title:面雨量数据处理
 * Description：对本文件的详细描述，原则上不能少于30字
 * @author dylan
 * @version 1.0
 * Remark：认为有必要的其他信息
 */
@Component
@Slf4j
@EnableScheduling
public class SurfDataProcessTask {
	private RocksDB rocksDb;

	private static ThreadFactory publicThreadFactory = new ThreadFactoryBuilder().setNameFormat("surf-thread-pool-%d").build();
	private static ExecutorService publicThreadPool = new ThreadPoolExecutor(1, 1,
		0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<Runnable>(1024), publicThreadFactory, new ThreadPoolExecutor.AbortPolicy());

	@Resource
	private RedisTemplate redisTemplate;

	@Resource
	private IRtuBaseInfoService rtuBaseInfoService;

	@Value("${ftp-config.ip}")
	private String ftpIp;

	@Value("${ftp-config.port}")
	private int port;

	@Value("${ftp-config.username}")
	private String username;

	@Value("${ftp-config.password}")
	private String password;

	@Value("${ftp-config.workdir}")
	private String workdir;

	@Value("${ftp-config.targetdir}")
	private String targetdir;

//	@Value("${rockdb-config.save-path}")
//	private String rockdbSavePath;

//	static {
//		RocksDB.loadLibrary();
//	}
//
//	@PostConstruct
//	public void init() {
//		try {
//			log.info("rockdb path {}", rockdbSavePath);
//			rocksDb = RocksDB.open(rockdbSavePath);
//		} catch (RocksDBException e) {
//			e.printStackTrace();
//		}
//	}

	//@Scheduled(cron = "0 0/30 * * * * ")
	public void surfDataFtpTask() {
		try {
			FtpSurfDataTask task = new FtpSurfDataTask();
			FutureTask<Integer> futureTask = new FutureTask<>(task);
			publicThreadPool.execute(futureTask);
		} catch (Exception e) {
			log.error("{}", e.getMessage());
		}
	}

	private int getLonIndex(float lon, List<Float> lonArray) {
		for (int i = 0; i < lonArray.size(); i++) {
			if (lonArray.get(i) > lon) {
				return i - 1;
			}
		}
		return -1;
	}

	private int getLatIndex(float lat, List<Float> latArray) {
		for (int i = 0; i < latArray.size(); i++) {
			if (latArray.get(i) > lat) {
				return i - 1;
			}
		}
		return -1;
	}

	private class DataProcessTask implements Callable<Integer> {
		private String fileName;
		private String upTm;

		public DataProcessTask(String fileName, String upTm) {
			this.fileName = fileName;
			this.upTm = upTm;
		}

		@Override
		public Integer call() throws Exception {
			try {
				log.info("surf data process task start ***************************");
				log.info("surf data 文件生成时间 {}  文件名 {}",upTm,fileName);

				NetcdfFile netcdfFile = NetcdfFiles.openInMemory(fileName);
				ImmutableList<Variable> variables = netcdfFile.getVariables();
				List<Double> timeArray = new ArrayList<>();
				List<Float> lonArray = new ArrayList<>();
				List<Float> latArray = new ArrayList<>();
				for (Variable v : variables) {
//					log.info("***************************");
//					log.info("{},{},{}", v.getFullName(), v.getDataType().toString(), v.getDimensionsString());
//					log.info("***************************");
					if (v.getFullName().equals("lon")) {
						log.info("lon size {}", v.getSize());
						Array datas = v.read();
						Index index = datas.getIndex();
						int[] shapelist = v.getShape();
						for (int i = 0; i < shapelist[0]; i++) {
							float lonv = datas.getFloat(index.set(i));
							lonArray.add(lonv);
							String key = "lon." + Func.toStr(i);
							rocksDb.put(key.getBytes(), Func.toStr(lonv).getBytes());
						}
					}
				//	log.info("***************************");
					if (v.getFullName().equals("lat")) {
						log.info("lat size {}", v.getSize());
						Array datas = v.read();
						Index index = datas.getIndex();
						int[] shapelist = v.getShape();
						for (int i = 0; i < shapelist[0]; i++) {
							float latv = datas.getFloat(index.set(i));
							latArray.add(latv);
							String key = "lat." + Func.toStr(i);
							rocksDb.put(key.getBytes(), Func.toStr(latv).getBytes());
						}
					}
					//log.info("***************************");
					if (v.getFullName().equals("time")) {
						log.info("time size {}", v.getSize());
						Array datas = v.read();
						Index index = datas.getIndex();
						int[] shapelist = v.getShape();
						for (int i = 0; i < shapelist[0]; i++) {
							double timev = datas.getDouble(index.set(i));
							timeArray.add(timev);
						}
					}
					//log.info("***************************");
					if (v.getFullName().equals("reftime")) {
						log.info("reftime size {}", v.getSize());
					}
					//log.info("***************************");
					if (v.getFullName().equals("Total_precipitation_surface")) {
						log.info("surface size {}", v.getSize());
						Array datas = v.read();
						int[] shapelist = v.getShape();
						Index index = datas.getIndex();
						//按时间
						for (int i = 0; i < shapelist[0]; i++) {
							// 按维度
							for (int n = 0; n < shapelist[1]; n++) {
								// 按经度
								for (int k = 0; k < shapelist[2]; k++) {
									// 按照对应索引获取数据并转换为string类型添加到数组中
									float dval = datas.getFloat(index.set(i, n, k));
									String key = "value." + Func.toStr(i) + "." + Func.toStr(n) + "." + Func.toStr(k);
									rocksDb.put(key.getBytes(), Func.toStr(dval).getBytes());
								}
							}
						}
					}
				}
				//log.info("save to redis *****************************");
				HashOperations<String, String,String> ops = redisTemplate.opsForHash();
				RtuInfoDTO rtuInfoDTO = new RtuInfoDTO();
				List<RtuInfoEntity> rtuInfoEntityList = rtuBaseInfoService.rtuList(rtuInfoDTO);
				for (RtuInfoEntity entity : rtuInfoEntityList) {
					if (entity.getIsRain() != null && entity.getIsRain() == 1) {
						String rediskey = "surf.data."+entity.getRtuCode();
						String lng = entity.getLng();
						String lat = entity.getLat();
						if (lng != null && lat != null) {
							int lonIndex = getLonIndex(Func.toFloat(lng), lonArray);
							int latIndex = getLatIndex(Func.toFloat(lat), latArray);
							if (lonIndex >= 0 && latIndex >= 0) {
								int lonIndex2 = -1;
								if (lonIndex < lonArray.size() - 1) {
									lonIndex2 = lonIndex + 1;
								}
								int latIndex2 = -1;
								if (latIndex < latArray.size() - 1) {
									latIndex2 = latIndex + 1;
								}
								if (lonIndex2 > 0 && latIndex2 > 0) {
									String key = "value." + Func.toStr(0) + "." + Func.toStr(latIndex) + "." + Func.toStr(lonIndex);
									byte[] buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"lefttop.value", new String(buff));
									key = "lon." + Func.toStr(lonIndex);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"lefttop.lon", new String(buff));
									key = "lat." + Func.toStr(latIndex);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"lefttop.lat", new String(buff));
									key = "value." + Func.toStr(0) + "." + Func.toStr(latIndex) + "." + Func.toStr(lonIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"leftbottom.value", new String(buff));
									key = "lon." + Func.toStr(lonIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey, "leftbottom.lon",new String(buff));
									key = "lat." + Func.toStr(latIndex);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"leftbottom.lat", new String(buff));
									key = "value." + Func.toStr(0) + "." + Func.toStr(latIndex2) + "." + Func.toStr(lonIndex);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey, "righttop.value",new String(buff));
									key = "lon." + Func.toStr(lonIndex);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey, "righttop.lon",new String(buff));
									key = "lat." + Func.toStr(latIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"righttop.lat", new String(buff));
									key = "value." + Func.toStr(0) + "." + Func.toStr(latIndex2) + "." + Func.toStr(lonIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"rightbottom.value", new String(buff));
									key = "lon." + Func.toStr(lonIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"rightbottom.lon", new String(buff));
									key = "lat." + Func.toStr(latIndex2);
									buff = rocksDb.get(key.getBytes());
									ops.put(rediskey,"rightbottom.lat", new String(buff));
									ops.put(rediskey,"upTime", upTm);
								}
							}
						}
					}
				}
				log.info("surf data process task end ***************************");
			} catch (Exception e) {
				log.error("surf data process task ",e);
			}
			return 0;
		}
	}


	private class FtpSurfDataTask implements Callable<Integer> {


		@Override
		public Integer call() throws Exception {
			try {
				log.info("ftp task start ***************************");
				ValueOperations<String, String> ops = redisTemplate.opsForValue();
				String lastProcessFile = null;
				String surfDataLastTimeKey = "surfdata.last.file";
				if (redisTemplate.hasKey(surfDataLastTimeKey)) {
					String file = ops.get(surfDataLastTimeKey);
					if (null != file) {
						lastProcessFile = file;
						log.info("last surf data file {}",lastProcessFile);
					}
				}
				Ftp ftp = new Ftp(ftpIp, port, username, password, Charset.forName("utf-8"));
				ftp.setMode(FtpMode.Passive);
				ftp.cd(targetdir);
				Date baseTm = new Date();
				Calendar calendar = Calendar.getInstance();
				String dt = Func.formatDate(calendar.getTime());
				dt = dt.replace("-", "");
				String targetDir = dt;
				boolean dirExist = false;
				if (ftp.exist(targetDir)) {
					//log.info("targetDir {}", targetDir);
					dirExist = true;
					calendar.set(Calendar.HOUR_OF_DAY, 0);
					calendar.set(Calendar.MINUTE, 0);
					calendar.set(Calendar.SECOND, 0);
					baseTm = calendar.getTime();
				} else {
					calendar.add(Calendar.DAY_OF_MONTH, -1);
					dt = Func.formatDate(calendar.getTime());
					dt = dt.replace("-", "");
					targetDir = dt;
					if (ftp.exist(targetDir)) {
						//log.info("targetDir {}", targetDir);
						dirExist = true;
						calendar.set(Calendar.HOUR_OF_DAY, 0);
						calendar.set(Calendar.MINUTE, 0);
						calendar.set(Calendar.SECOND, 0);
						baseTm = calendar.getTime();
					}
				}
				if (dirExist) {
					String lastFile = null;
					FTPFile[] ftpFiles = ftp.lsFiles(targetDir);
					for (FTPFile f : ftpFiles) {
						if (f.isFile()) {
							log.info("file name {}", f.getName());
							//Z_SURF_C_BABJ_20230517120815_P_CMPA_RT_BEHT_0P01_HOR-PRE-2023051712.GRB2
							String fileName = f.getName();
							String[] names = fileName.split("_");
							Date upTime = Func.parse(names[4], ConcurrentDateFormat.of("yyyyMMddHHmmss"));
							if (upTime.after(baseTm)) {
								baseTm = upTime;
								lastFile = fileName;
							}
						}
					}
					//log.info("last file {},{}", targetDir, lastFile);
					if (null != lastFile) {
						log.info("last file {},{}", targetDir, lastFile);
						if (null != lastProcessFile) {
							if (!lastFile.equals(lastProcessFile)) {
								log.info("find new file *********");
								File oldFile = new File(workdir + File.separator + lastProcessFile);
								if (oldFile.exists()) {
									oldFile.delete();
								}
								//找到新的文件，开始处理
								ops.set(surfDataLastTimeKey, lastFile);
								String localFile = workdir + File.separator + lastFile;
								File lFile = FileUtil.file(localFile);
								ftp.download(targetDir, lastFile, lFile);
								log.info("download file  *********");
								DataProcessTask task = new DataProcessTask(localFile, Func.formatDateTime(baseTm));
								FutureTask<Integer> futureTask = new FutureTask<>(task);
								publicThreadPool.execute(futureTask);
							}
						} else {
							log.info("find new file *********");
							//找到新的文件，开始处理
							ops.set(surfDataLastTimeKey, lastFile);
							String localFile = workdir + File.separator + lastFile;
							File lFile = FileUtil.file(localFile);
							ftp.download(targetDir, lastFile, lFile);
							log.info("download file  *********");
							DataProcessTask task = new DataProcessTask(localFile, Func.formatDateTime(baseTm));
							FutureTask<Integer> futureTask = new FutureTask<>(task);
							publicThreadPool.execute(futureTask);
						}
					}
				}
				ftp.close();
				log.info("ftp task end ***************************");
			}catch (Exception e){
				log.error("ftp task ",e);
			}
			return 0;
		}
	}
}
