基本写完了工单流程数据的同步流程

README.md

@@ -0,0 +1 @@
+# tsl-data

pom.xml

@@ -11,7 +11,7 @@
         <relativePath />
     </parent>
 
-    <groupId>org.example</groupId>
+    <groupId>com.nokia</groupId>
     <artifactId>tsl-data</artifactId>
     <version>2.0-SNAPSHOT</version>
 
@@ -36,12 +36,24 @@
     </properties>
 
     <dependencies>
+        <!-- common-utils -->
+        <dependency>
+            <groupId>com.nokia</groupId>
+            <artifactId>common-utils</artifactId>
+            <version>1.0.0-SNAPSHOT</version>
+        </dependency>
         <!-- cache-starter -->
         <dependency>
             <groupId>com.nokia</groupId>
             <artifactId>cache-starter</artifactId>
             <version>1.0.0</version>
         </dependency>
+        <!-- fastjson -->
+        <dependency>
+            <groupId>com.alibaba</groupId>
+            <artifactId>fastjson</artifactId>
+            <version>2.0.40</version>
+        </dependency>
         <!-- postgresql驱动，版本由spring-boot-parent指定 -->
         <dependency>
             <groupId>org.postgresql</groupId>

src/main/java/com/nokia/tsl_data/config/DefaultJdbcTemplateConfig.java

@@ -0,0 +1,16 @@
+package com.nokia.tsl_data.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import javax.sql.DataSource;
+
+@Configuration
+public class DefaultJdbcTemplateConfig {
+
+    @Bean
+    public JdbcTemplate jdbcTemplate(DataSource dataSource) {
+        return new JdbcTemplate(dataSource);
+    }
+}

src/main/java/com/nokia/tsl_data/config/WorkFlowJdbcTemplateConfig.java

@@ -16,7 +16,7 @@ public class WorkFlowJdbcTemplateConfig {
      * 连接到工作流数据库的jdbcTemplate
      */
     @Bean("workFlowJdbcTemplate")
-    public JdbcTemplate jdbcTemplate() {
+    public JdbcTemplate workFlowJdbcTemplate() {
         HikariConfig config = new HikariConfig();
         config.setJdbcUrl("jdbc:postgresql://192.168.70.136:9999/flow_hb?characterEncoding=utf8");
         config.setUsername("flow_web");

src/main/java/com/nokia/tsl_data/dao/TaskRecordDao.java

@@ -0,0 +1,19 @@
+package com.nokia.tsl_data.dao;
+
+import com.nokia.tsl_data.entity.TaskRecord;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+import java.util.List;
+
+@Repository
+public interface TaskRecordDao extends JpaRepository<TaskRecord, Long> {
+
+    TaskRecord findFirstByTaskNameAndEndStatusOrderByStartTimeDesc(String taskName, String endStatus);
+
+    default TaskRecord findLastSuccessUpdateWorkFlowBasicDataTask() {
+        return findFirstByTaskNameAndEndStatusOrderByStartTimeDesc("更新工作流数据", "成功");
+    }
+
+    List<TaskRecord> findTop2ByOrderByStartTimeDesc();
+}

src/main/java/com/nokia/tsl_data/dao/TslDataDao.java

@@ -0,0 +1,100 @@
+package com.nokia.tsl_data.dao;
+
+import com.nokia.tsl_data.entity.WorkFlowBasicData;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.object.BatchSqlUpdate;
+import org.springframework.stereotype.Component;
+
+import java.sql.Timestamp;
+import java.sql.Types;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * tslData 直接使用jdbcTemplate完成的操作
+ */
+@Component
+public class TslDataDao {
+
+    private final JdbcTemplate jdbcTemplate;
+
+    public TslDataDao(JdbcTemplate jdbcTemplate) {
+        this.jdbcTemplate = jdbcTemplate;
+    }
+
+    /**
+     * 采用BatchSqlUpdate的方法批量insert到表tableName(可以是临时表)
+     */
+    public void batchInsertWorkFlowBasicData(List<WorkFlowBasicData> data, String tableName) {
+        String sqlFormat = "INSERT INTO tsl_data.%s\n" +
+                "(city_id, kfsn, region_id, work_flow_create_time, work_flow_update_time)\n" +
+                "VALUES(?,?,?,?,?)";
+        BatchSqlUpdate batchSqlUpdate = new BatchSqlUpdate(Objects.requireNonNull(jdbcTemplate.getDataSource()), String.format(sqlFormat, tableName));
+        batchSqlUpdate.setBatchSize(1000);
+        batchSqlUpdate.setTypes(new int[]{
+                Types.VARCHAR, Types.VARCHAR, Types.VARCHAR, Types.TIMESTAMP, Types.TIMESTAMP
+        });
+        for (WorkFlowBasicData item : data) {
+            batchSqlUpdate.update(
+                    item.getCityId(),
+                    item.getKfsn(),
+                    item.getRegionId(),
+                    item.getWorkFlowCreateTime() == null ? null : Timestamp.from(item.getWorkFlowCreateTime()),
+                    item.getWorkFlowUpdateTime() == null ? null : Timestamp.from(item.getWorkFlowUpdateTime())
+            );
+        }
+        batchSqlUpdate.flush();
+    }
+
+    /**
+     * 创建临时表，并返回临时表表名
+     */
+    public String createTempTableOfWorkFlowBasicData() {
+        // 临时表添加一个时间戳
+        String tempTableName = "work_flow_basic_data_temp_" + System.currentTimeMillis();
+        String sqlFormat = "CREATE TABLE tsl_data.%s (\n" +
+                "\tcity_id varchar(50) NULL,\n" +
+                "\tkfsn varchar(50) NOT NULL,\n" +
+                "\tregion_id varchar(50) NULL,\n" +
+                "\twork_flow_create_time timestamp NULL,\n" +
+                "\twork_flow_update_time timestamp NULL\n" +
+                ");";
+        jdbcTemplate.execute(String.format(sqlFormat, tempTableName));
+        return tempTableName;
+    }
+
+    /**
+     * 删除临时表
+     */
+    public void dropTempTable(String tempTableName) {
+        if (!tempTableName.contains("_temp_")) {
+            throw new RuntimeException("仅支持删除命名带有 _temp_ 的临时表");
+        }
+        jdbcTemplate.execute("drop table tsl_data." + tempTableName);
+    }
+
+    /**
+     * 从临时表中查找新增内容并插入work_flow_basic_data表
+     */
+    public int insertWorkFlowBasicDataFromTempTable(String tempTableName) {
+        String sqlFormat = "insert into tsl_data.work_flow_basic_data \n" +
+                "(city_id, kfsn, region_id, work_flow_create_time, work_flow_update_time, create_date, last_update_date) \n" +
+                "select t.city_id, t.kfsn, t.region_id, t.work_flow_create_time, t.work_flow_update_time, now(), now() \n" +
+                "from tsl_data.%s t \n" +
+                "left join tsl_data.work_flow_basic_data b \n" +
+                "on t.kfsn = b.kfsn \n" +
+                "where b.kfsn is null";
+        return jdbcTemplate.update(String.format(sqlFormat, tempTableName));
+    }
+
+    /**
+     * 从临时表中查找需要更新内容并更新到work_flow_basic_data表
+     */
+    public int updateWorkFlowBasicDataFromTempTable(String tempTableName) {
+        String sqlFormat = "update tsl_data.work_flow_basic_data b\n" +
+                "set city_id = t.city_id, region_id = t.region_id, work_flow_update_time = t.work_flow_update_time, last_update_date = now()\n" +
+                "from (select city_id, kfsn, region_id, work_flow_create_time, work_flow_update_time from tsl_data.%s) t\n" +
+                "where b.kfsn = t.kfsn and (b.city_id != t.city_id or b.region_id != t.region_id)";
+        return jdbcTemplate.update(String.format(sqlFormat, tempTableName));
+    }
+}

src/main/java/com/nokia/tsl_data/dao/WorkFlowBasicDataDao.java

@@ -0,0 +1,9 @@
+package com.nokia.tsl_data.dao;
+
+import com.nokia.tsl_data.entity.WorkFlowBasicData;
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.stereotype.Repository;
+
+@Repository
+public interface WorkFlowBasicDataDao extends JpaRepository<WorkFlowBasicData, Long> {
+}

src/main/java/com/nokia/tsl_data/dao/WorkFlowDao.java

@@ -1,6 +1,7 @@
 package com.nokia.tsl_data.dao;
 
 import com.nokia.tsl_data.entity.SysDataDictionary;
+import com.nokia.tsl_data.entity.WorkFlowBasicData;
 import org.springframework.beans.factory.annotation.Qualifier;
 import org.springframework.jdbc.core.JdbcTemplate;
 import org.springframework.jdbc.core.RowMapper;
@@ -8,6 +9,10 @@ import org.springframework.stereotype.Component;
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.util.Date;
 import java.util.List;
 
 /**
@@ -22,16 +27,41 @@ public class WorkFlowDao {
         this.jdbcTemplate = jdbcTemplate;
     }
 
+    // 获取全部地市信息
     public List<SysDataDictionary> findSysDataDictionaryForCity() {
         String sql = "select id, name, type_code from sys_data_dictionary sdd where type_code = 'city'";
         return jdbcTemplate.query(sql, new SysDataDictionaryRowMapper());
     }
 
+    // 获取全部区县信息
     public List<SysDataDictionary> findSysDataDictionaryForRegion() {
         String sql = "select id, name, type_code from sys_data_dictionary sdd where type_code = 'region'";
         return jdbcTemplate.query(sql, new SysDataDictionaryRowMapper());
     }
 
+    // 获取update_time在start之后的全部工单信息
+    public List<WorkFlowBasicData> findWorkFlowBasicDataByUpdateTime(Date start) {
+        return findWorkFlowBasicDataByUpdateTime(start, new Date());
+    }
+
+    // 获取update_time在两个范围之间的全部工单信息
+    public List<WorkFlowBasicData> findWorkFlowBasicDataByUpdateTime(Date start, Date end) {
+        String sqlFormat = "select create_time as work_flow_create_time, update_time as work_flow_update_time, \n" +
+                "city as city_id, form_basic_data->>'yh_region' as region_id, form_basic_data->>'kd_kfsn' as kfsn \n" +
+                "from flow_form_basic where update_time >= to_timestamp('%s', 'yyyy-mm-dd hh24:mi:ss') \n" +
+                "and update_time <= to_timestamp('%s', 'yyyy-mm-dd hh24:mi:ss')";
+        DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+        return jdbcTemplate.query(String.format(sqlFormat, dateFormat.format(start), dateFormat.format(end)), (rs, rowNum) -> {
+            WorkFlowBasicData basicData = new WorkFlowBasicData();
+            basicData.setWorkFlowCreateTime(Instant.ofEpochMilli(rs.getTimestamp("work_flow_create_time").getTime()));
+            basicData.setWorkFlowUpdateTime(Instant.ofEpochMilli(rs.getTimestamp("work_flow_update_time").getTime()));
+            basicData.setCityId(rs.getString("city_id"));
+            basicData.setRegionId(rs.getString("region_id"));
+            basicData.setKfsn(rs.getString("kfsn"));
+            return basicData;
+        });
+    }
+
     private static class SysDataDictionaryRowMapper implements RowMapper<SysDataDictionary> {
 
         @Override

src/main/java/com/nokia/tsl_data/entity/SysDataDictionary.java

@@ -18,21 +18,30 @@ public class SysDataDictionary {
     @GeneratedValue(strategy = GenerationType.IDENTITY)
     private Long id;
 
-    @Column(name = "work_flow_id", columnDefinition = "varchar(32)")
+    // 工作流id
+    @Column(name = "work_flow_id", columnDefinition = "varchar(50)")
     private String workFlowId;
 
-    @Column(name = "work_flow_type_code", columnDefinition = "varchar(32)")
+    // 工作流类型
+    @Column(name = "work_flow_type_code", columnDefinition = "varchar(50)")
     private String workFlowTypeCode;
 
+    // 工作流命名
     @Column(name = "work_flow_name", columnDefinition = "varchar(100)")
     private String workFlowName;
 
-    @Column(name = "type_name", columnDefinition = "varchar(32)")
+    // 类型名
+    @Column(name = "type_name", columnDefinition = "varchar(50)")
     private String typeName;
 
+    // 命名
     @Column(name = "name", columnDefinition = "varchar(100)")
     private String name;
 
+    // 码
+    @Column(name = "code", columnDefinition = "varchar(50)")
+    private String code;
+
     @CreatedDate
     private Instant createDate;
 

src/main/java/com/nokia/tsl_data/entity/TaskRecord.java

@@ -0,0 +1,53 @@
+package com.nokia.tsl_data.entity;
+
+import com.alibaba.fastjson2.JSONObject;
+import com.nokia.common.spring.jpa.converter.JSONObjectConverter;
+import lombok.Data;
+import org.springframework.data.annotation.CreatedDate;
+import org.springframework.data.annotation.LastModifiedDate;
+import org.springframework.data.jpa.domain.support.AuditingEntityListener;
+
+import javax.persistence.*;
+import java.time.Instant;
+
+@Data
+@Entity
+@EntityListeners(AuditingEntityListener.class)
+@Table(name = "task_record", schema = "tsl_data")
+public class TaskRecord {
+
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
+    private Long id;
+
+    // 任务名称
+    @Column
+    private String taskName;
+
+    // 任务执行开始时间
+    @Column
+    private Instant startTime;
+
+    // 任务执行结束时间
+    @Column
+    private Instant endTime;
+
+    // 任务结束状态 成功/失败
+    @Column
+    private String endStatus;
+
+    // 任务耗时
+    @Column
+    private Long timeCost;
+
+    // 任务详情，采用json字符串记录
+    @Column(name = "task_info", columnDefinition = "text")
+    @Convert(converter = JSONObjectConverter.class)
+    private JSONObject taskInfo;
+
+    @CreatedDate
+    private Instant createDate;
+
+    @LastModifiedDate
+    private Instant lastUpdateDate;
+}

src/main/java/com/nokia/tsl_data/entity/WorkFlowBasicData.java

@@ -1,18 +1,47 @@
 package com.nokia.tsl_data.entity;
 
+import lombok.Data;
+import org.springframework.data.annotation.CreatedDate;
+import org.springframework.data.annotation.LastModifiedDate;
+import org.springframework.data.jpa.domain.support.AuditingEntityListener;
+
+import javax.persistence.*;
+import java.time.Instant;
+
 /**
  * 数据源是工单流程的flow_form_basic表 码表 sys_data_dictionary
  */
+@Data
+@Entity
+@EntityListeners(AuditingEntityListener.class)
+@Table(name = "work_flow_basic_data", schema = "tsl_data")
 public class WorkFlowBasicData {
 
+    @Id
+    @GeneratedValue(strategy = GenerationType.IDENTITY)
     private Long id;
 
-    // 客服工单号 唯一
+    // 客服工单号 理论上不应有重复，但是实际上会出现重复
+    @Column(name = "kfsn", columnDefinition = "varchar(50)", nullable = false)
     private String kfsn;
 
     // 归属地市
-    private String city;
+    @Column(name = "city_id", columnDefinition = "varchar(50)")
+    private String cityId;
 
     // 归属区县
-    private String region;
+    @Column(name = "region_id", columnDefinition = "varchar(50)")
+    private String regionId;
+
+    @Column
+    private Instant workFlowCreateTime;
+
+    @Column
+    private Instant workFlowUpdateTime;
+
+    @CreatedDate
+    private Instant createDate;
+
+    @LastModifiedDate
+    private Instant lastUpdateDate;
 }

src/main/java/com/nokia/tsl_data/service/CronTaskService.java

@@ -0,0 +1,33 @@
+package com.nokia.tsl_data.service;
+
+import org.springframework.stereotype.Service;
+
+/**
+ * 定时任务
+ */
+@Service
+public class CronTaskService {
+
+    private final WorkFlowService workFlowService;
+
+    public CronTaskService(WorkFlowService workFlowService) {
+        this.workFlowService = workFlowService;
+    }
+
+    /**
+     * 从工单系统定时更新数据
+     * 调度周期每小时
+     */
+    public void updateWorkFlowBasicDataCronTask() {
+        // TODO 每小时执行一次，从工单系统获取数据并更新
+
+    }
+
+    /**
+     * 定时删除过期的临时表
+     * 调度周期，每天
+     */
+    public void tempTableCleanCronTask() {
+        // TODO 定时删除过期的临时表
+    }
+}

src/main/java/com/nokia/tsl_data/service/WorkFlowService.java

@@ -1,11 +1,65 @@
 package com.nokia.tsl_data.service;
 
+import com.alibaba.fastjson2.JSONObject;
+import com.nokia.tsl_data.dao.TslDataDao;
+import com.nokia.tsl_data.dao.WorkFlowDao;
+import com.nokia.tsl_data.entity.TaskRecord;
+import com.nokia.tsl_data.entity.WorkFlowBasicData;
+import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
+import java.text.DateFormat;
+import java.text.SimpleDateFormat;
+import java.time.Instant;
+import java.time.temporal.ChronoUnit;
+import java.util.Date;
+import java.util.List;
+
 /**
  * 用于所有查询工单流程的任务
  */
+@Slf4j
 @Service
 public class WorkFlowService {
 
+    private final WorkFlowDao workFlowDao;
+    private final TslDataDao tslDataDao;
+
+    private final DateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
+
+    public WorkFlowService(WorkFlowDao workFlowDao, TslDataDao tslDataDao) {
+        this.workFlowDao = workFlowDao;
+        this.tslDataDao = tslDataDao;
+    }
+
+    /**
+     * 任务名称：更新工作流数据
+     */
+    public TaskRecord updateWorkFlowBasicData(Date start, Date end) {
+        TaskRecord record = new TaskRecord();
+        record.setTaskName("更新工作流数据");
+        JSONObject jsonObject = new JSONObject();
+        record.setTaskInfo(jsonObject);
+        jsonObject.put("start", dateFormat.format(start));
+        jsonObject.put("end", dateFormat.format(end));
+        record.setStartTime(Instant.ofEpochMilli(System.currentTimeMillis()));
+        try {
+            List<WorkFlowBasicData> workFlowBasicDataList = workFlowDao.findWorkFlowBasicDataByUpdateTime(start, end);
+            String tempTableName = tslDataDao.createTempTableOfWorkFlowBasicData();
+            jsonObject.put("tempTableName", tempTableName);
+            tslDataDao.batchInsertWorkFlowBasicData(workFlowBasicDataList, tempTableName);
+            int insertCount = tslDataDao.insertWorkFlowBasicDataFromTempTable(tempTableName);
+            int updateCount = tslDataDao.updateWorkFlowBasicDataFromTempTable(tempTableName);
+            log.info("已完成工作流数据更新, 插入 {} 项, 更新 {} 项, 起: {} 止: {}", insertCount, updateCount, start, end);
+            record.setEndStatus("成功");
+            record.setEndTime(Instant.ofEpochMilli(System.currentTimeMillis()));
+        } catch (Exception e) {
+            e.printStackTrace();
+            record.setEndStatus("失败");
+            record.setEndTime(Instant.ofEpochMilli(System.currentTimeMillis()));
+        } finally {
+            record.setTimeCost(ChronoUnit.MILLIS.between(record.getStartTime(), record.getEndTime()));
+        }
+        return record;
+    }
 }

src/main/resources/application.yml

@@ -1,15 +1,6 @@
 server:
   port: 22222
 
-common:
-  util:
-    cache:
-      enable: true
-
-cache:
-  type: redis
-  time-to-live-expiration: 2
-
 Spring:
   profiles:
     active: dev

src/test/java/com/nokia/tsl_data/MainTest.java

@@ -0,0 +1,11 @@
+package com.nokia.tsl_data;
+
+import org.junit.jupiter.api.Test;
+
+public class MainTest {
+
+    @Test
+    void test() {
+        System.out.println("TS202310071127443617".length());
+    }
+}

src/test/java/com/nokia/tsl_data/TslDataApplicationTest.java

@@ -1,8 +1,8 @@
 package com.nokia.tsl_data;
 
-import com.nokia.tsl_data.dao.SysDataDictionaryDao;
-import com.nokia.tsl_data.dao.WorkFlowDao;
-import com.nokia.tsl_data.entity.SysDataDictionary;
+import com.nokia.tsl_data.dao.TaskRecordDao;
+import com.nokia.tsl_data.entity.TaskRecord;
+import com.nokia.tsl_data.service.WorkFlowService;
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
@@ -12,27 +12,21 @@ import java.util.List;
 @SpringBootTest
 class TslDataApplicationTest {
 
-
-    @Autowired(required = false)
-    private WorkFlowDao workFlowDao;
-
     @Autowired
-    private SysDataDictionaryDao sysDataDictionaryDao;
+    private WorkFlowService workFlowService;
 
+    @Autowired
+    private TaskRecordDao taskRecordDao;
 
     @Test
     void test() {
-        List<SysDataDictionary> sysDataDictionaryForCity = workFlowDao.findSysDataDictionaryForCity();
-        System.out.println(sysDataDictionaryForCity);
-        sysDataDictionaryDao.saveAll(sysDataDictionaryForCity);
+        TaskRecord lastSuccessUpdateWorkFlowBasicDataTask = taskRecordDao.findLastSuccessUpdateWorkFlowBasicDataTask();
+        System.out.println(lastSuccessUpdateWorkFlowBasicDataTask);
     }
 
     @Test
-    void test2() {
-        List<SysDataDictionary> sysDataDictionaryForRegion = workFlowDao.findSysDataDictionaryForRegion();
-        for (SysDataDictionary sysDataDictionary : sysDataDictionaryForRegion) {
-            System.out.println(sysDataDictionary);
-        }
+    void test1() {
+        List<TaskRecord> top2ByOrderByStartTimeDesc = taskRecordDao.findTop2ByOrderByStartTimeDesc();
+        System.out.println(top2ByOrderByStartTimeDesc);
     }
-
 }