package com.chenyang.nse.bussiness.tools.dataprocess.hive;

import java.lang.String;
import com.chenyang.nse.bussiness.entity.orm.table.core.TCoreDatasystem;
import com.chenyang.nse.bussiness.entity.orm.table.core.encryption.ColumnPrimaryKeyInfo;
import com.chenyang.nse.bussiness.entity.vo.strategy.ColumnInfoVO;
import com.chenyang.nse.bussiness.service.core.encryption.TCoreEncryptionService;
import com.chenyang.nse.bussiness.service.core.encryption.impl.TCoreEncryptionContext;
import com.chenyang.nse.bussiness.tools.dataprocess.ApplicationContextProvider;
import com.chenyang.nse.bussiness.tools.dataprocess.EncTaskStateContainer;
import com.chenyang.nse.bussiness.tools.dataprocess.EncTaskTracker;
import com.chenyang.nse.bussiness.tools.encryption.EncProvider;
import com.chenyang.nse.bussiness.tools.jdbc.JdbcTool;
import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class HiveLoadingTask implements Runnable {
   private static final Logger log = LoggerFactory.getLogger(HiveLoadingTask.class);
   private final Connection connection;
   private static final int BATCH_SIZE = 20;
   private static final int FETCH_SIZE = 20;
   protected static final String COMPLETE = "5";
   protected static final String RUNNING = "-1";
   protected static final String FAIL = "4";
   private String dataSystemId;
   private List<ColumnInfoVO> columnInfoList;
   private String schema;
   private String projectId;
   private String tableName;
   private String columnName;
   private EncTaskStateContainer container;

   public HiveLoadingTask(String projectId, String dataSystemId, TCoreDatasystem tCoreDatasystem, String schema, String tableName, String[] primaryKeys, List<ColumnInfoVO> columnInfoList, List<com.chenyang.nse.bussiness.entity.vo.maskingtask.ColumnInfoVO> allColumn, List<ColumnPrimaryKeyInfo> columnPrimaryKeyInfoList, List<com.chenyang.nse.bussiness.entity.vo.maskingtask.ColumnInfoVO> allColumnList, boolean isReload) {
      this.projectId = projectId;
      this.dataSystemId = dataSystemId;
      this.schema = schema;
      this.tableName = tableName;
      this.columnInfoList = columnInfoList;
      this.connection = JdbcTool.getConnection(tCoreDatasystem);
      this.container = EncTaskTracker.instance().container();
      this.container.add(projectId, schema, tableName, columnInfoList);
      EncTaskTracker.instance().register(this.container);
   }

   public void run() {
      try {
         this.runTask();
      } catch (Exception e) {
         log.error(e.getMessage());
         throw new RuntimeException(e);
      } finally {
         EncTaskTracker.instance().over(this.container.getTask());
      }

   }

   public void runTask() throws Exception {
      try {
         String backupTable = this.tableName + "_backup";
         log.info("HIVE-备份表名: {}", backupTable);
         if (this.isTableExists(backupTable)) {
            throw new IllegalStateException("备份表 " + backupTable + " 已存在，请先删除或重命名该表后再执行迁移操作");
         }

         this.updateEncryptionProgress(this.columnInfoList, "-1");
         log.info("HIVE-开始备份原表 {} 到 {}", this.tableName, backupTable);
         this.executeUpdate("ALTER TABLE " + this.tableName + " RENAME TO " + backupTable);
         log.info("HIVE-备份原表完成");
         log.info("HIVE-开始创建新表 {}", this.tableName);
         this.executeUpdate("CREATE TABLE " + this.tableName + " LIKE " + backupTable);
         log.info("HIVE-创建新表完成");
         log.info("HIVE-开始处理数据迁移和加密");
         this.processData(backupTable, this.columnInfoList);
         this.updateEncryptionProgress(this.columnInfoList, "5");
         this.executeUpdate("DROP TABLE " + backupTable);
         log.info("HIVE-备份表删除完毕");
      } catch (Exception e) {
         log.error(e.getMessage());
      } finally {
         this.connection.close();
         log.info("HIVE-操作完毕，关闭连接");
      }

   }

   private void updateEncryptionProgress(List<ColumnInfoVO> columnInfoList, String status) {
      TCoreEncryptionService tCoreEncryptionService = (TCoreEncryptionService)ApplicationContextProvider.getBean(TCoreEncryptionContext.class);

      for(int i = 0; i < columnInfoList.size(); ++i) {
         this.columnName = ((ColumnInfoVO)columnInfoList.get(i)).getColumnName();
         tCoreEncryptionService.updateFlag(this.projectId, this.dataSystemId, this.schema, this.tableName, this.columnName, status);
      }

   }

   private boolean isTableExists(String tableName) throws SQLException {
      ResultSet rs = this.connection.getMetaData().getTables((String)null, this.schema, tableName, (String[])null);
      Throwable var3 = null;

      boolean var5;
      try {
         boolean exists = rs.next();
         log.info("HIVE-表 {} 是否存在: {}", tableName, exists);
         var5 = exists;
      } catch (Throwable var14) {
         var3 = var14;
         throw var14;
      } finally {
         if (rs != null) {
            if (var3 != null) {
               try {
                  rs.close();
               } catch (Throwable var13) {
                  var3.addSuppressed(var13);
               }
            } else {
               rs.close();
            }
         }

      }

      return var5;
   }

   private void processData(String sourceTable, List<ColumnInfoVO> columnInfoList) throws Exception {
      List<String> columns = this.getColumnNames(sourceTable);
      log.info("HIVE-获取到的列名: {}", columns);
      Map<String, Integer> encryptColumns = this.getEncryptColumnIndices(columns);
      log.info("HIVE-需要加密的列及其索引: {}", encryptColumns);
      String insertSql = this.buildInsertSql(columns);
      long total = this.getRowCount(sourceTable);
      log.info("HIVE-源表 {} 的总行数: {}", sourceTable, total);
      long processed = 0L;

      while(processed < total) {
         List<String[]> batch = this.fetchBatch(sourceTable, columns, processed);
         log.info("HIVE-获取到一批数据，大小: {}", batch.size());
         List<String[]> encryptedBatch = this.encryptBatch(batch, encryptColumns, columnInfoList);
         log.info("HIVE-加密完成，加密后的数据大小: {}", encryptedBatch.size());
         this.insertBatch(encryptedBatch, insertSql);
         log.info("HIVE-插入一批数据完成");
         processed += (long)batch.size();
         double progress = (double)processed / (double)total * (double)100.0F;
         log.info("HIVE-执行进度: {}/{}", new Object[]{processed, total, progress});
      }

   }

   private List<String> getColumnNames(String table) throws SQLException {
      List<String> columns = new ArrayList();
      Statement stmt = this.connection.createStatement();
      Throwable var4 = null;

      try {
         ResultSet rs = stmt.executeQuery("SELECT * FROM " + table + " LIMIT 1");
         Throwable var6 = null;

         try {
            ResultSetMetaData meta = rs.getMetaData();

            for(int i = 1; i <= meta.getColumnCount(); ++i) {
               String columnName = meta.getColumnName(i);
               if (columnName.contains(".")) {
                  columnName = columnName.substring(columnName.indexOf(".") + 1);
               }

               columns.add(columnName);
            }
         } catch (Throwable var31) {
            var6 = var31;
            throw var31;
         } finally {
            if (rs != null) {
               if (var6 != null) {
                  try {
                     rs.close();
                  } catch (Throwable var30) {
                     var6.addSuppressed(var30);
                  }
               } else {
                  rs.close();
               }
            }

         }
      } catch (Throwable var33) {
         var4 = var33;
         throw var33;
      } finally {
         if (stmt != null) {
            if (var4 != null) {
               try {
                  stmt.close();
               } catch (Throwable var29) {
                  var4.addSuppressed(var29);
               }
            } else {
               stmt.close();
            }
         }

      }

      return columns;
   }

   private Map<String, Integer> getEncryptColumnIndices(List<String> columns) {
      Map<String, Integer> indices = new HashMap();

      for(int i = 0; i < columns.size(); ++i) {
         String columnName = (String)columns.get(i);

         for(ColumnInfoVO columnInfo : this.columnInfoList) {
            if (columnInfo.getColumnName().equalsIgnoreCase(columnName)) {
               indices.put(columnName, i);
               break;
            }
         }
      }

      return indices;
   }

   private long getRowCount(String table) throws SQLException {
      try(Statement stmt = this.connection.createStatement();
          ResultSet rs = stmt.executeQuery("SELECT COUNT(*) FROM " + table))
      {
         rs.next();
         return rs.getLong(1);
      }
   }

   private List<String[]> fetchBatch(String table, List<String> columns, long offset) throws SQLException {
      List<String[]> batch = new ArrayList(20);
      String sql = String.format("SELECT %s FROM %s LIMIT %d OFFSET %d", String.join(",", columns), table, 20, offset);
      Statement stmt = this.connection.createStatement();
      Throwable var8 = null;

      try {
         stmt.setFetchSize(20);
         ResultSet rs = stmt.executeQuery(sql);
         Throwable var10 = null;

         try {
            while(rs.next()) {
               String[] row = new String[columns.size()];

               for(int i = 0; i < columns.size(); ++i) {
                  row[i] = rs.getString(i + 1);
               }

               batch.add(row);
            }
         } catch (Throwable var34) {
            var10 = var34;
            throw var34;
         } finally {
            if (rs != null) {
               if (var10 != null) {
                  try {
                     rs.close();
                  } catch (Throwable var33) {
                     var10.addSuppressed(var33);
                  }
               } else {
                  rs.close();
               }
            }

         }
      } catch (Throwable var36) {
         var8 = var36;
         throw var36;
      } finally {
         if (stmt != null) {
            if (var8 != null) {
               try {
                  stmt.close();
               } catch (Throwable var32) {
                  var8.addSuppressed(var32);
               }
            } else {
               stmt.close();
            }
         }

      }

      return batch;
   }

   private List<String[]> encryptBatch(List<String[]> batch, Map<String, Integer> encryptColumns, List<ColumnInfoVO> allColumnList) throws Exception {
      List<String[]> encrypted = new ArrayList(batch.size());

      for(String[] row : batch) {
         String[] newRow = (String[])row.clone();

         for(ColumnInfoVO columnInfo : allColumnList) {
            String columnName = columnInfo.getColumnName();
            String secretKey = columnInfo.getSecretKey();
            String encryptionType = columnInfo.getEncryptionType();
            String isLike = columnInfo.getIsLike();
            String keepfirst = columnInfo.getKeepfirst();
            String encdigit = columnInfo.getEncdigit();
            String twoindex = columnInfo.getTwoindex();
            Integer columnIndex = (Integer)encryptColumns.get(columnName);
            if (columnIndex != null) {
               newRow[columnIndex] = this.encrypt(newRow[columnIndex], secretKey, encryptionType, isLike, keepfirst, encdigit, twoindex);
            }
         }

         encrypted.add(newRow);
      }

      return encrypted;
   }

   private String encrypt(String value, String secretKey, String encryptionType, String isLike, String keepfirst, String encdigit, String twoindex) throws Exception {
      if (null != encryptionType && !"".equals(encryptionType)) {
         value = EncProvider.encryptEcb(encryptionType, secretKey, value, isLike, keepfirst, encdigit, twoindex);
      }

      return value;
   }

   private String buildInsertSql(List<String> columns) {
      String sql = "INSERT INTO " + this.tableName + " (" + String.join(",", columns) + ") VALUES (" + String.join(",", Collections.nCopies(columns.size(), "?")) + ")";
      return sql;
   }

   private void insertBatch(List<String[]> batch, String sql) throws SQLException {
      if (!batch.isEmpty()) {
         StringBuilder multiValueSql = new StringBuilder();
         multiValueSql.append(sql.substring(0, sql.indexOf("VALUES") + 6));

         for(int i = 0; i < batch.size(); ++i) {
            if (i > 0) {
               multiValueSql.append(",");
            }

            multiValueSql.append("(");
            String[] row = (String[])batch.get(i);

            for(int j = 0; j < row.length; ++j) {
               if (j > 0) {
                  multiValueSql.append(",");
               }

               multiValueSql.append("'").append(this.escapeSql(row[j])).append("'");
            }

            multiValueSql.append(")");
         }

         Statement stmt = this.connection.createStatement();
         Throwable var17 = null;

         try {
            stmt.executeUpdate(multiValueSql.toString());
         } catch (Throwable var14) {
            var17 = var14;
            throw var14;
         } finally {
            if (stmt != null) {
               if (var17 != null) {
                  try {
                     stmt.close();
                  } catch (Throwable var13) {
                     var17.addSuppressed(var13);
                  }
               } else {
                  stmt.close();
               }
            }

         }

      }
   }

   private String escapeSql(String str) {
      return str == null ? "" : str.replace("'", "''");
   }

   private void executeUpdate(String sql) throws SQLException {
      Statement stmt = this.connection.createStatement();
      Throwable var3 = null;

      try {
         stmt.executeUpdate(sql);
      } catch (Throwable var12) {
         var3 = var12;
         throw var12;
      } finally {
         if (stmt != null) {
            if (var3 != null) {
               try {
                  stmt.close();
               } catch (Throwable var11) {
                  var3.addSuppressed(var11);
               }
            } else {
               stmt.close();
            }
         }

      }

   }
}
