Merge #49 into master from pingkang

Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

* pingkang: (4 commits)
  点众获取下载地址
  点众获取下载地址
  点众
  Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

Signed-off-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>
Merged-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>

CR-link: https://codeup.aliyun.com/60c06368dbbed34fe0f0c263/job/hr_job-bytedance/change/49

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/DianZhongConfig.java

@@ -4,16 +4,28 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 @Configuration
 @ConfigurationProperties(prefix = "dianzhong.client")
 public class DianZhongConfig {
     private static List<DZConfig> dzConfig;
+    private static final Map<String, String> tokenMap = new ConcurrentHashMap<>();
 
     public List<DZConfig> getDzConfig() {
         return dzConfig;
     }
 
+    public String getToken(String clientId) {
+        if (tokenMap.isEmpty()) {
+            for (DZConfig config : dzConfig) {
+                tokenMap.put(config.getClientId(), config.getToken());
+            }
+        }
+        return tokenMap.get(clientId);
+    }
+
     public void setDzConfig(List<DZConfig> dzConfig) {
         DianZhongConfig.dzConfig = dzConfig;
     }
@@ -21,6 +33,7 @@ public class DianZhongConfig {
     public static class DZConfig {
         private String clientId;
         private String token;
+        private String type;
 
         public String getClientId() {
             return clientId;
@@ -37,5 +50,13 @@ public class DianZhongConfig {
         public void setToken(String token) {
             this.token = token;
         }
+
+        public String getType() {
+            return type;
+        }
+
+        public void setType(String type) {
+            this.type = type;
+        }
     }
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/RestTemplateConfig.java

@@ -0,0 +1,61 @@
+package cn.com.ctop.job.bytedance.config;
+
+import org.apache.http.NoHttpResponseException;
+import org.apache.http.client.HttpRequestRetryHandler;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.protocol.HttpContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class RestTemplateConfig {
+
+
+    @Bean(name = "downloadFileRestTemplate")
+    public RestTemplate restTemplate(){
+        HttpComponentsClientHttpRequestFactory httpRequestFactory = new HttpComponentsClientHttpRequestFactory();
+        httpRequestFactory.setConnectionRequestTimeout(50000);
+        httpRequestFactory.setConnectTimeout(5000);
+        httpRequestFactory.setReadTimeout(300000);
+
+        CloseableHttpClient httpClient = HttpClientBuilder.create()
+                .setConnectionTimeToLive(300, TimeUnit.SECONDS)
+                .setRetryHandler(new HttpRequestRetryHandler() {
+                    @Override
+                    public boolean retryRequest(IOException exception, int executionCount, HttpContext context) {
+                        if (executionCount >= 3) {
+                            // 最多重试3次
+                            return false;
+                        } else if (exception instanceof NoHttpResponseException) {
+                            // 对于服务器未响应的情况进行重试
+                            requestWait();
+                            return true;
+                        } else if (exception instanceof InterruptedIOException) {
+                            // 对于连接中断的情况进行重试
+                            requestWait();
+                            return true;
+                        }
+                        return false;
+                    }
+                }).build();
+
+        httpRequestFactory.setHttpClient(httpClient);
+
+        return new RestTemplate(httpRequestFactory);
+    }
+
+    public void requestWait() {
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/controller/DianZhongController.java

@@ -1,12 +1,8 @@
 package cn.com.ctop.job.bytedance.controller;
 
 import cn.com.ctop.job.bytedance.config.DianZhongConfig;
-import cn.com.ctop.job.bytedance.data.service.IBytedancePromoterReportService;
-import cn.com.ctop.job.bytedance.data.service.IChannelService;
 import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
-import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.DateUtils;
-import com.alibaba.fastjson.JSONObject;
 import io.swagger.annotations.Api;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,17 +32,20 @@ public class DianZhongController {
                             @RequestParam(required = false, defaultValue = "false") Boolean allDay) {
         List<DianZhongConfig.DZConfig> dzConfig = dianZhongConfig.getDzConfig();
         for (DianZhongConfig.DZConfig config : dzConfig) {
+            if ("iaa".equals(config.getType())) {
+                continue;
+            }
             String clientId = config.getClientId();
             String token = config.getToken();
             String queryStartTime = DateUtils.getQueryStartTime(statDate, "yyyy-MM-dd HH:mm:ss", allDay);
             String queryEndTime = DateUtils.getQueryEndTime(statDate, "yyyy-MM-dd HH:mm:ss", allDay);
-            System.out.println(type);
-            System.out.println(clientId);
-            System.out.println(token);
-            System.out.println(queryStartTime);
-            System.out.println(queryEndTime);
-            //dianZhongService.getDownloadUrl(type, clientId, token, statDate,queryStartTime,queryEndTime);
+            dianZhongService.getDownloadUrl(type, clientId, token, queryStartTime, queryEndTime, null, false);
         }
     }
 
+    @GetMapping("/getUrlData")
+    public void urlData(String statDate) {
+        dianZhongService.getUrlData(statDate);
+    }
+
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/mapper/DianZhongMapper.java

@@ -3,6 +3,14 @@ package cn.com.ctop.job.bytedance.data.mapper;
 import com.alibaba.fastjson.JSONObject;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 public interface DianZhongMapper {
     void insertDownloadTask(@Param("item") JSONObject jsonObject);
+
+    List<JSONObject> getDZRetry();
+
+    List<JSONObject> getTaskUrl(@Param("statDate") String statDate);
+
+    void updateTaskDownloaded(@Param("taskId") String taskId);
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/mapper/xml/DianZhongMapper.xml

@@ -3,13 +3,35 @@
 <mapper namespace="cn.com.ctop.job.bytedance.data.mapper.DianZhongMapper">
 
     <insert id="insertDownloadTask">
-        replace into media_api.dianzhong_data_download_url
-        (stat_date,task_id,`type`,client_id) values
+        insert into media_api.dianzhong_data_download_url
+        (stat_date,task_id,`type`,client_id,start_time,end_time) values
         (
         #{item.stat_date},
         #{item.task_id},
         #{item.type},
-        #{item.client_id}
+        #{item.client_id},
+        #{item.start_time},
+        #{item.end_time}
         )
+        on duplicate key update
+        stat_date= VALUES(stat_date),`type`= VALUES(`type`),client_id= VALUES(client_id),
+        start_time= VALUES(start_time),end_time= VALUES(end_time)
     </insert>
+    <update id="updateTaskDownloaded">
+        update media_api.dianzhong_data_download_url
+        set downloaded = 1 where task_id = #{taskId}
+    </update>
+
+    <select id="getTaskUrl" resultType="com.alibaba.fastjson.JSONObject">
+        select task_id,url,type,client_id
+        from media_api.dianzhong_data_download_url
+        where stat_date= ${statDate} and downloaded = 0 and url is not null and url != '-1'
+    </select>
+
+    <select id="getDZRetry" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT t1.is_send AS 'isSend', t1.id AS 'id', t1.advertiser_id AS 'advertiserId', t1.start_date AS 'startDate', t1.end_date AS 'endDate', t1.data_dimension AS 'dataDimension', t1.data_type AS 'dataType', t3.access_token AS 'accessToken'
+        FROM bytedance_error_log t1
+        where data_dimension like 'dianzhong%'
+    </select>
+
 </mapper>

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/IDianZhongService.java

@@ -1,6 +1,12 @@
 package cn.com.ctop.job.bytedance.data.service;
 
+import com.alibaba.fastjson.JSONObject;
+
+import java.util.List;
+
 public interface IDianZhongService {
     void getDownloadUrl(String dataType, String clientId, String token,
-                        String statDay, String startTime, String endTime);
+                        String startTime, String endTime,Long errId,boolean isLog);
+    List<JSONObject> getDZRetry();
+    void getUrlData(String statDate);
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/service/impl/DianZhongServiceImpl.java

@@ -4,25 +4,67 @@ import cn.com.ctop.job.bytedance.data.entity.BytedanceErrorLog;
 import cn.com.ctop.job.bytedance.data.mapper.BytedancePromoterReportMapper;
 import cn.com.ctop.job.bytedance.data.mapper.DianZhongMapper;
 import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
+import cn.com.ctop.job.bytedance.data.utils.Check;
+import cn.com.ctop.job.bytedance.data.utils.DateUtils;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import cn.hutool.core.date.DateUtil;
 import cn.hutool.crypto.SecureUtil;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.ResponseExtractor;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 @Service
 @Slf4j
 public class DianZhongServiceImpl implements IDianZhongService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
 
     @Autowired
+    private RestTemplate downloadFileRestTemplate;
+
+    @Autowired
     private DianZhongMapper dianZhongMapper;
 
+    private static final Map<Integer, Integer> domain2Platform = new HashMap<Integer, Integer>() {{
+        put(14, 2);//抖小
+        put(16, 1);//微小
+        put(17, 3);//快小
+    }};
+
+    private static final Map<String, String> clientDomain2AppId = new HashMap<String, String>() {{
+        put("10003129" + "14", "dzdyIAP");//抖小
+    }};
+
+    private static final Map<String, String> media2Id = new HashMap<String, String>() {{
+        put("adq", "3");//腾讯广点通
+        put("dy", "1");//字节巨量
+        put("ks", "2");//快手-磁力
+    }};
+
+    public List<JSONObject> getDZRetry() {
+        return dianZhongMapper.getDZRetry();
+    }
+
+    /*
+    获取点众IAP数据下载url
+     */
     public void getDownloadUrl(String dataType, String clientId, String token,
-                                String statDay, String startTime, String endTime) {
+                               String startTime, String endTime, Long errId, boolean isLog) {
         long l = System.currentTimeMillis();
         JSONObject param = new JSONObject()
                 .fluentPut("clientId", clientId)
@@ -30,47 +72,226 @@ public class DianZhongServiceImpl implements IDianZhongService {
                 .fluentPut("edate", String.valueOf(DateUtil.parse(endTime, "yyyy-MM-dd HH:mm:ss").getTime() / 1000))
                 .fluentPut("timestamp", l)
                 .fluentPut("signKey", getSignKey(clientId, token, l));
+        if ("user".equals(dataType)) {
+            param.put("queryField", "loginTime");
+        }
         try {
             String url = "https://routine.wqxsw.com/flames/channel/query/" + dataType;
             String response = HttpUtils.httpPostRequest(url, param, null);
             JSONObject responseJson = JSONObject.parseObject(response);
-            System.out.println(response);
             if ("0".equals(responseJson.getString("retCode"))) {
                 String taskId = responseJson.getString("data");
                 JSONObject jsonObject = new JSONObject()
-                        .fluentPut("stat_date", statDay.replace("-", ""))
+                        .fluentPut("stat_date", DateUtils.transformFormat(endTime, "yyyy-MM-dd HH:mm:ss", "yyyyMMdd"))
                         .fluentPut("task_id", taskId)
                         .fluentPut("type", dataType)
-                        .fluentPut("client_id", clientId);
+                        .fluentPut("client_id", clientId)
+                        .fluentPut("start_time", startTime)
+                        .fluentPut("end_time", endTime);
                 dianZhongMapper.insertDownloadTask(jsonObject);
-                log.info("获取点众数据下载地址成功 {} {} {}~{} ", dataType, clientId, startTime, endTime);
+                log.info("获取点众{}数据下载地址成功 {} {}~{} taskId:{}", dataType, clientId, startTime, endTime, taskId);
+
+                if (errId != null) {
+                    promoterReportMapper.deleteInfoByErrorId(errId);
+                }
             } else {
-                log.error("获取点众数据下载地址错误 {} {} {}~{} msg:{}", dataType, clientId, startTime, endTime, responseJson.getString("retMsg"));
+                log.error("获取点众{}数据下载地址错误 {} {}~{} msg:{}", dataType, clientId, startTime, endTime, responseJson.getString("retMsg"));
+                if (isLog) {
+                    promoterReportMapper.replaceErrorInfo(
+                            BytedanceErrorLog.builder()
+                                    .msg(responseJson.getString("retMsg"))
+                                    .advertiserId(Long.valueOf(clientId))
+                                    .dataDimension("dianzhong_url")
+                                    .dataType(dataType)
+                                    .startDate(startTime)
+                                    .endDate(endTime)
+                                    .build());
+                }
+            }
+        } catch (Exception e) {
+            log.error("获取点众{}数据下载地址错误 {} {}~{} ", dataType, clientId, startTime, endTime, e);
+            if (isLog) {
                 promoterReportMapper.replaceErrorInfo(
                         BytedanceErrorLog.builder()
-                                .msg(responseJson.getString("retMsg"))
+                                .msg(e.getMessage())
                                 .advertiserId(Long.valueOf(clientId))
-                                .dataDimension("dianzhong")
+                                .dataDimension("dianzhong_url")
                                 .dataType(dataType)
                                 .startDate(startTime)
                                 .endDate(endTime)
                                 .build());
             }
-        } catch (Exception e) {
-            log.error("获取点众数据下载地址错误 {} {} {}~{} ", dataType, clientId, startTime, endTime, e);
-            promoterReportMapper.replaceErrorInfo(
-                    BytedanceErrorLog.builder()
-                            .msg(e.getMessage())
-                            .advertiserId(Long.valueOf(clientId))
-                            .dataDimension("dianzhong")
-                            .dataType(dataType)
-                            .startDate(startTime)
-                            .endDate(endTime)
-                            .build());
+        }
+    }
+
+    /*
+    下载点众IAP数据
+     */
+    public void getUrlData(String statDate) {
+        List<JSONObject> tasks = dianZhongMapper.getTaskUrl(statDate.replace("-", ""));
+        if (!Check.isNull(tasks)) {
+            for (JSONObject task : tasks) {
+                String taskId = task.getString("task_id");
+                String url = task.getString("url");
+                String type = task.getString("type");
+                String clientId = task.getString("client_id");
+
+                downloadFileRestTemplate.execute(url, org.springframework.http.HttpMethod.GET, null,
+                        (ResponseExtractor<Void>) response -> {
+                            try (BufferedReader reader = new BufferedReader(
+                                    new InputStreamReader(response.getBody(), StandardCharsets.UTF_8))) {
+                                String line;
+                                while ((line = reader.readLine()) != null) {
+                                    System.out.println(line);
+                                    if ("user".equals(type)) {
+                                        JSONObject dzUser = getDZUser(line, clientId);
+                                        System.out.println(dzUser);
+                                        dorisLoadManager.add("hr_playlet", "user_event", dzUser);
+                                        JSONObject dzLogin = getDZLogin(line, clientId);
+                                        System.out.println(dzLogin);
+                                        dorisLoadManager.add("hr_playlet", "server_event", dzLogin);
+                                    } else if ("order".equals(type)) {
+                                        JSONObject dzOrder = getDZOrder(line, clientId);
+                                        System.out.println(dzOrder);
+                                        dorisLoadManager.add("hr_playlet", "server_event", dzOrder);
+                                    }
+                                }
+                            }
+                            return null;
+                        });
+                dianZhongMapper.updateTaskDownloaded(taskId);
+            }
         }
     }
 
     private static String getSignKey(String clientId, String token, long timeStamp) {
         return SecureUtil.md5(clientId + token + timeStamp).toLowerCase();
     }
+
+    private static JSONObject getDZUser(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject userJson = new JSONObject();
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long ctime = jsonObject.getLong("ctime");
+        String statDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
+        String registerTime = DateUtils.formatDate(ctime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        userJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+        userJson.put("uid", jsonObject.getString("userId"));
+        userJson.put("stat_date", statDate);
+        userJson.put("register_time", registerTime);
+        userJson.put("created_at", ctime);
+        userJson.put("open_id", jsonObject.getString("openId"));
+        userJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        userJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        userJson.put("channel_id", jsonObject.getString("channelId"));
+        userJson.put("link_id", jsonObject.getString("referralId"));
+        userJson.put("expiration_time", jsonObject.getString("vipEndDate"));
+        userJson.put("pay_amount", jsonObject.getString("payAmount"));
+        userJson.put("pay_count", jsonObject.getString("payCount"));
+        userJson.put("last_login_time", jsonObject.getString("loginTime"));
+        userJson.put("ip", jsonObject.getString("ip"));
+        userJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        userJson.put("person_id", jsonObject.getString("kdrId"));
+        userJson.put("a_id", jsonObject.getString("projectId"));
+        userJson.put("ad_id", adId);
+        userJson.put("activation_time", activationTime);
+        userJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+        return userJson;
+    }
+
+    private static JSONObject getDZLogin(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject loginJson = new JSONObject();
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long loginTime = jsonObject.getLong("loginTime");
+        String statDate = DateUtils.formatDate(loginTime * 1000, "yyyyMMdd");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        String userId = jsonObject.getString("userId");
+        loginJson.put("stat_date", statDate);
+        loginJson.put("event", "LOGIN");
+        loginJson.put("event_pk", SecureUtil.md5(String.format("dianzhong_login_%s_%s", userId, loginTime)));
+        loginJson.put("event_tag", "dianzhong_login");
+        loginJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+
+        loginJson.put("uid", userId);
+        loginJson.put("open_id", jsonObject.getString("openId"));
+        loginJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        loginJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        loginJson.put("channel_id", jsonObject.getString("channelId"));
+        loginJson.put("link_id", jsonObject.getString("referralId"));
+        loginJson.put("register_date", jsonObject.getString("registerDate"));
+        loginJson.put("expiration_time", jsonObject.getString("vipEndDate"));
+        loginJson.put("last_login_time", jsonObject.getString("loginTime"));
+        loginJson.put("created_at", loginTime);
+        loginJson.put("ip", jsonObject.getString("ip"));
+        loginJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        loginJson.put("person_id", jsonObject.getString("kdrId"));
+        loginJson.put("a_id", jsonObject.getString("projectId"));
+        loginJson.put("ad_id", adId);
+        loginJson.put("activation_time", activationTime);
+        loginJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+        return loginJson;
+    }
+
+    private static JSONObject getDZOrder(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject orderJson = new JSONObject();
+        String orderId = jsonObject.getString("id");
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long ctime = jsonObject.getLong("ctime");
+        String statDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        orderJson.put("stat_date", statDate);
+        orderJson.put("event", "PAY");
+        orderJson.put("event_pk", SecureUtil.md5("dianzhong_order_id:" + orderId));
+        orderJson.put("event_tag", "dianzhong_order");
+        orderJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+        orderJson.put("order_no", orderId);
+        orderJson.put("platform_order_no", jsonObject.getString("outTradeNo"));
+        orderJson.put("total_fee", jsonObject.getString("discount"));
+        orderJson.put("type", jsonObject.getString("type"));
+        orderJson.put("status", jsonObject.getString("statusNotify"));
+        orderJson.put("created_at", ctime);
+        orderJson.put("pay_time", jsonObject.getLong("finishTime"));
+        orderJson.put("uid", jsonObject.getString("userId"));
+        orderJson.put("channel_id", jsonObject.getString("channelId"));
+        orderJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        orderJson.put("video_id", jsonObject.getString("sourceInfo"));
+        orderJson.put("video_detail_id", jsonObject.getString("chapterId"));
+        orderJson.put("video_name", jsonObject.getString("sourceDesc"));
+        orderJson.put("register_date", jsonObject.getString("registerDate"));
+        orderJson.put("open_id", jsonObject.getString("openId"));
+        orderJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        orderJson.put("link_id", jsonObject.getString("referralId"));
+        orderJson.put("a_id", jsonObject.getString("projectId"));
+        orderJson.put("ad_id", adId);
+        orderJson.put("person_id", jsonObject.getString("fromDrId") != null ? jsonObject.getString("fromDrId") : jsonObject.getString("kdrId"));
+        orderJson.put("scene", jsonObject.getString("scene"));
+        orderJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        orderJson.put("activation_time", activationTime);
+        orderJson.put("pay_type", jsonObject.getString("orderSubType"));
+        orderJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+        orderJson.put("money_benefit", jsonObject.getDouble("moneyBenefit"));
+        orderJson.put("params", jsonObject.toJSONString());
+        return orderJson;
+    }
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/utils/DateUtils.java

@@ -221,7 +221,11 @@ public class DateUtils {
     }
 
     public static String formatDate(Long timestamp, String formatPattern) {
-        return formatDate(new Date(timestamp), formatPattern);
+        try {
+            return formatDate(new Date(timestamp), formatPattern);
+        }catch (Exception e) {
+            return null;
+        }
     }
 
     public static String formatDate(Date date, String formatPattern) {

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/BytedanceReportRetryJob.java

@@ -1,6 +1,7 @@
 package cn.com.ctop.job.bytedance.handler;
 
 
+import cn.com.ctop.job.bytedance.config.DianZhongConfig;
 import cn.com.ctop.job.bytedance.data.enums.AccountTypeEnum;
 import cn.com.ctop.job.bytedance.data.service.*;
 import cn.com.ctop.job.bytedance.data.utils.Check;
@@ -10,6 +11,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -47,6 +49,12 @@ public class BytedanceReportRetryJob {
     @Autowired
     private IBytedanceMaterialReportService bytedanceMaterialReportService;
 
+    @Autowired
+    private IDianZhongService dianZhongService;
+
+    @Resource
+    private DianZhongConfig dianZhongConfig;
+
 
     private static ExecutorService hourAccountRetrExecutorService = Executors.newFixedThreadPool(2);
     private static ExecutorService dailyAccountRetrExecutorService = Executors.newFixedThreadPool(2);
@@ -245,6 +253,32 @@ public class BytedanceReportRetryJob {
             log.info("巨量星图重试列表为空");
         }
 
+        List<JSONObject> dzRetry = dianZhongService.getDZRetry();
+        if (!Check.isNull(dzRetry)) {
+            for (int i = 0; i < dzRetry.size(); i++) {
+                JSONObject jsonObject = dzRetry.get(i);
+                Long id = jsonObject.getLong("id");
+                String clientId = jsonObject.getString("advertiserId");
+                String startDate = jsonObject.getString("startDate");
+                String endDate = jsonObject.getString("endDate");
+                String dataDimension = jsonObject.getString("dataDimension");
+                String dataType = jsonObject.getString("dataType");
+                String token = dianZhongConfig.getToken(clientId);
+                if ("dianzhong_url".equals(dataDimension)) {
+                    changduExecutorService.submit(new Runnable() {
+                        @Override
+                        public void run() {
+                            if (!Check.isNull(token)) {
+                                dianZhongService.getDownloadUrl(dataType, clientId, token, startDate, endDate, id, false);
+                            }
+                        }
+                    });
+                }
+            }
+        } else {
+            log.info("点众重试列表为空");
+        }
+
     }
 
 

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/handler/DianZhongJob.java

@@ -0,0 +1,55 @@
+package cn.com.ctop.job.bytedance.handler;
+
+
+import cn.com.ctop.job.bytedance.config.DianZhongConfig;
+import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
+import cn.com.ctop.job.bytedance.data.utils.DateUtils;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+import javax.annotation.Resource;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+
+@Slf4j
+@Component
+public class DianZhongJob {
+    @Autowired
+    private IDianZhongService dianZhongService;
+
+    @Resource
+    private DianZhongConfig dianZhongConfig;
+
+    /**
+     * 获取点众IAP数据url
+     */
+    @XxlJob("dzDownloadUrl")
+    public void dzDownloadUrl() {
+        List<DianZhongConfig.DZConfig> dzConfig = dianZhongConfig.getDzConfig();
+        for (DianZhongConfig.DZConfig config : dzConfig) {
+            if ("iaa".equals(config.getType())) {
+                continue;
+            }
+            String statDate = DateUtils.dateAddByType(new Date(), -10, Calendar.MINUTE, "yyyy-MM-dd");
+            String clientId = config.getClientId();
+            String token = config.getToken();
+            String queryStartTime = DateUtils.getQueryStartTime(statDate, "yyyy-MM-dd HH:mm:ss", false);
+            String queryEndTime = DateUtils.getQueryEndTime(statDate, "yyyy-MM-dd HH:mm:ss", false);
+            dianZhongService.getDownloadUrl("user", clientId, token, queryStartTime, queryEndTime, null, true);
+            dianZhongService.getDownloadUrl("order", clientId, token, queryStartTime, queryEndTime, null, true);
+        }
+    }
+
+    /**
+     * 点众IAP数据下载
+     */
+    @XxlJob("dzUrlData")
+    public void dzUrlData() {
+        String statDate = DateUtils.dateAddByType(new Date(), -10, Calendar.MINUTE, "yyyy-MM-dd");
+        dianZhongService.getUrlData(statDate);
+    }
+
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris/sink/DorisMultiTableStreamLoadManager.java

@@ -5,11 +5,11 @@ import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
 
 import javax.annotation.PreDestroy;
 import java.util.Collection;
-import java.util.List;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class DorisMultiTableStreamLoadManager {
@@ -48,5 +48,15 @@ public class DorisMultiTableStreamLoadManager {
         log.info("Shutting down DorisMultiTableStreamLoadManager");
         bufferManagers.values().forEach(TableBufferManager::flushNow);
         uploaderExecutor.shutdown();
+        try {
+            if (!uploaderExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
+                log.info("doris stream load 任务未完成，强制关闭线程池...");
+                uploaderExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            uploaderExecutor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+        log.info("doris stream load 线程池已关闭");
     }
 }

job-bytedance/src/main/resources/application.yml

@@ -195,4 +195,9 @@ dianzhong:
     dzConfig[0]:
       clientId: 10003129
       token: ckogrgCE4PEDxT1oL5
+      type: iap
+    dzConfig[1]:
+      clientId: 10004501
+      token: HgmmEQM4ddcLQ8mn5i
+      type: iaa