package com.liquidnet.service.platform.service.impl.adam.dm.processor;

import com.liquidnet.service.adam.dto.vo.AdamRealInfoVo;
import com.liquidnet.service.adam.entity.AdamRealName;
import com.liquidnet.service.platform.service.impl.adam.dm.DMRealNameService;
import com.liquidnet.service.platform.service.impl.adam.dm.DataMigrationProcessorService;
import com.mysql.cj.jdbc.result.ResultSetImpl;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;

@Slf4j
@Service
public class DMRealNameProcessor extends DataMigrationProcessorService {
    @Autowired
    DMRealNameService dmRealNameService;

    @SneakyThrows
    @Override
    public void dataProcessing(Integer dg, Integer dG, String incrDt, String flg, String flu) {
        if (StringUtils.isNotEmpty(flu)) {
            log.info("DM.flush.real_name:{}", userMapper.executeForDM("adam_real_name"));

//            log.info("DM.flush.AdamRealInfoVo:{}", mongoTemplate.remove(Query.query(Criteria.where("_id").exists(true)), AdamRealInfoVo.class.getSimpleName()).getDeletedCount());
        }

        String sqlCount = "select count(1) from user_real_name where id not in (9145,7456,7448,7197,7194,4137,4138,4139,269,261,262,263,264,265,266,267,251) and created_at" + (StringUtils.isBlank(incrDt) ? "<curdate()" : (">=curdate()-" + incrDt));
        if (null != dg) {
            sqlCount = sqlCount + " and id%" + dG + "=" + dg;
        }
        String field = "id,uid,real_name,id_card,ifnull(created_at, '1970-01-01 00:00:00') created_at,updated_at";
//                     1-id,2-uid,3-real_name,4-id_card,5-created_at,6-updated_at
        String sql = sqlCount.replace("count(1)", field);
        boolean smFlg = "1".equals(flg);
        if (smFlg) {
            sqlCount += " and uid in (" + DG_UID + ")";
            sql += " and uid in (" + DG_UID + ")";
        }
        sql += " order by id limit ?,?";

        Connection connection = null;
        PreparedStatement statement = null;
        ResultSetImpl row = null;
        try {
            Class.forName(DB_DRIVER);
            connection = DriverManager.getConnection(DB_URL + DB_NAME_PASSPORT, DB_USER, DB_PWD);
            statement = connection.prepareStatement(sqlCount);
            row = (ResultSetImpl) statement.executeQuery();
            row.first();
            int ct = row.getInt(1), pSize = 1000, num = 0, tl = 0, pl = ct;
            log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> start.sql:{}", pl, num, pSize, tl, sql);

            row.close();
            statement.close();
            while (ct > 0) {
                statement = connection.prepareStatement(sql);
                int ls = pSize * num;
                statement.setInt(1, ls);
                statement.setInt(2, pSize);
                row = (ResultSetImpl) statement.executeQuery();
                log.info("DM.execute.limit {},{} - begin", ls, pSize);

                List<AdamRealName> realNameList = new ArrayList<>();
                List<AdamRealInfoVo> vos = new ArrayList<>();
                while (row.next()) {
                    String uid = row.getString(2);
                    if (!reUidList.contains(uid)) {
                        AdamRealName realName = new AdamRealName();
                        realName.setRealNameId(row.getString(1));
                        realName.setUid(uid);
                        realName.setType(1);
                        realName.setName(row.getString(3));
                        realName.setIdCard(row.getString(4));
                        realName.setState(1);
                        realName.setCreatedAt(row.getLocalDateTime(5));
                        realName.setUpdatedAt(row.getLocalDateTime(6));

                        AdamRealInfoVo vo = AdamRealInfoVo.getNew();
                        vo.setUid(uid);
                        vo.setType(1);
                        vo.setName(realName.getName());
                        vo.setIdCard(realName.getIdCard());
                        vo.setState(1);

                        realNameList.add(realName);
                        vos.add(vo);

                        dmRdmService.setRealInfoVoByUid(uid, vo);
                        dmRdmService.setCertification(1, realName.getIdCard(), realName.getName());
                    }
                    if (realNameList.size() == 1000 || (realNameList.size() > 0 && smFlg) || ct <= 1000) {
                        tl += realNameList.size();
//                        mongoTemplate.insert(vos, AdamRealInfoVo.class.getSimpleName());
                        log.info("DM.execute.limit.result:{}", dmRealNameService.saveBatch(realNameList));
                        vos.clear();
                        realNameList.clear();
                    }
                }
                if (realNameList.size() > 0) {
                    tl += realNameList.size();
                    log.info("DM.execute.limit.result:{}", dmRealNameService.saveBatch(realNameList));
                }
                row.close();
                statement.close();

                num++;
                ct -= pSize;

                log.info("DM.execute.limit {},{} - handle.counts:{}", ls, pSize, tl);
            }
            log.info("DM.execute.counts:{}/{},{} ~ {} ----------------------------> end", pl, num, pSize, tl);
        } catch (Exception e) {
            log.error("ex:getConnection,dG/dg:{}/{}", dG, dg, e);
        }
        try {
            if (null != row) row.close();
        } catch (SQLException ignored) {
        }
        try {
            if (null != statement) statement.close();
        } catch (SQLException ignored) {
        }
//        try {
//            if (null != connection) connection.close();
//        } catch (SQLException ignored) {
//        }
    }
}
