数据回溯

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/entity/BytedanceErrorLog.java

@@ -39,58 +39,66 @@ public class BytedanceErrorLog {
      * hour:时报 daily：日报
      */
     private String dataType;
+    private Integer isSend;
+
+    public Long getId() {
+        return id;
+    }
 
     public void setId(Long id) {
         this.id = id;
     }
 
-    public Long getId() {
-        return id;
+    public Long getAdvertiserId() {
+        return advertiserId;
     }
 
     public void setAdvertiserId(Long advertiserId) {
         this.advertiserId = advertiserId;
     }
 
-    public Long getAdvertiserId() {
-        return advertiserId;
+    public String getStartDate() {
+        return startDate;
     }
 
     public void setStartDate(String startDate) {
         this.startDate = startDate;
     }
 
-    public String getStartDate() {
-        return startDate;
+    public String getEndDate() {
+        return endDate;
     }
 
     public void setEndDate(String endDate) {
         this.endDate = endDate;
     }
 
-    public String getEndDate() {
-        return endDate;
+    public String getDataDimension() {
+        return dataDimension;
     }
 
     public void setDataDimension(String dataDimension) {
         this.dataDimension = dataDimension;
     }
 
-    public String getDataDimension() {
-        return dataDimension;
+    public String getDataType() {
+        return dataType;
     }
 
     public void setDataType(String dataType) {
         this.dataType = dataType;
     }
 
-    public String getDataType() {
-        return dataType;
+    public Integer getIsSend() {
+        return isSend;
+    }
+
+    public void setIsSend(Integer isSend) {
+        this.isSend = isSend;
     }
 
     @Override
-    public String
-    toString() {
+    public String toString() {
         return "BytedanceErrorLog{" +
                 "id=" + id +
                 ", advertiserId=" + advertiserId +
@@ -98,6 +106,7 @@ public class BytedanceErrorLog {
                 ", endDate='" + endDate + '\'' +
                 ", dataDimension='" + dataDimension + '\'' +
                 ", dataType='" + dataType + '\'' +
+                ", isSend=" + isSend +
                 '}';
     }
 }

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

@@ -8,7 +8,7 @@
     </delete>
 
     <select id="getAccountList" resultType="com.alibaba.fastjson.JSONObject">
-        SELECT 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'
+        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
                  LEFT JOIN deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
                  LEFT JOIN deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id

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

@@ -8,9 +8,9 @@ import java.util.List;
 public interface IBytedancePromoterReportService {
     List<JSONObject> getAccountList();
 
-    void getBytedanceRetryPromoterHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page);
+    void getBytedanceRetryPromoterHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page,Integer isSend);
 
-    void getBytedancePromoterRetryDailyReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page);
+    void getBytedancePromoterRetryDailyReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page,Integer isSend);
 
     List<JSONObject> getChannelInfoList();
 

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

@@ -18,12 +18,16 @@ import java.util.ArrayList;
 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 BytedancePromoterReportServiceImpl implements IBytedancePromoterReportService {
     @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
+    private static String sendUrl = "http://172.16.0.108:8765/webhook/media_report";
+    private static ExecutorService sendExecutorService = Executors.newFixedThreadPool(5);
 
     @Override
     public List<JSONObject> getAccountList() {
@@ -31,7 +35,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
     }
 
     @Override
-    public void getBytedanceRetryPromoterHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page) {
+    public void getBytedanceRetryPromoterHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isSend) {
         try {
             String url = "https://api.oceanengine.com/open_api/v3.0/report/custom/get/";
             Map<String, Object> paramsMap = new HashMap<>();
@@ -97,10 +101,13 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
-                getBytedanceRetryPromoterHourReport(errorId, advertiserId, accessToken, startDate, endDate, page + 1);
+                getBytedanceRetryPromoterHourReport(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
             } else {
                 log.info("账户ID：{}，日期:{},广告时报（最细粒度）数据请求成功", advertiserId, startDate);
                 promoterReportMapper.deleteInfoByErrorId(errorId);
+                if (isSend == 1) {
+                    sendMessage(advertiserId, startDate, endDate, 1, "hour", 1);
+                }
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -109,7 +116,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
     }
 
     @Override
-    public void getBytedancePromoterRetryDailyReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page) {
+    public void getBytedancePromoterRetryDailyReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isSend) {
         try {
             String url = "https://api.oceanengine.com/open_api/v3.0/report/custom/get/";
             Map<String, Object> paramsMap = new HashMap<>();
@@ -174,11 +181,13 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
-                getBytedancePromoterRetryDailyReport(errorId, advertiserId, accessToken, startDate, endDate, page + 1);
+                getBytedancePromoterRetryDailyReport(errorId, advertiserId, accessToken, startDate, endDate, page + 1, isSend);
             } else {
                 log.info("账户ID：{}，日期:{},广告日报（最细粒度）数据请求成功", advertiserId, startDate);
                 promoterReportMapper.deleteInfoByErrorId(errorId);
-
+                if (isSend == 1) {
+                    sendMessage(advertiserId, startDate, endDate, 1, "daily", 1);
+                }
             }
         } catch (Exception e) {
             e.printStackTrace();
@@ -390,4 +399,34 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
     public List<JSONObject> getAllList() {
         return promoterReportMapper.getAllList();
     }
+
+    private static void sendMessage(Long account_id, String start_date, String end_date, Integer media_type, String type, Integer count) {
+        if (count > 5) {
+            log.error("超过最大次数，account_id：{}", count);
+            return;
+        }
+        sendExecutorService.submit(new Runnable() {
+            @Override
+            public void run() {
+                Map<String, Object> param = new HashMap<>();
+                param.put("account_id", account_id);
+                param.put("start_date", start_date);
+                param.put("end_date", end_date);
+                param.put("media_type", media_type);
+                param.put("type", type);
+                String s = HttpUtils.httpPostRequest(sendUrl, param, new HashMap<>());
+                JSONObject jsonObject = JSONObject.parseObject(s);
+                if (!Check.isNull(jsonObject)) {
+                    Integer code = jsonObject.getInteger("code");
+                    if (code == 200) {
+                        log.info("请求数据清洗服务成功,account_id:{}", account_id);
+
+                    } else {
+                        sendMessage(account_id, start_date, end_date, media_type, type, count + 1);
+                    }
+                }
+            }
+        });
+    }
+
 }

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

@@ -49,6 +49,7 @@ public class BytedancePromoterReportJob {
                 String dataDimension = jsonObject.getString("dataDimension");
                 String dataType = jsonObject.getString("dataType");
                 String accessToken = jsonObject.getString("accessToken");
+                Integer isSend = jsonObject.getInteger("isSend");
 
                 if ("promoter".equals(dataDimension)) {
                     if (dataType.equals("hour")) { // 广告时报
@@ -56,7 +57,7 @@ public class BytedancePromoterReportJob {
                             @Override
                             public void run() {
                                 if (!Check.isNull(accessToken)) {
-                                    promoterReportService.getBytedanceRetryPromoterHourReport(id, advertiserId, accessToken, startDate, endDate, 1);
+                                    promoterReportService.getBytedanceRetryPromoterHourReport(id, advertiserId, accessToken, startDate, endDate, 1,isSend);
                                 }
                             }
                         });
@@ -66,7 +67,7 @@ public class BytedancePromoterReportJob {
                             @Override
                             public void run() {
                                 if (!Check.isNull(accessToken)) {
-                                    promoterReportService.getBytedancePromoterRetryDailyReport(id, advertiserId, accessToken, startDate, endDate, 1);
+                                    promoterReportService.getBytedancePromoterRetryDailyReport(id, advertiserId, accessToken, startDate, endDate, 1,isSend);
 
                                 }
                             }