Merge #55 into master from pingkang

点众

* pingkang: (5 commits)
  点众
  点众
  点众
  点众
  点众

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/55

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

@@ -44,7 +44,7 @@ public class DianZhongController {
     }
 
     @GetMapping("/getUrlData")
-    public void urlData(String statDate) {
+    public void urlData(@RequestParam(required = false) String statDate) {
         dianZhongService.getUrlData(statDate);
     }
 

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

@@ -23,9 +23,13 @@
     </update>
 
     <select id="getTaskUrl" resultType="com.alibaba.fastjson.JSONObject">
-        select task_id,url,type,client_id
+        select task_id,url,type,client_id,stat_date
         from media_api.dianzhong_data_download_url
-        where stat_date= ${statDate} and downloaded = 0 and url is not null and url != '-1'
+        where downloaded = 0 and url is not null and url != '-1'
+        <if test="statDate != null">
+            and stat_date = #{statDate}
+        </if>
+        order by stat_date, type desc , create_time
     </select>
 
     <select id="getDZRetry" resultType="com.alibaba.fastjson.JSONObject">

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

@@ -26,10 +26,14 @@ import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
 @Service
 @Slf4j
 public class DianZhongServiceImpl implements IDianZhongService {
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(5);
+
     @Autowired
     private ICacheService cacheService;
 
@@ -129,33 +133,44 @@ public class DianZhongServiceImpl implements IDianZhongService {
     下载点众IAP数据
      */
     public void getUrlData(String statDate) {
-        List<JSONObject> tasks = dianZhongMapper.getTaskUrl(statDate.replace("-", ""));
+        List<JSONObject> tasks = dianZhongMapper.getTaskUrl(statDate == null ? null : 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) {
-                                    if ("user".equals(type)) {
-                                        getDZUser(line);
-                                        getDZLogin(line);
-                                        getDZActivation(line);
-                                        getDZRegister(line, statDate);
-                                    } else if ("order".equals(type)) {
-                                        getDZOrder(line);
-                                    }
-                                }
-                            }
-                            return null;
-                        });
-                dianZhongMapper.updateTaskDownloaded(taskId);
+                executorService.submit(new Runnable() {
+                    @Override
+                    public void run() {
+                        String taskId = task.getString("task_id");
+                        String url = task.getString("url");
+                        String type = task.getString("type");
+                        String clientId = task.getString("client_id");
+                        String dataDate = task.getString("stat_date");
+                        try {
+                            log.info("开始下载点众数据 taskId:{}, clientId:{}, type:{}", taskId, clientId, type);
+                            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) {
+                                                if ("user".equals(type)) {
+                                                    getDZUser(line);
+                                                    getDZLogin(line);
+                                                    getDZActivation(line);
+                                                    getDZRegister(line, dataDate);
+                                                } else if ("order".equals(type)) {
+                                                    getDZOrder(line);
+                                                }
+                                            }
+                                        }
+                                        return null;
+                                    });
+                            dianZhongMapper.updateTaskDownloaded(taskId);
+                            log.info("下载点众数据成功 taskId:{}, clientId:{}, type:{}", taskId, clientId, type);
+                        } catch (Exception e) {
+                            log.error("下载点众数据失败 taskId:{}, clientId:{}, type:{}", taskId, clientId, type, e);
+                        }
+                    }
+                });
             }
         }
     }
@@ -180,7 +195,7 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 activationTime = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss") : null;
         String fromMedia = jsonObject.getString("from");
         userJson.put("app_id", appIdByChannelId);
         userJson.put("uid", jsonObject.getString("userId"));
@@ -230,8 +245,8 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 activationDate = DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd");
+        String activationTime = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss") : null;
+        String activationDate = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd") : null;
         Long ctime = jsonObject.getLong("ctime");
         String registerTime = DateUtils.formatDate(ctime * 1000, "yyyy-MM-dd HH:mm:ss");
         String registerDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
@@ -287,6 +302,9 @@ public class DianZhongServiceImpl implements IDianZhongService {
         String promotionId = jsonObject.getString("promotionId");
         adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
         Long dyeTime = jsonObject.getLong("dyeTime");
+        if (dyeTime == null) {
+            return;
+        }
         String statDate = DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd");
         String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
         Long ctime = jsonObject.getLong("ctime");
@@ -346,8 +364,8 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 activationDate = DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd");
+        String activationTime = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss") : null;
+        String activationDate = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd") : null;
         String registerTime = DateUtils.formatDate(ctime * 1000, "yyyy-MM-dd HH:mm:ss");
         String registerDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
         String fromMedia = jsonObject.getString("from");
@@ -407,8 +425,8 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 activationDate = DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd");
+        String activationTime = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss") : null;
+        String activationDate = dyeTime != null ? DateUtils.formatDate(dyeTime * 1000, "yyyyMMdd") : null;
         Long registerT = jsonObject.getLong("registerDate");
         String registerTime = DateUtils.formatDate(registerT * 1000, "yyyy-MM-dd HH:mm:ss");
         String registerDate = DateUtils.formatDate(registerT * 1000, "yyyyMMdd");

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

@@ -68,6 +68,7 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
             put.setHeader("Authorization", "Basic " + auth);
         }
 
+        put.setHeader("max_filter_ratio", "1");
         put.setHeader("format", format); // json
         put.setHeader("strip_outer_array", "true"); // json array
         put.setHeader("fuzzy_parse", "true"); // json array