根据接入井表中的数据来插入机构表和基础表

src/main/java/com/gct/tools/etlcamelhuge/camelconfig/MyDataSourceConfiguration.java

@@ -24,20 +24,14 @@ public class MyDataSourceConfiguration {
         return DataSourceBuilder.create().build();
     }
 
-    @Bean(name = "diagnosis")
-    @ConfigurationProperties(prefix = "spring.datasource.ds2")
-    public DataSource dataSource2() {
-        return DataSourceBuilder.create().build();
-    }
-
     @Bean(name = "oracle")
-    @ConfigurationProperties(prefix = "spring.datasource.ds3")
+    @ConfigurationProperties(prefix = "spring.datasource.ds2")
     public DataSource dataSource3() {
         return DataSourceBuilder.create().build();
     }
 
     @Bean(name = "gtsj")
-    @ConfigurationProperties(prefix = "spring.datasource.ds4")
+    @ConfigurationProperties(prefix = "spring.datasource.ds3")
     public DataSource dataSource4() {
         return DataSourceBuilder.create().build();
     }

src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelJDBCCofRealTimeConfiguration.java

@@ -125,7 +125,7 @@ public class CamelJDBCCofRealTimeConfiguration  {
                         aRow.put("susp_min_load",susp_min_load);
                     }
                     //对于位移没有数据,所有数据都在载荷中的特殊数据做特别处理
-                    if ((aRow.get("displacement") == null || aRow.get("displacement").equals("")) && (aRow.get("disp_load") != null && !aRow.get("disp_load").equals(""))){
+                    else if ((aRow.get("displacement") == null || aRow.get("displacement").equals("")) && (aRow.get("disp_load") != null && !aRow.get("disp_load").equals(""))){
                         String disp_load = aRow.get("disp_load").toString().replaceAll(";", ",");
                         Double[][] doubles = SGTUtil.encodeToDoubleArray(disp_load);
                         aRow.put("sgt", SGTUtil.encodeToString(doubles));

src/main/java/com/gct/tools/etlcamelhuge/routeconfig/CamelJDBCConfiguration.java

@@ -1,28 +1,13 @@
 package com.gct.tools.etlcamelhuge.routeconfig;
 
-import com.alibaba.fastjson.JSONObject;
-import com.gct.common.util.SGTUtil;
-import com.gct.tools.etlcamelhuge.MQ.DefaultMsgSendSuccessCallBack;
-import com.gct.tools.etlcamelhuge.MQ.MessageBody;
-import com.gct.tools.etlcamelhuge.MQ.MessageProducer;
-import com.gct.tools.etlcamelhuge.camelconfig.MyDataSourceConfiguration;
-import com.gct.tools.etlcamelhuge.entity.DiagnoseMsg;
-import lombok.Data;
 import org.apache.camel.*;
 import org.apache.camel.builder.RouteBuilder;
 //import org.apache.rocketmq.common.message.Message;
-import org.apache.rocketmq.spring.core.RocketMQTemplate;
-import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
-import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.scheduling.annotation.Async;
+import org.springframework.util.StringUtils;
 
-import javax.annotation.Resource;
-import javax.sql.DataSource;
 import java.math.BigDecimal;
-import java.text.DecimalFormat;
-import java.text.SimpleDateFormat;
 import java.time.LocalDateTime;
 import java.time.format.DateTimeFormatter;
 import java.util.*;
@@ -49,97 +34,157 @@ public class CamelJDBCConfiguration {
             private Map<String, Integer> orgIDs;
             private Integer orgID;
             private Map<String, Integer> stringIntegerMap;
+            private List<Object> wellIdList;
+            private Set<Object> orgIdPreList;
+            private Integer org;
             //全部执行完成的大概时间在30-40分钟
             @Override
             public void configure() throws Exception {
-                /*from("timer:mytimer1?period=604800000")
-                        .routeId("oracle-1")
+
+                from("timer:insert-OrgAndWellSource?period=86400000")
+                        .routeId("insert-OrgAndWellSource")
                         .setHeader("date", constant(getDate()))
-                        .setBody(simple("select  distinct jh,cydmc,zyq,zk,qyrq,sccw,qk,bz from DBA01 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
-                        .to("jdbc:oracle")
-                        .log("${header.date}"+"routeId:oracle-1->  select cb_cd_well_source need data failed")
-                        .transform()
-                        .body((result) -> {
-                            organization = new TreeSet<>();
-                            orgID = 0;
-                            orgIDs = new HashMap<>();
-                            return result;
+                        .process(exchange -> {
+                            org = 0;
+                            wellIdList = new ArrayList<>();
+                            orgIdPreList =new HashSet<>();
                         })
-                        .step("1")
-                        .split(body()).process(exchange -> {
-                    Message in = exchange.getIn();
-                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    String org_level3 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC") + "@" + aRow.get("ZK");
-                    String org_level2 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC");
-                    String org_level1 = aRow.get("ZYQ").toString();
-                    aRow.put("station_id", org_level3);
-                    if (organization.add(org_level1)) {
-                        orgID++;
-                        orgIDs.put(org_level1, orgID);
-                    }
-                    if (organization.add(org_level2)) {
-                        orgID++;
-                        orgIDs.put(org_level2, orgID);
-                    }
-                    if (organization.add(org_level3)) {
-                        orgID++;
-                        orgIDs.put(org_level3, orgID);
-                    }
-                })
-                        .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,station_id,station_name,completion_name,PRODUCING_AREA_name,remarks) " +
-                                "values ('${body[JH]}','${body[JH]}','${body[QYRQ]}'::timestamp,'${body[station_id]}','${body[ZK]}','${body[SCCW]}','${body[QK]}','${body[BZ]}')" +
-                                " ON conflict(well_id) DO UPDATE set remarks = '${body[BZ]}' "))
+                        .setBody(simple("select max(org_id) from centralbase.cb_pc_organization"))
                         .to("jdbc:centralbase")
-                        .log("${header.date}"+" routeId:oracle-1->  centralbase.cb_cd_well_source insert data failed")
-                        .end()
-                        .transform().body((re) -> {
-                    List<Map<String, Object>> rows = new ArrayList<>();
-                    int code = 0;
-                    for (String s : organization) {
-                        code++;// code is same as org_id
-                        String[] orgs = s.split("@");
-                        Map<String, Object> row = new HashMap<>();
-                        row.put("org_id_pre", s);
-                        row.put("org_code", code);
-                        row.put("org_id", "" + code);
-                        switch (orgs.length) {
-                            case 1:
-                                row.put("org_name", orgs[0]);
-                                row.put("org_level", 1);
-                                row.put("org_parent", "0");
-                                break;
-                            case 2:
-                                row.put("org_name", orgs[1]);
-                                row.put("org_level", 2);
-                                row.put("org_parent", orgIDs.get(orgs[0]).toString());
-                                break;
-                            case 3:
-                                row.put("org_name", orgs[2]);
-                                row.put("org_level", 3);
-                                row.put("org_parent", orgIDs.get(orgs[0] + "@" + orgs[1]).toString());
-                                break;
-                        }
-                        rows.add(row);
-                    }
-                    return rows;
-                }).split(body())
-                        .doTry()
-                        .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_code,org_name,org_level,parent_id,org_id_pre)" +
-                                "values('${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}')" +
-                                "ON conflict(org_id_pre) DO UPDATE set org_code = '${body[org_code]}' "))
+                        .process(exchange -> {
+                            HashMap body = exchange.getIn().getBody(HashMap.class);
+                            if(body==null|| StringUtils.isEmpty(body.get("max"))) org=0;
+                            else
+                            org = Integer.valueOf(body.get("max").toString());
+                        })
+                        .setBody(simple("select well_id from centralbase.sys_access_well_control  where  access_status='1'  "))
                         .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-1->  centralbase.cb_pc_organization insert data failed")
-                        .end()
-                        .setBody(simple("select org_id,org_name from centralbase.cb_pc_organization where org_level = '3' "))
+                        .split(body()).process(exchange -> {
+                            HashMap body = exchange.getIn().getBody(HashMap.class);
+                            exchange.getIn().setHeader("well_id",body.get("well_id"));
+                        })
+                        .setBody(simple("select org_id_pre from centralbase.cb_pc_organization"))
+                        .to("jdbc:centralbase")
+                        .split(body()).process(exchange -> {
+                            HashMap body = exchange.getIn().getBody(HashMap.class);
+                            orgIdPreList.add(body.get("org_id_pre"));
+                        }).end()
+                       .setBody(simple("select  distinct jh,cydmc,zyq,zk,qyrq,sccw,qk,bz from DBA01 where jh ='${header.well_id}' "))
+                       .to("jdbc:oracle")
+                         .transform()
+                          .body((result) -> {
+                              organization = new TreeSet<>();
+                              orgID = 0;
+                              orgIDs = new HashMap<>();
+                              return result;
+                          })
+                          .step("1")
+                          .split(body()).process(exchange -> {
+                      Message in = exchange.getIn();
+                      HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                      String org_level3 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC") + "@" + aRow.get("ZK");
+                      String org_level2 = aRow.get("ZYQ") + "@" + aRow.get("CYDMC");
+                      String org_level1 = aRow.get("ZYQ").toString();
+                      aRow.put("station_id", org_level3);
+                      if (organization.add(org_level1)) {
+                          orgID++;
+                          orgIDs.put(org_level1, orgID);
+                      }
+                      if (organization.add(org_level2)) {
+                          orgID++;
+                          orgIDs.put(org_level2, orgID);
+                      }
+                      if (organization.add(org_level3)) {
+                          orgID++;
+                          orgIDs.put(org_level3, orgID);
+                      }
+
+                  })
+                          .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,station_id,station_name,completion_name,PRODUCING_AREA_name,remarks) " +
+                                  "select '${body[JH]}','${body[JH]}','${body[QYRQ]}'::timestamp,'${body[station_id]}','${body[ZK]}','${body[SCCW]}','${body[QK]}','${body[BZ]}' " +
+                                  "where NOT EXISTS ( SELECT * FROM centralbase.cb_cd_well_source WHERE well_id = '${body[JH]}' )"))
+                          .to("jdbc:centralbase")
+                          .end()
+                          .transform().body((re) -> {
+                      List<Map<String, Object>> rows = new ArrayList<>();
+                      Map<String, Object> row = new HashMap<>();
+//                      int code = org;
+                      for (String s : organization) {
+                          // code is same as org_id
+                          String[] orgs = s.split("@");
+                          row.put("org_id_pre", s);
+                          switch (orgs.length) {
+                              case 1:
+                                  row.put("org_name", orgs[0]);
+                                  row.put("org_level", 1);
+                                  row.put("org_parent", "0");
+                                  break;
+                              case 2:
+                                  row.put("org_name", orgs[1]);
+                                  row.put("org_level", 2);
+                                  row.put("org_parent", orgIDs.get(orgs[0]).toString());
+                                  break;
+                              case 3:
+                                  row.put("org_name", orgs[2]);
+                                  row.put("org_level", 3);
+                                  row.put("org_parent", orgIDs.get(orgs[0] + "@" + orgs[1]).toString());
+                                  break;
+                          }
+                          if (orgIdPreList.add(s)) {
+                              org++;
+                              row.put("org_code", org);
+                              row.put("org_id", "" + org);
+                              rows.add(row);
+                          }
+                      }
+                      return rows;
+                  }).split(body())
+                          .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_code,org_name,org_level,parent_id,org_id_pre)" +
+                                  "select '${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}' " +
+                                  "where NOT EXISTS ( SELECT * FROM centralbase.cb_pc_organization WHERE org_id = '${body[org_id]}' and org_code='${body[org_code]}' and org_name = '${body[org_name]}' and org_level = '${body[org_level]}' and parent_id ='${body[org_parent]}' and org_id_pre ='${body[org_id_pre]}')"))
+                          .doTry()
+                          .to("jdbc:centralbase")
+                          .doCatch(Exception.class)
+                          .log("${header.date}"+" routeId:insert-OrgAndWellSource->  centralbase.cb_pc_organization insert data failed")
+                          .end()
+                          .setBody(simple("select org_id,org_id_pre from centralbase.cb_pc_organization where org_level = '3' "))
+                          .to("jdbc:centralbase")
+                          .split(body())
+                          .doTry()
+                          .setBody(simple("update centralbase.cb_cd_well_source set org_id = '${body[org_id]}' where station_id = '${body[org_id_pre]}'"))
+                          .to("jdbc:centralbase")
+                          .doCatch(Exception.class)
+                          .log("${header.date}"+" routeId:insert-OrgAndWellSource->  centralbase.cb_pc_organization update data failed")
+                        .end();
+
+                from("timer:update-wellControl?period=3600000")
+                        .routeId("update-wellControl")
+                        .setBody(simple("select scc.well_id,wo.well_common_name,op.org_id from centralbase.sys_access_well_control scc\n" +
+                                "left join centralbase.cb_cd_well_source wo on scc.well_id = wo.well_id\n" +
+                                "left join centralbase.cb_pc_organization op on wo.org_id = op.org_id\n" +
+                                "where scc.access_status ='1' "))
                         .to("jdbc:centralbase")
                         .split(body())
-                        .doTry()
-                        .setBody(simple("update centralbase.cb_cd_well_source set org_id = '${body[org_id]}' where station_name = '${body[org_name]}'"))
+                        .split(body()).process(exchange -> {
+                            HashMap<String, Object> body = exchange.getIn().getBody(HashMap.class);
+                            String format = LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
+                            if (body.get("well_common_name") == null || body.get("well_common_name").equals("")) {
+                                body.put("remarks","暂无井信息");
+                                body.put("error_id",0);
+                                body.put("updateTime",format);
+                            }else if (body.get("org_id") == null || body.get("org_id").equals("")){
+                                body.put("remarks","暂无机构信息");
+                                body.put("error_id",0);
+                                body.put("updateTime",format);
+                            }else {
+                                body.put("remarks","");
+                                body.put("error_id",1);
+                                body.put("updateTime",format);
+                            }
+                        })
+                        .setBody(simple("update centralbase.sys_access_well_control set well_common_name='${body[well_common_name]}',org_id='${body[org_id]}',update_time = '${body[updateTime]}'::timestamp, remarks ='${body[remarks]}' ,error_id ='${body[error_id]}' where well_id ='${body[well_id]}' "))
                         .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:oracle-1->  centralbase.cb_pc_organization update data failed")
-                        .end();*/
+                        .end();
 
 
                 from("timer:mytimer-insert-statusDaily?period=3600000")
@@ -229,27 +274,6 @@ public class CamelJDBCConfiguration {
                         .log("${header.date}"+" routeId:update-statusDaily-submergenceDepth ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
                         .end();
 
-
-              /*  from("timer:mytimer-update-statusDaily-oil_nozzle?period=1800000")
-                        .routeId("update-statusDaily-oil_nozzle")
-                        .setHeader("date", constant(getDate() + " 00:00:00"))
-                        .setBody(simple("select distinct rn.well_id,cb.prod_date,rn.pump_diameter  from centralbase.cb_temp_well_mech_runtime rn ,centralbase.cb_pc_pro_wellbore_status_daily cb where cb.well_id = rn.well_id " +
-                                "and cb.prod_date = '${header.date}' "))
-                        .to("jdbc:centralbase")//.log("${body}")
-                        .split(body()).process(exchange -> {
-                            Message in = exchange.getIn();
-                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                            aRow.putIfAbsent("pump_diameter", "0.0");
-                        })
-                        .setBody(simple("update centralbase.cb_pc_pro_wellbore_status_daily set oil_nozzle = '${body[pump_diameter]}' where well_id ='${body[well_id]}' and prod_date='${body[prod_date]}' "))
-                        .doTry()
-                        .to("jdbc:centralbase")
-                        .doCatch(Exception.class)
-                        .log("${header.date}"+" routeId:update-statusDaily-oil_nozzle ->  centralbase.cb_pc_pro_wellbore_status_daily update data failed")
-                        .end();*/
-
-
-
                 from("timer:mytimer-insert-volDaily?period=3600000")
                         .routeId("insert-volDaily")
                         .process(exchange -> {

src/main/resources/application.yml

@@ -26,72 +26,50 @@ spring:
         max-lifetime: 1800000
         ## 数据库连接超时时间,默认30秒，即30000
         connection-timeout: 30000
-    ds2:
-      ## Hikari连接池的设置 Hikari 时间单位都是毫秒
-      type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:postgresql://10.72.143.2:54321/diagnosis
-      username: root
-      password: 123456
-      driver-class-name: org.postgresql.Driver
-      hikari:
-        ## 连接池名字
-        pool-name: SystemHikariCP
-        ## 最小空闲连接数量
-        minimum-idle: 5
-        ## 空闲连接存活最大时间，默认600000（10分钟）
-        idle-timeout: 60000
-        ## 连接池最大连接数，默认是10
-        maximum-pool-size: 15
-        ## 此属性控制从池返回的连接的默认自动提交行为,默认值：true
-        auto-commit: true
-        ## 此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
-        max-lifetime: 1800000
-        ## 数据库连接超时时间,默认30秒，即30000
-        connection-timeout: 30000
-    ds3:
-      ## Hikari连接池的设置 Hikari 时间单位都是毫秒
-      type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:oracle:thin:@11.71.15.28:1521:oracycn2
-      username: ZD_FGD
-      password: ZJGSDDS#2014ZD
-      driver-class-name: oracle.jdbc.driver.OracleDriver
-      hikari:
-        ## 连接池名字
-        pool-name: SystemHikariCP
-        ## 最小空闲连接数量
-        minimum-idle: 5
-        ## 空闲连接存活最大时间，默认600000（10分钟）
-        idle-timeout: 60000
-        ## 连接池最大连接数，默认是10
-        maximum-pool-size: 15
-        ## 此属性控制从池返回的连接的默认自动提交行为,默认值：true
-        auto-commit: true
-        ## 此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
-        max-lifetime: 1800000
-        ## 数据库连接超时时间,默认30秒，即30000
-        connection-timeout: 30000
-    ds4:
-      ## Hikari连接池的设置 Hikari 时间单位都是毫秒
-      type: com.zaxxer.hikari.HikariDataSource
-      jdbc-url: jdbc:postgresql://11.72.187.11:5432/GTDATE
-      username: postgres
-      password: Postgres_2021#
-      driver-class-name: org.postgresql.Driver
-      hikari:
-        ## 连接池名字
-        pool-name: SystemHikariCP
-        ## 最小空闲连接数量
-        minimum-idle: 5
-        ## 空闲连接存活最大时间，默认600000（10分钟）
-        idle-timeout: 60000
-        ## 连接池最大连接数，默认是10
-        maximum-pool-size: 15
-        ## 此属性控制从池返回的连接的默认自动提交行为,默认值：true
-        auto-commit: true
-        ## 此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
-        max-lifetime: 1800000
-        ## 数据库连接超时时间,默认30秒，即30000
-        connection-timeout: 60000
+#    ds2:
+#      ## Hikari连接池的设置 Hikari 时间单位都是毫秒
+#      type: com.zaxxer.hikari.HikariDataSource
+#      jdbc-url: jdbc:oracle:thin:@11.71.15.28:1521:oracycn2
+#      username: ZD_FGD
+#      password: ZJGSDDS#2014ZD
+#      driver-class-name: oracle.jdbc.driver.OracleDriver
+#      hikari:
+#        ## 连接池名字
+#        pool-name: SystemHikariCP
+#        ## 最小空闲连接数量
+#        minimum-idle: 5
+#        ## 空闲连接存活最大时间，默认600000（10分钟）
+#        idle-timeout: 60000
+#        ## 连接池最大连接数，默认是10
+#        maximum-pool-size: 15
+#        ## 此属性控制从池返回的连接的默认自动提交行为,默认值：true
+#        auto-commit: true
+#        ## 此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
+#        max-lifetime: 1800000
+#        ## 数据库连接超时时间,默认30秒，即30000
+#        connection-timeout: 30000
+#    ds3:
+#      ## Hikari连接池的设置 Hikari 时间单位都是毫秒
+#      type: com.zaxxer.hikari.HikariDataSource
+#      jdbc-url: jdbc:postgresql://11.72.187.11:5432/GTDATE
+#      username: postgres
+#      password: Postgres_2021#
+#      driver-class-name: org.postgresql.Driver
+#      hikari:
+#        ## 连接池名字
+#        pool-name: SystemHikariCP
+#        ## 最小空闲连接数量
+#        minimum-idle: 5
+#        ## 空闲连接存活最大时间，默认600000（10分钟）
+#        idle-timeout: 60000
+#        ## 连接池最大连接数，默认是10
+#        maximum-pool-size: 15
+#        ## 此属性控制从池返回的连接的默认自动提交行为,默认值：true
+#        auto-commit: true
+#        ## 此属性控制池中连接的最长生命周期，值0表示无限生命周期，默认1800000即30分钟
+#        max-lifetime: 1800000
+#        ## 数据库连接超时时间,默认30秒，即30000
+#        connection-timeout: 60000
 
 management:
   info: