点众

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

@@ -33,6 +33,7 @@ public class DianZhongConfig {
     public static class DZConfig {
         private String clientId;
         private String token;
+        private String type;
 
         public String getClientId() {
             return clientId;
@@ -49,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

@@ -22,11 +22,11 @@ public class RestTemplateConfig {
     public RestTemplate restTemplate(){
         HttpComponentsClientHttpRequestFactory httpRequestFactory = new HttpComponentsClientHttpRequestFactory();
         httpRequestFactory.setConnectionRequestTimeout(50000);
-        httpRequestFactory.setConnectTimeout(50000);
-        httpRequestFactory.setReadTimeout(50000);
+        httpRequestFactory.setConnectTimeout(5000);
+        httpRequestFactory.setReadTimeout(300000);
 
         CloseableHttpClient httpClient = HttpClientBuilder.create()
-                .setConnectionTimeToLive(50, TimeUnit.SECONDS) // 连接存活时间为30秒
+                .setConnectionTimeToLive(300, TimeUnit.SECONDS)
                 .setRetryHandler(new HttpRequestRetryHandler() {
                     @Override
                     public boolean retryRequest(IOException exception, int executionCount, HttpContext context) {
@@ -48,8 +48,7 @@ public class RestTemplateConfig {
 
         httpRequestFactory.setHttpClient(httpClient);
 
-        RestTemplate restTemplate = new RestTemplate(httpRequestFactory);
-        return restTemplate;
+        return new RestTemplate(httpRequestFactory);
     }
 
     public void requestWait() {

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

@@ -32,12 +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);
-            dianZhongService.getDownloadUrl(type, clientId, token, queryStartTime, queryEndTime,null,false);
+            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/service/IDianZhongService.java

@@ -8,4 +8,5 @@ public interface IDianZhongService {
     void getDownloadUrl(String dataType, String clientId, String token,
                         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

@@ -7,10 +7,12 @@ 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;
@@ -19,12 +21,17 @@ 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
@@ -33,10 +40,29 @@ public class DianZhongServiceImpl implements IDianZhongService {
     @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 startTime, String endTime, Long errId, boolean isLog) {
         long l = System.currentTimeMillis();
@@ -98,6 +124,9 @@ public class DianZhongServiceImpl implements IDianZhongService {
         }
     }
 
+    /*
+    下载点众IAP数据
+     */
     public void getUrlData(String statDate) {
         List<JSONObject> tasks = dianZhongMapper.getTaskUrl(statDate.replace("-", ""));
         if (!Check.isNull(tasks)) {
@@ -115,16 +144,21 @@ public class DianZhongServiceImpl implements IDianZhongService {
                                 while ((line = reader.readLine()) != null) {
                                     System.out.println(line);
                                     if ("user".equals(type)) {
-                                        //todo
+                                        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)) {
-                                        //todo
-
+                                        JSONObject dzOrder = getDZOrder(line, clientId);
+                                        System.out.println(dzOrder);
+                                        dorisLoadManager.add("hr_playlet", "server_event", dzOrder);
                                     }
                                 }
                             }
                             return null;
                         });
-
                 dianZhongMapper.updateTaskDownloaded(taskId);
             }
         }
@@ -133,4 +167,131 @@ public class DianZhongServiceImpl implements IDianZhongService {
     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/DianZhongJob.java

@@ -24,12 +24,15 @@ public class DianZhongJob {
     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();
@@ -40,4 +43,13 @@ public class DianZhongJob {
         }
     }
 
+    /**
+     * 点众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