feat: 导入数据由gpload改成copy

scripts/import.sh

@@ -0,0 +1,15 @@
+#!/bin/expect
+
+set timeout 300
+set host [lindex $argv 0]
+set port [lindex $argv 1]
+set username [lindex $argv 2]
+set password [lindex $argv 3]
+set dbname [lindex $argv 4]
+set table [lindex $argv 5]
+set filename [lindex $argv 6]
+spawn psql -h ${host} -p ${port} -U ${username} -d ${dbname} -c "\\copy ${table} from ${filename} with csv"
+expect "Password" {
+  send "${password}\r"
+}
+expect eof

scripts/test/rollback.sh

@@ -1,6 +0,0 @@
-#!/bin/bash
-
-sh stop.sh
-rm -rf aclTousuTask.jar
-mv aclTousuTask.jar.bak aclTousuTask.jar
-sh run.sh

scripts/test/run.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-
-nohup java -jar aclTousuTask.jar >/dev/null 2>&1 &

scripts/test/stop.sh

@@ -1,5 +0,0 @@
-#!/bin/bash
-
-for i in $(pgrep -f aclTousuTask.jar); do
-  kill -9 "$i"
-done

scripts/test/update.sh

@@ -1,7 +0,0 @@
-#!/bin/bash
-
-rm -rf aclTousuTask.jar.bak
-sh stop.sh
-mv aclTousuTask.jar aclTousuTask.jar.bak
-mv aclTousuTask.jar.new aclTousuTask.jar
-sh run.sh

src/main/java/com/nokia/common/command/CommandUtil.java

@@ -0,0 +1,62 @@
+package com.nokia.common.command;
+
+import com.xxl.job.core.context.XxlJobHelper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.util.CollectionUtils;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * 执行命令工具
+ */
+@Slf4j
+public class CommandUtil {
+    /**
+     * 执行
+     *
+     * @param command 命令
+     * @return {@link List}<{@link String}>
+     * @throws IOException          ioexception
+     * @throws InterruptedException 中断异常
+     */
+    public static List<String> exec(String command) throws IOException, InterruptedException {
+        List<String> error = new ArrayList<>();
+        List<String> input = new ArrayList<>();
+        Process process = null;
+        try {
+            process = Runtime.getRuntime().exec(command);
+            try (BufferedReader inputReader = new BufferedReader(new InputStreamReader(process.getInputStream()));
+                 BufferedReader errorReader = new BufferedReader(new InputStreamReader(process.getErrorStream()))) {
+                String line;
+                // 读取错误流
+                while ((line = errorReader.readLine()) != null) {
+                    error.add(line);
+                }
+                // 读取标准流
+                while ((line = inputReader.readLine()) != null) {
+                    input.add(line);
+                }
+            }
+
+            if (!CollectionUtils.isEmpty(input)) {
+                log.info("inputStream: " + input);
+                XxlJobHelper.log("inputStream: " + input);
+            }
+            int code = process.waitFor();
+            // 非正常退出或错误流不为空
+            if (code != 0 || !CollectionUtils.isEmpty(error)) {
+                throw new RuntimeException(error.toString());
+            }
+            return input;
+        } finally {
+            if (process != null) {
+                log.info("销毁进程");
+                process.destroy();
+            }
+        }
+    }
+}

src/main/java/com/nokia/common/psql/PsqlUtil.java

@@ -0,0 +1,66 @@
+package com.nokia.common.psql;
+
+import com.nokia.common.command.CommandUtil;
+import com.xxl.job.core.context.XxlJobHelper;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * psql命令工具
+ */
+@Slf4j
+public class PsqlUtil {
+    /**
+     * 匹配psql copy成功结果
+     */
+    private static final Pattern PATTERN = Pattern.compile("^(COPY) (\\d+)$");
+
+    /**
+     * 导入csv
+     *
+     * @param script         脚本
+     * @param dbHost         数据库主机
+     * @param dbPort         数据库端口
+     * @param dbUsername     数据库用户名
+     * @param dbPassword     数据库密码
+     * @param dbName         数据库名字
+     * @param dbTable        数据库表
+     * @param csv            csv
+     * @param minInsertCount 最小值插入数
+     * @throws IOException          ioexception
+     * @throws InterruptedException 中断异常
+     */
+    public static void copyCsv(String script, String dbHost, Integer dbPort, String dbUsername, String dbPassword,
+                               String dbName, String dbTable, String csv, Long minInsertCount)
+            throws IOException, InterruptedException {
+        String command = "expect -f"
+                + " " + script
+                + " " + dbHost
+                + " " + dbPort
+                + " " + dbUsername
+                + " " + dbPassword
+                + " " + dbName
+                + " " + dbTable
+                + " " + csv;
+        List<String> list = CommandUtil.exec(command);
+        Long count = null;
+        for (String s : list) {
+            Matcher matcher = PATTERN.matcher(s);
+            while (matcher.find()) {
+                count = Long.parseLong(matcher.group(2));
+            }
+        }
+        if (count == null) {
+            throw new RuntimeException("导入数据失败: " + list);
+        }
+        log.info("插入 {} 条数据", count);
+        XxlJobHelper.log("插入 {} 条数据", count);
+        if (count < minInsertCount) {
+            throw new RuntimeException(csv + " 数据异常，少于 " + minInsertCount);
+        }
+    }
+}

src/main/java/com/nokia/config/TaskConfig.java

@@ -0,0 +1,75 @@
+package com.nokia.config;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.context.annotation.Configuration;
+
+@ConfigurationProperties("task")
+@Configuration
+@Data
+public class TaskConfig {
+    /**
+     * sftp ip
+     */
+    private String sshHost;
+    /**
+     * sftp端口
+     */
+    private Integer sshPort;
+    /**
+     * sftp账号
+     */
+    private String sshUsername;
+    /**
+     * sftp密码
+     */
+    private String sshPassword;
+    /**
+     * top用户数据文件夹
+     */
+    private String sourceDir;
+    /**
+     * 下载文件夹
+     */
+    private String downloadDir;
+    /**
+     * 去重文件名
+     */
+    private String distinctFilename;
+    /**
+     * 导入数据脚本路径
+     */
+    private String importScript;
+    /**
+     * 导入的数据库ip
+     */
+    private String dbHost;
+    /**
+     * 数据库端口
+     */
+    private Integer dbPort;
+    /**
+     * 数据库账号
+     */
+    private String dbUsername;
+    /**
+     * 数据库密码
+     */
+    private String dbPassword;
+    /**
+     * 数据库名称
+     */
+    private String dbName;
+    /**
+     * 数据库表
+     */
+    private String dbTable;
+    /**
+     * 最小插入数据
+     */
+    private Long minInsertCount;
+    /**
+     * 超时分钟
+     */
+    private Long timeout;
+}

src/main/java/com/nokia/controller/TestController.java

@@ -0,0 +1,20 @@
+package com.nokia.controller;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
+
+@Slf4j
+@RequestMapping("/test")
+@RestController
+public class TestController {
+    /**
+     * 短信告警测试
+     */
+    @GetMapping("/alert")
+    public Object alert() {
+        log.error("短信告警测试");
+        return "ok";
+    }
+}

src/main/java/com/nokia/task/CheckTask.java

@@ -0,0 +1,162 @@
+package com.nokia.task;
+
+import com.alibaba.fastjson2.JSON;
+import com.nokia.common.exception.MyRuntimeException;
+import com.nokia.config.TaskConfig;
+import com.xxl.job.core.context.XxlJobHelper;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.Data;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVParser;
+import org.apache.commons.csv.CSVRecord;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
+import org.springframework.stereotype.Component;
+import org.springframework.util.CollectionUtils;
+
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.*;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+
+@Data
+@Component
+@Slf4j
+public class CheckTask {
+    private final JdbcTemplate jdbcTemplate;
+    private final TaskConfig taskConfig;
+
+    public CheckTask(JdbcTemplate jdbcTemplate, TaskConfig taskConfig) {
+        this.jdbcTemplate = jdbcTemplate;
+        this.taskConfig = taskConfig;
+    }
+
+    /**
+     * 同步top用户信息定时任务
+     */
+    @XxlJob("check")
+    public void check() {
+        try {
+            CompletableFuture.runAsync(() -> {
+                try {
+                    singleTask();
+                } catch (Exception e) {
+                    throw new MyRuntimeException(e);
+                }
+            }).get(5, TimeUnit.MINUTES);
+        } catch (InterruptedException e) {
+            log.error("线程中断: {}", e.getMessage(), e);
+            XxlJobHelper.log("线程中断: {}", e.getMessage(), e);
+            Thread.currentThread().interrupt();
+            XxlJobHelper.handleFail(e.getMessage());
+        } catch (Exception e) {
+            log.error("发生异常了: {}", e.getMessage(), e);
+            XxlJobHelper.log("发生异常了: {}", e.getMessage(), e);
+            XxlJobHelper.handleFail(e.getMessage());
+        }
+    }
+
+    /**
+     * 单一任务
+     */
+    public void singleTask() throws IOException {
+        List<Object> modifiedUsers = new ArrayList<>();
+        try (CSVParser parser = CSVFormat.DEFAULT.builder().build()
+                .parse(new InputStreamReader(Files.newInputStream(Paths.get(taskConfig.getDistinctFilename())),
+                        StandardCharsets.UTF_8))) {
+            Set<String> loginNames = new HashSet<>();
+            Map<Object, Map<String, Object>> userMap = getUserMap();
+            for (CSVRecord t : parser) {
+                String loginName = t.get(0);
+                String orgName = t.get(2);
+                String cityId = t.get(8);
+                loginNames.add(loginName);
+                checkUserInfoChanged(modifiedUsers, userMap, orgName, loginName, cityId);
+            }
+            checkDeletedUser(loginNames);
+            checkRoleCity();
+            if (!CollectionUtils.isEmpty(modifiedUsers)) {
+                log.error("信息变化用户: {}", JSON.toJSONString(modifiedUsers));
+            }
+        }
+    }
+
+    /**
+     * 获取用户地图
+     *
+     * @return {@link Map}<{@link Object}, {@link Map}<{@link String}, {@link Object}>>
+     */
+    private Map<Object, Map<String, Object>> getUserMap() {
+        // 查询所有用户，过滤已删除和测试用户
+        String sql = "select user_id, login_name, city_id, org from sqmdb_rpt.acl_user"
+                + " where deleted = 0 and test_user = 0 order by user_id";
+        List<Map<String, Object>> allUsers = jdbcTemplate.queryForList(sql);
+        Map<Object, Map<String, Object>> userMap = new HashMap<>();
+        allUsers.forEach(t -> userMap.put(t.get("login_name"), t));
+        return userMap;
+    }
+
+    /**
+     * 检查用户信息改变了
+     *
+     * @param modifiedUsers 修改用户
+     * @param userMap       用户映射
+     * @param orgName       组织名字
+     * @param loginName     登录名
+     * @param cityId        城市标识
+     */
+    private void checkUserInfoChanged(List<Object> modifiedUsers, Map<Object, Map<String, Object>> userMap,
+                                      String orgName, String loginName, String cityId) {
+        // 检查用户信息变化
+        Map<String, Object> user = userMap.get(loginName);
+        boolean infoChanged = user != null
+                && (!cityId.equals(String.valueOf(user.get("city_id"))) || !orgName.equals(user.get("org")));
+        if (infoChanged) {
+            modifiedUsers.add(user.get("user_id"));
+            log.warn("用户 {} 信息变化: {} -> {}, {} -> {}", user.get("user_id"), user.get("city_id"), cityId,
+                    user.get("org"), orgName);
+            XxlJobHelper.log("用户 {} 信息变化: {} -> {}, {} -> {}", user.get("user_id"),
+                    user.get("city_id"), cityId, user.get("org"), orgName);
+        }
+    }
+
+    /**
+     * 检查角色城市和归属地市是否一致
+     */
+    private void checkRoleCity() {
+        String sql = "select distinct au.user_id"
+                + " from sqmdb_rpt.acl_user au"
+                + " inner join sqmdb_rpt.acl_user_role_city aurc on au.user_id = aurc.user_id"
+                + " where au.city_id != aurc.city_id"
+                + " and aurc.role_id != 3"
+                + " order by au.user_id";
+        List<Integer> diffUsers = jdbcTemplate.queryForList(sql, Integer.class);
+        if (!CollectionUtils.isEmpty(diffUsers)) {
+            log.error("权限地市不一致用户: {}", JSON.toJSONString(diffUsers));
+        }
+    }
+
+    /**
+     * 检查删除用户
+     *
+     * @param loginNames 账号列表
+     */
+    private void checkDeletedUser(Set<String> loginNames) {
+        String sql = "select user_id from sqmdb_rpt.acl_user"
+                + " where deleted = 0 and test_user = 0 and login_name not in (:loginNames)"
+                + " order by user_id";
+        Map<String, Object> paramMap = new HashMap<>();
+        paramMap.put("loginNames", loginNames);
+        NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
+        List<Integer> deletedUsers = namedParameterJdbcTemplate.queryForList(sql, paramMap, Integer.class);
+        if (!CollectionUtils.isEmpty(deletedUsers)) {
+            log.error("已删除用户: {}", JSON.toJSONString(deletedUsers));
+            XxlJobHelper.log("已删除用户: {}", JSON.toJSONString(deletedUsers));
+        }
+    }
+}

src/main/java/com/nokia/task/SyncTask.java

@@ -1,12 +1,13 @@
 package com.nokia.task;
 
-import com.alibaba.fastjson2.JSON;
 import com.jcraft.jsch.JSchException;
 import com.jcraft.jsch.SftpException;
 import com.nokia.common.exception.MyRuntimeException;
 import com.nokia.common.gpload.GploadUtil;
 import com.nokia.common.gpload.entity.GploadResult;
+import com.nokia.common.psql.PsqlUtil;
 import com.nokia.common.ssh.SSHUtil;
+import com.nokia.config.TaskConfig;
 import com.xxl.job.core.context.XxlJobHelper;
 import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.Data;
@@ -15,9 +16,7 @@ import org.apache.commons.csv.CSVFormat;
 import org.apache.commons.csv.CSVParser;
 import org.apache.commons.csv.CSVPrinter;
 import org.apache.commons.csv.CSVRecord;
-import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.core.namedparam.NamedParameterJdbcTemplate;
 import org.springframework.stereotype.Component;
 import org.springframework.util.CollectionUtils;
 
@@ -32,46 +31,24 @@ import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
 
 @Data
 @Component
 @Slf4j
-@ConfigurationProperties("task")
 public class SyncTask {
-
-    /**
-     * sftp ip
-     */
-    private String host;
-    /**
-     * sftp端口
-     */
-    private Integer port;
-    /**
-     * sftp账号
-     */
-    private String username;
     /**
-     * sftp密码
+     * 上次扫描到的文件
      */
-    private String password;
-    /**
-     * top用户数据文件夹
-     */
-    private String sourceDir;
-    /**
-     * 下载文件夹
-     */
-    private String downloadDir;
-    /**
-     * 去重文件名
-     */
-    private String distinctFilename;
+    private String lastFilename;
     private SSHUtil sshUtil;
     private final JdbcTemplate jdbcTemplate;
+    private final TaskConfig taskConfig;
 
-    public SyncTask(JdbcTemplate jdbcTemplate) {
+    public SyncTask(JdbcTemplate jdbcTemplate, TaskConfig taskConfig) {
         this.jdbcTemplate = jdbcTemplate;
+        this.taskConfig = taskConfig;
     }
 
     /**
@@ -81,26 +58,29 @@ public class SyncTask {
     @XxlJob("syncTopUser")
     public void syncTopUser() {
         try {
-            // 创建文件夹
-            Files.createDirectories(Paths.get(downloadDir));
-            sshUtil = new SSHUtil(host, port, username, password);
-            // 获取文件列表
-            List<String> list = sshUtil.ls(sourceDir);
-            if (CollectionUtils.isEmpty(list)) {
-                log.error("没有文件");
-                XxlJobHelper.log("没有文件");
-                XxlJobHelper.handleFail("没有文件");
-                return;
-            }
-
-            String filename = list.get(list.size() - 1);
-            log.info("同步的文件: {}", filename);
-            XxlJobHelper.log("同步的文件: {}", filename);
-            singleTask(filename);
+            CompletableFuture.runAsync(() -> {
+                try {
+                    singleTask();
+                } catch (InterruptedException e) {
+                    log.error("线程中断: {}", e.getMessage(), e);
+                    XxlJobHelper.log("线程中断: {}", e.getMessage(), e);
+                    Thread.currentThread().interrupt();
+                    throw new MyRuntimeException(e);
+                } catch (Exception e) {
+                    throw new MyRuntimeException(e);
+                }
+            }).get(5, TimeUnit.MINUTES);
+        } catch (InterruptedException e) {
+            log.error("线程中断: {}", e.getMessage(), e);
+            XxlJobHelper.log("线程中断: {}", e.getMessage(), e);
+            Thread.currentThread().interrupt();
+            XxlJobHelper.handleFail(e.getMessage());
+            rollback();
         } catch (Exception e) {
             log.error("发生异常了: {}", e.getMessage(), e);
             XxlJobHelper.log("发生异常了: {}", e.getMessage(), e);
             XxlJobHelper.handleFail(e.getMessage());
+            rollback();
         } finally {
             try {
                 sshUtil.disconnect();
@@ -115,12 +95,31 @@ public class SyncTask {
     /**
      * 单一任务
      *
-     * @param filename 文件名
      */
-    public void singleTask(String filename) throws JSchException, IOException, SftpException {
+    public void singleTask() throws JSchException, IOException, SftpException, InterruptedException {
+        // 创建文件夹
+        Files.createDirectories(Paths.get(taskConfig.getDownloadDir()));
+        sshUtil = new SSHUtil(taskConfig.getSshHost(), taskConfig.getSshPort(), taskConfig.getSshUsername(),
+                taskConfig.getSshPassword());
+        // 获取文件列表
+        List<String> list = sshUtil.ls(taskConfig.getSourceDir());
+        if (CollectionUtils.isEmpty(list)) {
+            log.error("没有文件");
+            XxlJobHelper.log("没有文件");
+            XxlJobHelper.handleFail("没有文件");
+            return;
+        }
+
+        String filename = list.get(list.size() - 1);
+        log.info("同步的文件: {}", filename);
+        XxlJobHelper.log("同步的文件: {}", filename);
+        if (filename.equals(lastFilename)) {
+            throw new MyRuntimeException("没有新文件");
+        }
         download(filename);
         distinct(filename);
-        gpload();
+        importData();
+        lastFilename = filename;
     }
 
     /**
@@ -131,8 +130,8 @@ public class SyncTask {
     public void download(String filename) throws JSchException, IOException, SftpException {
         log.info("下载文件: {}", filename);
         XxlJobHelper.log("下载文件: {}", filename);
-        String src = sourceDir + filename;
-        String dst = downloadDir + filename;
+        String src = taskConfig.getSourceDir() + filename;
+        String dst = taskConfig.getDownloadDir() + filename;
         sshUtil.get(src, dst);
     }
 
@@ -144,15 +143,13 @@ public class SyncTask {
     public void distinct(String filename) throws IOException {
         log.info("文件 {} 去重", filename);
         XxlJobHelper.log("文件 {} 去重", filename);
-        String inputFilePath = downloadDir + filename;
+        String inputFilePath = taskConfig.getDownloadDir() + filename;
         Path inputPath = Paths.get(inputFilePath);
-        List<Object> modifiedUsers = new ArrayList<>();
         try (CSVParser parser = CSVFormat.DEFAULT.builder().build()
                 .parse(new InputStreamReader(Files.newInputStream(inputPath), "gbk"));
-             OutputStreamWriter osw = new OutputStreamWriter(Files.newOutputStream(Paths.get(distinctFilename)),
-                     StandardCharsets.UTF_8);
+             OutputStreamWriter osw = new OutputStreamWriter(Files.newOutputStream(
+                     Paths.get(taskConfig.getDistinctFilename())), StandardCharsets.UTF_8);
              CSVPrinter printer = new CSVPrinter(osw, CSVFormat.DEFAULT)) {
-//            printer.printRecord("login_name","org_id","org_name","user_id","user_name","phone","employee_code","province_id","city_id","area_id", "province_name", "city_name", "area_name");
             Map<String, CSVRecord> csvRecordMap = new HashMap<>();
             // 按loginName去重
             for (CSVRecord t : parser) {
@@ -166,104 +163,121 @@ public class SyncTask {
             if (CollectionUtils.isEmpty(csvRecordMap)) {
                 throw new MyRuntimeException("数据为空");
             }
-            // 查询所有用户，过滤已删除和测试用户
-            String sql = "select user_id, login_name, city_id from sqmdb_rpt.acl_user"
-                    + " where deleted = 0 and test_user = 0 order by user_id";
-            List<Map<String, Object>> allUsers = jdbcTemplate.queryForList(sql);
-            Map<Object, Map<String, Object>> userMap = new HashMap<>();
-            allUsers.forEach(t -> userMap.put(t.get("login_name"), t));
-            // 查询所有地区
-            sql = "select * from sqmdb_rpt.acl_area order by type_code, area_id";
-            List<Map<String, Object>> areas = jdbcTemplate.queryForList(sql);
             Map<String, Map<String, Object>> cityMap = new HashMap<>();
             Map<Object, List<Map<String, Object>>> areaMap = new HashMap<>();
-            for (Map<String, Object> t : areas) {
-                if (t.get("type_code").equals(2)) {
-                    cityMap.put((String) t.get("area_name"), t);
-                    areaMap.putIfAbsent(t.get("area_id"), new ArrayList<>());
-                    for (Map<String, Object> tt : areas) {
-                        if (tt.get("type_code").equals(3) && tt.get("parent_id").equals(t.get("area_id"))) {
-                            areaMap.get(t.get("area_id")).add(tt);
-                        }
-                    }
-                }
-            }
-            for (CSVRecord t : csvRecordMap.values()) {
-                String orgId = t.get(0);
-                String orgName = t.get(1);
-                String userId = t.get(2);
-                String userName = t.get(3);
-                String loginName = t.get(4);
-                String phone = t.get(5);
-                String employeeCode = t.get(6);
-                Integer provinceId = -1;
-                Object cityId = -1;
-                Object areaId = null;
-//                String provinceName = "河北省";
-//                Object cityName = null;
-//                Object areaName = null;
-                for (Map.Entry<String, Map<String, Object>> tt : cityMap.entrySet()) {
-                    // 匹配地市
-                    if (orgName.contains(tt.getKey())) {
-                        Map<String, Object> area = tt.getValue();
-                        cityId = area.get("area_id");
-//                        cityName = area.get("area_name");
-                        List<Map<String, Object>> areaList = areaMap.get(area.get("area_id"));
-                        for (Map<String, Object> ttt : areaList) {
-                            // 匹配区县
-                            if (orgName.contains((CharSequence) ttt.get("area_name"))) {
-                                areaId = ttt.get("area_id");
-//                                areaName = ttt.get("area_name");
-                                break;
-                            }
-                        }
-                        break;
-                    }
-                }
-//                printer.printRecord(loginName, orgId, orgName, userId, userName, phone, employeeCode, provinceId, cityId, areaId, provinceName, cityName, areaName);
-                printer.printRecord(loginName, orgId, orgName, userId, userName, phone, employeeCode, provinceId,
-                        cityId, areaId);
-                // 检查用户信息变化
-                Map<String, Object> user = userMap.get(loginName);
-                if (user != null && !cityId.equals(user.get("city_id"))) {
-                    modifiedUsers.add(user.get("user_id"));
-                    log.debug("用户 {} 地市变化: {} -> {}", user.get("user_id"), user.get("city_id"), cityId);
-                }
-            }
+            getAreaMap(cityMap, areaMap);
+            matchArea(printer, csvRecordMap, cityMap, areaMap);
             log.info("文件 {} 去重完成", filename);
             XxlJobHelper.log("文件 {} 去重完成", filename);
             // 删除本地源文件
             Files.deleteIfExists(inputPath);
             log.info("删除本地源文件 {}", filename);
             XxlJobHelper.log("删除本地源文件 {}", filename);
-            // 检查删除用户
-            sql = "select user_id from sqmdb_rpt.acl_user"
-                    + " where deleted = 0 and test_user = 0 and login_name not in (:loginNames)"
-                    + " order by user_id";
-            Map<String, Object> paramMap = new HashMap<>();
-            paramMap.put("loginNames", csvRecordMap.keySet());
-            NamedParameterJdbcTemplate namedParameterJdbcTemplate = new NamedParameterJdbcTemplate(jdbcTemplate);
-            List<Integer> deletedUsers = namedParameterJdbcTemplate.queryForList(sql, paramMap, Integer.class);
-            if (!CollectionUtils.isEmpty(deletedUsers)) {
-                log.error("已删除用户: {}", JSON.toJSONString(deletedUsers));
-            }
-            // 检查权限地市
-            sql = "select distinct au.user_id"
-                    + " from sqmdb_rpt.acl_user au"
-                    + " inner join sqmdb_rpt.acl_user_role_city aurc on au.user_id = aurc.user_id"
-                    + " where au.city_id != aurc.city_id"
-                    + " and aurc.role_id != 3"
-                    + " order by au.user_id";
-            List<Integer> diffUsers = jdbcTemplate.queryForList(sql, Integer.class);
-            if (!CollectionUtils.isEmpty(diffUsers)) {
-                log.error("权限地市不一致用户: {}", JSON.toJSONString(diffUsers));
+        }
+    }
+
+    /**
+     * 得到区域地图
+     *
+     * @param cityMap 城市地图
+     * @param areaMap 区域地图
+     */
+    private void getAreaMap(Map<String, Map<String, Object>> cityMap, Map<Object, List<Map<String, Object>>> areaMap) {
+        // 查询所有地区
+        String sql = "select * from sqmdb_rpt.acl_area order by type_code, area_id";
+        List<Map<String, Object>> areas = jdbcTemplate.queryForList(sql);
+        for (Map<String, Object> t : areas) {
+            if (t.get("type_code").equals(2)) {
+                cityMap.put((String) t.get("area_name"), t);
+                areaMap.putIfAbsent(t.get("area_id"), new ArrayList<>());
+                for (Map<String, Object> tt : areas) {
+                    if (tt.get("type_code").equals(3) && tt.get("parent_id").equals(t.get("area_id"))) {
+                        areaMap.get(t.get("area_id")).add(tt);
+                    }
+                }
             }
-            if (!CollectionUtils.isEmpty(modifiedUsers)) {
-                log.error("地市变化用户: {}", JSON.toJSONString(modifiedUsers));
+        }
+    }
+
+    /**
+     * 匹配区域
+     *
+     * @param printer       打印机
+     * @param csvRecordMap  csv记录地图
+     * @param cityMap       城市地图
+     * @param areaMap       区域地图
+     */
+    private void matchArea(CSVPrinter printer, Map<String, CSVRecord> csvRecordMap,
+                           Map<String, Map<String, Object>> cityMap, Map<Object, List<Map<String, Object>>> areaMap)
+            throws IOException {
+        for (CSVRecord t : csvRecordMap.values()) {
+            String orgId = t.get(0);
+            String orgName = t.get(1);
+            String userId = t.get(2);
+            String userName = t.get(3);
+            String loginName = t.get(4);
+            String phone = t.get(5);
+            String employeeCode = t.get(6);
+            Integer provinceId = -1;
+            Object cityId = -1;
+            Object areaId = null;
+            for (Map.Entry<String, Map<String, Object>> tt : cityMap.entrySet()) {
+                // 匹配地市
+                if (orgName.contains(tt.getKey())) {
+                    Map<String, Object> area = tt.getValue();
+                    cityId = area.get("area_id");
+                    List<Map<String, Object>> areaList = areaMap.get(area.get("area_id"));
+                    for (Map<String, Object> ttt : areaList) {
+                        // 匹配区县
+                        if (orgName.contains((CharSequence) ttt.get("area_name"))) {
+                            areaId = ttt.get("area_id");
+                            break;
+                        }
+                    }
+                    break;
+                }
             }
+            printer.printRecord(loginName, orgId, orgName, userId, userName, phone, employeeCode, provinceId,
+                    cityId, areaId);
         }
     }
 
+    /**
+     * 导入数据
+     *
+     */
+    public void importData() throws IOException, InterruptedException {
+        String sql = "drop table if exists sqmdb_rpt.acl_top_user_bak";
+        log.info(sql);
+        XxlJobHelper.log(sql);
+        jdbcTemplate.execute(sql);
+        sql = "create table sqmdb_rpt.acl_top_user_bak as select * from sqmdb_rpt.acl_top_user";
+        log.info(sql);
+        XxlJobHelper.log(sql);
+        jdbcTemplate.execute(sql);
+        sql = "truncate table sqmdb_rpt.acl_top_user";
+        log.info(sql);
+        XxlJobHelper.log(sql);
+        jdbcTemplate.execute(sql);
+        PsqlUtil.copyCsv(taskConfig.getImportScript(), taskConfig.getDbHost(), taskConfig.getDbPort(),
+                taskConfig.getDbUsername(), taskConfig.getDbPassword(), taskConfig.getDbName(), taskConfig.getDbTable(),
+                taskConfig.getDistinctFilename(), taskConfig.getMinInsertCount());
+    }
+
+    /**
+     * 回滚
+     */
+    public void rollback() {
+        String sql = "drop table if exists sqmdb_rpt.acl_top_user";
+        log.warn(sql);
+        XxlJobHelper.log(sql);
+        jdbcTemplate.execute(sql);
+        sql = "alter table sqmdb_rpt.acl_top_user_bak rename to acl_top_user";
+        log.warn(sql);
+        XxlJobHelper.log(sql);
+        jdbcTemplate.execute(sql);
+    }
+
     public void gpload() throws IOException {
         String gploadCommand = "sh gpload/gpload.sh";
         GploadResult gpload = GploadUtil.gpload(gploadCommand);
@@ -271,7 +285,7 @@ public class SyncTask {
             log.info("gpload完成: {}", gpload);
             XxlJobHelper.log("gpload完成: {}", gpload);
             // 删除文件
-            Files.deleteIfExists(Paths.get(distinctFilename));
+            Files.deleteIfExists(Paths.get(taskConfig.getDistinctFilename()));
         } else {
             throw new MyRuntimeException("gpload失败: " + gpload.getMessage());
         }

src/main/resources/application-product.properties

@@ -10,13 +10,22 @@ spring.datasource.driver-class-name=org.postgresql.Driver
 spring.datasource.url=jdbc:postgresql://192.168.70.109:5432/sqmmt
 spring.datasource.username=sqmdb
 spring.datasource.password=sqmdb_1QAZ
-task.host=192.168.70.130
-task.port=22
-task.username=do
-task.password=Richr00t
-task.sourceDir=/data1/esbftp/top_user/
-task.downloadDir=download/
-task.distinctFilename=top-user.csv
+task.ssh-host=192.168.70.130
+task.ssh-port=22
+task.ssh-username=do
+task.ssh-password=Richr00t
+task.source-dir=/data1/esbftp/top_user/
+task.download-dir=download/
+task.distinct-filename=top-user.csv
+task.min-insert-count=1
+task.import-script=import.sh
+task.db-host=192.168.70.109
+task.db-port=5432
+task.db-username=sqmdb
+task.db-password=sqmdb_1QAZ
+task.db-name=sqmmt
+task.db-table=sqmdb_rpt.acl_top_user
+task.timeout=5
 ### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
 xxl.job.admin.addresses=http://192.168.10.7:8087/xxl-job-admin
 ### xxl-job, access token

src/main/resources/application-test.properties

@@ -8,13 +8,22 @@ spring.datasource.driver-class-name=org.postgresql.Driver
 spring.datasource.url=jdbc:postgresql://192.168.50.5:5432/sqmmt
 spring.datasource.username=sqmdb
 spring.datasource.password=sqmdb_1QAZ
-task.host=192.168.70.130
-task.port=22
-task.username=do
-task.password=Richr00t
-task.sourceDir=/data1/esbftp/top_user/
-task.downloadDir=download/
-task.distinctFilename=top-user.csv
+task.ssh-host=192.168.70.130
+task.ssh-port=22
+task.ssh-username=do
+task.ssh-password=Richr00t
+task.source-dir=/data1/esbftp/top_user/
+task.download-dir=download/
+task.distinct-filename=top-user.csv
+task.min-insert-count=1
+task.import-script=import.sh
+task.db-host=192.168.50.5
+task.db-port=5432
+task.db-username=sqmdb
+task.db-password=sqmdb_1QAZ
+task.db-name=sqmmt
+task.db-table=sqmdb_rpt.acl_top_user
+task.timeout=5
 ### xxl-job admin address list, such as "http://address" or "http://address01,http://address02"
 xxl.job.admin.addresses=http://192.168.10.7:8087/xxl-job-admin
 ### xxl-job, access token

src/main/resources/application.properties

@@ -1,2 +1,2 @@
 # 启用的配置
-spring.profiles.active=test
+spring.profiles.active=product

src/main/resources/logback-spring.xml

@@ -33,7 +33,7 @@
             <totalSizeCap>20GB</totalSizeCap>
         </rollingPolicy>
         <encoder>
-            <Pattern>%d %highlight(%-5level) %magenta([%thread]) %cyan(%logger:%line) %msg%n</Pattern>
+            <Pattern>%d %-5level %X{traceId} [%thread] %logger:%line %msg%n</Pattern>
             <charset>UTF-8</charset>
         </encoder>
         <filter class="ch.qos.logback.classic.filter.LevelFilter">

src/test/java/com/nokia/acl_tousu_task/AclTousuTaskApplicationTests.java

@@ -1,5 +1,6 @@
 package com.nokia.acl_tousu_task;
 
+import com.nokia.task.CheckTask;
 import com.nokia.task.SyncTask;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.jupiter.api.Test;
@@ -13,9 +14,16 @@ import org.springframework.test.context.ActiveProfiles;
 class AclTousuTaskApplicationTests {
     @Autowired
     private SyncTask syncTask;
+    @Autowired
+    private CheckTask checkTask;
 
     @Test
     void testSync() {
         syncTask.syncTopUser();
     }
+
+    @Test
+    void testCheck() {
+        checkTask.check();
+    }
 }