Merge remote-tracking branch 'origin/master'

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

@@ -153,14 +153,14 @@ public class BytedanceCdpProject {
     /**
      * 创建时间
      */
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    private Date createTime;
+    //@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    //private Date createTime;
 
     /**
      * 更新时间
      */
-    @JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
-    private Date updateTime;
+    //@JsonFormat(pattern = "yyyy-MM-dd HH:mm:ss")
+    //private Date updateTime;
 
     public BytedanceCdpProject() {
     }

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

@@ -13,6 +13,6 @@ import java.util.List;
  */
 public interface UserLoginRecordMapper {
 
-    List<UserLoginRecord> queryUserLoginRecord();
+    List<UserLoginRecord> queryUserLoginRecord(String date);
 
 }

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

@@ -12,7 +12,7 @@
                 LEFT JOIN thirtpart.ly_user lu ON lulr.uid = lu.id
                 LEFT JOIN thirtpart.ly_user_black_list b ON lulr.uid = b.uid
         WHERE
-            lulr.created_at >= UNIX_TIMESTAMP(CURDATE())
+            lulr.date = #{date}
           AND b.uid IS NULL
         GROUP BY lulr.uid
         HAVING login_times > 500

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

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.service.IByteDanceFlowDetailsReportService
 import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -32,6 +33,8 @@ import java.util.Map;
 @Service
 @Slf4j
 public class ByteDanceFlowDetailsReportServiceImpl implements IByteDanceFlowDetailsReportService {
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
     @Resource
     private ByteDanceFlowDetailsReportMapper byteDanceFlowDetailsReportMapper;
@@ -114,16 +117,13 @@ public class ByteDanceFlowDetailsReportServiceImpl implements IByteDanceFlowDeta
                                 .transferOut(transfer_out)
                                 .build();
                         dailies.add(daily);
+                        dorisLoadManager.add("media_api","bytedance_flow_details_report_daily", daily);
                     }
 
                     if (!Check.isNull(dailies)) {
                         log.info("账户id:{}，流水日报请求结果返回成功", advertiserId);
                         byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailies(dailies);
-                        try {
-                            byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailiesSD(dailies);
-                        }catch (Exception e) {
-                            log.error("",e);
-                        }
+                        //byteDanceFlowDetailsReportMapper.replaceDanceFlowDetailsReportDailiesSD(dailies);
                     }
                 }
 

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

@@ -7,18 +7,14 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
-import cn.com.ctop.job.bytedance.utils.doris_sink.DorisMultiTableStreamLoadManager;
-import cn.com.ctop.job.bytedance.utils.doris_sink.RecordWithMeta;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.scheduling.annotation.EnableScheduling;
-import org.springframework.scheduling.annotation.Scheduled;
 import org.springframework.stereotype.Service;
 
-import javax.annotation.PreDestroy;
-import java.text.DecimalFormat;
 import java.util.*;
 
 @Service
@@ -74,6 +70,9 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setStatTimeDay(statTimeDay);
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_hour", advertiserReportHour);
+
                         }
                     }
                 }
@@ -84,8 +83,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
-                accountReportMapper.replaceAccountReportHourSD(hours);
-                //advertiserReportHourBuffer.addAll(hours);
+                //accountReportMapper.replaceAccountReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -146,6 +144,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportDaily.setAdvertiserId(advertiserId);
                             advertiserReportDaily.setStatTimeDay(statTimeDay);
                             dailyList.add(advertiserReportDaily);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_daily", advertiserReportDaily);
                         }
                     }
                 }
@@ -156,8 +156,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                accountReportMapper.replaceAccountReportDailySD(dailyList);
-                //advertiserReportDailyBuffer.addAll(dailyList);
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -229,7 +228,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
 
-                            dorisLoadManager.add(new RecordWithMeta("media_api","bytedance_advertiser_report_hour",advertiserReportHour));
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_hour", advertiserReportHour);
                         }
                     } else {
                         log.info("账户id:{}，日期:{}广告时报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -317,6 +316,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportDaily.setAdvertiserId(advertiserId);
                             advertiserReportDaily.setStatTimeDay(statTimeDay);
                             dailyList.add(advertiserReportDaily);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_daily", advertiserReportDaily);
                         }
                     } else {
                         log.info("账户id:{}，日期:{}，广告日报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -339,8 +340,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                accountReportMapper.replaceAccountReportDailySD(dailyList);
-                //advertiserReportDailyBuffer.addAll(dailyList);
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -357,83 +357,4 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
         }
     }
 
-    /*@Scheduled(fixedRate = 12000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行账户报表定时批处理任务....");
-        synchronized (advertiserReportHourBuffer) {
-            long now = System.currentTimeMillis();
-            // advertiserReportHourBuffer.size() >= MAX_BATCH_SIZE ||                    (now - advertiserReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if (!advertiserReportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed advertiserReportHour batch size: {}，执行时间：{}", advertiserReportHourBuffer.size(), seconds);
-                    advertiserReportHourBuffer.clear();
-                    advertiserReportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (advertiserReportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            // advertiserReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - advertiserReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if (!advertiserReportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed advertiserReportDaily batch size: {}，执行时间：{}", advertiserReportDailyBuffer.size(), seconds);
-                    advertiserReportDailyBuffer.clear();
-                    advertiserReportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启，开始执行广告报表批处理任务....");
-        synchronized (advertiserReportHourBuffer) {
-            if (!advertiserReportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportHour batch size: {}", advertiserReportHourBuffer.size());
-                    accountReportMapper.replaceAccountReportHourSD(new ArrayList<>(advertiserReportHourBuffer));
-                    advertiserReportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (advertiserReportDailyBuffer) {
-            if (!advertiserReportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed advertiserReportDaily batch size: {}", advertiserReportDailyBuffer.size());
-                    accountReportMapper.replaceAccountReportDailySD(new ArrayList<>(advertiserReportDailyBuffer));
-                    advertiserReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing advertiserReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
 }

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

@@ -5,6 +5,7 @@ import cn.com.ctop.job.bytedance.data.mapper.BytedanceCdpProjectMapper;
 import cn.com.ctop.job.bytedance.data.mapper.BytedanceRefreshTokenMapper;
 import cn.com.ctop.job.bytedance.data.service.IBytedanceCdpProjectService;
 import cn.com.ctop.job.bytedance.data.utils.*;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -29,6 +30,9 @@ import java.util.*;
 @Service
 public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceCdpProjectMapper bytedanceCdpProjectMapper;
 
     @Autowired
@@ -109,15 +113,13 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray array = data.getJSONArray("list");
                     for (int i = 0; i < array.size(); i++) {
-                        checkPromotionInfo(array.getJSONObject(i), advertiserId);
+                        JSONObject jsonObject = array.getJSONObject(i);
+                        checkPromotionInfo(jsonObject, advertiserId);
+                        dorisLoadManager.add("media_api", "bytedance_promotion_list", jsonObject);
                     }
 
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedancePromotionListSD(array);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedancePromotionListSD(array);
                         if (bytedanceCdpProjectMapper.replaceBytedancePromotionList(array) > 0)
                             log.info("---账户（{}）广告更新成功", advertiserId);
                     }
@@ -201,13 +203,11 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         BytedanceCdpProject advertiserToolsSite = new BytedanceCdpProject(array.getJSONObject(i));
                         projectList.add(advertiserToolsSite);
+
+                        dorisLoadManager.add("media_api", "bytedance_cdp_project_list", advertiserToolsSite);
                     }
                     if (!Check.isNull(projectList)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceCdpProjectListSD(projectList);
-                        } catch (Exception e) {
-                            log.error("replaceBytedanceCdpProjectListSD", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedanceCdpProjectListSD(projectList);
                         if (bytedanceCdpProjectMapper.replaceBytedanceCdpProjectList(projectList) > 0)
                             log.info("---账户（{}）项目信息更新成功", advertiserId);
                     }
@@ -326,11 +326,8 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray list = data.getJSONArray("list");
                     if (!Check.isNull(list)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfoSD(list);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        dorisLoadManager.add("media_api", "bytedance_tools_asset_link_list", list);
+                        //bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfoSD(list);
                         if (bytedanceCdpProjectMapper.replaceBytedanceByteAppLinkInfo(list) > 0)
                             log.info("---账户（{}）抖小详情信息更新成功", advertiserId);
                     }
@@ -394,13 +391,15 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     }
                     JSONArray list = data.getJSONArray("list");
                     if (!Check.isNull(list)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceBytedanceWxAppInfoSD(list, advertiserId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceBytedanceWxAppInfoSD(list, advertiserId);
                         if (bytedanceCdpProjectMapper.replaceBytedanceWxAppInfo(list, advertiserId) > 0)
                             log.info("---账户（{}）微小信息更新成功", advertiserId);
+
+                        for (Object o : list) {
+                            JSONObject json = (JSONObject) o;
+                            json.put("advertiser_id", advertiserId);
+                            dorisLoadManager.add("media_api", "bytedance_wx_app_info", json);
+                        }
                     }
 
                 } else {
@@ -435,7 +434,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
     @Override
-    public void getAdqCreatives(Long advertiserId, String startDate,String endDate) {
+    public void getAdqCreatives(Long advertiserId, String startDate, String endDate) {
         try {
             String accessToken = tokenMapper.onlyGetADQToken(advertiserId);
             if (Check.isNull(accessToken)) {
@@ -507,13 +506,13 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         JSONObject object = array.getJSONObject(i);
                         getAppId(object);
+
+                        object.put("advertiser_id", accountId);
+                        object.put("creative_create_time", object.getString("created_time"));
+                        dorisLoadManager.add("media_api", "adq_dynamic_createive_list", object);
                     }
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceAdqDynamicCreateivesSD(array, accountId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceAdqDynamicCreateivesSD(array, accountId);
                         if (bytedanceCdpProjectMapper.replaceAdqDynamicCreateives(array, accountId) > 0)
                             log.info("---账户（{}）广点通创意更新成功", accountId);
                     }
@@ -529,19 +528,19 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             }
         }
         if (nextPage) {
-            rebackadqCreativesVo(accountId, accessToken, startDate ,endDate, page + 1, isDeleted, 1);
+            rebackadqCreativesVo(accountId, accessToken, startDate, endDate, page + 1, isDeleted, 1);
         }
     }
 
     @Override
-    public void getAdqWechatPages(Long advertiserId,Long pageId) {
+    public void getAdqWechatPages(Long advertiserId, Long pageId) {
         try {
             String accessToken = tokenMapper.onlyGetADQToken(advertiserId);
             if (Check.isNull(accessToken)) {
                 log.error("账户（{}）accessToken is null", advertiserId);
                 return;
             }
-            this.getAdqWechatPagesVo(advertiserId,pageId, accessToken, 1, true, 1);
+            this.getAdqWechatPagesVo(advertiserId, pageId, accessToken, 1, true, 1);
         } catch (Exception e) {
             e.printStackTrace();
             log.error("账户（{}）获取原生落地页异常", e.getMessage());
@@ -549,7 +548,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
 
-    private void getAdqWechatPagesVo(Long accountId,Long pageId, String accessToken, Integer page, Boolean isDeleted, int count) throws Exception {
+    private void getAdqWechatPagesVo(Long accountId, Long pageId, String accessToken, Integer page, Boolean isDeleted, int count) throws Exception {
         boolean nextPage = false;
         Map<String, Object> param = new HashMap<>();
         long timestamp = Instant.now().getEpochSecond();
@@ -588,13 +587,12 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     for (int i = 0; i < array.size(); i++) {
                         JSONObject object = array.getJSONObject(i);
                         getPageAppId(object);
+
+                        object.put("advertiser_id", accountId);
+                        dorisLoadManager.add("media_api", "adq_dynamic_page_list", object);
                     }
                     if (!Check.isNull(array)) {
-                        try {
-                            bytedanceCdpProjectMapper.replaceAdqDynamicPagesSD(array, accountId);
-                        } catch (Exception e) {
-                            log.error("", e);
-                        }
+                        //bytedanceCdpProjectMapper.replaceAdqDynamicPagesSD(array, accountId);
                         if (bytedanceCdpProjectMapper.replaceAdqDynamicPages(array, accountId) > 0)
                             log.info("---账户（{}）广点通落地页更新成功", accountId);
                     }
@@ -603,14 +601,14 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             } else {
                 if (count <= 4) {
                     Thread.sleep(500);
-                    getAdqWechatPagesVo(accountId,pageId, accessToken, page, isDeleted, count + 1);
+                    getAdqWechatPagesVo(accountId, pageId, accessToken, page, isDeleted, count + 1);
                 } else {
                     log.error("账户（{}）获取动态创意失败，入参：{}\n，结果：{}", accountId, JSONObject.toJSONString(param), obj);
                 }
             }
         }
         if (nextPage) {
-            getAdqWechatPagesVo(accountId,pageId, accessToken, page + 1, isDeleted, 1);
+            getAdqWechatPagesVo(accountId, pageId, accessToken, page + 1, isDeleted, 1);
         }
     }
 

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

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -24,6 +25,9 @@ import java.util.Map;
 @Slf4j
 public class BytedanceProductReportServiceImpl implements IBytedanceProductReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
 
     @Autowired
@@ -83,6 +87,7 @@ public class BytedanceProductReportServiceImpl implements IBytedanceProductRepor
                             BytedanceProductReportDaily reportDaily = JSONObject.parseObject(metricsJson.toJSONString(), BytedanceProductReportDaily.class);
                             reportDaily.setAdvertiserId(advertiserId);
                             reportDailies.add(reportDaily);
+                            dorisLoadManager.add("media_api","bytedance_product_report_daily", reportDaily);
                         }
                     }
                 }
@@ -93,11 +98,7 @@ public class BytedanceProductReportServiceImpl implements IBytedanceProductRepor
             }
             if (!Check.isNull(reportDailies)) {
                 productReportMapper.replaceProductReportDailies(reportDailies);
-                try {
-                    productReportMapper.replaceProductReportDailiesSD(reportDailies);
-                }catch (Exception e){
-                    log.error("",e);
-                }
+                //productReportMapper.replaceProductReportDailiesSD(reportDailies);
                 log.info("账户ID：{}，产品数据日报请求成功,开始日期：{}，结束日期：{}，page:{}", advertiserId, startDate, endDate, page);
             }
             if (page < totalPage) {

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

@@ -9,6 +9,7 @@ import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.FieldsConfig;
 import cn.com.ctop.job.bytedance.data.utils.HttpUtils;
 import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
 import com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -26,14 +27,10 @@ import java.util.*;
 @EnableScheduling
 public class BytedancePromoterReportServiceImpl implements IBytedancePromoterReportService {
     @Autowired
-    private BytedancePromoterReportMapper promoterReportMapper;
-    /*private static final int MAX_BATCH_SIZE = 10000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
-    private final List<BytedancePromoterReportHour> promoterReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<BytedancePromoterReportDaily> promoterReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long promoterReportHourLastFlushTime = System.currentTimeMillis();
-    private long promoterReportDailyLastFlushTime = System.currentTimeMillis();*/
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
+    @Autowired
+    private BytedancePromoterReportMapper promoterReportMapper;
 
     @Override
     public List<JSONObject> getAccountList() {
@@ -90,6 +87,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             bytedancePromoterReportHour.setStatTimeDay(statTimeDay);
                             bytedancePromoterReportHour.setHour(hour);
                             hours.add(bytedancePromoterReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_promoter_report_hour", bytedancePromoterReportHour);
                         }
                     }
                 }
@@ -100,8 +99,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                promoterReportMapper.replacePromoterReportHourSD(hours);
-                //promoterReportHourBuffer.addAll(hours);
+                //promoterReportMapper.replacePromoterReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -171,6 +169,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             promoterReportDaily.setCdpPromotionName(cdpPromotionName);
                             promoterReportDaily.setStatTimeDay(statTimeDay);
                             promoterReportDailies.add(promoterReportDaily);
+                            dorisLoadManager.add("media_api", "bytedance_promoter_report_daily", promoterReportDaily);
                         }
                     }
                 }
@@ -181,8 +180,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                //promoterReportDailyBuffer.addAll(promoterReportDailies);
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
@@ -265,6 +263,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             bytedancePromoterReportHour.setStatTimeDay(statTimeDay);
                             bytedancePromoterReportHour.setHour(hour);
                             hours.add(bytedancePromoterReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_promoter_report_hour", bytedancePromoterReportHour);
                         }
                     }
                 }
@@ -284,8 +284,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                promoterReportMapper.replacePromoterReportHourSD(hours);
-                //promoterReportHourBuffer.addAll(hours);
+                //promoterReportMapper.replacePromoterReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -359,6 +358,8 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                             promoterReportDaily.setCdpPromotionName(cdpPromotionName);
                             promoterReportDaily.setStatTimeDay(statTimeDay);
                             promoterReportDailies.add(promoterReportDaily);
+
+                            dorisLoadManager.add("media_api", "bytedance_promoter_report_daily", promoterReportDaily);
                         }
                     }
                 }
@@ -378,8 +379,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                //promoterReportDailyBuffer.addAll(promoterReportDailies);
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
@@ -406,86 +406,4 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
         return promoterReportMapper.getAllList();
     }
 
-    /*@Scheduled(fixedRate = 11000) // 每30秒检查一次
-    public void flushIfNeeded() {
-        log.info("开始执行广告报表定时批处理任务....");
-        synchronized (promoterReportHourBuffer) {
-            long now = System.currentTimeMillis();
-            // promoterReportHourBuffer.size() >= MAX_BATCH_SIZE || (now - promoterReportHourLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !promoterReportHourBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed PromoterReportHour batch size: {}，执行时间：{}", promoterReportHourBuffer.size(),seconds);
-
-                    promoterReportHourBuffer.clear();
-                    promoterReportHourLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (promoterReportDailyBuffer) {
-            long now = System.currentTimeMillis();
-            // promoterReportDailyBuffer.size() >= MAX_BATCH_SIZE ||  (now - promoterReportDailyLastFlushTime) >= MAX_WAIT_TIME_MS &&
-            if ( !promoterReportDailyBuffer.isEmpty()) {
-                try {
-                    long startTime = System.currentTimeMillis();
-                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
-                    long endTime = System.currentTimeMillis();
-                    long diffInMilliseconds = endTime - startTime;
-                    double diffInSeconds = diffInMilliseconds / 1000.0;
-                    DecimalFormat df = new DecimalFormat("#.00");
-                    String seconds = df.format(diffInSeconds);
-                    log.info("Flushed PromoterReportDaily batch size: {}，执行时间：{}", promoterReportDailyBuffer.size(),seconds);
-
-                    promoterReportDailyBuffer.clear();
-                    promoterReportDailyLastFlushTime = now;
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportDaily batch", e);
-                }
-            }
-        }
-
-
-    }
-
-
-    @PreDestroy
-    public void flushPreDestroy() {
-        log.info("项目重启，开始执行广告报表批处理任务....");
-        synchronized (promoterReportHourBuffer) {
-            if (!promoterReportHourBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed PromoterReportHour batch size: {}", promoterReportHourBuffer.size());
-                    promoterReportMapper.replacePromoterReportHourSD(new ArrayList<>(promoterReportHourBuffer));
-                    promoterReportHourBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing PromoterReportHour batch", e);
-                }
-            }
-        }
-
-
-        synchronized (promoterReportDailyBuffer) {
-            if (!promoterReportDailyBuffer.isEmpty()) {
-                try {
-                    log.info("Flushed promoterReportDaily batch size: {}", promoterReportDailyBuffer.size());
-                    promoterReportMapper.replacePromoterReportDailiesSD(new ArrayList<>(promoterReportDailyBuffer));
-                    promoterReportDailyBuffer.clear();
-                } catch (Exception e) {
-                    log.error("Error flushing promoterReportDaily batch", e);
-                }
-            }
-        }
-    }*/
-
-
 }

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

@@ -11,6 +11,7 @@ import cn.com.ctop.job.bytedance.data.utils.AlertUtils;
 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 com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -29,6 +30,9 @@ import java.util.Map;
 @Slf4j
 public class BytedanceXingTuServiceImpl implements IBytedanceXingTuService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceXingTuMapper bytedanceXingTuMapper;
 
     @Autowired
@@ -387,19 +391,13 @@ public class BytedanceXingTuServiceImpl implements IBytedanceXingTuService {
 
             if (!Check.isNull(reports)) {
                 bytedanceXingTuMapper.insertReport(reports);
-                try {
-                    bytedanceXingTuMapper.insertReportSD(reports);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //bytedanceXingTuMapper.insertReportSD(reports);
+                dorisLoadManager.add("media_api", "bytedance_xingtu_task_report" ,reports);
             }
             if (!Check.isNull(dailyReports)) {
                 bytedanceXingTuMapper.insertReportDaily(dailyReports);
-                try {
-                    bytedanceXingTuMapper.insertReportDailySD(dailyReports);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //bytedanceXingTuMapper.insertReportDailySD(dailyReports);
+                dorisLoadManager.add("media_api", "bytedance_xingtu_task_report_daily" ,dailyReports);
             }
 
             if (hasMore) {

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

@@ -26,9 +26,10 @@ public class UserLoginRecordServiceImpl implements IUserLoginRecordService {
     private UserLoginRecordMapper userLoginRecordMapper;
     @Override
     public String queryUserLoginRecord() {
-        List<UserLoginRecord> userLoginRecords = userLoginRecordMapper.queryUserLoginRecord();
+        String date = DateUtils.getNowDate("yyyy-MM-dd");
+        List<UserLoginRecord> userLoginRecords = userLoginRecordMapper.queryUserLoginRecord(date);
 
-        String date = DateUtils.getNowDate("YYYY-MM-dd HH:mm:ss");
+        date = DateUtils.getNowDate("YYYY-MM-dd HH:mm:ss");
         StringBuilder text = new StringBuilder(date);
         text.append("\n");
         if (userLoginRecords != null && userLoginRecords.size() > 0){

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

@@ -13,7 +13,7 @@ public class JsonUtils {
     // 正则预编译
     private static final Pattern INVALID_PREFIX = Pattern.compile("^[^a-zA-Z0-9]+");
     private static final Pattern INVALID_SUFFIX = Pattern.compile("[^a-zA-Z0-9]+$");
-    private static final Pattern LETTER_NUMBER = Pattern.compile("([a-zA-Z])([0-9])");
+    private static final Pattern LETTER_NUMBER = Pattern.compile("([a-z])([0-9])");
     //private static final Pattern NUMBER_LETTER = Pattern.compile("([0-9])([a-zA-Z])");
     private static final Pattern CAMEL_CASE = Pattern.compile("([a-z])([A-Z])");
 

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisConfiguration.java

@@ -1,11 +1,15 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils;
 
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisBufferConfig;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisMultiTableStreamLoadManager;
+import cn.com.ctop.job.bytedance.utils.doris.sink.DorisSinkClient;
+import cn.com.ctop.job.bytedance.utils.doris.sink.SimpleHttpDorisSinkClient;
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
 @Configuration
-public class DorisConfiguration {
+public class DorisConfig {
     @Value("${spring.doris.fe-host}")
     private String feHost;
 

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisBufferConfig.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

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

@@ -1,9 +1,11 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.extern.slf4j.Slf4j;
 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;
@@ -17,8 +19,8 @@ public class DorisMultiTableStreamLoadManager {
     private final DorisBufferConfig defaultBufferConfig;
 
     public DorisMultiTableStreamLoadManager(int uploaderThreads,
-                                             DorisBufferConfig defaultBufferConfig,
-                                             DorisSinkClient sinkClient) {
+                                            DorisBufferConfig defaultBufferConfig,
+                                            DorisSinkClient sinkClient) {
         log.info("init DorisMultiTableStreamLoadManager");
         this.defaultBufferConfig = defaultBufferConfig;
         this.sinkClient = sinkClient;
@@ -26,12 +28,21 @@ public class DorisMultiTableStreamLoadManager {
                 new CustomizableThreadFactory("doris-stream-uploader-"));
     }
 
-    public void add(RecordWithMeta record) {
-        TableKey key = new TableKey(record.getDatabase(), record.getTable());
-        bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
-                .add(record);
+    public void add(String database, String table, Object obj) {
+        TableKey key = new TableKey(database, table);
+        if (obj instanceof Collection) {
+            Collection<?> collection = (Collection<?>) obj;
+            for (Object o : collection) {
+                bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
+                        .add(new RecordWithMeta(database, table, o));
+            }
+        } else {
+            bufferManagers.computeIfAbsent(key, k -> new TableBufferManager(k, defaultBufferConfig, sinkClient, uploaderExecutor))
+                    .add(new RecordWithMeta(database, table, obj));
+        }
     }
 
+
     @PreDestroy
     public void shutdown() {
         log.info("Shutting down DorisMultiTableStreamLoadManager");

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/DorisSinkClient.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import java.util.List;
 @FunctionalInterface

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/RecordWithMeta.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

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

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
 import lombok.extern.slf4j.Slf4j;
@@ -40,7 +40,7 @@ public class SimpleHttpDorisSinkClient implements DorisSinkClient {
         String db = sample.getDatabase();
         String table = sample.getTable();
 
-        String label = labelPrefix + db + "_" + table + "_" + UUID.randomUUID();
+        String label = labelPrefix + "_" + db + "_" + table + "_" + UUID.randomUUID();
         String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
 
         // 提取 data

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/StrictRedirectStrategy.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import org.apache.http.HttpEntityEnclosingRequest;
 import org.apache.http.HttpRequest;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/TableBufferManager.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang3.StringUtils;

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/utils/doris_sink/TableKey.java

@@ -1,4 +1,4 @@
-package cn.com.ctop.job.bytedance.utils.doris_sink;
+package cn.com.ctop.job.bytedance.utils.doris.sink;
 
 import lombok.AllArgsConstructor;
 import lombok.Data;