package com.product.data.center.service;


import cn.hutool.core.date.DateUtil;
import cn.hutool.core.date.TimeInterval;
import cn.hutool.core.util.ArrayUtil;
import cn.hutool.core.util.NumberUtil;
import com.alibaba.fastjson.JSONObject;
import com.product.core.cache.util.RedisUtil;
import com.product.core.config.Global;
import com.product.core.dao.BaseDao;
import com.product.core.entity.DataTableEntity;
import com.product.core.entity.FieldSetEntity;
import com.product.core.exception.BaseException;
import com.product.core.spring.context.SpringMVCContextHolder;
import com.product.data.center.config.CmnConst;
import com.product.data.center.config.ErrorCode;
import com.product.data.center.entity.ExtractUniqueEntity;
import com.product.data.center.entity.JournalEntity;
import com.product.data.center.utils.CustomLock;
import com.product.data.center.utils.WriteExtractUtil;
import com.product.datasource.dao.Dao;
import com.product.datasource.entity.BatchResultEntity;
import com.product.datasource.entity.DataBaseEntity;
import com.product.datasource.entity.UpdateFilterEntity;
import com.product.datasource.service.RedisService;
import com.product.util.BaseUtil;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import javax.annotation.Resource;
import java.sql.Connection;
import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @Author cheng
 * @Date 2022/7/15 13:25
 * @PackageName:com.product.data.center.service
 * @ClassName: DataExtractService
 * @Description: 数据提取 从Redis 到 数据库
 * @Version 1.0
 */
@Service
public class DataExtractService {

	final String FIX_KEY = "2022年9月2日092542";

	@Autowired
	BaseDao baseDao;

	@Autowired
	JournalManagerService journalManagerService;

	@Resource
	MesExternalService mesExternalService;

	public final static String UNIQUE_FIX_KEY = "MES_MATER_TABLE_KEY:";


	private static CustomLock lock = new CustomLock(500L);

	/**
	 * 检测采集成功但未提取的记录并标记为执行失败
	 *
	 * @param errorMinute 检测规定时间范围内没有被提取的时间（分钟）
	 */
	public void detectioncExtractIntegrality(String errorMinute) {
		StringBuilder sql = new StringBuilder(128);
		sql.append("\n SELECT ");
		sql.append("\n l.uuid,l.config_uuid ");
		sql.append("\n FROM ");
		sql.append("\n product_sys_data_center_log l ");
		sql.append("\n LEFT JOIN product_sys_data_center_log g ON l.uuid = g.pre_step_uuid AND g.type = 2 ");
		sql.append("\n WHERE ");
		sql.append("\n l.type = 1 AND l.result!=0 AND l.deal_flag!=1 and l.count>0 ");
		sql.append("\n AND g.uuid IS NULL ");
		sql.append("\n AND  TIMESTAMPDIFF(MINUTE  ,l.created_utc_datetime,NOW())> ?");
		DataTableEntity dt = baseDao.listTable(sql.toString(), new Object[]{Integer.parseInt(errorMinute)});
		if (!DataTableEntity.isEmpty(dt)) {
			baseDao.executeUpdate("UPDATE product_sys_data_center_log SET error=? ,result=0,deal_flag=0,updated_utc_datetime=now() WHERE " +
							BaseUtil.buildQuestionMarkFilter("uuid", dt.getRows(), true),
					ArrayUtil.addAll(new Object[]{"采集成功，未能在 " + errorMinute + " 分钟内成功提取！"}, dt.getUuids()));
		}

		sql.setLength(0);
		sql.append("\nSELECT");
		sql.append("\nb.uuid");
		sql.append("\n        FROM");
		sql.append("\nproduct_sys_data_center_log a");
		sql.append("\nJOIN product_sys_data_center_log b ON a.type = 2");
		sql.append("\nAND b.type = 1");
		sql.append("\nAND a.pre_step_uuid = b.uuid");
		sql.append("\nWHERE");
		sql.append("\nb.error LIKE '采集成功，未能在%分钟内成功提取！'");
		sql.append("\nAND a.result = 1");
		sql.append("\nAND b.result = 0");
		sql.append("\nAND b.deal_flag !=1");
		dt = baseDao.listTable(sql.toString(), new Object[]{});
		if (!DataTableEntity.isEmpty(dt)) {
			baseDao.executeUpdate("update product_sys_data_center_log set result=1,error=null,updated_utc_datetime=now() where  " + BaseUtil.buildQuestionMarkFilter("uuid", dt.getRows(), true), dt.getUuids());
		}
	}

	/**
	 * 初始化表唯一值
	 *
	 * @param tableName
	 * @param uniqueFieldName
	 * @param collectId
	 * @param collectSourceField
	 * @param targetSource
	 * @param autoIncrementField
	 */
	public void initTableUniqueValue(String tableName, String uniqueFieldName, String collectId, String collectSourceField, String targetSource, String autoIncrementField) {
		ExtractUniqueEntity extractUniqueEntity = new ExtractUniqueEntity();
		extractUniqueEntity.setCollectSourceField(collectSourceField);
		extractUniqueEntity.setExtractTargetTable(tableName);
		extractUniqueEntity.setUniqueFieldName(uniqueFieldName);
		extractUniqueEntity.setExtractTargetSource(targetSource);
		extractUniqueEntity.setAutoIncrement(autoIncrementField);
		extractUniqueEntity.addSourceInfo(collectId);
//        SpringMVCContextHolder.getSystemLogger().info("开始初始化表唯一值：" + tableName);
//        WriteExtractUtil.append("开始初始化表唯一值：" + tableName);
		String[] lockKey = {tableName, collectId};
		try {
			tableName = tableName.toUpperCase();
			String loadKey = UNIQUE_FIX_KEY + "LOADING-" + tableName + ":" + collectId;
			if ("1".equals(((String) RedisUtil.get(loadKey)))) {
				WriteExtractUtil.append("唯一值缓存已存在跳过：" + tableName);
				return;
			}
			this.initUniqueValue(extractUniqueEntity);
//            SpringMVCContextHolder.getSystemLogger().info("初始化表唯一值结束：" + tableName + "，耗时" + tempTestTimer.intervalMs() + " ms");
//            WriteExtractUtil.append("初始化表唯一值结束：" + tableName + "，耗时" + tempTestTimer.intervalMs() + " ms");
		} catch (Exception e) {
			WriteExtractUtil.append("缓存唯一值出错：" + tableName);
			e.printStackTrace();
			SpringMVCContextHolder.getSystemLogger().error(e);
			throw e;
		}
	}

	/**
	 * 初始化历史数据唯一值
	 * 在系统启动完成后自动调用
	 */
	@Async //异步执行
	public void initUniqueValue() {
		StringBuilder sql = new StringBuilder();
		sql.append(" SELECT ");
		sql.append("collect.id collect_id,\n");
		sql.append("extract_target_source,\n");
		sql.append("extract_target_table,\n");
		sql.append("extract.extract_unique_field,\n");
		sql.append("extract.collect_source_field,\n");
		sql.append("extract.auto_increment \n");
		sql.append(" \n FROM ");
		sql.append(" \n product_sys_data_collect collect ");
		sql.append(" \n JOIN product_sys_data_extract_config extract ON UPPER( extract.extract_source_table )= UPPER( case when length(collect.target_table)>0 then collect.target_table else collect.source_table end ) ");
		sql.append(" \n AND extract.is_used = 1 ");
		sql.append(" \n AND collect.is_used =1 ");

		DataTableEntity dt = baseDao.listTable(sql.toString(), (Object[]) null);
		if (!DataTableEntity.isEmpty(dt)) {
			Map<String, ExtractUniqueEntity> extractUniqueEntityMap = new HashMap<>();
			for (int i = 0; i < dt.getRows(); i++) {
				FieldSetEntity fse = dt.getFieldSetEntity(i);
				String extractTargetSource = fse.getString("extract_target_source");
				String extractTargetTable = fse.getString("extract_target_table");
				String uniqueFieldName = fse.getString("extract_unique_field");
				String collectSourceField = fse.getString("collect_source_field");
				String autoIncrement = fse.getString("auto_increment");
				String key = extractTargetSource + ":" + extractTargetTable;
				ExtractUniqueEntity extractUniqueEntity = extractUniqueEntityMap.get(key);
				if (extractUniqueEntity == null) {
					extractUniqueEntity = new ExtractUniqueEntity();
					extractUniqueEntityMap.put(key, extractUniqueEntity);
					extractUniqueEntity.setExtractTargetSource(extractTargetSource);
					extractUniqueEntity.setAutoIncrement(autoIncrement);
					extractUniqueEntity.setExtractTargetTable(extractTargetTable);
					extractUniqueEntity.setUniqueFieldName(uniqueFieldName);
					extractUniqueEntity.setCollectSourceField(collectSourceField);
				}
				extractUniqueEntity.addSourceInfo(fse.getString("collect_id"));
			}
			ExecutorService threadPool = Executors.newFixedThreadPool(extractUniqueEntityMap.size()
					> 10 ? 10 : extractUniqueEntityMap.size());
			for (ExtractUniqueEntity value : extractUniqueEntityMap.values()) {
				WriteExtractUtil.append("提交到线程池：" + value.getExtractTargetTable());
				threadPool.submit(() -> this.initUniqueValue(value));
			}
			threadPool.shutdown();
			try {//等待直到所有任务完成
				threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.MINUTES);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
			SpringMVCContextHolder.getSystemLogger().info("初始化加载提取表唯一值完成！！！！！！！！");
		}
	}


	void initUniqueValue(ExtractUniqueEntity extractUnique) {
		if (StringUtils.isEmpty(extractUnique.getExtractTargetSource()) || StringUtils.isEmpty(extractUnique.getExtractTargetTable())) {
			return;
		}
		Set<String> collectIdSet = extractUnique.getCollectIdSet();
		String loadKeyTemplate = UNIQUE_FIX_KEY + "LOADING-" + extractUnique.getExtractTargetTable() + ":";
		Iterator<String> iterator = collectIdSet.iterator();
		while (iterator.hasNext()) {
			//初始化过了跳过
			if ("1".equals(((String) RedisUtil.get(loadKeyTemplate + iterator.next())))) {
				iterator.remove();
			}
		}
		if (collectIdSet == null || collectIdSet.isEmpty()) {
			return;
		}
		for (String collect : collectIdSet) {
			lock.lock(extractUnique.getExtractTargetTable(), collect);
		}
		DataBaseEntity dbe = new DataBaseEntity(extractUnique.getExtractTargetSource());
		Dao dao = dbe.getDao();
		try {
			StringBuilder sql = new StringBuilder();
			sql.append(" SELECT MAX(TO_NUMBER(").append(extractUnique.getUniqueFieldName()).append(")) max_value, ");
			sql.append(extractUnique.getCollectSourceField()).append(" source_info");
			sql.append(" FROM ").append(extractUnique.getExtractTargetTable());
			sql.append(" GROUP BY ").append(extractUnique.getCollectSourceField());
			WriteExtractUtil.append("【开始】开始初始化表唯一值：" + extractUnique.getExtractTargetTable() + ",分组： " + extractUnique.getCollectIdSet());
			DataTableEntity dt = dao.getList(sql.toString(), (Object[]) null);
			if (!DataTableEntity.isEmpty(dt)) {
				for (int i = 0; i < dt.getRows(); i++) {
					String maxValue = dt.getString(i, "max_value");
					String sourceInfo = dt.getString(i, "source_info");
					if (StringUtils.isEmpty(maxValue) || StringUtils.isEmpty(sourceInfo)) {
						continue;
					}
					//放入最大值
					RedisUtil.set(UNIQUE_FIX_KEY + extractUnique.getExtractTargetTable().toUpperCase() + ":" + sourceInfo, maxValue);
				}
			}
			for (String collect : collectIdSet) {
				RedisUtil.set(loadKeyTemplate + collect, "1");
			}
			WriteExtractUtil.append("【结束】初始化表唯一值结束：" + extractUnique.getExtractTargetTable() + ",分组： " + extractUnique.getCollectIdSet());
		} catch (Exception e) {
			e.printStackTrace();
			SpringMVCContextHolder.getSystemLogger().error(e);
		} finally {
			dao.closeConnection();
			for (String collect : collectIdSet) {
				lock.unLock(extractUnique.getExtractTargetTable(), collect);
			}
		}
	}

	/**
	 * 删除表对应的唯一值
	 *
	 * @param tableName              表名
	 * @param tableUniqueByCollectId 表对应的唯一值
	 */
	@Deprecated
	public static void removeTableUniqueValue(String tableName, Map<String, Set<String>> tableUniqueByCollectId) {
		if (StringUtils.isEmpty(tableName)) {
			return;
		}
		tableName = tableName.toUpperCase();

		String[] lockKey = {tableName, null};
		if (tableUniqueByCollectId != null && tableUniqueByCollectId.size() > 0) {
			for (Map.Entry<String, Set<String>> v : tableUniqueByCollectId.entrySet()) {
				Set<String> values = v.getValue();
				String collectId = v.getKey();
				lockKey[1] = collectId;
				lock.lock(lockKey);
				try {
					String loadKey = UNIQUE_FIX_KEY + "LOADING-" + tableName + ":" + collectId;
					if (!"1".equals((RedisUtil.get(loadKey)))) {
						//表+采集id对应的唯一值集合还未初始化
						continue;
					}
					RedisUtil.delSetAimValue(UNIQUE_FIX_KEY + tableName + ":" + collectId, values.toArray(new String[0]));
				} finally {
					lock.unLock(lockKey);
				}
			}
		}

	}


	/**
	 * 添加表唯一值
	 *
	 * @param tableName
	 * @param uniqueValues
	 * @param collectId
	 */
	private void addTableUniqueValue(String tableName, String[] uniqueValues, String collectId) {
		tableName = tableName.toUpperCase();
		String[] lockKey = {tableName, collectId};
		Optional<String> max = Arrays.stream(uniqueValues).max(Comparator.comparingInt(item -> Integer.parseInt(item)));
		Integer maxValue = Integer.parseInt(max.get());
		lock.lock(lockKey);
		try {
			String value = (String) RedisUtil.get(UNIQUE_FIX_KEY + tableName + ":" + collectId);
			if (!StringUtils.isEmpty(value) && Integer.parseInt(value) < maxValue) {
				RedisUtil.set(UNIQUE_FIX_KEY + tableName + ":" + collectId, String.valueOf(maxValue));
			}
		} finally {
			lock.unLock(lockKey);
		}
	}

	/**
	 * 验证表唯一值是否存在
	 *
	 * @param tableName
	 * @param value
	 * @param collectId
	 * @return
	 */
	private boolean isExists(String tableName, String value, String collectId) {
		tableName = tableName.toUpperCase();
		String[] lockKey = {tableName, collectId};
		lock.lock(lockKey);
		try {
			String valueOld = (String) RedisUtil.get(UNIQUE_FIX_KEY + tableName + ":" + collectId);

			if (StringUtils.isEmpty(valueOld)) {
				return false;
			} else if (!NumberUtil.isNumber(valueOld)) {
				throw new BaseException(ErrorCode.UNIQUE_VALUE_NUMBER);
			} else {
				return Integer.parseInt(value) <= Integer.parseInt(valueOld);
			}
		} catch (Exception e) {
			throw e;
		} finally {
			lock.unLock(lockKey);
		}
	}

	private CustomLock extractLock = new CustomLock();


	/**
	 * 提取数据定时任务入口方法
	 *
	 * @param uuid 提取配置uuid
	 */
	public void startExtractData(String uuid) throws BaseException {
		String[] lockKey = {"startExtractData", uuid};
		if (!extractLock.tryLock(lockKey)) {
			WriteExtractUtil.append("已有任务在运行跳过此次运行！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！:::" + uuid);
			return;
		}
		WriteExtractUtil.append("开始提取！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！！:::" + uuid);
		TimeInterval tempTestTimer = DateUtil.timer();
		WriteExtractUtil.append("提取-等待耗时：" + tempTestTimer.intervalMs() + ",:::" + uuid);
		tempTestTimer = DateUtil.timer();
		try {
			//查询提取数据配置
			FieldSetEntity fse = baseDao.getFieldSetEntity(CmnConst.DATA_EXTRACT_TABLE, uuid, true);
			if (FieldSetEntity.isEmpty(fse) || DataTableEntity.isEmpty(fse.getSubDataTable(CmnConst.DATA_EXTRACT_SUB_TABLE))) {
				writeLog(uuid, false, ErrorCode.GET_EXTRACT_DATA_CONFIG_FAIL.getText(), -1, null);
				throw new BaseException(ErrorCode.GET_EXTRACT_DATA_CONFIG_FAIL);
			}
			DataTableEntity fieldMappingConfig = fse.getSubDataTable(CmnConst.DATA_EXTRACT_SUB_TABLE);
			//目标表唯一标识字段
			String extractUniqueField = fse.getString("extract_unique_field");
			//来源唯一标识字段
			String sourceUniqueField = null;
			//自动增长字段
			String autoIncrementField = fse.getString("auto_increment");
			for (int i = 0; i < fieldMappingConfig.getRows(); i++) {
				if (extractUniqueField.equals(fieldMappingConfig.getString(i, "target_table_field"))) {
					sourceUniqueField = fieldMappingConfig.getString(i, "source_redis_field");
//                    break;
				}
			}
			if (StringUtils.isEmpty(sourceUniqueField)) {
				throw new BaseException(ErrorCode.GET_EXTRACT_UNIQUE_FIELD_FAIL);
			}

			//redis 主
			String extractSource = fse.getString("extract_source");
			//redis 从
//                String extractSourceBak = fse.getString("bak_redis_source");
			//目标源配置uuid
			String extractTargetSource = fse.getString("extract_target_source");
			//redis数据源
//                DataBaseEntity redisDataBase = new DataBaseEntity(extractSourceBak);
			//目标数据源
			DataBaseEntity targetDataBase = new DataBaseEntity(extractTargetSource);
			//集群模式
			DataBaseEntity redisDataBase = new DataBaseEntity(extractSource);
			RedisService readRedisService = new RedisService(redisDataBase);
			RedisService writeRedisService = readRedisService;
//                redisDataBase = new DataBaseEntity(extractSource);
//                RedisService writeRedisService = new RedisService(redisDataBase);
			String templatePattern = fse.getString("extract_prefix_key") + ":" + fse.getString("extract_source_table").toLowerCase() + ":";
			String pattern = templatePattern + "*";

			WriteExtractUtil.append("keyp-pattern：" + pattern);
			String[] keys = readRedisService.getKeys(pattern);
			if (ArrayUtil.isEmpty(keys)) {
				writeLog(uuid, true, "", 0, null);
				readRedisService.close();
				WriteExtractUtil.append("DE-没有找到对应的redis数据," + pattern);
				return;
			}
			Dao targetDao = targetDataBase.getDao();
			//提取更新时间标识子弹
			String extractUpdateTimeField = fse.getString("extract_update_time_field");
			//采集来源id存储字段
			String collectSourceField = fse.getString("collect_source_field");
			//提取目标表表名
			String extractTargetTable = fse.getString("extract_target_table");
			// 提取时间字段
			String extractTimeField = fse.getString(CmnConst.EXTRACT_TIME_FIELD);
			Dao targetNewDao = null;
			WriteExtractUtil.append("提取-准备耗时：" + tempTestTimer.intervalMs());
			try {
				for (String key : keys) {
					List<String> hashFields = readRedisService.getHashFields(key);
					hashFields = redisHashFields(hashFields);
					if (hashFields != null && hashFields.size() > 0) {
						//采集ID
						String collectId = key.substring(key.lastIndexOf(":") + 1);
						initTableUniqueValue(extractTargetTable, extractUniqueField, collectId, collectSourceField, targetDataBase.getUuid(), autoIncrementField);
						for (int ij = 0; ij < hashFields.size(); ij++) {
							String field = hashFields.get(ij);
							tempTestTimer = DateUtil.timer();
							Object value = readRedisService.getHash(key, field);
							JournalEntity logFse = new JournalEntity();
							WriteExtractUtil.append("提取-redis数据获取耗时：" + tempTestTimer.intervalMs());
							logFse.setOther_info("提取-redis数据获取耗时：" + tempTestTimer.intervalMs());
							if (field == null || value == null || ("impossible_" + FIX_KEY).equals(field)) {
								continue;
							}
							tempTestTimer = DateUtil.timer();
							String collectLogUid = field;
//                            FieldSetEntity collectInfo = baseDao.getFieldSetEntity(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG, collectLogUid, false);

							int totalCount = -1;
							try {
//                                DataTableEntity extractLogs = new DataTableEntity();
								DataTableEntity data = (DataTableEntity) value;
								totalCount = data.getRows();
								DataTableEntity maybeUpdate = new DataTableEntity();
								TimeInterval tempTestTimer1 = DateUtil.timer();
								List<String> newAutoIncrementValues = new ArrayList<>();
								//制令单特殊处理
								updateMoBase(data, extractTargetTable, targetDao);
								//字段映射处理
								data = getFieldMappingDispose(collectId,
										collectSourceField,
										extractTargetTable,
										extractTimeField,
										data,
										fieldMappingConfig,
										extractUniqueField,
										extractUpdateTimeField,
										maybeUpdate,
										autoIncrementField,
										targetDataBase,
										newAutoIncrementValues
								);
								WriteExtractUtil.append("提取-字段映射耗时：" + tempTestTimer1.intervalMs());
								tempTestTimer1 = DateUtil.timer();

								BatchResultEntity batchResultEntity = null;
								if (!DataTableEntity.isEmpty(maybeUpdate)) {
									//有需要查询确认更新的数据
									StringBuilder filter = new StringBuilder(32);
									filter.append(extractUpdateTimeField).append(" < ?  ");
									filter.append(" AND ").append(collectSourceField).append(" =? AND ");
									filter.append(extractUniqueField).append("= ?  ");
									targetNewDao = targetNewDao != null ? targetNewDao : targetDataBase.newDao();
									TimeInterval tempTestTimer2 = DateUtil.timer();
									UpdateFilterEntity updateFilterEntity = new UpdateFilterEntity(autoIncrementField + "=?",
											new String[]{autoIncrementField});
									WriteExtractUtil.append("开始查询数据是否需要更新：" + maybeUpdate.getRows() + ",:" + extractTargetTable);
									Map<JSONObject, JSONObject> exist = isExist(maybeUpdate, collectId, extractTargetTable, autoIncrementField,
											collectSourceField, extractUniqueField, extractUpdateTimeField, targetNewDao);
									WriteExtractUtil.append("查询数据是否需要更新结束共耗时：" + tempTestTimer2.intervalMs() + ",:" + extractTargetTable);
									for (int i = 0; i < maybeUpdate.getRows(); i++) {
										FieldSetEntity item = maybeUpdate.getFieldSetEntity(i);
										String sourceInfo = item.getString(collectSourceField);
										String preMasterKey = item.getString(extractUniqueField);
										JSONObject jsonObject = new JSONObject();
										jsonObject.put(collectSourceField, sourceInfo);
										jsonObject.put(extractUniqueField, preMasterKey);
										jsonObject = exist.get(jsonObject);
										if ((jsonObject == null || jsonObject.isEmpty())) {
//                                            WriteExtractUtil.append("将数据放入新增：" + item.getString(extractUniqueField) + "," + extractTargetTable);
//                                                //判定为新增的数据
											data.addFieldSetEntity(item);
											newAutoIncrementValues.add(item.getString(extractUniqueField));
											maybeUpdate.removeFieldSetEntity(i);
											i--;
											continue;
										}
										item.setValue(autoIncrementField, jsonObject.get(autoIncrementField));
										//数据库更新时间
										Date oldDate = jsonObject.getDate(extractUpdateTimeField);
										//数据的更新时间
										Date newDate = item.getDate(extractUpdateTimeField);
										int compare = DateUtil.compare(oldDate, newDate);
										if (compare >= 0) {
//                                            extractLogs.addFieldSetEntity(getFieldSetEntity(item.getString(extractUniqueField), item.getTableName(), 2, item.getString(collectSourceField)));
											//不需要更新
											maybeUpdate.removeFieldSetEntity(i);
											i--;
											continue;
//                                            WriteExtractUtil.append("跳过不需要更新的数据项，主键：" + item.getString(autoIncrementField) + ",数据库时间：" + DateUtil.format(oldDate, "yyyy-MM-dd HH:mm:ss") + ",数据库时间：" + DateUtil.format(newDate, "yyyy-MM-dd HH:mm:ss"));
										}
//                                        extractLogs.addFieldSetEntity(getFieldSetEntity(item.getString(extractUniqueField), item.getTableName(), 1, item.getString(collectSourceField)));
									}
									if (maybeUpdate.getRows() > 0) {
										maybeUpdate.setMeta(maybeUpdate.getFieldSetEntity(0).getMeta());
									}
									WriteExtractUtil.append("更新提取过滤后数据：" + extractTargetTable + ",需要更新的条数：" + maybeUpdate.getRows());

									maybeUpdate.getData().sort((o1, o2) -> DateUtil.compare(o1.getDate(extractUpdateTimeField), o2.getDate(extractUpdateTimeField)));

									batchResultEntity = targetNewDao.updateBatch(maybeUpdate, updateFilterEntity, false);

									WriteExtractUtil.append("更新提取过滤后数据：" + extractTargetTable + ",需要更新的条数：" + maybeUpdate.getRows() + ",耗时：" + tempTestTimer2.intervalMs());
									targetNewDao.closeConnection();
								}

								if (!DataTableEntity.isEmpty(data)) {
									tempTestTimer1 = DateUtil.timer();
									targetDao.addBatch(data, autoIncrementField);
//                                    data.getData().stream().forEach(item -> {
//                                        extractLogs.addFieldSetEntity(getFieldSetEntity(item.getString(extractUniqueField), item.getTableName(), 0, item.getString(collectSourceField)));
//                                    });
									WriteExtractUtil.append("新增数据耗时：" + tempTestTimer1.intervalMs());
									this.addTableUniqueValue(extractTargetTable, newAutoIncrementValues.stream().toArray(String[]::new), collectId);
								}
//                                if (!DataTableEntity.isEmpty(extractLogs)) {
//                                    baseDao.add(extractLogs);
//                                }
//                                    writeLog(uuid, true, "", totalCount, collectLogUid);
								WriteExtractUtil.append("提取-新增更新数据耗时：" + tempTestTimer1.intervalMs());
								WriteExtractUtil.append("提取-写入数据库耗时：" + tempTestTimer.intervalMs());
								writeLog(logFse, uuid, true, "", totalCount, collectLogUid);
							} catch (Exception e) {
								e.printStackTrace();
								SpringMVCContextHolder.getSystemLogger().error(e);
								writeLog(uuid, false, e, totalCount, collectLogUid);
							} finally {
								//redis数据源
								writeRedisService.setHash(key, "impossible_" + FIX_KEY, new DataTableEntity());
								writeRedisService.delHash(key, field);
							}
						}
					}
				}
			} catch (Exception e) {
				throw e;
			} finally {
				if (targetNewDao != null) {
					targetNewDao.closeConnection();
				}
				readRedisService.close();
				writeRedisService.close();
				targetDao.closeConnection();
			}
		} catch (Exception e) {
			e.printStackTrace();
			SpringMVCContextHolder.getSystemLogger().error(e);
			throw e;
		} finally {
			extractLock.unLock(lockKey);
		}
	}


	/**
	 * 制令单特殊处理 将找目标库中的工单表的id字段写入project_id
	 *
	 * @param dt
	 * @param targetTable
	 * @param targetDao
	 * @throws BaseException
	 */
	private void updateMoBase(DataTableEntity dt, String targetTable, Dao targetDao) throws BaseException {
		if (DataTableEntity.isEmpty(dt) || !"T_PM_MO_BASE".equalsIgnoreCase(targetTable)) {
			return;
		}
		Map<String, FieldSetEntity> map = new HashMap<>();
		List<Object> moNumberList = new ArrayList<>();
		for (int i = 0; i < dt.getRows(); i++) {
			String moNumber = dt.getString(i, "mo_number");
			if (StringUtils.isEmpty(moNumber)) {
				continue;
			}
			moNumberList.add(moNumber);
			map.put(moNumber, dt.getFieldSetEntity(i));
		}
		//查询制令单关联的工单数据
		DataTableEntity projectBaseDt = targetDao.getList("T_PM_PROJECT_BASE", new String[]{"PROJECT_ID,PROJECT_BASE_ID"},
				BaseUtil.buildQuestionMarkFilter("project_id", moNumberList.size(), true), moNumberList.toArray(), null, 1, dt.getRows());
		if (!DataTableEntity.isEmpty(projectBaseDt)) {
			for (int i = 0; i < projectBaseDt.getRows(); i++) {
				FieldSetEntity fs = projectBaseDt.getFieldSetEntity(i);
				String projectId = fs.getString("project_id");
				FieldSetEntity fieldSetEntity = map.get(projectId);
				if (!FieldSetEntity.isEmpty(fieldSetEntity)) {
					//将工单表id写入制令单的project_id字段
					fieldSetEntity.setValue("project_id", fs.getString("project_base_id"));
					map.remove(projectId);
				}
			}
		}
		if (!map.isEmpty()) {
			throw new BaseException(ErrorCode.MO_NUMBER_MASTER_PROJECT_BASE_EMPTY);
		}
	}

	/**
	 * @param primaryValue  主键值
	 * @param tableName     表名
	 * @param operationType 操作类型 0 insert 1 update 2 数据已存在不需要更新
	 * @return
	 */
	private FieldSetEntity getFieldSetEntity(String primaryValue, String tableName, int operationType, String sourceInfo) {
		FieldSetEntity fs = new FieldSetEntity();
		fs.setTableName("product_mes_extract_log");
		fs.setValue("primary_value", primaryValue);
		fs.setValue("table_name", tableName);
		fs.setValue("operation_type", operationType);
		fs.setValue("source_info", sourceInfo);
		fs.setValue(CmnConst.CREATED_UTC_DATETIME, new Date());
		return fs;
	}

	/**
	 * 将采集uuid按采集表id排序返回
	 *
	 * @param fields
	 * @return
	 */
	private List<String> redisHashFields(List<String> fields) {
		if (fields != null && !fields.isEmpty()) {
			DataTableEntity dt = baseDao.listTable(CmnConst.PRODUCT_SYS_DATA_CENTER_LOG,
					"type = 1 and " + BaseUtil.buildQuestionMarkFilter("uuid", fields.toArray(), true),
					new Object[]{}, new Object[]{CmnConst.UUID}, "id");
			if (!DataTableEntity.isEmpty(dt)) {
				return dt.getData().stream().map(item -> item.getString(CmnConst.UUID)).collect(Collectors.toList());
			}
		}

		return Collections.emptyList();
	}

	private void writeLog(String configUid, boolean isSuccess, String errorMsg, int count, String collectLogUid) {
		JournalEntity fse = new JournalEntity();
		fse.setType(2);
		fse.setError(errorMsg);
		fse.setCount(count);
		fse.setResult(isSuccess ? 1 : 0);
		fse.setConfigUuid(configUid);
		fse.setPre_step_uuid(collectLogUid);
		fse.setCreated_utc_datetime(new Date());
		if (!StringUtils.isEmpty(errorMsg) || count > 0 || !isSuccess) {
			FieldSetEntity fs = journalManagerService.autoCreateJournal(fse);
			mesExternalService.remoteSaveExtractLog(fs);
		}
	}

	private void writeLog(JournalEntity fse, String configUid, boolean isSuccess, String errorMsg, int count, String collectLogUid) {
		fse.setType(2);
		fse.setError(errorMsg);
		fse.setCount(count);
		fse.setResult(isSuccess ? 1 : 0);
		fse.setConfigUuid(configUid);
		fse.setPre_step_uuid(collectLogUid);
		fse.setCreated_utc_datetime(new Date());
		FieldSetEntity fs = journalManagerService.autoCreateJournal(fse);
		mesExternalService.remoteSaveExtractLog(fs);
	}

	private void writeLog(String configUid, boolean isSuccess, Throwable throwable, int count, String collectLogUid) {
		JournalEntity fse = new JournalEntity();
		String errorMsg = journalManagerService.getStackTrace(throwable);
		fse.setType(2);
		fse.setError(errorMsg);
		fse.setCount(count);
		fse.setResult(isSuccess ? 1 : 0);
		fse.setConfigUuid(configUid);
		fse.setPre_step_uuid(collectLogUid);
		fse.setCreated_utc_datetime(new Date());
		FieldSetEntity fs = journalManagerService.autoCreateJournal(fse);
		mesExternalService.remoteSaveExtractLog(fs);

	}

	/**
	 * @param dt
	 * @param collectId              采集id
	 * @param targetTable            目标表表名
	 * @param idField                id字段名称
	 * @param sourceInfoField        采集来源字段名称
	 * @param preMasterKeyField      子库主键字段名称
	 * @param extractUpdateTimeField 修改时间字段
	 * @param dao
	 * @return
	 */
	private Map<JSONObject, JSONObject> isExist(DataTableEntity dt, String collectId, String targetTable, String idField, String sourceInfoField, String preMasterKeyField, String extractUpdateTimeField, Dao dao) {
		List<FieldSetEntity> data = dt.getData();
		String dataSystemName = Global.getSystemConfig("data.system.name", "");
		if (StringUtils.isEmpty(dataSystemName)) {
			//数据来源系统名称不能为空
			throw new BaseException(ErrorCode.DATA_SYSTEM_NAME_CAN_NOT_EMPTY);
		}
		Map<String, List<String>> collect = data.stream().collect(Collectors.groupingBy(item -> item.getString(sourceInfoField), Collectors.mapping(item -> item.getString(preMasterKeyField), Collectors.toList())));
		//子库采集数据的主键集合
//        Set<String> collectData = data.stream().filter(item -> !dataSystemName.equals(item.getString(sourceInfoField))).map(item -> item.getString(preMasterKeyField)).collect(Collectors.toSet());
		int i = 0;
		List<String> params = new ArrayList<>();
		StringBuilder sql = new StringBuilder();
		for (Map.Entry<String, List<String>> v : collect.entrySet()) {
			if (collectId.equals(v.getKey()) || dataSystemName.equals(v.getKey())) {
				if (i > 0) {
					sql.append(" UNION ALL ");
				}
				sql.append(" SELECT ").append(preMasterKeyField).append(",").append(idField).append(",").append(extractUpdateTimeField);
				sql.append(" ,").append(sourceInfoField);
				sql.append(" FROM ").append(targetTable);
				sql.append(" WHERE ").append(sourceInfoField).append(" = ? and ").append(BaseUtil.buildQuestionMarkFilter(preMasterKeyField, v.getValue().size(), true));
				params.add(v.getKey());
				params.addAll(v.getValue());
			}
			i++;
		}
		DataTableEntity result = dao.getList(sql.toString(), params.toArray());
		if (DataTableEntity.isEmpty(result)) {
			return Collections.emptyMap();
		}
		Map<JSONObject, JSONObject> resultMap = new HashMap<>();
		for (int j = 0; j < result.getRows(); j++) {
			FieldSetEntity item = result.getFieldSetEntity(j);
			JSONObject jsonObject = new JSONObject();
			jsonObject.put(preMasterKeyField, item.getString(preMasterKeyField));
			jsonObject.put(sourceInfoField, item.getString(sourceInfoField));
			resultMap.put(jsonObject, new JSONObject((Map) item.getValues()));
		}
		return resultMap;
	}

	/**
	 * 字段映射
	 * *
	 *
	 * @param collectId          采集id
	 * @param collectSourceField 采集来源字段
	 * @param extractTargetTable 提取目标表
	 * @param data               要映射的元数据集
	 * @param fieldMappingConfig 映射关系
	 * @param extractUniqueField 目标表唯一标识字段
	 * @param updateTimeField    更新标识字段
	 * @param updateDt           可能需要更新的数据
	 * @return
	 */
	private DataTableEntity getFieldMappingDispose(String collectId,
												   String collectSourceField,
												   String extractTargetTable,
												   String extractTimeField,
												   DataTableEntity data,
												   DataTableEntity fieldMappingConfig,
												   String extractUniqueField,
												   String updateTimeField,
												   DataTableEntity updateDt,
												   String autoIncrementField,
												   DataBaseEntity targetDataBase,
												   List<String> newAutoIncrementValues) {
		String dataSystemName = Global.getSystemConfig("data.system.name", "");
		if (StringUtils.isEmpty(dataSystemName)) {
			//数据来源系统名称不能为空
			throw new BaseException(ErrorCode.DATA_SYSTEM_NAME_CAN_NOT_EMPTY);
		}
		if (!DataTableEntity.isEmpty(data)) {
			//从数据库中查询有没有存在的主键值
			DataTableEntity result = new DataTableEntity();
			for (int i = 0; i < data.getRows(); i++) {
				FieldSetEntity fse1 = data.getFieldSetEntity(i);
				FieldSetEntity fseNew = new FieldSetEntity();
				fseNew.setTableName(extractTargetTable);
				//历史数据
				String historyCollect = fse1.getString(collectSourceField);
				for (int j = 0; j < fieldMappingConfig.getRows(); j++) {
					String targetTableField = fieldMappingConfig.getString(j, "target_table_field");
					String sourceRedisField = fieldMappingConfig.getString(j, "source_redis_field");
					fseNew.setValue(targetTableField, fse1.getObject(sourceRedisField));
				}
				// 提取时间
				fseNew.setValue(extractTimeField, new Date());
				//数据系统名称与历史采集id不一致才复制当前采集的id 否则判断为历史系统数据将进行更新操作 2022年12月1日22:33:00 cheng
				if (!dataSystemName.equals(historyCollect)) {
					fseNew.setValue(collectSourceField, collectId);
				} else {
					fseNew.setValue(collectSourceField, historyCollect);
					fseNew.setValue(extractUniqueField, fse1.getString(autoIncrementField));
				}
				if (StringUtils.isEmpty(fseNew.getString(extractUniqueField))) {
					//唯一值为空
					throw new BaseException(ErrorCode.EXTRACT_DATA_ROW_UNIQUE_VALUE_CAN_NOT_EMPTY);
				}
				if (StringUtils.isEmpty(updateTimeField)) {
					throw new BaseException(new Exception("更新时间字段为空"));
				}
				//系统数据名称和采集来源字段的值相同 默认为更新
				if (fseNew.getObject(updateTimeField) != null && (dataSystemName.equals(historyCollect) ||
//                        exist.contains(fseNew.getString(uniqueField))
						isExists(extractTargetTable, fseNew.getString(extractUniqueField), collectId)
				)) {
					//更新字段不为空可能会有更新
					fseNew.remove(autoIncrementField);
					updateDt.addFieldSetEntity(fseNew);
				} else {
					//新增
					result.addFieldSetEntity(fseNew);
					newAutoIncrementValues.add(fseNew.getString(extractUniqueField));
				}
			}
			return result;

		}
		return data;
	}
}