package cc.smtweb.system.canal.file.common;

import cc.smtweb.framework.core.common.CanalVO;
import cc.smtweb.framework.core.db.DbEngine;
import cc.smtweb.framework.core.db.jdbc.IDbWorker;
import cc.smtweb.framework.core.exception.SwException;
import cc.smtweb.framework.core.util.*;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.io.FileUtils;
import org.springframework.stereotype.Component;

import java.io.File;
import java.io.FileFilter;
import java.util.*;

@Slf4j
@Component
public abstract class AbstractFileWork {
    private final static String suffix = ".cf";
    protected final static String idField = "idField";

    protected FileDecodeConfig fileConfig;

    protected FieldMappingVo fieldMapping;

    protected DbEngine getDbEngine() {
        return DbEngine.getInstance();
    }


    public AbstractFileWork() {
        fileConfig = SpringUtil.getBean(FileDecodeConfig.class);
        fieldMapping = FileDecodeFactory.getInstance().getFieldMapping();
    }

    /**
     * @Author yaoq
     * @Date 2022/9/20 9:25
     * @Params
     * @Return
     * @Description 解析文件
     */
    public void doWork() {
        String path = fileConfig.getPath();
        if (!path.endsWith("/")) {
            path = path + "/";
        }
        path = path + "increment/";
        log.debug("[解析canal文件] path：" + path);
        File dir = new File(path);
        File[] fs = dir.listFiles(new FileFilter() {
            @Override
            public boolean accept(File file) {
                return !file.isDirectory() && file.getName().endsWith(suffix);
            }
        });
        if (fs == null) return;
        List<File> files = Arrays.asList(fs);
        Collections.sort(files, (o1, o2) -> StringUtil.chineseCompare(o1.getName(), o2.getName()));
        log.debug("[解析canal文件] size：" + files.size());
        for (File f : files) {
            try {
                //为保证数据一致性，发现错误，不能继续，卡在这
                getDbEngine().doTrans(new IDbWorker() {
                    @Override
                    public void work() {
                        try {
                            decodeFile(f);
                        } catch (Exception e) {
                            throw new SwException(e);
                        }
                    }
                });
                //备份文件
                FileUtil.copyFile(f, path + "/bak/" + DateUtil.getNowYm() + "/" + f.getName(), true);
            } catch (Exception e) {
                //出错了跳出，避免数据遗失
                log.error("[解析canal文件] file：" + f.getName() + " 写入数据失败", e);
                break;
            }
        }
    }

    private void decodeFile(File file) throws Exception {
        String fileStr = FileUtils.readFileToString(file, "UTF-8");
        if (StringUtil.isEmpty(fileStr)) return;
        List<CanalVO> list = JsonUtil.parseList(fileStr, CanalVO.class);
        int insertSize = 0;
        int updateSize = 0;
        int deleteSize = 0;
        int createSize = 0;
        int alterSize = 0;
        for (CanalVO canal : list) {
            if (canal.createIs()) {
                doCreateSql(canal);
                createSize++;
                continue;
            }
            if (canal.alterIs()) {
                doAlterSql(canal);
                alterSize++;
                continue;
            }
            if (canal.insertIs()) {
                doInsertSql(canal);
                insertSize++;
                continue;
            }
            if (canal.updateIs()) {
                doUpdateSql(canal);
                updateSize++;
                continue;
            }
            if (canal.deleteIs()) {
                doDeleteSql(canal);
                deleteSize++;
            }
        }
        log.debug("[解析canal文件] file：" + file.getName() + " insert size：[" + insertSize + "] update size：[" + updateSize + "] delete size：[" + deleteSize + "] create size：[" + createSize + "] alter size：[" + alterSize + "]");
    }

    protected abstract String getDbType();

    protected abstract void doCreateSql(CanalVO canalVO);

    protected abstract void doAlterSql(CanalVO canalVO);

    protected void doInsertSql(CanalVO canalVO) {
        Map<String, String> data = canalVO.getData();
        Set<String> fields = data.keySet();
        StringBuilder sql = new StringBuilder();
        List<Object> args = new ArrayList<>();
        sql.append("insert into ").append(canalVO.getTableName()).append(" ( ");
        for (String field : fields) {
            sql.append(field).append(",");
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(")").append("values").append(" ( ");
        for (String field : fields) {
            sql.append("?").append(",");
            String value = data.get(field);
            if (StringUtil.isEmpty(value)) {
                args.add(null);
            } else {
                args.add(value);
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(")");
        getDbEngine().update(sql.toString(), args.toArray());
    }

    protected void doUpdateSql(CanalVO canalVO) {
        Map<String, String> data = canalVO.getData();
        Set<String> fields = data.keySet();
        StringBuilder sql = new StringBuilder();
        List<Object> args = new ArrayList<>();
        sql.append("update ").append(canalVO.getTableName()).append(" set ");
        for (String field : fields) {
            if (field.equals(canalVO.getIdField())) continue;
            sql.append(field).append("=").append("?").append(",");
            String value = data.get(field);
            if (StringUtil.isEmpty(value)) {
                args.add(null);
            } else {
                args.add(value);
            }
        }
        sql.deleteCharAt(sql.length() - 1);
        sql.append(" where ").append(canalVO.getIdField()).append("=").append(canalVO.getId());
        getDbEngine().update(sql.toString(), args.toArray());
    }

    protected void doDeleteSql(CanalVO canalVO) {
        getDbEngine().update("delete from " + canalVO.getTableName() + " where " + canalVO.getIdField() + " = ? ", canalVO.getId());
    }

    /**
     * @Author yaoq
     * @Date j 9:55
     * @Params sql
     * @Return java.util.Map<java.lang.String, java.lang.String>
     * @Description 解析mysql create 语句， 转换成 {filed,type} 格式
     */
    protected Map<String, String> decodeCreateSql(String sql) {
        Map<String, String> map = new HashMap<>();
        String data = sql.substring(sql.indexOf("(") + 1, sql.lastIndexOf(")")).trim().replaceAll("`", "");
        String[] rows = data.split(",");
        for (String row : rows) {
            //索引不管
            if (row.contains("INDEX")) continue;
            if (row.contains("PRIMARY KEY")) {
                map.put(idField, row.substring(row.indexOf("(") + 1, row.lastIndexOf(")")));
                continue;
            }
            String[] fields = row.trim().split(" ");
            map.put(fields[0], fields[1]);
        }
        return map;
    }
}
