/**
 * 
 */
package org.springblade.etl.engine.datax;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.collection.CollectionUtil;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.io.IoUtil;
import cn.hutool.core.io.LineHandler;
import cn.hutool.core.lang.PatternPool;
import cn.hutool.core.util.IdUtil;
import cn.hutool.core.util.ReUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.nacos.common.utils.ExceptionUtil;
import com.jcraft.jsch.Session;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.calcite.shaded.org.apache.commons.io.output.ByteArrayOutputStream;
import org.springblade.core.tool.exception.ServiceException;
import org.springblade.etl.engine.datax.dto.DataxReq;
import org.springblade.etl.engine.datax.dto.DataxRes;
import org.springblade.etl.engine.datax.util.DataxSftpKit;
import org.springblade.etl.entity.DsyJobInstance;
import org.springblade.etl.enums.InstanceStatus;
import org.springblade.etl.properties.DsyProperties;
import org.springblade.etl.properties.DsyProperties.ServerInfo;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.ByteArrayInputStream;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.*;
import java.util.regex.Pattern;

/**
 * @author wookvn
 *  ps -ef | grep /opt/bigdata/datax_data/2023-10-14/fee0b0ceec044e6ba1d14c28b062f1f1.json | grep -v grep | awk '{print $2}' | xargs kill -9
 */
@Slf4j
@Component
public class DataxJobService {
	
	@Autowired
	private DsyProperties dsyConfig;
	
	private final static String SUBMIT_COMMAND = "source /etc/profile && nohup python %s %s %s > %s 2>&1 &";
	private final static String CANCEL_COMMAND = "ps -ef | grep %s | grep -v grep | awk '{print $2}' | xargs -r kill -9";
	private final static Pattern DATAX_PATTERN =  PatternPool.get("Total (\\d+) records,.*?\\| Speed(.*?), (\\d+) records/s \\| Error (\\d+) records,", Pattern.DOTALL);
	
	static final DateTimeFormatter LOCAL_DATE_TIME = DateTimeFormatter.ofPattern("yyyyMMddHHmmss");

	/**
	 * 
	 * @param dataxReq
	 * @return
	 */
	public DataxRes submit(DataxReq dataxReq, ServerInfo serverInfo) {
		List<DataxRes> resultList = this.submit(Arrays.asList(dataxReq), serverInfo);
		return CollectionUtil.isEmpty(resultList) ? null : resultList.get(0);
	}
	
	/**
	 * 
	 * @return
	 * @throws Exception
	 */
	public List<DataxRes> submit(List<DataxReq> dataxDtoList, ServerInfo serverInfo) {
		if(CollectionUtil.isEmpty(dataxDtoList)) {
			return null;
		}
		String dataxTemplateJson = IoUtil.readUtf8(DataxJobService.class.getResourceAsStream("/datax/template.json"));
		if(serverInfo == null) {
			serverInfo = dsyConfig.getSshServers().get(0);
		}
		Session session = DataxSftpKit.getSession(serverInfo);
		List<DataxRes> dataxResList = new ArrayList<>();
		LocalDateTime localDateTime = LocalDateTime.now();
		for(DataxReq dataxReq : dataxDtoList) {
			DataxRes dataxRes = new DataxRes();
			String uuid = IdUtil.fastSimpleUUID();
			Map<String, Object> beanToMap = BeanUtil.beanToMap(dataxReq);
			String dataxConfigJson = StrUtil.format(dataxTemplateJson, beanToMap, false);
			//这里没有使用java.nio.file.Paths 因为windows环境的分隔符是反斜杠，提交到linux环境有问题。  这里全部以linux环境为准
			String dataxExecDir = StrUtil.removeSuffixIgnoreCase(dsyConfig.getDataxExecHome(), "/") + "/" + DateTimeFormatter.ISO_LOCAL_DATE.format(localDateTime);
			//配置文件名
			String dataxConfigName = uuid + ".json";
			String dataxConfigPath = dataxExecDir + "/" + dataxConfigName;
			//日志文件名
			String dataxLogName = uuid + "_" + LOCAL_DATE_TIME.format(localDateTime) + ".log";
			String dataxLogPath = dataxExecDir + "/" + dataxLogName;
			//datax.py路径
			String dataxPy = StrUtil.removeSuffixIgnoreCase(dsyConfig.getDataxHome(), "/") + "/bin/datax.py";
			//增量数值作为参数传递
			String dataVar = "";
//			if(dataxReq.getSyncType() == DataSyncType.INCREMENT) {
//				dataVar = String.format("-p \"-D%s='%s' -D%s='%s'\"", DataxConstants.START_VAR, dataxReq.getStartIncrementData(), DataxConstants.END_VAR, dataxReq.getEndIncrementData());
//			}
			String errorMsg = null;
			try(ByteArrayInputStream bais = new ByteArrayInputStream(dataxConfigJson.getBytes(StandardCharsets.UTF_8));
					ByteArrayOutputStream os = new ByteArrayOutputStream()) {
				log.info("数据同步:datax配置文件:{}-{}\n{}", dataxReq.getJobTableId(), dataxConfigPath, dataxConfigJson);
				//上传配置文件并执行命令
				if(DataxSftpKit.upload(session, dataxExecDir, dataxConfigName, bais)) {
					String destCommand = String.format(SUBMIT_COMMAND, dataxPy, dataxConfigPath, dataVar, dataxLogPath);
					log.info("数据同步:datax命令:{}\n{}", dataxReq.getJobTableId(), destCommand);
					//设置任务开始时间
					dataxRes.setStartTime(new Date());
					String result = DataxSftpKit.execute(session, destCommand, os);
					errorMsg = new String(os.toByteArray(), StandardCharsets.UTF_8);
					log.info("数据同步:datax命令执行结果：{}, error: {}", result, errorMsg);
					dataxRes.setPid(result);
					dataxRes.setConfigName(dataxConfigPath);
					dataxRes.setLogName(dataxLogPath);
				}
			} catch (Exception e) {
				log.error(e.getMessage(), e);
				errorMsg = ExceptionUtil.getAllExceptionMsg(e);
			}
			dataxRes.setSyncType(dataxReq.getSyncType());
			dataxRes.setStartIncrementData(dataxReq.getStartIncrementData());
			dataxRes.setEndIncrementData(dataxReq.getEndIncrementData());
			dataxRes.setErrorMsg(errorMsg);
			dataxRes.setServerIp(serverInfo.getIp());
			dataxRes.setJobTableId(dataxReq.getJobTableId());
			dataxResList.add(dataxRes);
		}
		return dataxResList;
	}
	
	/**
	 * 读取datax的统计信息
	 * @param updateEntity
	 * @param serverInfo
	 * @param dataxLogName
	 */
	public void readMetrics(DsyJobInstance updateEntity, String serverIp, String dataxLogName) {
		List<ServerInfo> sshServers = dsyConfig.getSshServers();
		Optional<ServerInfo> optional = sshServers.stream().filter(item -> item.getIp().equals(serverIp)).findAny();
		if(!optional.isPresent()) {
			throw new ServiceException("未找到该实例所在服务器");
		}
		ServerInfo serverInfo = optional.get();
		InstanceStatus[] instanceStatus = new InstanceStatus[] {InstanceStatus.RUNNING};
		StringBuilder errorStrBuilder = new StringBuilder();
		InputStream is = null;
		try {
			//获取log日志文件流
			Session session = DataxSftpKit.getSession(serverInfo);
			is = DataxSftpKit.download(session, dataxLogName);
			
			IoUtil.readUtf8Lines(is, (LineHandler)line -> {
				//获取异常信息
				if(instanceStatus[0] == InstanceStatus.FAILURE || (instanceStatus[0] == InstanceStatus.RUNNING && StrUtil.containsAny(line, "Exception", "[Errno", "command not found", "datax.py: error", "ERROR StdoutPluginCollector", "java: not found"))) { //有异常
					instanceStatus[0] = InstanceStatus.FAILURE;
					errorStrBuilder.append(line + "\n");
				}
				
				//获取中间统计信息
				if(instanceStatus[0] == InstanceStatus.RUNNING && line.contains("StandAloneJobContainerCommunicator")) {
					//Total 52 records, 356 bytes | Speed 35B/s, 5 records/s | Error 0 records, 0 bytes |  All Task WaitWriterTime 0.000s |  All Task WaitReaderTime 0.000s | Percentage 100.00%
					ReUtil.findAll(DATAX_PATTERN, line, matcher -> {
						int groupCount = matcher.groupCount();
						if(groupCount == 4) {
							updateEntity.setReadRowCount(Long.valueOf(matcher.group(1).trim()));
//							updateEntity.setWriteRowCount(Long.valueOf(matcher.group(1).trim()));
							updateEntity.setDataFlowAvg(matcher.group(2).trim());
							updateEntity.setWriteSpeed(Integer.valueOf(matcher.group(3).trim()));
							updateEntity.setErrorRowCount(Long.valueOf(matcher.group(4).trim()));
							updateEntity.setWriteRowCount(updateEntity.getReadRowCount() - updateEntity.getErrorRowCount());
						}
					});
				}
				//获取结果统计信息
				if(instanceStatus[0] == InstanceStatus.RUNNING  && StrUtil.containsAny(line, "任务启动时刻", "任务结束时刻")) {
					String startTimeStr = line.contains("任务启动时刻") ? line.replace("任务启动时刻", "").replace(": ", "").trim() : null;
					if(StrUtil.isNotBlank(startTimeStr)) {
						updateEntity.setStartTime(DateUtil.parse(startTimeStr, "yyyy-MM-dd HH:mm:ss"));
					}
					if(line.contains("任务结束时刻")) {
						String endTimeStr = line.replace("任务结束时刻", "").replace(": ", "").trim();
						updateEntity.setEndTime(DateUtil.parse(endTimeStr, "yyyy-MM-dd HH:mm:ss"));
						instanceStatus[0] = InstanceStatus.SUCCESS;
					}
				}
			});
		}catch(Exception e) {
			errorStrBuilder.append(ExceptionUtil.getAllExceptionMsg(e));
			instanceStatus[0] = InstanceStatus.FAILURE;
		} finally {
			IoUtil.close(is);
		}
		updateEntity.setErrMsg(errorStrBuilder.toString());
		updateEntity.setInstanceStatus(instanceStatus[0]);
	}
	
	/**
	 * 
	 * @param jobId
	 * @param webInterfaceUrl
	 * @throws Exception
	 */
	public String cancel(Integer instanceId, String dataConfigName, String serverIp) throws Exception {
		List<ServerInfo> sshServers = dsyConfig.getSshServers();
		Optional<ServerInfo> optional = sshServers.stream().filter(item -> item.getIp().equals(serverIp)).findAny();
		if(!optional.isPresent()) {
			throw new ServiceException("未找到该实例所在服务器");
		}
		ServerInfo serverInfo = optional.get();
		Session session = DataxSftpKit.getSession(serverInfo);
		try (ByteArrayOutputStream os = new ByteArrayOutputStream()){
			String destCommand = String.format(CANCEL_COMMAND, dataConfigName);
			log.info("数据同步:取消命令｜{}", destCommand);
			//执行取消命令
			String resp = DataxSftpKit.execute(session, destCommand, os);
			String errorMsg = new String(os.toByteArray(), StandardCharsets.UTF_8);
			log.info("数据同步:执行取消命令完成|instanceId={},result={},errorMsg={}", instanceId, resp, errorMsg);
			return errorMsg;
		} finally {
			//nothing
		}
	}
}
