Merge remote-tracking branch 'origin/master'

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

@@ -191,10 +191,32 @@ public class BytedanceController {
         }
     }
 
+    @RequestMapping("/adq/creatives")
+    public void adqCreatives(Long accountId, String startDate, String endDate) {
+        cdpProjectService.getAdqCreatives(accountId, startDate, endDate);
+    }
+
 
     @GetMapping("/testB")
     public void testB(Long id) {
-        cdpProjectService.getAdqCreatives(id, "2025-02-15");
+        cdpProjectService.getAdqCreatives(id, "2025-02-15", "2025-02-15");
+    }
+
+    @GetMapping("/adqWechatPages")
+    public void adqWechatPages() {
+        List<JSONObject> iids = cdpProjectService.queryPageAdvertiserIds();
+        if (!Check.isNull(iids)) {
+            for (JSONObject iid : iids) {
+                Long pageId = iid.getLong("page_id");
+                Long advertiserId = iid.getLong("advertiser_id");
+                dailyRetrExecutorService.submit(() -> cdpProjectService.getAdqWechatPages(advertiserId, pageId));
+            }
+        }
     }
 
-}
+
+    @GetMapping("/accountHour")
+    public void bytedanceAccountHour(Long advertiserId, String accessToken, String date) {
+        accountReportService.getBytedanceAccountHour(advertiserId, accessToken, date, date, 1, 0);
+    }
+}

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

@@ -19,6 +19,6 @@ public interface ByteDanceFlowDetailsReportMapper {
     List<AdvertiserAccount> getAdAccount();
 
     void replaceDanceFlowDetailsReportDailies(@Param("list") List<ByteDanceFlowDetailsReportDaily> dailyList);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replaceDanceFlowDetailsReportDailiesSD(@Param("list") List<ByteDanceFlowDetailsReportDaily> dailyList);
 }

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

@@ -10,11 +10,11 @@ import java.util.List;
 
 public interface BytedanceAccountReportMapper {
     void replaceAccountReportHour(@Param("list") List<BytedanceAdvertiserReportHour> hours);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replaceAccountReportHourSD(@Param("list") List<BytedanceAdvertiserReportHour> hours);
 
     void replaceAccountReportDaily(@Param("list") List<BytedanceAdvertiserReportDaily> dailyList);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replaceAccountReportDailySD(@Param("list") List<BytedanceAdvertiserReportDaily> dailyList);
 
     void replaceErrorInfo(BytedanceErrorLog errorInfo);

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

@@ -2,6 +2,7 @@ package cn.com.ctop.job.bytedance.data.mapper;
 
 import cn.com.ctop.job.bytedance.data.entity.BytedanceCdpProject;
 import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
@@ -16,11 +17,11 @@ import java.util.List;
 public interface BytedanceCdpProjectMapper {
 
     int replaceBytedanceCdpProjectList(@Param("projectList") List<BytedanceCdpProject> projectList);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceBytedanceCdpProjectListSD(@Param("projectList") List<BytedanceCdpProject> projectList);
 
     int replaceBytedancePromotionList(@Param("list") JSONArray array);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceBytedancePromotionListSD(@Param("list") JSONArray array);
 
     void insertErrorAdvertiserId(@Param("advertiserId") Long advertiserId, @Param("type") String type);
@@ -28,7 +29,7 @@ public interface BytedanceCdpProjectMapper {
     int replaceBytedanceByteAppInfo(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
 
     int replaceBytedanceByteAppLinkInfo(@Param("list") JSONArray list);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceBytedanceByteAppLinkInfoSD(@Param("list") JSONArray list);
 
     List<Long> queryAllAdverIdList(@Param("mediaType") String mediaType);
@@ -38,7 +39,7 @@ public interface BytedanceCdpProjectMapper {
     List<Long> queryTTErrApps();
 
     @DS("selectdb")
-    List<Long> queryPageAdvertiserIds();
+    List<JSONObject> queryPageAdvertiserIds();
 
     boolean delErrApps(@Param("type") String type);
 
@@ -48,18 +49,18 @@ public interface BytedanceCdpProjectMapper {
     List<Long> queryAdvertiserIds(@Param("db") String db, @Param("date") String date);
 
     int replaceBytedanceWxAppInfo(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceBytedanceWxAppInfoSD(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
 
     @DS("selectdb")
     List<Long> queryMediaAdverIds(@Param("date") String date, @Param("mediaType") String mediaType);
 
     int replaceAdqDynamicCreateives(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceAdqDynamicCreateivesSD(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
 
     int replaceAdqDynamicPages(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     int replaceAdqDynamicPagesSD(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
 
     List<Long> queryADQAdverIds(String date);

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

@@ -13,7 +13,7 @@ public interface BytedanceProductReportMapper {
     List<JSONObject> queryCostAccount(@Param("advertiserId") Long advertiserId, @Param("startDate") String startDate, @Param("endDate") String endDate);
 
     void replaceProductReportDailies(List<BytedanceProductReportDaily> reports);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replaceProductReportDailiesSD(List<BytedanceProductReportDaily> reports);
 
     void deleteProductReportDailies(@Param("advertiserId") Long advertiserId,@Param("startDate") Integer startDate,@Param("endDate")  Integer endDate);

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

@@ -14,13 +14,13 @@ public interface BytedancePromoterReportMapper {
     List<JSONObject> getAccountList();
 
     void replacePromoterReportHour(@Param("list") List<BytedancePromoterReportHour> hours);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replacePromoterReportHourSD(@Param("list") List<BytedancePromoterReportHour> hours);
 
     void deleteInfoByErrorId(@Param("errorId") Long errorId);
 
     void replacePromoterReportDailies(@Param("list") List<BytedancePromoterReportDaily> promoterReportDailies);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void replacePromoterReportDailiesSD(@Param("list") List<BytedancePromoterReportDaily> promoterReportDailies);
 
     void replaceErrorInfo(BytedanceErrorLog errorInfo);

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

@@ -30,11 +30,11 @@ public interface BytedanceXingTuMapper {
     void updateTaskInfoSD(BytedanceXingTuTask task);
 
     void insertReport(List<BytedanceXingTuReport> reports);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void insertReportSD(List<BytedanceXingTuReport> reports);
 
     void insertReportDaily(List<BytedanceXingTuReportDaily> dailyReports);
-    @DS("selectdb_group_commit")
+    @DS("sd_group_commit")
     void insertReportDailySD(List<BytedanceXingTuReportDaily> dailyReports);
 
     @DS("selectdb")

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

@@ -8,12 +8,10 @@ import java.util.List;
 
 public interface CPSReportMapper {
 
-    boolean cleanCPSPayDailyReports(@Param("nowDate") Integer nowDate);
-
-    boolean cleanCPSIaaDailyReports(@Param("nowDate") Integer nowDate);
+    boolean cleanCPSPayDailyReports(@Param("nowDate") String nowDate);
 
     @DS("selectdb")
-    List<JSONObject> queryCPSIaaDaily(@Param("nowDate") Integer nowDate);
+    List<JSONObject> queryCPSIaaDaily(@Param("nowDate") Integer nowDate, @Param("dateStr") String dateStr);
 
     boolean cleanCPSIaaDailyReportsDS(@Param("list") List<JSONObject> list);
 

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

@@ -346,8 +346,8 @@
         order by create_time
     </select>
 
-    <select id="queryPageAdvertiserIds" resultType="Long">
-        SELECT DISTINCT advertiser_id
+    <select id="queryPageAdvertiserIds" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT DISTINCT advertiser_id,page_id
         FROM media_api.adq_dynamic_createive_list
         where page_type = 'PAGE_TYPE_WECHAT_CANVAS'
           and mini_program_id is null
@@ -421,7 +421,9 @@
         mini_program_id,
         page_type,
         page_id,
-        mini_program_path
+        mini_program_path,
+        creative_create_time,
+        last_modified_time
         )
         values
         <foreach collection="list" item="item" separator=",">
@@ -432,7 +434,9 @@
             #{item.mini_program_id},
             #{item.page_type},
             #{item.page_id},
-            #{item.mini_program_path}
+            #{item.mini_program_path},
+            #{item.created_time},
+            #{item.last_modified_time}
             )
         </foreach>
     </insert>

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

@@ -90,10 +90,7 @@
                        LEFT JOIN thirtpart.ly_official_accounts t2 ON t0.official_account_id = t2.id
               WHERE official_account_id IS NOT NULL
               GROUP BY t1.link_id
-
-
-              union all
-
+              UNION ALL
               SELECT t3.date,
                      t0.official_account_id,
                      t0.responsibility_person_id,
@@ -113,7 +110,8 @@
               FROM (SELECT DATE_FORMAT(activation_time, '%Y%m%d') AS 'date', link_id,
                            count(DISTINCT uid) AS 'new_users'
                     FROM thirtpart.ly_user_activation_info
-                    WHERE DATE_FORMAT(activation_time, '%Y%m%d') = #{nowDate}
+                    WHERE created_at >= unix_timestamp(#{nowDate})
+                      AND created_at < unix_timestamp(DATE_ADD(#{nowDate}, INTERVAL 1 day))
                     GROUP BY link_id) t3
                        LEFT JOIN thirtpart.ly_link t0 ON t0.id = t3.link_id
                        LEFT JOIN (SELECT consume_date,
@@ -165,107 +163,6 @@
         GROUP BY t.id
     </insert>
 
-    <insert id="cleanCPSIaaDailyReports">
-        replace
-        into thirtpart.cps_link_data_report
-        (
-                stat_date,
-                official_account_id,
-                responsibility_person_id,
-                ma_appid,
-                link_id,
-                new_users,
-                new_pay_users,
-                new_pay_counts,
-                new_order_counts,
-                new_pay_amount,
-                new_xn_pay_amount,
-                pay_amount,
-                xn_pay_amount,
-                pay_users,pay_counts,
-                order_counts,
-                new_iaa_users,
-                new_iaa_counts,
-                new_iaa_amount,
-                iaa_users,
-                iaa_counts,
-                iaa_amount,
-				active_users
-                )
-        SELECT IFNULL(t.date, t0.date),
-               tt.official_account_id,
-               tt.responsibility_person_id,
-               tt.app_id,
-               tt.id,
-               t1.new_users,
-               t3.new_pay_users,
-               t3.new_pay_counts,
-               t3.new_order_counts,
-               t3.new_pay_amount,
-               t3.new_xn_pay_amount,
-               t3.pay_amount,
-               t3.xn_pay_amount,
-               t3.pay_users,
-               t3.pay_counts,
-               t3.order_counts,
-               t0.new_iaa_users,
-               t0.new_iaa_counts,
-               t0.new_iaa_amount,
-               t0.iaa_users,
-               t0.iaa_counts,
-               t0.iaa_amount,
-               t.active_users
-        FROM (SELECT t0.id,
-                     t0.responsibility_person_id,
-                     t0.official_account_id,
-                     t2.app_id
-              FROM thirtpart.ly_link t0
-                       LEFT JOIN thirtpart.ly_official_accounts t2 ON t0.official_account_id = t2.id) tt
-                 LEFT JOIN
-             (SELECT t.date, t.link_id, COUNT(DISTINCT t.uid) as 'active_users'
-              FROM thirtpart.ly_user_login_record t
-              where date = #{nowDate}
-                AND link_id
-                  >0
-              GROUP BY link_id) t ON tt.id = t.link_id
-                 LEFT JOIN
-             (SELECT t.stat_date as 'date', t.link_id,
-                     count(distinct case
-                                        when SUBSTR(dying_time, 1, 10) = SUBSTR(event_time, 1, 10)
-                                            then uid end) 'new_iaa_users', sum(if(SUBSTR(dying_time, 1, 10) = SUBSTR(event_time, 1, 10), 1, 0)) 'new_iaa_counts', IFNULL(ROUND(
-                             sum(if(SUBSTR(dying_time, 1, 10) = SUBSTR(event_time, 1, 10), cost / 100000, 0)),
-                             2),0.00) 'new_iaa_amount', count(distinct uid) 'iaa_users', sum(1) 'iaa_counts', IFNULL(ROUND(sum(cost / 100000), 2),0.00) 'iaa_amount'
-              FROM media_api.bytedance_webhook_ma_ecpm t
-                       LEFT JOIN thirtpart.ly_link t0 ON t.link_id = t0.id
-              where t.stat_date = #{nowDate}
-              GROUP BY t.link_id
-              ORDER BY t.link_id) t0 ON tt.id = t0.link_id
-                 LEFT JOIN (SELECT t0.id,
-                                   DATE_FORMAT(t.activation_time, '%Y%m%d') as 'date', t0.official_account_id,
-                                   t0.responsibility_person_id,
-                                   count(DISTINCT t.uid) AS 'new_users'
-                            FROM thirtpart.ly_user_activation_info t
-                                     LEFT JOIN thirtpart.ly_link t0 ON t.link_id = t0.id
-                            WHERE DATE_FORMAT(t.activation_time, '%Y%m%d') = #{nowDate}
-                            GROUP BY t0.id) t1 ON tt.id = t1.id
-                 LEFT JOIN
-             (SELECT link_id,
-                     new_pay_users,
-                     new_pay_counts,
-                     new_order_counts,
-                     new_pay_amount,
-                     new_xn_pay_amount,
-                     pay_amount,
-                     xn_pay_amount,
-                     pay_users,
-                     pay_counts,
-                     order_counts
-              FROM thirtpart.cps_link_data_report
-              WHERE stat_date = #{nowDate}) t3 ON t3.link_id = tt.id
-        WHERE t0.iaa_amount is not null
-        GROUP BY tt.id
-    </insert>
-
     <select id="queryCPSIaaDaily" resultType="com.alibaba.fastjson.JSONObject">
         SELECT IFNULL(t.date, t0.date) AS 'date', tt.official_account_id,
                tt.responsibility_person_id,
@@ -294,7 +191,7 @@
                        LEFT JOIN tidb.thirtpart.ly_official_accounts t2 ON t0.official_account_id = t2.id) tt
                  LEFT JOIN (SELECT t.date, t.link_id, COUNT(DISTINCT t.uid) AS 'active_users'
                             FROM tidb.thirtpart.ly_user_login_record t
-                            WHERE date = #{nowDate} AND link_id > 0
+                            WHERE date = #{dateStr} AND link_id > 0
                             GROUP BY link_id, date) t ON tt.id = t.link_id
                  LEFT JOIN (SELECT t.stat_date AS 'date', t.link_id,
                                    count(DISTINCT CASE
@@ -313,7 +210,8 @@
                                    count(DISTINCT t.uid) AS 'new_users'
                             FROM tidb.thirtpart.ly_user_activation_info t
                                      LEFT JOIN tidb.thirtpart.ly_link t0 ON t.link_id = t0.id
-                            WHERE DATE_FORMAT(t.activation_time, '%Y%m%d') = #{nowDate}
+                            WHERE t.created_at >= unix_timestamp(#{nowDate})
+                              AND t.created_at &lt; unix_timestamp(DATE_ADD(#{nowDate}, INTERVAL 1 day))
                             GROUP BY 1, 2, 3, 4) t1 ON tt.id = t1.id
                  LEFT JOIN (SELECT link_id,
                                    new_pay_users,

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

@@ -1,6 +1,7 @@
 package cn.com.ctop.job.bytedance.data.service;
 
 import com.alibaba.fastjson.JSONArray;
+import com.alibaba.fastjson.JSONObject;
 
 import java.util.List;
 
@@ -36,11 +37,11 @@ public interface IBytedanceCdpProjectService {
 
     List<Long> queryADQAdverIds(String date);
 
-    void getAdqCreatives(Long advertiserId, String date);
+    void getAdqCreatives(Long advertiserId, String startDate,String endDate);
 
-    void getAdqWechatPages(Long advertiserId);
+    void getAdqWechatPages(Long advertiserId,Long pageId);
 
     List<Long> queryTTErrApps();
 
-    List<Long> queryPageAdvertiserIds();
+    List<JSONObject> queryPageAdvertiserIds();
 }

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

@@ -2,5 +2,5 @@ package cn.com.ctop.job.bytedance.data.service;
 
 public interface ICPSReportService {
 
-    void cleanCPSDailyReports(Integer nowDate);
+    void cleanCPSDailyReports(String nowDate);
 }

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

@@ -7,6 +7,8 @@ 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 com.alibaba.fastjson.JSONArray;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
@@ -24,15 +26,10 @@ import java.util.*;
 @EnableScheduling
 public class BytedanceAccountReportServiceImpl implements IBytedanceAccountReportService {
     @Autowired
-    private BytedanceAccountReportMapper accountReportMapper;
-
-    /*private static final int MAX_BATCH_SIZE = 5000;
-    private static final long MAX_WAIT_TIME_MS = 2 * 60 * 1000;
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
 
-    private final List<BytedanceAdvertiserReportHour> advertiserReportHourBuffer = Collections.synchronizedList(new ArrayList<>());
-    private final List<BytedanceAdvertiserReportDaily> advertiserReportDailyBuffer = Collections.synchronizedList(new ArrayList<>());
-    private long advertiserReportHourLastFlushTime = System.currentTimeMillis();
-    private long advertiserReportDailyLastFlushTime = System.currentTimeMillis();*/
+    @Autowired
+    private BytedanceAccountReportMapper accountReportMapper;
 
 
     @Override
@@ -231,6 +228,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setStatTimeDay(statTimeDay);
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
+
+                            dorisLoadManager.add(new RecordWithMeta("media_api","bytedance_advertiser_report_hour",advertiserReportHour));
                         }
                     } else {
                         log.info("账户id:{}，日期:{}广告时报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -253,8 +252,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 {
@@ -350,7 +348,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
-                getBytedanceAccountHour(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
+                getBytedanceAccountDaily(advertiserId, accessToken, startDate, endDate, page + 1, isLog);
             } else {
                 log.info("账户ID：{}，日期:{},广告日报（账户粒度）数据请求成功", advertiserId, startDate);
             }

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

@@ -12,10 +12,7 @@ import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
 
 import java.time.Instant;
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 /**
  * 头条项目列Service业务层处理
@@ -53,7 +50,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
     @Override
-    public List<Long> queryPageAdvertiserIds() {
+    public List<JSONObject> queryPageAdvertiserIds() {
         return bytedanceCdpProjectMapper.queryPageAdvertiserIds();
     }
 
@@ -438,14 +435,15 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
     @Override
-    public void getAdqCreatives(Long advertiserId, String date) {
+    public void getAdqCreatives(Long advertiserId, String startDate,String endDate) {
         try {
             String accessToken = tokenMapper.onlyGetADQToken(advertiserId);
             if (Check.isNull(accessToken)) {
                 log.error("账户（{}）accessToken is null", advertiserId);
                 return;
             }
-            this.rebackadqCreativesVo(advertiserId, accessToken, date, 1, true, 1);
+            this.rebackadqCreativesVo(advertiserId, accessToken, startDate, endDate, 1, true, 1);
+            this.rebackadqCreativesVo(advertiserId, accessToken, startDate, endDate, 1, false, 1);
         } catch (Exception e) {
             e.printStackTrace();
             log.error("账户（{}）获取动态创意异常", e.getMessage());
@@ -453,9 +451,8 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
 
-    private void rebackadqCreativesVo(Long accountId, String accessToken, String date, Integer page, Boolean isDeleted, int count) throws Exception {
+    private void rebackadqCreativesVo(Long accountId, String accessToken, String startDate, String endDate, Integer page, Boolean isDeleted, int count) throws Exception {
         boolean nextPage = false;
-        JSONObject returnJson = new JSONObject();
         Map<String, Object> param = new HashMap<>();
         long timestamp = Instant.now().getEpochSecond();
         param.put("timestamp", String.valueOf(timestamp));
@@ -464,29 +461,36 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
         param.put("account_id", accountId);
         param.put("page", page);
         param.put("page_size", 100);
-        if (!Check.isNull(date)) {
-            JSONArray filtering = new JSONArray();
+        JSONArray filtering = new JSONArray();
+        if (!Check.isNull(startDate)) {
             JSONObject greObj = new JSONObject();
-            greObj.put("field", "created_time");
+            greObj.put("field", "last_modified_time");
             greObj.put("operator", "GREATER_EQUALS");
             JSONArray startDateVls = new JSONArray();
-            startDateVls.add(String.valueOf(DateUtils.dt2milli(date + " 00:00:00")));
+            startDateVls.add(String.valueOf(DateUtils.dt2milli(startDate + " 00:00:00")));
             greObj.put("values", startDateVls);
             filtering.add(greObj);
+        }
+
+        if (!Check.isNull(endDate)) {
             JSONObject lesObj = new JSONObject();
-            lesObj.put("field", "created_time");
+            lesObj.put("field", "last_modified_time");
             lesObj.put("operator", "LESS_EQUALS");
             JSONArray endDateVls = new JSONArray();
-            endDateVls.add(String.valueOf(DateUtils.dt2milli(date + " 23:59:59")));
+            endDateVls.add(String.valueOf(DateUtils.dt2milli(endDate + " 23:59:59")));
             lesObj.put("values", endDateVls);
             filtering.add(lesObj);
+        }
+        if (!Check.isNull(filtering)) {
             param.put("filtering", filtering);
         }
-//        param.put("is_deleted", isDeleted);
+        param.put("is_deleted", isDeleted);
         JSONArray fields = new JSONArray();
         fields.add("adgroup_id");
         fields.add("dynamic_creative_id");
         fields.add("creative_components");
+        fields.add("created_time");
+        fields.add("last_modified_time");
         param.put("fields", fields);
         String result = HttpUtils.httpGet("https://api.e.qq.com/v3.0/dynamic_creatives/get", param, new HashMap<>());
         JSONObject obj = JSONObject.parseObject(result);
@@ -518,26 +522,26 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             } else {
                 if (count <= 4) {
                     Thread.sleep(500);
-                    rebackadqCreativesVo(accountId, accessToken, date, page, isDeleted, count + 1);
+                    rebackadqCreativesVo(accountId, accessToken, startDate, endDate, page, isDeleted, count + 1);
                 } else {
                     log.error("账户（{}）获取动态创意失败，入参：{}\n，结果：{}", accountId, JSONObject.toJSONString(param), obj);
                 }
             }
         }
         if (nextPage) {
-            rebackadqCreativesVo(accountId, accessToken, date, page + 1, isDeleted, 1);
+            rebackadqCreativesVo(accountId, accessToken, startDate ,endDate, page + 1, isDeleted, 1);
         }
     }
 
     @Override
-    public void getAdqWechatPages(Long advertiserId) {
+    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, accessToken, 1, true, 1);
+            this.getAdqWechatPagesVo(advertiserId,pageId, accessToken, 1, true, 1);
         } catch (Exception e) {
             e.printStackTrace();
             log.error("账户（{}）获取原生落地页异常", e.getMessage());
@@ -545,9 +549,8 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
 
-    private void getAdqWechatPagesVo(Long accountId, 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;
-        JSONObject returnJson = new JSONObject();
         Map<String, Object> param = new HashMap<>();
         long timestamp = Instant.now().getEpochSecond();
         param.put("timestamp", String.valueOf(timestamp));
@@ -561,9 +564,15 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
         greObj.put("field", "page_type");
         greObj.put("operator", "EQUALS");
         greObj.put("values", "PAGE_TYPE_WECHAT_CANVAS");
+
+        JSONObject pageFilter = new JSONObject();
+        pageFilter.put("field", "page_id");
+        pageFilter.put("operator", "EQUALS");
+        pageFilter.put("values", Collections.singletonList(String.valueOf(pageId)));
         filtering.add(greObj);
+        filtering.add(pageFilter);
 
-//        param.put("filtering", filtering);
+        param.put("filtering", filtering);
         String result = HttpUtils.httpGet("https://api.e.qq.com/v3.0/wechat_pages/get", param, new HashMap<>());
         JSONObject obj = JSONObject.parseObject(result);
         if (!Check.isNull(obj)) {
@@ -594,14 +603,14 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
             } else {
                 if (count <= 4) {
                     Thread.sleep(500);
-                    getAdqWechatPagesVo(accountId, 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, 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/CPSReportServiceImpl.java

@@ -13,10 +13,10 @@ public class CPSReportServiceImpl implements ICPSReportService {
     private CPSReportMapper cpsReportMapper;
 
     @Override
-    public void cleanCPSDailyReports(Integer nowDate) {
+    public void cleanCPSDailyReports(String nowDate) {
         try {
             if (cpsReportMapper.cleanCPSPayDailyReports(nowDate))
-                if (cpsReportMapper.cleanCPSIaaDailyReportsDS(cpsReportMapper.queryCPSIaaDaily(nowDate)))
+                if (cpsReportMapper.cleanCPSIaaDailyReportsDS(cpsReportMapper.queryCPSIaaDaily(Integer.valueOf(nowDate), nowDate)))
                     log.info("CPS数据清洗完成:{}", nowDate);
         } catch (Exception e) {
             e.printStackTrace();

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

@@ -0,0 +1,56 @@
+package cn.com.ctop.job.bytedance.data.utils;
+
+import com.alibaba.fastjson.JSON;
+import com.alibaba.fastjson.PropertyNamingStrategy;
+import com.alibaba.fastjson.serializer.NameFilter;
+import com.alibaba.fastjson.serializer.SerializeConfig;
+import com.alibaba.fastjson.serializer.SerializerFeature;
+
+import java.util.regex.Pattern;
+
+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 NUMBER_LETTER = Pattern.compile("([0-9])([a-zA-Z])");
+    private static final Pattern CAMEL_CASE = Pattern.compile("([a-z])([A-Z])");
+
+    // 全局 SerializeConfig
+    private static final SerializeConfig SNAKE_CASE_CONFIG = new SerializeConfig();
+
+    static {
+        SNAKE_CASE_CONFIG.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
+    }
+
+    // NameFilter
+    private static final NameFilter CAMEL_TO_SNAKE_FILTER = (object, name, value) -> {
+        // 去除非法首尾字符
+        name = INVALID_PREFIX.matcher(name).replaceAll("");
+        name = INVALID_SUFFIX.matcher(name).replaceAll("");
+
+        // a3 -> a_3，3a -> 3_a
+        name = LETTER_NUMBER.matcher(name).replaceAll("$1_$2");
+        //name = NUMBER_LETTER.matcher(name).replaceAll("$1_$2");
+
+        // camelCase -> camel_case
+        name = CAMEL_CASE.matcher(name).replaceAll("$1_$2");
+
+        return name.toLowerCase();
+    };
+
+    /**
+     * 将对象转为 JSON 字符串，字段名转为下划线命名。
+     */
+    public static String toSnakeCaseJsonIgnoreNull(Object obj) {
+        return JSON.toJSONString(obj, SNAKE_CASE_CONFIG);
+    }
+
+    /**
+     * 将对象转为 JSON 字符串，字段名转为下划线命名
+     */
+    public static String toFilterJSON(Object obj) {
+        return JSON.toJSONString(obj, CAMEL_TO_SNAKE_FILTER, SerializerFeature.WriteMapNullValue, SerializerFeature.MapSortField);
+    }
+}

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

@@ -4,6 +4,7 @@ package cn.com.ctop.job.bytedance.handler;
 import cn.com.ctop.job.bytedance.data.service.IBytedanceCdpProjectService;
 import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.DateUtils;
+import com.alibaba.fastjson.JSONObject;
 import com.xxl.job.core.handler.annotation.XxlJob;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -42,8 +43,10 @@ public class BytedanceCdpProjectJob {
         List<Long> ids = cdpProjectService.queryAllAdverIdList("");
         if (!Check.isNull(ids)) {
             log.info("总数：{}", ids.size());
-            ids.forEach(id -> allService.submit(() -> cdpProjectService.getProjectList(id, null, null, null)
-            ));
+//            ids.forEach(id -> allService.submit(() -> cdpProjectService.getProjectList(id, null, null, null)));
+            ids.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, "2025-07-09", "2025-07-09")));
+
+
         }
     }
 
@@ -156,14 +159,17 @@ public class BytedanceCdpProjectJob {
             String today = DateUtils.getNowDate("yyyy-MM-dd");
             List<Long> ids = cdpProjectService.queryAdvertiserIds("adq_advertiser_report_hour", DateUtils.getYYMMDDTime(today));
             if (!Check.isNull(ids)) {
-                ids.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, today)));
+                ids.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, today, today)));
             }
 
-            List<Long> iids = cdpProjectService.queryPageAdvertiserIds();
+            List<JSONObject> iids = cdpProjectService.queryPageAdvertiserIds();
             if (!Check.isNull(iids)) {
-                iids.forEach(id -> pageAppsExecutorService.submit(() -> cdpProjectService.getAdqWechatPages(id)));
+                for (JSONObject iid : iids) {
+                    Long pageId = iid.getLong("page_id");
+                    Long advertiserId = iid.getLong("advertiser_id");
+                    pageAppsExecutorService.submit(() -> cdpProjectService.getAdqWechatPages(advertiserId,pageId));
+                }
             }
-
         } catch (Exception e) {
             e.printStackTrace();
             log.error("[nowDynamicCreateives]" + e.getMessage());
@@ -178,13 +184,13 @@ public class BytedanceCdpProjectJob {
             String yesterday = DateUtils.getNowAddDate("yyyy-MM-dd", -1);
             List<Long> ids = cdpProjectService.queryAdvertiserIds("adq_advertiser_report_daily", DateUtils.getYYMMDDTime(yesterday));
             if (!Check.isNull(ids)) {
-                ids.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, yesterday)));
+                ids.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, yesterday, yesterday)));
             }
             //拉取前天数据
             String yesterday2 = DateUtils.getNowAddDate("yyyy-MM-dd", -2);
             List<Long> ids2 = cdpProjectService.queryAdvertiserIds("adq_advertiser_report_daily", DateUtils.getYYMMDDTime(yesterday2));
             if (!Check.isNull(ids2)) {
-                ids2.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, yesterday2)));
+                ids2.forEach(id -> createiveAppsExecutorService.submit(() -> cdpProjectService.getAdqCreatives(id, yesterday2, yesterday2)));
             }
         } catch (Exception e) {
             e.printStackTrace();

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

@@ -204,8 +204,6 @@ public class BytedanceReportRetryJob {
                 String dataDimension = jsonObject.getString("dataDimension");
                 Integer isSend = jsonObject.getInteger("isSend");
                 String accessToken = jsonObject.getString("accessToken");
-                Long taskId = jsonObject.getJSONObject("msg").getLong("taskId");
-
                 if ("xingtu_task_list".equalsIgnoreCase(dataDimension)) {
                     xingTuExecutorService.submit(new Runnable() {
                         @Override
@@ -214,6 +212,7 @@ public class BytedanceReportRetryJob {
                         }
                     });
                 } else if ("xingtu_task_info".equalsIgnoreCase(dataDimension)) {
+                    Long taskId = jsonObject.getJSONObject("msg").getLong("taskId");
                     xingTuExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
@@ -221,6 +220,7 @@ public class BytedanceReportRetryJob {
                         }
                     });
                 } else if ("xingtu_task_report".equalsIgnoreCase(dataDimension)) {
+                    Long taskId = jsonObject.getJSONObject("msg").getLong("taskId");
                     xingTuExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {

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

@@ -26,7 +26,7 @@ public class CPSReportJob {
         log.info("[dayCleanCPSDailyReports] 任务启动...");
         try {
             String nowDate = DateUtils.getNowDate("yyyyMMdd");
-            cpsReportService.cleanCPSDailyReports(Integer.valueOf(nowDate));
+            cpsReportService.cleanCPSDailyReports(nowDate);
             log.info("[dayCleanCPSDailyReports] 任务完成");
         } catch (Exception e) {
             e.printStackTrace();
@@ -43,7 +43,7 @@ public class CPSReportJob {
         try {
             String nowDate = DateUtils.getNowDate("yyyyMMdd");
             String yesterday = DateUtils.getNowAddDate("yyyyMMdd", -1);
-            cpsReportService.cleanCPSDailyReports(Integer.valueOf(yesterday));
+            cpsReportService.cleanCPSDailyReports(yesterday);
             log.info("[yesCleanCPSDailyReports] 任务完成");
         } catch (Exception e) {
             e.printStackTrace();

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

@@ -0,0 +1,16 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class DorisBufferConfig {
+    private long bufferFlushMaxRows;
+    private long bufferFlushMaxBytes;
+    private long bufferFlushIntervalMs;
+}

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

@@ -0,0 +1,45 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+@Configuration
+public class DorisConfiguration {
+    @Value("${spring.doris.fe-host}")
+    private String feHost;
+
+    @Value("${spring.doris.user}")
+    private String userName;
+
+    @Value("${spring.doris.password}")
+    private String password;
+
+    @Value("${spring.doris.label-prefix}")
+    private String labelPrefix;
+
+    @Value("${spring.doris.sink-threads}")
+    private int uploaderThreads;
+
+    @Value("${spring.doris.sink-batch-rows}")
+    private int flushMaxRows;
+
+    @Value("${spring.doris.sink-batch-mb}")
+    private long flushMaxMb;
+
+    @Value("${spring.doris.sink-batch-seconds}")
+    private long flushIntervalSeconds;
+
+    @Bean
+    public DorisMultiTableStreamLoadManager dorisStreamLoadManager() {
+        DorisBufferConfig defaultConfig = DorisBufferConfig.builder()
+                .bufferFlushMaxRows(flushMaxRows)
+                .bufferFlushMaxBytes(flushMaxMb * 1024 * 1024)
+                .bufferFlushIntervalMs(flushIntervalSeconds * 1000)
+                .build();
+
+        DorisSinkClient sinkClient = new SimpleHttpDorisSinkClient(feHost, userName, password, labelPrefix);
+
+        return new DorisMultiTableStreamLoadManager(uploaderThreads, defaultConfig, sinkClient);
+    }
+}

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

@@ -0,0 +1,41 @@
+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.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+@Slf4j
+public class DorisMultiTableStreamLoadManager {
+    private final Map<TableKey, TableBufferManager> bufferManagers = new ConcurrentHashMap<>();
+    private final ExecutorService uploaderExecutor;
+    private final DorisSinkClient sinkClient;
+    private final DorisBufferConfig defaultBufferConfig;
+
+    public DorisMultiTableStreamLoadManager(int uploaderThreads,
+                                             DorisBufferConfig defaultBufferConfig,
+                                             DorisSinkClient sinkClient) {
+        log.info("init DorisMultiTableStreamLoadManager");
+        this.defaultBufferConfig = defaultBufferConfig;
+        this.sinkClient = sinkClient;
+        this.uploaderExecutor = Executors.newFixedThreadPool(uploaderThreads,
+                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);
+    }
+
+    @PreDestroy
+    public void shutdown() {
+        log.info("Shutting down DorisMultiTableStreamLoadManager");
+        bufferManagers.values().forEach(TableBufferManager::flushNow);
+        uploaderExecutor.shutdown();
+    }
+}

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

@@ -0,0 +1,7 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import java.util.List;
+@FunctionalInterface
+public interface DorisSinkClient {
+    void streamLoad(List<RecordWithMeta> batch) throws Exception;
+}

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

@@ -0,0 +1,17 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+
+@Data
+@Builder
+@AllArgsConstructor
+@NoArgsConstructor
+public class RecordWithMeta {
+    private String database;
+    private String table;
+    private Object data;
+}

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

@@ -0,0 +1,86 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import cn.com.ctop.job.bytedance.data.utils.JsonUtils;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.http.HttpEntity;
+import org.apache.http.client.methods.CloseableHttpResponse;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.entity.StringEntity;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClients;
+import org.apache.http.util.EntityUtils;
+
+import java.nio.charset.StandardCharsets;
+import java.util.Base64;
+import java.util.List;
+import java.util.UUID;
+import java.util.stream.Collectors;
+
+@Slf4j
+public class SimpleHttpDorisSinkClient implements DorisSinkClient {
+    private final String feHost;      // http://fe_host:8030
+    private final String user;        // doris 用户名
+    private final String password;    // doris 密码
+    private final String format;      // json/csv, 推荐 json
+    private final String labelPrefix; // stream load label 前缀（可选）
+
+    public SimpleHttpDorisSinkClient(String feHost, String user, String password, String labelPrefix) {
+        this.feHost = feHost;
+        this.user = user;
+        this.password = password;
+        this.format = "json";
+        this.labelPrefix = labelPrefix;
+    }
+
+    @Override
+    public void streamLoad(List<RecordWithMeta> records) throws Exception {
+        if (records == null || records.isEmpty()) return;
+
+        RecordWithMeta sample = records.get(0);
+        String db = sample.getDatabase();
+        String table = sample.getTable();
+
+        String label = labelPrefix + db + "_" + table + "_" + UUID.randomUUID();
+        String url = String.format("%s/api/%s/%s/_stream_load", feHost, db, table);
+
+        // 提取 data
+        List<Object> dataList = records.stream()
+                .map(RecordWithMeta::getData)
+                .collect(Collectors.toList());
+
+        String jsonBody = JsonUtils.toFilterJSON(dataList);
+        //log.info("jsonBody:{}", jsonBody);
+
+        HttpPut put = new HttpPut(url);
+        put.setHeader("Expect", "100-continue");
+        put.setHeader("label", label);
+        put.setHeader("Content-Type", "application/json");
+        if (user != null && password != null) {
+            String auth = Base64.getEncoder().encodeToString((user + ":" + password).getBytes(StandardCharsets.UTF_8));
+            put.setHeader("Authorization", "Basic " + auth);
+        }
+
+        put.setHeader("format", format); // json
+        put.setHeader("strip_outer_array", "true"); // json array
+        put.setHeader("fuzzy_parse", "true"); // json array
+
+        put.setEntity(new StringEntity(jsonBody, StandardCharsets.UTF_8));
+
+        try (CloseableHttpClient httpClient = HttpClients.custom()
+                .setRedirectStrategy(new StrictRedirectStrategy())  // 支持 307/301 重定向
+                .build();
+             CloseableHttpResponse response = httpClient.execute(put)) {
+            int code = response.getStatusLine().getStatusCode();
+            HttpEntity entity = response.getEntity();
+            String responseBody = EntityUtils.toString(entity, "UTF-8");
+            if (code != 200 && code != 201) {
+                throw new RuntimeException("Doris stream load failed. Status: " + code + ", response: " + responseBody);
+            }
+            // 可选: 解析返回内容判断 success
+            if (!responseBody.contains("\"Status\": \"Success\"")) {
+                throw new RuntimeException("Stream load returned non-success: " + responseBody);
+            }
+            log.info("Doris stream load success: {}", responseBody);
+        }
+    }
+}

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

@@ -0,0 +1,38 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import org.apache.http.HttpEntityEnclosingRequest;
+import org.apache.http.HttpRequest;
+import org.apache.http.HttpResponse;
+import org.apache.http.ProtocolException;
+import org.apache.http.client.methods.HttpPut;
+import org.apache.http.client.methods.HttpUriRequest;
+import org.apache.http.impl.client.DefaultRedirectStrategy;
+import org.apache.http.protocol.HttpContext;
+
+import java.net.URI;
+
+public class StrictRedirectStrategy extends DefaultRedirectStrategy {
+
+    @Override
+    protected boolean isRedirectable(String method) {
+        // 允许 GET, POST, PUT 等都可重定向
+        return true;
+    }
+
+    @Override
+    public HttpUriRequest getRedirect(HttpRequest request, HttpResponse response, HttpContext context)
+            throws ProtocolException {
+        URI uri = getLocationURI(request, response, context);
+
+        String method = request.getRequestLine().getMethod();
+        if (method.equalsIgnoreCase(HttpPut.METHOD_NAME)) {
+            HttpPut put = new HttpPut(uri);
+            put.setHeaders(request.getAllHeaders());
+            if (request instanceof HttpEntityEnclosingRequest) {
+                put.setEntity(((HttpEntityEnclosingRequest) request).getEntity());
+            }
+            return put;
+        }
+        return super.getRedirect(request, response, context);
+    }
+}

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

@@ -0,0 +1,96 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
+import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
+
+import java.nio.charset.StandardCharsets;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+@Slf4j
+public class TableBufferManager {
+
+    private final TableKey tableKey;
+    private final DorisBufferConfig config;
+    private final DorisSinkClient sinkClient;
+    private final ExecutorService uploaderExecutor;
+
+    private final List<RecordWithMeta> buffer = new ArrayList<>();
+    private final Object lock = new Object();
+    private final AtomicLong currentBytes = new AtomicLong(0);
+
+    private final ScheduledExecutorService scheduler =
+            Executors.newSingleThreadScheduledExecutor(new CustomizableThreadFactory("flusher-" + UUID.randomUUID()));
+
+    public TableBufferManager(TableKey tableKey,
+                              DorisBufferConfig config,
+                              DorisSinkClient sinkClient,
+                              ExecutorService uploaderExecutor) {
+        this.tableKey = tableKey;
+        this.config = config;
+        this.sinkClient = sinkClient;
+        this.uploaderExecutor = uploaderExecutor;
+
+        scheduler.scheduleAtFixedRate(this::tryFlush,
+                config.getBufferFlushIntervalMs(),
+                config.getBufferFlushIntervalMs(),
+                TimeUnit.MILLISECONDS);
+    }
+
+    public void add(RecordWithMeta record) {
+        if (record == null || StringUtils.isBlank(record.getDatabase()) || StringUtils.isBlank(record.getTable())) {
+            return;
+        }
+        synchronized (lock) {
+            buffer.add(record);
+            currentBytes.addAndGet(estimateSize(record));
+
+            if (buffer.size() >= config.getBufferFlushMaxRows() ||
+                    currentBytes.get() >= config.getBufferFlushMaxBytes()) {
+                flush();
+            }
+        }
+    }
+
+    public void tryFlush() {
+        synchronized (lock) {
+            if (!buffer.isEmpty()) {
+                flush();
+            }
+        }
+    }
+
+    public void flushNow() {
+        synchronized (lock) {
+            flush();
+        }
+        scheduler.shutdownNow();
+    }
+
+    private void flush() {
+        if (buffer.isEmpty()) return;
+
+        List<RecordWithMeta> toSend = new ArrayList<>(buffer);
+        buffer.clear();
+        currentBytes.set(0);
+
+        uploaderExecutor.submit(() -> {
+            try {
+                sinkClient.streamLoad(toSend);
+            } catch (Exception e) {
+                log.error("Stream load failed for {}.{}: {}", tableKey.getDatabase(), tableKey.getTable(), e.getMessage(), e);
+            }
+        });
+    }
+
+    private long estimateSize(RecordWithMeta record) {
+        return record.toString().getBytes(StandardCharsets.UTF_8).length;
+    }
+}

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

@@ -0,0 +1,26 @@
+package cn.com.ctop.job.bytedance.utils.doris_sink;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+
+import java.util.Objects;
+
+@Data
+@AllArgsConstructor
+public class TableKey {
+    private String database;
+    private String table;
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(database, table);
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) return true;
+        if (!(obj instanceof TableKey)) return false;
+        TableKey other = (TableKey) obj;
+        return Objects.equals(database, other.database) && Objects.equals(table, other.table);
+    }
+}

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

@@ -85,7 +85,7 @@ spring:
         master:
           url: jdbc:mysql://172.16.0.117:3390/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false
           #url: jdbc:mysql://8.147.135.80:3390/media_api?characterEncoding=UTF-8&useUnicode=true&allowMultiQueries=true&useSSL=false
-          username: data
+          username: yanfa
           password: hr@2025..
           driver-class-name: com.mysql.jdbc.Driver
         selectdb:
@@ -93,7 +93,7 @@ spring:
           username: data
           password: hr@2025..
           driver-class-name: com.mysql.jdbc.Driver
-        selectdb_group_commit:
+        sd_group_commit:
           ## 注意此selectdb数据源不支持列更新，insert into 对于未指定列会被置为null，类似tidb replace into
           url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false&useServerPrepStmts=true&useLocalSessionState=true&rewriteBatchedStatements=true&cachePrepStmts=true&prepStmtCacheSqlLimit=99999&prepStmtCacheSize=500&sessionVariables=group_commit=async_mode,enable_nereids_planner=false
           username: data
@@ -134,6 +134,15 @@ spring:
       min-idle: 8
     port: 6379
 
+  doris:
+    fe-host: http://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:8080
+    user: data
+    password: hr@2025..
+    label-prefix: spring-bytedance
+    sink-threads: 3
+    sink-batch-rows: 50000
+    sink-batch-mb: 20
+    sink-batch-seconds: 20
 
 #mybatis plus 设置
 mybatis-plus: