1.0 整合了api和定时任务

.gitignore

@@ -0,0 +1,5 @@
+/target/
+/result/
+/log/
+*.iml
+comp-connector-1.0.jar

other/gpload配置表/g5.yml

@@ -0,0 +1,42 @@
+VERSION: 1.0.0.1
+DATABASE: localdb
+USER: gpadmin
+HOST: 10.170.42.48
+PORT: 5432
+GPLOAD:
+  INPUT:
+    - SOURCE:
+        LOCAL_HOSTNAME:
+          - 10.170.42.93
+        PORT: 1234
+        FILE:
+          - /data/terminal/distinct/g5_terminal_20220123.csv
+    - FORMAT: csv
+    - DELIMITER: '|'
+    - HEADER: false
+    - ENCODING: utf-8
+    - ERROR_LIMIT: 20000000
+    - LOG_ERRORS: true
+  OUTPUT:
+    - TABLE: terminal_info.o_info_list_5g_terminal
+    - MODE: merge
+    - MATCH_COLUMNS:
+        - msisdn
+    - UPDATE_COLUMNS:
+        - imsi
+        - imei
+        - tac
+        - fact_name
+        - terminal_name
+        - hprovince
+        - hprovince_name
+        - hcity
+        - hcity_name
+        - vprovince
+        - vprovince_name
+        - vcity_name
+        - roaming_city_name
+        - cell_most
+        - used_5g
+        - status_5g
+        - day

other/gpload配置表/volte.yml

@@ -0,0 +1,40 @@
+VERSION: 1.0.0.1
+DATABASE: localdb
+USER: gpadmin
+HOST: 10.170.42.48
+PORT: 5432
+GPLOAD:
+  INPUT:
+    - SOURCE:
+        LOCAL_HOSTNAME:
+          - 10.170.42.93
+        PORT: 1234
+        FILE:
+          - /data/terminal/distinct/volte_terminal_20220123.csv
+    - FORMAT: csv
+    - DELIMITER: '|'
+    - HEADER: false
+    - ENCODING: utf-8
+    - ERROR_LIMIT: 20000000
+    - LOG_ERRORS: true
+  OUTPUT:
+    - TABLE: terminal_info.o_info_list_volte_terminal
+    - MODE: merge
+    - MATCH_COLUMNS:
+        - msisdn
+    - UPDATE_COLUMNS:
+        - imsi
+        - imei
+        - tac
+        - fact_name
+        - terminal_name
+        - hprovince
+        - hprovince_name
+        - hcity
+        - hcity_name
+        - vprovince
+        - vprovince_name
+        - vcity
+        - vcity_name
+        - is_volte
+        - day_date

other/http/http测试.http

@@ -0,0 +1,12 @@
+POST http://localhost:12900/terminal/5g?phoneNumber=13231899751
+Content-Type: application/x-www-form-urlencoded
+
+###
+POST http://localhost:12900/terminal/volte?phoneNumber=13231899751
+Content-Type: application/x-www-form-urlencoded
+
+###
+GET http://localhost:12900/
+Accept: application/json
+
+###

other/shell脚本/gpload.sh

@@ -0,0 +1,25 @@
+#!/bin/bash
+
+date=$1
+
+if [ $2 == 5g ]; then
+  # 修改gpload配置文件
+  sed -i 's/\/data\/terminal\/distinct\/g5_terminal_[0-9]\{8\}.csv/\/data\/terminal\/distinct\/g5_terminal_'${date}'.csv/' /data/terminal/gpload/g5.yml
+  yml=g5.yml
+elif [ $2 == volte ]; then
+  # 修改gpload配置文件
+  sed -i 's/\/data\/terminal\/distinct\/volte_terminal_[0-9]\{8\}.csv/\/data\/terminal\/distinct\/volte_terminal_'${date}'.csv/' /data/terminal/gpload/volte.yml
+  yml=volte.yml
+fi
+
+# 由于gpload需要输入密码，这里需要使用expect执行
+password=Richr00t!
+expect -c "
+set timeout 300
+spawn gpload -f /data/terminal/gpload/${yml}
+expect {
+\"connecting (yes/no)?\" { send \"yes\n\";exp_continue }
+\"Password:\" { send \"${password}\n\"; exp_continue}
+timeout { puts \"超时\" exit 2}
+}
+"

other/shell脚本/gpload_5g.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+
+########
+# 废弃
+########
+
+
+date=$1
+
+# 修改gpload配置文件
+sed -i 's/\/data\/gpload\/g5_terminal_[0-9]\{8\}.csv/\/data\/gpload\/g5_terminal_'${date}'.csv/' /data/gpload/g5.yml
+
+# 由于gpload需要输入密码，这里需要使用expect执行
+password=Richr00t
+expect -c "
+set timeout 300
+spawn gpload -f /data/gpload/g5.yml
+expect {
+\"connecting (yes/no)?\" { send \"yes\n\";exp_continue }
+\"Password:\" { send \"${password}\n\"; exp_continue}
+timeout { puts \"超时\" exit 2}
+}
+"

other/shell脚本/gpload_volte.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+
+########
+# 废弃
+########
+
+date=$1
+
+# 修改gpload配置文件
+sed -i 's/\/data\/gpload\/volte_terminal_[0-9]\{8\}.csv/\/data\/gpload\/volte_terminal_'${date}'.csv/' /data/gpload/volte.yml
+
+# 由于gpload需要输入密码，这里需要使用expect执行
+password=Richr00t
+expect -c "
+set timeout 300
+spawn gpload -f /data/gpload/volte.yml
+expect {
+\"connecting (yes/no)?\" { send \"yes\n\";exp_continue }
+\"Password:\" { send \"${password}\n\"; exp_continue}
+timeout { puts \"超时\" exit 2}
+}
+"

other/shell脚本/scp_file.sh

@@ -0,0 +1,32 @@
+#!/bin/bash
+
+## 验证参数大于2个参数
+#if [ $# -lt 2 ]; then
+#  echo "USAGE: $0 需要且只能携带两个日期参数，如20220101 20220102"
+#  exit 1
+#fi
+
+#if [ $# -eq 2 ];
+#then
+#  filename="814153986171547648_5G终端及5G用户数据含5G终端开关状态_hprovince_DB服务_${1}_1_12788_${2}1220.dat.gz"
+#elif [ $# -eq 1 ];
+#then
+#  filename="752951054268059649_intf_list_volte_users_day_${1}_0_127.dat"
+#else
+#  echo "输入1个时间参数如20220101表示scp volte文件，输入2个时间参数如20220101 20220102表示scp 5G文件"
+#  exit
+#fi
+
+filename=$1
+localPath=$2
+
+# scp需要密码
+password=Richr00t!fast
+expect -c "
+set timeout 600
+spawn scp do@133.96.92.136:/data/esbdata/WY/${filename} ${localPath}
+expect {
+\"password:\" { send \"${password}\n\"; exp_continue}
+timeout {send \"echo timeout\n\"; exit 2}
+}
+"

other/sql脚本/g5.sql

@@ -0,0 +1,29 @@
+create table terminal_info.o_info_list_5g_terminal
+(
+    msisdn            varchar not null
+        constraint o_info_list_5g_terminal_pk
+        primary key,
+    imsi              varchar,
+    imei              varchar,
+    tac               varchar,
+    fact_name         varchar,
+    terminal_name     varchar,
+    hprovince         varchar,
+    hprovince_name    varchar,
+    hcity             varchar,
+    hcity_name        varchar,
+    vprovince         varchar,
+    vprovince_name    varchar,
+    vcity_name        varchar,
+    roaming_city_name varchar,
+    cell_most         varchar,
+    used_5g           varchar,
+    status_5g         varchar,
+    day               varchar
+)
+    distributed by (msisdn);
+
+alter table terminal_info.o_info_list_5g_terminal
+    owner to lifuq;
+
+

other/sql脚本/task_record.sql

@@ -0,0 +1,19 @@
+create table terminal_info.task_record
+(
+    id          bigint not null
+        constraint task_record_pk
+            primary key,
+    task_type        varchar,
+    task_date        varchar,
+    task_status        varchar,
+    last_update_time        timestamp,
+    inserted    integer,
+    updated     integer,
+    error       integer,
+    create_time timestamp default now()
+)
+    distributed by (id);
+
+alter table terminal_info.task_record
+    owner to lifuq;
+

other/sql脚本/volte.sql

@@ -0,0 +1,27 @@
+-- 建表语句
+create table terminal_info.o_info_list_volte_terminal
+(
+    msisdn         varchar not null
+        constraint o_info_list_volte_terminal_pk
+        primary key,
+    imsi           varchar,
+    imei           varchar,
+    tac            varchar,
+    fact_name      varchar,
+    terminal_name  varchar,
+    hprovince      varchar,
+    hprovince_name varchar,
+    hcity          varchar,
+    hcity_name     varchar,
+    vprovince      varchar,
+    vprovince_name varchar,
+    vcity          varchar,
+    vcity_name     varchar,
+    is_volte       varchar,
+    day_date       varchar
+)
+    distributed by (msisdn);
+
+alter table terminal_info.o_info_list_volte_terminal
+    owner to lifuq;
+

other/sql脚本/创建数据库、模式和用户.sql

@@ -0,0 +1,13 @@
+-- 创建数据库
+create database localdb;
+
+-- 创建用户
+create role lifuq with login encrypted password 'Mdasil789!@#';
+
+-- 创建模式
+create schema terminal_info;
+
+-- 赋予权限
+grant all on schema terminal_info to lifuq;
+grant create on schema terminal_info to lifuq;
+grant usage on schema terminal_info to lifuq;

pom.xml

@@ -0,0 +1,70 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <groupId>top.lifuquan</groupId>
+    <artifactId>terminal</artifactId>
+    <version>1.0</version>
+
+    <packaging>jar</packaging>
+
+    <parent>
+        <artifactId>spring-boot-starter-parent</artifactId>
+        <groupId>org.springframework.boot</groupId>
+        <version>2.1.18.RELEASE</version>
+        <relativePath/>
+    </parent>
+
+    <properties>
+        <maven.compiler.source>8</maven.compiler.source>
+        <maven.compiler.target>8</maven.compiler.target>
+    </properties>
+
+    <dependencies>
+
+        <dependency>
+            <groupId>com.baomidou</groupId>
+            <artifactId>mybatis-plus-boot-starter</artifactId>
+            <version>3.3.2</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.postgresql</groupId>
+            <artifactId>postgresql</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-csv</artifactId>
+            <version>1.9.0</version>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-web</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.projectlombok</groupId>
+            <artifactId>lombok</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.springframework.boot</groupId>
+            <artifactId>spring-boot-starter-test</artifactId>
+        </dependency>
+    </dependencies>
+
+    <build>
+        <plugins>
+            <plugin>
+                <groupId>org.springframework.boot</groupId>
+                <artifactId>spring-boot-maven-plugin</artifactId>
+            </plugin>
+        </plugins>
+        <finalName>terminalApplication</finalName>
+    </build>
+
+</project>

readme.md

@@ -0,0 +1,26 @@
+# 思路
+
+## 20220304
+
+首先要将文件从本地搬运到集团服务器
+
+早上测试过，两侧速度基本都在8MBps，传2G多的文件用时6分钟，可以接受
+
+方案一：从本地定时扫描，发现文件生成后上传到集团
+
+缺点：部署分散，维护比较难，还要考虑过一段时间本地断网的情况
+
+方案二：集团定时从本地scp文件
+
+程序部署位置：最好是10.170.42.91，但是91上没有gpload客户端,考虑部署在10.170.42.93
+
+现在先考虑方案二
+
+每日任务步骤：
+
+1. scp文件(09->93)
+
+2. 文件去重
+
+3. gpload入数据库
+

src/main/java/top/lifuquan/TerminalApplication.java

@@ -0,0 +1,15 @@
+package top.lifuquan;
+
+import org.mybatis.spring.annotation.MapperScan;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.scheduling.annotation.EnableScheduling;
+
+@SpringBootApplication
+@MapperScan("top.lifuquan.dao")
+@EnableScheduling
+public class TerminalApplication {
+    public static void main(String[] args) {
+        SpringApplication.run(TerminalApplication.class, args);
+    }
+}

src/main/java/top/lifuquan/config/ScheduledConfig.java

@@ -0,0 +1,46 @@
+package top.lifuquan.config;
+
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.scheduling.annotation.SchedulingConfigurer;
+import org.springframework.scheduling.config.ScheduledTaskRegistrar;
+import org.springframework.scheduling.support.CronTrigger;
+import top.lifuquan.service.CronService;
+
+import javax.annotation.Resource;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.concurrent.Executors;
+
+@Slf4j
+// @Configuration
+public class ScheduledConfig implements SchedulingConfigurer {
+    // 由配置文件指定线程数
+    @Value("${scheduled.pool.size}")
+    private Integer poolSize;
+    @Value("${scheduled.cron}")
+    private String cron;
+    @Value("${scheduled.delay}")
+    private long delay;
+
+    private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy年MM月dd日");
+
+    @Resource
+    private CronService cronService;
+
+    @Override
+    public void configureTasks(ScheduledTaskRegistrar scheduledTaskRegistrar) {
+        scheduledTaskRegistrar.setScheduler(Executors.newScheduledThreadPool(poolSize));
+        scheduledTaskRegistrar.addTriggerTask(
+                () -> {
+                    // 回溯延迟时间
+                    Date date = new Date(new Date().getTime() - delay * 1000 * 3600 * 24);
+                    log.info("开始执行定时任务--" + dateFormat.format(date));
+                    cronService.run5gTask(date);
+                    cronService.runVolteTask(date);
+                },
+                triggerContext -> new CronTrigger(cron).nextExecutionTime(triggerContext)
+        );
+    }
+}

src/main/java/top/lifuquan/controller/TerminalInfoController.java

@@ -0,0 +1,44 @@
+package top.lifuquan.controller;
+
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RequestMethod;
+import org.springframework.web.bind.annotation.RequestParam;
+import org.springframework.web.bind.annotation.ResponseBody;
+import top.lifuquan.pojo.G5TerminalInfo;
+import top.lifuquan.pojo.VolteTerminalInfo;
+import top.lifuquan.service.TerminalInfoService;
+import top.lifuquan.vo.G5TerminalInfoVo;
+import top.lifuquan.vo.R;
+import top.lifuquan.vo.VolteTerminalInfoVo;
+
+import javax.annotation.Resource;
+import java.text.ParseException;
+
+@Controller
+@RequestMapping("terminal")
+public class TerminalInfoController {
+
+    @Resource
+    private TerminalInfoService terminalInfoService;
+
+    @ResponseBody
+    @RequestMapping(value = "volte", method = RequestMethod.POST)
+    public R getVoLteTerminalInfo(@RequestParam String phoneNumber) throws ParseException {
+        VolteTerminalInfo volteTerminalInfo = terminalInfoService.findVolteInfoByPhoneNumber(phoneNumber);
+        if (volteTerminalInfo != null) {
+            return R.ok().data("data", new VolteTerminalInfoVo(volteTerminalInfo));
+        }
+        return R.error().message("未找到该号码VoLte相关终端信息！！");
+    }
+
+    @ResponseBody
+    @RequestMapping(value = "5g", method = RequestMethod.POST)
+    public R get5GTerminalInfo(@RequestParam String phoneNumber) throws ParseException {
+        G5TerminalInfo terminalInfo = terminalInfoService.find5GInfoByPhoneNumber(phoneNumber);
+        if (terminalInfo != null) {
+            return R.ok().data("data", new G5TerminalInfoVo(terminalInfo));
+        }
+        return R.error().message("未找到该号码5G相关终端信息！！");
+    }
+}

src/main/java/top/lifuquan/controller/TestController.java

@@ -0,0 +1,15 @@
+package top.lifuquan.controller;
+
+import org.springframework.stereotype.Controller;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.ResponseBody;
+
+@Controller
+public class TestController {
+
+    @ResponseBody
+    @RequestMapping("")
+    public String test() {
+        return "当你看到这句话,说明这个应用已经正常启动了";
+    }
+}

src/main/java/top/lifuquan/dao/G5TerminalInfoDao.java

@@ -0,0 +1,8 @@
+package top.lifuquan.dao;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import top.lifuquan.pojo.G5TerminalInfo;
+
+public interface G5TerminalInfoDao extends BaseMapper<G5TerminalInfo> {
+
+}

src/main/java/top/lifuquan/dao/TaskRecordDao.java

@@ -0,0 +1,7 @@
+package top.lifuquan.dao;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import top.lifuquan.pojo.TaskRecord;
+
+public interface TaskRecordDao extends BaseMapper<TaskRecord> {
+}

src/main/java/top/lifuquan/dao/VolteTerminalInfoDao.java

@@ -0,0 +1,8 @@
+package top.lifuquan.dao;
+
+import com.baomidou.mybatisplus.core.mapper.BaseMapper;
+import top.lifuquan.pojo.VolteTerminalInfo;
+
+public interface VolteTerminalInfoDao extends BaseMapper<VolteTerminalInfo> {
+
+}

src/main/java/top/lifuquan/pojo/G5TerminalInfo.java

@@ -0,0 +1,28 @@
+package top.lifuquan.pojo;
+
+import com.baomidou.mybatisplus.annotation.TableField;
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+@Data
+@TableName("terminal_info.o_info_list_5g_terminal")
+public class G5TerminalInfo {
+    private String msisdn;
+    /**
+     * 字段说明：
+     * 在统计时段内，用户是否接入过5G网络
+     * 0：未登网   1：登网
+     */
+    @TableField("used_5g")
+    private Integer used5g;
+    /**
+     * 字段说明：
+     * 统计当天5G开关是否曾打开
+     * true：打开   false：未打开
+     */
+    @TableField("status_5g")
+    private String status5g;
+    private String day;
+    private String factName;
+    private String terminalName;
+}

src/main/java/top/lifuquan/pojo/TaskRecord.java

@@ -0,0 +1,20 @@
+package top.lifuquan.pojo;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+import java.util.Date;
+
+@Data
+@TableName("terminal_info.task_record")
+public class TaskRecord {
+    private Long id;
+    private String taskType;
+    private String taskDate;
+    private String taskStatus;
+    private Date lastUpdateTime;
+    private Integer inserted;
+    private Integer updated;
+    private Integer error;
+    private Date createTime;
+}

src/main/java/top/lifuquan/pojo/VolteTerminalInfo.java

@@ -0,0 +1,27 @@
+package top.lifuquan.pojo;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+
+@Data
+@TableName("terminal_info.o_info_list_volte_terminal")
+public class VolteTerminalInfo {
+
+    private String msisdn;
+    /**
+     * 字段说明：
+     * 硬件无法识别：-3
+     * 其它终端硬件支持但无法识别软件是否支持或开关是否打开：-2
+     * 苹果终端硬件支持但无法识别软件是否支持或开关是否打开：-1
+     * 硬件不支持：0
+     * 硬件支持软件不支持：1
+     * 硬件软件支持但开关未打开：2
+     * 硬件软件支持且开关打开：3
+     */
+    private Integer isVolte;
+    private String dayDate;
+    private String factName;
+    private String terminalName;
+    // 归属省
+    private String hprovinceName;
+}

src/main/java/top/lifuquan/service/CronService.java

@@ -0,0 +1,118 @@
+package top.lifuquan.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import top.lifuquan.pojo.TaskRecord;
+import top.lifuquan.util.CsvUtil;
+import top.lifuquan.util.ProcessUtil;
+
+import javax.annotation.Resource;
+import java.nio.file.Paths;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+@Slf4j
+@Service
+public class CronService {
+
+    private final String localPath = "/data/terminal/source";
+    private final String distinctPath = "/data/terminal/distinct";
+
+    private final String distinctPrefix_5g = "g5_terminal_";
+    private final String taskType_5g = "5g";
+
+    private final String distinctPrefix_volte = "volte_terminal_";
+    private final String taskType_volte = "volte";
+
+    private final SimpleDateFormat dateFormat = new SimpleDateFormat("yyyyMMdd");
+
+    @Resource
+    private TaskRecordService recordService;
+
+    public void run5gTask(Date date) {
+        log.debug("开始执行{}任务", taskType_5g);
+        String dateStr = dateFormat.format(date);
+        // 获取任务记录
+        TaskRecord taskRecord = new TaskRecord();
+        taskRecord.setTaskType(taskType_5g);
+        taskRecord.setTaskDate(dateStr);
+        TaskRecord record = recordService.getOne(new QueryWrapper<>(taskRecord));
+        if (record != null) {
+            log.debug("已存在任务记录" + record.toString());
+            return;
+        }
+        // 开始执行任务
+        String dateString2 = dateFormat.format(date.getTime() + 1000 * 3600 * 24);
+        String fileName = "814153986171547648_5G终端及5G用户数据含5G终端开关状态_hprovince_DB服务_" + dateStr + "_1_12788_" + dateString2 + "1220.dat.gz";
+        // 1. scp文件到本地
+        log.info("准备下载文件 " + fileName);
+        boolean b = ProcessUtil.scpFile(fileName, localPath);
+        if (!b) {
+            log.info("文件{}scp失败", fileName);
+            taskRecord.setTaskStatus("scp失败");
+            recordService.save(taskRecord);
+            return;
+        }
+        taskRecord.setTaskStatus("scp完成");
+        // 2. 对文件进行去重
+        String localFilePath = Paths.get(localPath, fileName).toString();
+        String distinctFilePath = Paths.get(distinctPath, distinctPrefix_5g + dateStr + ".csv").toString();
+        log.debug("准备筛选文件 " + localFilePath + " 并将不重复的记录写入 " + distinctFilePath);
+        b = CsvUtil.distinct(localFilePath, distinctFilePath, 0, "|");
+        if (!b) {
+            log.info("文件{}去重失败", localFilePath);
+            taskRecord.setTaskStatus("去重失败");
+            recordService.save(taskRecord);
+            return;
+        }
+        taskRecord.setTaskStatus("去重完成");
+        // 3. gpload
+        taskRecord = ProcessUtil.gpLoad(dateStr, taskRecord);
+        taskRecord.setLastUpdateTime(new Date());
+        recordService.save(taskRecord);
+    }
+
+    public void runVolteTask(Date date) {
+        log.debug("开始执行{}任务", taskType_volte);
+        String dateStr = dateFormat.format(date);
+        // 获取任务记录
+        TaskRecord taskRecord = new TaskRecord();
+        taskRecord.setTaskType(taskType_volte);
+        taskRecord.setTaskDate(dateStr);
+        TaskRecord record = recordService.getOne(new QueryWrapper<>(taskRecord));
+        if (record != null) {
+            log.debug("已存在任务记录" + record.toString());
+            return;
+        }
+        // 开始执行任务
+        String fileName = "752951054268059649_intf_list_volte_users_day_" + dateStr + "_0_127.dat";
+        // 1. scp文件到本地
+        log.info("准备下载文件 " + fileName);
+        boolean b = ProcessUtil.scpFile(fileName, localPath);
+        if (!b) {
+            log.info("文件{}scp失败", fileName);
+            taskRecord.setTaskStatus("scp失败");
+            recordService.save(taskRecord);
+            return;
+        }
+        taskRecord.setTaskStatus("scp完成");
+        // 2. 对文件进行去重
+        String localFilePath = Paths.get(localPath, fileName).toString();
+        String distinctFilePath = Paths.get(distinctPath, distinctPrefix_volte + dateStr + ".csv").toString();
+        log.debug("准备筛选文件 " + localFilePath + " 并将不重复的记录写入 " + distinctFilePath);
+        b = CsvUtil.distinct(localFilePath, distinctFilePath, 0, "|");
+        if (!b) {
+            log.info("文件{}去重失败", localFilePath);
+            taskRecord.setTaskStatus("去重失败");
+            recordService.save(taskRecord);
+            return;
+        }
+        taskRecord.setTaskStatus("去重完成");
+        // 3. gpload
+        taskRecord = ProcessUtil.gpLoad(dateStr, taskRecord);
+        taskRecord.setLastUpdateTime(new Date());
+        recordService.save(taskRecord);
+    }
+}

src/main/java/top/lifuquan/service/TaskRecordService.java

@@ -0,0 +1,33 @@
+package top.lifuquan.service;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
+import org.springframework.stereotype.Service;
+import top.lifuquan.dao.TaskRecordDao;
+import top.lifuquan.pojo.TaskRecord;
+
+@Service
+public class TaskRecordService extends ServiceImpl<TaskRecordDao, TaskRecord> {
+
+    /**
+     * 查询某类型的某天的任务记录
+     *
+     * @param taskType 任务类型
+     * @param taskDate 任务时间
+     * @return 任务记录
+     */
+    public TaskRecord getRecord(String taskType, String taskDate) {
+        TaskRecord taskRecord = new TaskRecord();
+        taskRecord.setTaskType(taskType);
+        taskRecord.setTaskDate(taskDate);
+        TaskRecord record = getOne(new QueryWrapper<>(taskRecord));
+        if (record != null) {
+            // 记录已存在，直接返回
+            return record;
+        } else {
+            // 记录不存在，保存一条记录
+            taskRecord.setTaskStatus("任务开始");
+            return taskRecord;
+        }
+    }
+}

src/main/java/top/lifuquan/service/TerminalInfoService.java

@@ -0,0 +1,32 @@
+package top.lifuquan.service;
+
+import com.baomidou.mybatisplus.core.toolkit.Wrappers;
+import org.springframework.stereotype.Service;
+import top.lifuquan.dao.G5TerminalInfoDao;
+import top.lifuquan.dao.VolteTerminalInfoDao;
+import top.lifuquan.pojo.G5TerminalInfo;
+import top.lifuquan.pojo.VolteTerminalInfo;
+
+import javax.annotation.Resource;
+
+@Service
+public class TerminalInfoService {
+
+    @Resource
+    private VolteTerminalInfoDao volteTerminalInfoDao;
+
+    @Resource
+    private G5TerminalInfoDao g5TerminalInfoDao;
+
+    public VolteTerminalInfo findVolteInfoByPhoneNumber(String phoneNumber) {
+        VolteTerminalInfo info = new VolteTerminalInfo();
+        info.setMsisdn(phoneNumber);
+        return volteTerminalInfoDao.selectOne(Wrappers.query(info));
+    }
+
+    public G5TerminalInfo find5GInfoByPhoneNumber(String phoneNumber) {
+        G5TerminalInfo info = new G5TerminalInfo();
+        info.setMsisdn(phoneNumber);
+        return g5TerminalInfoDao.selectOne(Wrappers.query(info));
+    }
+}

src/main/java/top/lifuquan/util/CsvUtil.java

@@ -0,0 +1,116 @@
+package top.lifuquan.util;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.csv.CSVFormat;
+import org.apache.commons.csv.CSVPrinter;
+import org.apache.commons.csv.CSVRecord;
+
+import java.io.*;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.zip.GZIPInputStream;
+
+@Slf4j
+public class CsvUtil {
+
+    private static Reader reader = null;
+    private static CSVPrinter printer = null;
+
+    /**
+     * 对csv文件进行去重
+     *
+     * @param csvIn                输入文件路径
+     * @param csvOut               输出文件路径
+     * @param distinctColumnNumber 去重依据的列的编号，从0开始
+     * @param delimiter            csv文件的分割符
+     */
+    public static boolean distinct(String csvIn, String csvOut, int distinctColumnNumber, String delimiter) {
+        Iterable<CSVRecord> records = null;
+        try {
+            reader = getReader(csvIn);
+            records = CSVFormat.DEFAULT.builder()
+                    .setDelimiter(delimiter)
+                    .build().parse(reader);
+            log.debug("第一次遍历文件，筛选第 " + distinctColumnNumber + " 列不重复的项，重复值将保留最后一条记录....");
+            // 将文件加入map，对每个不同的 distinctColumnNumber ，仅会保留最后1条
+            Map<String, Integer> map = new HashMap<>();
+            int count = 0;
+            for (CSVRecord record : records) {
+                // map中记录的是distinctColumnNumber最后一行的行号
+                map.put(record.get(distinctColumnNumber), count);
+                count++;
+            }
+            log.debug("筛选完成，共筛选 " + count + " 记录,其中不重复的记录有 " + map.size() + " 条");
+            // 需要从头开始重新遍历文件
+            reader = getReader(csvIn);
+            records = read(csvIn, delimiter);
+            // 输出
+            printer = getPrinter(csvOut, delimiter);
+            log.debug("准备第二次遍历输入，将不重复记录写入 " + csvOut);
+            count = 0;
+            for (CSVRecord record : records) {
+                // 如果map中记录的行号是当前的行号，就可以写入输出文件
+                if (map.get(record.get(distinctColumnNumber)) == count) {
+                    printer.printRecord(record);
+                }
+                count++;
+            }
+            printer.flush();
+            log.debug("已完成去重");
+            return true;
+        } catch (IOException e) {
+            e.printStackTrace();
+        } finally {
+            close();
+        }
+        return false;
+    }
+
+    private static Iterable<CSVRecord> read(String csvIn, String delimiter) throws IOException {
+        reader = getReader(csvIn);
+        return CSVFormat.DEFAULT.builder()
+                .setDelimiter(delimiter)
+                .build().parse(reader);
+    }
+
+
+    private static CSVPrinter getPrinter(String csvOut, String delimiter) throws IOException {
+        Appendable out = new PrintWriter(csvOut);
+        return CSVFormat.DEFAULT.builder()
+                .setDelimiter(delimiter)
+                .build().print(out);
+    }
+
+    private static void close() {
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        if (printer != null) {
+            try {
+                printer.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+    }
+
+    /**
+     * 需要处理的文件可能是文本或者gz压缩格式
+     *
+     * @param csvIn 输入文件路径
+     * @return Reader
+     * @throws IOException
+     */
+    private static Reader getReader(String csvIn) throws IOException {
+        if (csvIn.toLowerCase().endsWith(".gz")) {
+            // 这里识别一下gz压缩文件
+            return new InputStreamReader(new GZIPInputStream(new FileInputStream(csvIn)));
+        } else {
+            return new FileReader(csvIn);
+        }
+    }
+}

src/main/java/top/lifuquan/util/ProcessUtil.java

@@ -0,0 +1,144 @@
+package top.lifuquan.util;
+
+import lombok.extern.slf4j.Slf4j;
+import top.lifuquan.pojo.TaskRecord;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.file.Paths;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+@Slf4j
+public class ProcessUtil {
+
+    private static final String scpShell = "scp_file.sh";
+    private static final String gpLoad = "gpload.sh";
+
+    private static Process process = null;
+    private static BufferedReader reader = null;
+
+    public static TaskRecord gpLoad(String date, TaskRecord record) {
+        if (!new File(gpLoad).exists()) {
+            // scp的sh文件丢失
+            log.info("用于scp的sh文件丢失。。。。");
+        }
+        try {
+            log.debug("准备调用gpload");
+            process = Runtime.getRuntime().exec("sh " + gpLoad + " " + date + " " + record.getTaskType());
+            reader = getReader();
+            List<String> lines = new ArrayList<>();
+            String line;
+            while ((line = reader.readLine()) != null) {
+                log.debug(line);
+                lines.add(line);
+            }
+            return getTaskRecordFromResult(lines, record);
+        } catch (IOException e) {
+            e.printStackTrace();
+            record.setTaskStatus("GPLOAD失败");
+            return record;
+        } finally {
+            destroy();
+        }
+    }
+
+    private static TaskRecord getTaskRecordFromResult(List<String> lines, TaskRecord record) {
+        for (int i = 0; i < lines.size(); i++) {
+            if (lines.get(i).contains("|INFO|rows Inserted")) {
+                // 找到第一行 对应的是插入数据的数量
+                // s4可以判断gpload是否成功
+                String s4 = lines.get(i + 3);
+                if (s4.endsWith("succeeded")) {
+                    // 插入数量
+                    String s1 = lines.get(i);
+                    s1 = s1.substring(s1.indexOf('=') + 2);
+                    // 更新数量
+                    String s2 = lines.get(i + 1);
+                    s2 = s2.substring(s2.indexOf('=') + 2);
+                    // 错误数量
+                    String s3 = lines.get(i + 2);
+                    s3 = s3.substring(s3.indexOf('=') + 2);
+                    record.setInserted(Integer.parseInt(s1));
+                    record.setUpdated(Integer.parseInt(s2));
+                    record.setError(Integer.parseInt(s3));
+                    record.setTaskStatus("任务完成");
+                } else {
+                    record.setTaskStatus("GPLOAD失败");
+                }
+            }
+        }
+        return record;
+    }
+
+    public static boolean scpFile(String fileName, String localPath) {
+        if (!new File(scpShell).exists()) {
+            // scp的sh文件丢失
+            log.info("用于scp的sh文件丢失。。。。");
+            return false;
+        }
+        // 拼接本地文件路径
+        String localFilePath = Paths.get(localPath, fileName).toString();
+        try {
+            // 执行本地sh文件进行下载
+            process = Runtime.getRuntime().exec("sh " + scpShell + " " + fileName + " " + localPath);
+            reader = getReader();
+            // 识别完成度的百分比
+            Pattern pattern = Pattern.compile("\\d+%");
+            String line;
+            Matcher matcher;
+            while ((line = reader.readLine()) != null) {
+                if (line.contains("No such file or directory")) {
+                    // 在源端文件不存在
+                    log.debug("在源端不存在文件" + fileName);
+                }
+                matcher = pattern.matcher(line);
+                if (matcher.find()) {
+                    if (matcher.group().equals("100%")) {
+                        // 只有下载到100%才算完成
+                        log.info("文件SCP完成,已下载至 " + localFilePath);
+                        return true;
+                    }
+                }
+            }
+            // 文件未下载成功，删除文件
+            log.debug("文件未下载完成，可能的原因是下载超时，将删除未完全下载的文件 " + localFilePath);
+            rmFile(localFilePath);
+        } catch (IOException e) {
+            // 遇到异常也可能会遗留未完全下载的文件
+            e.printStackTrace();
+            rmFile(localFilePath);
+        } finally {
+            destroy();
+        }
+        return false;
+    }
+
+    private static BufferedReader getReader() {
+        return new BufferedReader(new InputStreamReader(process.getInputStream()));
+    }
+
+    private static void rmFile(String localFilePath) {
+        File file = new File(localFilePath);
+        if (file.exists()) {
+            file.delete();
+        }
+    }
+
+    private static void destroy() {
+        if (reader != null) {
+            try {
+                reader.close();
+            } catch (IOException e) {
+                e.printStackTrace();
+            }
+        }
+        if (process != null) {
+            process.destroy();
+        }
+    }
+}

src/main/java/top/lifuquan/vo/G5TerminalInfoVo.java

@@ -0,0 +1,35 @@
+package top.lifuquan.vo;
+
+import com.baomidou.mybatisplus.annotation.TableName;
+import lombok.Data;
+import top.lifuquan.pojo.G5TerminalInfo;
+
+@Data
+@TableName("tmlq.o_info_list_5g_terminal")
+public class G5TerminalInfoVo {
+    private String msisdn;
+    /**
+     * 字段说明：
+     * 在统计时段内，用户是否接入过5G网络
+     * 0：未登网   1：登网
+     */
+    private String used5g;
+    /**
+     * 字段说明：
+     * 统计当天5G开关是否曾打开
+     * true：打开   false：未打开
+     */
+    private String status5g;
+    private String day;
+    private String factName;
+    private String terminalName;
+
+    public G5TerminalInfoVo(G5TerminalInfo terminalInfo) {
+        this.msisdn = terminalInfo.getMsisdn();
+        this.used5g = terminalInfo.getUsed5g().equals(1) ? "使用过5G网络" : "未使用5G网络";
+        this.status5g = terminalInfo.getStatus5g().equals("true") ? "5G开关打开" : "5G终端未打开";
+        this.day = terminalInfo.getDay();
+        this.factName = terminalInfo.getFactName();
+        this.terminalName = terminalInfo.getTerminalName();
+    }
+}

src/main/java/top/lifuquan/vo/R.java

@@ -0,0 +1,101 @@
+package top.lifuquan.vo;
+
+import java.util.HashMap;
+import java.util.Map;
+
+public class R {
+    private Boolean success;
+    private Integer code;
+    private String message;
+
+    private Map<String, Object> data = new HashMap<>();
+
+    /**
+     * 私有化构造方法,不允许在外部实例化
+     */
+    private R() {
+    }
+
+    /**
+     * 成功的静态方法
+     *
+     * @return R实例
+     */
+    public static R ok() {
+        R r = new R();
+        r.setSuccess(true);
+        r.setCode(20000);
+        r.setMessage("成功");
+        return r;
+    }
+
+    /**
+     * 失败的静态方法
+     *
+     * @return R实例
+     */
+    public static R error() {
+        R r = new R();
+        r.setSuccess(false);
+        r.setCode(20001);
+        r.setMessage("失败");
+        return r;
+    }
+
+    public R success(Boolean success) {
+        this.setSuccess(success);
+        return this;
+    }
+
+    public R message(String message) {
+        this.setMessage(message);
+        return this;
+    }
+
+    public R code(Integer code) {
+        this.setCode(code);
+        return this;
+    }
+
+    public R data(String key, Object value) {
+        this.data.put(key, value);
+        return this;
+    }
+
+    public R data(Map<String, Object> map) {
+        this.setData(map);
+        return this;
+    }
+
+    public Boolean getSuccess() {
+        return success;
+    }
+
+    public void setSuccess(Boolean success) {
+        this.success = success;
+    }
+
+    public Integer getCode() {
+        return code;
+    }
+
+    public void setCode(Integer code) {
+        this.code = code;
+    }
+
+    public String getMessage() {
+        return message;
+    }
+
+    public void setMessage(String message) {
+        this.message = message;
+    }
+
+    public Map<String, Object> getData() {
+        return data;
+    }
+
+    public void setData(Map<String, Object> data) {
+        this.data = data;
+    }
+}

src/main/java/top/lifuquan/vo/VolteTerminalInfoVo.java

@@ -0,0 +1,64 @@
+package top.lifuquan.vo;
+
+import com.fasterxml.jackson.annotation.JsonFormat;
+import lombok.Data;
+import top.lifuquan.pojo.VolteTerminalInfo;
+
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+
+@Data
+public class VolteTerminalInfoVo {
+
+    private String msisdn;
+    /**
+     * 字段说明：
+     * 硬件无法识别：-3
+     * 其它终端硬件支持但无法识别软件是否支持或开关是否打开：-2
+     * 苹果终端硬件支持但无法识别软件是否支持或开关是否打开：-1
+     * 硬件不支持：0
+     * 硬件支持软件不支持：1
+     * 硬件软件支持但开关未打开：2
+     * 硬件软件支持且开关打开：3
+     */
+    private String isVolte;
+    @JsonFormat(pattern = "yyyy年M月d日")
+    private Date dayDate;
+    private String factName;
+    private String terminalName;
+    private String hprovinceName;
+
+    public VolteTerminalInfoVo(VolteTerminalInfo volteTerminalInfo) throws ParseException {
+        this.msisdn = volteTerminalInfo.getMsisdn();
+        switch (volteTerminalInfo.getIsVolte()) {
+            case -3:
+                this.isVolte = "硬件无法识别，不做参考";
+                break;
+            case -2:
+                this.isVolte = "其它终端硬件支持但无法识别软件是否支持或开关是否打开，建议用户尝试升级系统版本或打开volte开关";
+                break;
+            case -1:
+                this.isVolte = "苹果终端硬件支持但无法识别软件是否支持或开关是否打开，建议用户尝试升级系统版本或打开volte开关";
+                break;
+            case 0:
+                this.isVolte = "硬件不支持，用户无法使用volte功能，转向2/3G语音流程";
+                break;
+            case 1:
+                this.isVolte = "硬件支持软件不支持，建议用户升级系统版本";
+                break;
+            case 2:
+                this.isVolte = "硬件软件支持但开关未打开,建议用户打开volte开关";
+                break;
+            case 3:
+                this.isVolte = "硬件软件支持且开关打开，可正常使用volte功能";
+                break;
+            default:
+                this.isVolte = "出现未定义值，请记下号码并通知管理员";
+        }
+        this.dayDate = new SimpleDateFormat("yyyyMMdd").parse(volteTerminalInfo.getDayDate());
+        this.factName = volteTerminalInfo.getFactName();
+        this.terminalName = volteTerminalInfo.getTerminalName();
+        this.hprovinceName = volteTerminalInfo.getHprovinceName();
+    }
+}

src/main/resources/application.properties

@@ -0,0 +1,18 @@
+# 端口号
+server.port=12900
+
+# 数据源配置
+spring.datasource.driver-class-name=org.postgresql.Driver
+spring.datasource.url=jdbc:postgresql://10.170.42.48:5432/localdb
+spring.datasource.username=lifuq
+spring.datasource.password=Mdasil789!@#
+
+# 日志
+logging.level.top.lifuquan=debug
+logging.file=log/terminalApplication.log
+
+# 定时调度
+scheduled.pool.size=2
+scheduled.cron=0 45 * * * ?
+# 延迟天数
+scheduled.delay=1