其他需要的数据导入

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

@@ -1,14 +1,26 @@
 package com.gct.tools.etlcamelhuge.routeconfig;
 
+import com.alibaba.fastjson.JSONObject;
+import com.gct.tools.etlcamelhuge.entity.MQDiagnoseMsg;
 import org.apache.camel.*;
 import org.apache.camel.builder.RouteBuilder;
+import org.apache.rocketmq.client.producer.DefaultMQProducer;
+//import org.apache.rocketmq.common.message.Message;
 import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.slf4j.Logger;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.jdbc.support.SQLStateSQLExceptionTranslator;
 
+import java.math.BigDecimal;
+import java.nio.charset.StandardCharsets;
 import java.text.SimpleDateFormat;
 import java.util.*;
+import java.util.concurrent.atomic.AtomicInteger;
 
 /**
  * class name: CamelJDBCConfiguration
@@ -18,7 +30,7 @@ import java.util.*;
  * @since 2021/4/14 下午3:16
  */
 @Configuration
-public class CamelJDBCConfiguration /*extends RouteBuilder */ {
+public class CamelJDBCConfiguration /*extends RouteBuilder */{
 
     @Autowired
     private RocketMQTemplate rocketMQTemplate;
@@ -245,9 +257,10 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
     }*/
     @Bean
     public RouteBuilder routeBuilderWithOracle1() {
-        SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
-        Date date = new Date(System.currentTimeMillis());
-        String formatDate = formatter.format(date) + " 00:00:00";
+         SimpleDateFormat formatter = new SimpleDateFormat("yyyy-MM-dd");
+         Date date = new Date(System.currentTimeMillis());
+         String date1 = formatter.format(date);
+         String formatDate = formatter.format(date) + " 00:00:00";
         return new RouteBuilder() {
             private SortedSet<String> organization;
             private Map<String, Integer> orgIDs;
@@ -255,10 +268,10 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
 
             @Override
             public void configure() throws Exception {
-
                 from("timer:mytimer1?period=999999999")
                 .routeId("oracle-1")
-                .setBody(simple("select  jh,cydmc,zyq,zk,qyrq from zd_zdgs.dba01@A2 where rq  = to_date('2021-07-01','yyyy-MM-dd') and qyrq is not null "))
+                .setHeader("date",constant(date1))
+                .setBody(simple("select  jh,cydmc,zyq,zk,qyrq from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
                 .to("jdbc:oracle")
                 .transform()
                 .body((result) -> {
@@ -267,8 +280,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                     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");
@@ -289,16 +302,12 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         orgID++;
                         orgIDs.put(org_level3, orgID);
                     }
-
-
                 })
-                .log("${body}")
                 .setBody(simple("insert into centralbase.cb_cd_well_source (well_id,well_common_name,spud_date,station_id,station_name) " +
                         "values ('${body[JH]}','${body[JH]}','${body[QYRQ]}'::timestamp,'${body[station_id]}','${body[ZK]}')"))
                 .to("jdbc:centralbase")
+                .end()
 
-                .log("insert success")
-                /*******notice*********/
                 .transform().body((re) -> {
                     List<Map<String, Object>> rows = new ArrayList<>();
                     int code = 0;
@@ -329,24 +338,134 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         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)" +
-                                "values('${body[org_id],body[org_code],body[org_name],body[org_level],body[parent_id],body[org_id_pre]}"))
+                                "values('${body[org_id]}','${body[org_code]}','${body[org_name]}','${body[org_level]}','${body[org_parent]}','${body[org_id_pre]}')"))
                         .to("jdbc:centralbase")
-                        /*******************************/
+                        .end()
+                        .setBody(simple("select org_id,org_name from centralbase.cb_pc_organization where org_level = '3' "))
+                        .to("jdbc:centralbase")
+                        .split(body())
+                        .setBody(simple("update centralbase.cb_cd_well_source set org_id = '${body[org_id]}' where station_name = '${body[org_name]}'"))
+                        .to("jdbc:centralbase")
+                        .log("insert")
                         .end();
 
                 from("timer:mytimer2?period=999999999")
                         .routeId("oracle-2")
-                        .setBody(simple("select distinct station_id,station_name from centralbase.cb_cd_well_source"))
+                        .setHeader("date",constant(date1))
+                        .setBody(simple("select  jh,rq,cyfs,yz,hysx , yysx ,tysx,bs,dym from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                        .to("jdbc:oracle")
+                        .split(body()).process(exchange -> {
+                            Message in = exchange.getIn();
+                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                            if (aRow.get("YZ")==null)   aRow.put("YZ","0.0");
+                            if (aRow.get("HYSX")==null) aRow.put("HYSX","0.0");
+                            if (aRow.get("YYSX")==null) aRow.put("YYSX","0.0");
+                            if (aRow.get("TYSX")==null) aRow.put("TYSX","0.0");
+                            if (aRow.get("BS")==null)   aRow.put("BS","0.0");
+                            if (aRow.get("DYM")==null)  aRow.put("DYM","0.0");
+                })
+                        .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_status_daily(well_id,prod_date,oil_prod_method,oil_nozzle,back_pres,tubing_pres,casing_pres,pump_depth,start_pump_liq_level)" +
+                                "values ('${body[JH]}','${body[RQ]}','${body[CYFS]}','${body[YZ]}','${body[HYSX]}','${body[YYSX]}','${body[TYSX]}','${body[BS]}','${body[DYM]}')"))
                         .to("jdbc:centralbase")
+                        .log("insert !!!")
+                        .end();
+
+                from("timer:mytimer3?period=999999999")
+                        .routeId("oracle-3")
+                        .setHeader("date",constant(date1))
+                        .setBody(simple("select  jh,rq,scsj, rcyl1,rcyl,rcql,hs, bz from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null "))
+                        .to("jdbc:oracle")
                         .split(body()).process(exchange -> {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);
-                    String parent_id = aRow.get("station_id").toString().substring(0, aRow.get("station_id").toString().lastIndexOf("@"));
-                    aRow.put("parent_id", parent_id);
-                    String org_id = UUID.randomUUID().toString().replace("-", "").substring(0, 10);
-                    aRow.put("org_id", org_id);
+                    if (aRow.get("SCSJ")==null)   aRow.put("SCSJ","0.0");
+                    if (aRow.get("RCYL1")==null) aRow.put("RCYL1","0.0");
+                    if (aRow.get("RCYL")==null) aRow.put("RCYL","0.0");
+                    if (aRow.get("RCQL")==null) aRow.put("RCQL","0.0");
+                    if (aRow.get("HS")==null)   aRow.put("HS","0.0");
+                    if (aRow.get("BZ")==null)  aRow.put("DYM","");
+                })
+                        .setBody(simple("insert into centralbase.cb_pc_pro_wellbore_vol_daily(well_id,prod_date,prod_time,liq_prod_daily,oil_prod_daily,gas_prod_daily,water_cut,remarks) " +
+                                "values ('${body[JH]}','${body[RQ]}','${body[SCSJ]}','${body[RCYL1]}','${body[RCYL]}','${body[RCQL]}','${body[HS]}','${body[BZ]}')"))
+                        .to("jdbc:centralbase")
+                        .log("insert !!!")
+                        .end();
+
+                from("timer:mytimer4?period=999999999")
+                        .routeId("oracle-4")
+                        .setHeader("date",constant(date1))
+                        .setBody(simple("select  jh,rq,dym,jy,ly,bj,bs,bx,zs,cc,cs,blx,dl from zd_zdgs.dba01@A2 where rq  = to_date('${header.date}','yyyy-MM-dd') and qyrq is not null  "))
+                        .to("jdbc:oracle")
+                        .split(body()).process(exchange -> {
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    if (aRow.get("DYM")==null)   aRow.put("DYM","0.0");
+                    if (aRow.get("JY")==null) aRow.put("JY","0.0");
+                    if (aRow.get("LY")==null) aRow.put("LY","0.0");
+                    if (aRow.get("BJ")==null) aRow.put("BJ","0.0");
+                    if (aRow.get("BS")==null)   aRow.put("BS","0.0");
+                    if (aRow.get("BX")==null)  aRow.put("BX","0.0");
+                    if (aRow.get("ZS")==null)  aRow.put("ZS","0.0");
+                    if (aRow.get("CC")==null)  aRow.put("CC","0.0");
+                    if (aRow.get("CS")==null)  aRow.put("CS","0.0");
+                    if (aRow.get("BLX")==null)  aRow.put("BLX","");
+                    if (aRow.get("DL")==null)  aRow.put("DL","0.0");
+                })
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_daily(well_id,prod_date,dynamic_liq_level,static_pressure,flow_pres,pump_diameter,pump_depth,pump_efficiency,rotate_frequency,stroke_length,stroke_frequency,pump_type,elec_frequency)" +
+                                "values ('${body[JH]}','${body[RQ]}','${body[DYM]}','${body[JY]}','${body[LY]}','${body[BJ]}','${body[BS]}','${body[BX]}','${body[ZS]}','${body[CC]}','${body[CS]}','${body[BLX]}','${body[DL]}')"))
+                        .to("jdbc:centralbase")
+                        .log("insert !!!")
+                        .end();
+
+                //从天安哪里获取的数据
+                from("timer:mytimer1?period=10000")
+                        .routeId("jdbc-gtsj-?")
+                        .setHeader("dyna_create_time", constant(formatDate))
+                        .setBody(simple("select well_id,dyna_create_time,check_date,displacement,disp_load,stroke,frequency,susp_max_load,susp_min_load from public.pc_fd_pumpjack_dyna_dia_t where   dyna_create_time > '2021-07-01 00:00:00' "))
+                        .to("jdbc:gtsj")
+                        .split(body()).process(exchange -> {
+                    Message in = exchange.getIn();
+                    HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                    String prod_date = aRow.get("dyna_create_time").toString().split("\\+")[0];
+                    aRow.put("dyna_create_time", prod_date);
+                    if (aRow.get("displacement") != null && !aRow.get("displacement").equals("") && aRow.get("disp_load") != null && !aRow.get("disp_load").equals("")) {
+                        String[] displacements = aRow.get("displacement").toString().split(";");
+                        String[] disp_loads = aRow.get("disp_load").toString().split(";");
+                        String sgt = "";
+                        for (int i = 0; i < displacements.length; i++) {
+                            sgt = sgt + displacements[i] + "," + disp_loads[i] + ",";
+                        }
+                        String[] s = sgt.split(",");
+                        String w = "";
+                        for (int i = 0; i < s.length; i++) {
+                            w += new BigDecimal(Math.round(Double.parseDouble(s[i]) * 100)).stripTrailingZeros().toPlainString() + ",";
+                        }
+                        aRow.put("sgt", w);
+                    }
+                    if (aRow.get("stroke") == null) aRow.put("stroke","0.0");
+                    if (aRow.get("frequency") == null) aRow.put("frequency","0.0");
+                    if (aRow.get("susp_max_load") == null) aRow.put("susp_max_load","0.0");
+                    if (aRow.get("susp_min_load") == null) aRow.put("susp_min_load","0.0");
+
+                })//从这里开始
+                        .setBody(simple("insert into centralbase.cb_temp_well_mech_runtime(well_id,prod_date,stroke_length,stroke_frequency,susp_max_load,susp_min_load,sgt) " +
+                                "values ('${body[well_id]}','${body[dyna_create_time]}','${body[stroke]}','${body[frequency]}','${body[susp_max_load]}','${body[susp_min_load]}','${body[sgt]}')"))
+                        .to("jdbc:centralbase")
+                        .log("insert!!!").end();
+
+           /*     from("timer:mytimer2?period=999999999")
+                        .routeId("oracle-2")
+                        .setBody(simple("select distinct station_id,station_name from centralbase.cb_cd_well_source"))
+                        .to("jdbc:centralbase")
+                        .split(body()).process(exchange -> {
+                            Message in = exchange.getIn();
+                            HashMap<String, Object> aRow = in.getBody(HashMap.class);
+                            String parent_id = aRow.get("station_id").toString().substring(0, aRow.get("station_id").toString().lastIndexOf("@"));
+                            aRow.put("parent_id",parent_id);
+                            String org_id = UUID.randomUUID().toString().replace("-","").substring(0,10);
+                            aRow.put("org_id",org_id);
                 })
                         .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_name,org_level,parent_id)" +
                                 "values('${body[org_id]}','${body[station_name]}','3','${body[parent_id]}')"))
@@ -364,13 +483,13 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                     String parent_id = aRow.get("parent_id").toString().split("@")[0];
                     String org_name = aRow.get("parent_id").toString().split("@")[1];
                     //if (!aRow.containsValue(org_name)){
-                    aRow.put("org_name", org_name);
+                    aRow.put("org_name",org_name);
                     // }
                     //if (!aRow.containsValue(parent_id)){
-                    aRow.put("parent_id", parent_id);
+                    aRow.put("parent_id",parent_id);
                     //}
-                    String org_id = UUID.randomUUID().toString().replace("-", "").substring(0, 10);
-                    aRow.put("levelTwoOrgId", org_id);
+                    String org_id = UUID.randomUUID().toString().replace("-","").substring(0,10);
+                    aRow.put("levelTwoOrgId",org_id);
                 })
                         .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_name,org_level,parent_id)" +
                                 "values('${body[levelTwoOrgId]}','${body[org_name]}','2','${body[parent_id]}')"))
@@ -386,8 +505,8 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                     Message in = exchange.getIn();
                     HashMap<String, Object> aRow = in.getBody(HashMap.class);//沙南作业区@沙采二区
 
-                    String org_id = UUID.randomUUID().toString().replace("-", "").substring(0, 10);
-                    aRow.put("levelOneOrgId", org_id);
+                    String org_id = UUID.randomUUID().toString().replace("-","").substring(0,10);
+                    aRow.put("levelOneOrgId",org_id);
                 })
                         .setBody(simple("insert into centralbase.cb_pc_organization(org_id,org_name,org_level,parent_id)" +
                                 "values('${body[levelOneOrgId]}','${body[parent_id]}','1','0')"))
@@ -420,40 +539,10 @@ public class CamelJDBCConfiguration /*extends RouteBuilder */ {
                         .setBody(simple("update centralbase.cb_cd_well_source set org_id = '${body[org_id]}' where station_name = '${body[org_name]}'"))
                         .to("jdbc:centralbase")
                         .log("insert !!!")
-                        .end();
+                        .end();*/
             }
 
             ;
         };
     }
-}
-
-
-   /* from("timer:mytimer1?period=30000")
-                        .setHeader("acqui_time", constant(formatDate))
-                                .setBody(simple("select distinct cydmc jlz ,zyq from zd_zdgs.dba01@A2 where rq  > to_date('2021-06-30','yyyy-MM-dd')  group by zyq,cydmc"))
-                                .to("jdbc:oracle")
-                                .split(body()).process(exchange -> {
-                                Message in = exchange.getIn();
-                                HashMap<String, Object> aRow = in.getBody(HashMap.class);
-        })
-        .setBody(simple("insert into centralbase.cb_pc_organization(org_name,org_level,parent_id) " +
-        "select '${body[JLZ]}' ,'2' ,org_id  from centralbase.cb_pc_organization  where org_name = '${body[ZYQ]}' " +
-        "group by org_id "))
-        .to("jdbc:centralbase")
-        .log("insert !!!!")
-        .end();*/
-    /*from("timer:mytimer2?period=30000")
-                        .startupOrder(20)
-                                .setBody(simple("select distinct cydmc jlz ,zyq,zk zd from zd_zdgs.dba01@A2 where rq  > to_date('2021-06-30','yyyy-MM-dd')"))
-                                .to("jdbc:oracle")
-                                .split(body()).process(exchange -> {
-                                Message in = exchange.getIn();
-                                HashMap<String, Object> aRow = in.getBody(HashMap.class);
-        })
-        .setBody(simple("insert into centralbase.cb_pc_organization(org_name,org_level,parent_id) " +
-        "select '${body[ZD]}' ,'3' ,org_id  from centralbase.cb_pc_organization  where org_name = '${body[JLZ]}' " +
-        "group by org_id "))
-        .to("jdbc:centralbase")
-        .log("insert completed")
-        .end();*/
+}