Merge branch 'master' into zhaotx

# Conflicts:
#	job-bytedance/src/main/java/cn/com/ctop/job/bytedance/data/mapper/xml/BytedanceProductUpdateMapper.xml

job-bytedance/pom.xml

@@ -9,7 +9,7 @@
     <parent>
         <groupId>org.springframework.boot</groupId>
         <artifactId>spring-boot-starter-parent</artifactId>
-        <version>2.1.3.RELEASE</version>
+        <version>2.3.12.RELEASE</version>
         <relativePath/>
     </parent>
     <properties>
@@ -180,7 +180,7 @@
         <dependency>
             <groupId>redis.clients</groupId>
             <artifactId>jedis</artifactId>
-            <version>2.9.0</version>
+            <version>3.1.0</version>
         </dependency>
         <dependency>
             <groupId>org.apache.commons</groupId>

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/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/AdminAdvertiserListMapper.java

@@ -14,6 +14,6 @@ public interface AdminAdvertiserListMapper {
 
     void batchUpdateAdvertiserName(@Param("infos") List<JSONObject> infos);
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<JSONObject> getAdvertiserInfosByAdminIds();
 }

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")
+    @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")
+    @DS("sd_group_commit")
     void replaceAccountReportHourSD(@Param("list") List<BytedanceAdvertiserReportHour> hours);
 
     void replaceAccountReportDaily(@Param("list") List<BytedanceAdvertiserReportDaily> dailyList);
-    @DS("selectdb")
+    @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")
+    @DS("sd_group_commit")
     int replaceBytedanceCdpProjectListSD(@Param("projectList") List<BytedanceCdpProject> projectList);
 
     int replaceBytedancePromotionList(@Param("list") JSONArray array);
-    @DS("selectdb")
+    @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")
+    @DS("sd_group_commit")
     int replaceBytedanceByteAppLinkInfoSD(@Param("list") JSONArray list);
 
     List<Long> queryAllAdverIdList(@Param("mediaType") String mediaType);
@@ -37,29 +38,29 @@ public interface BytedanceCdpProjectMapper {
 
     List<Long> queryTTErrApps();
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
-    List<Long> queryPageAdvertiserIds();
+    @DS("selectdb")
+    List<JSONObject> queryPageAdvertiserIds();
 
     boolean delErrApps(@Param("type") String type);
 
     boolean delTTErrApps();
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<Long> queryAdvertiserIds(@Param("db") String db, @Param("date") String date);
 
     int replaceBytedanceWxAppInfo(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
-    @DS("selectdb")
+    @DS("sd_group_commit")
     int replaceBytedanceWxAppInfoSD(@Param("list") JSONArray list, @Param("advertiserId") Long advertiserId);
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @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")
+    @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")
+    @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/BytedanceMiniAppMapper.java

@@ -10,7 +10,7 @@ import org.apache.ibatis.annotations.Param;
 import java.util.List;
 
 public interface BytedanceMiniAppMapper {
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<String> getEcpmKeys(@Param("startDate") Integer startDate,
                              @Param("endDate") Integer endDate,
                              @Param("startTime") String startTime,

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

@@ -9,11 +9,11 @@ import org.apache.ibatis.annotations.Param;
 import java.util.List;
 
 public interface BytedanceProductReportMapper {
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<JSONObject> queryCostAccount(@Param("advertiserId") Long advertiserId, @Param("startDate") String startDate, @Param("endDate") String endDate);
 
     void replaceProductReportDailies(List<BytedanceProductReportDaily> reports);
-    @DS("selectdb")
+    @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,18 +14,18 @@ public interface BytedancePromoterReportMapper {
     List<JSONObject> getAccountList();
 
     void replacePromoterReportHour(@Param("list") List<BytedancePromoterReportHour> hours);
-    @DS("selectdb")
+    @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")
+    @DS("sd_group_commit")
     void replacePromoterReportDailiesSD(@Param("list") List<BytedancePromoterReportDaily> promoterReportDailies);
 
     void replaceErrorInfo(BytedanceErrorLog errorInfo);
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<JSONObject> getListBydate(@Param("date") Integer date);
 
     List<JSONObject> getAllList();

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

@@ -13,10 +13,10 @@ import java.util.List;
 public interface BytedanceXingTuMapper {
     List<JSONObject> queryXingTuAccount(@Param("advertiserId") Long advertiserId);
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<JSONObject> queryXingTuActiveTask(@Param("advertiserId") Long advertiserId);
 
-    //TODO 修改为从selectdb读取 @DS("selectdb")
+    @DS("selectdb")
     List<JSONObject> queryXingTuReportTask(@Param("advertiserId") Long advertiserId);
 
     List<JSONObject> queryXingTuRetryJob();
@@ -30,11 +30,11 @@ public interface BytedanceXingTuMapper {
     void updateTaskInfoSD(BytedanceXingTuTask task);
 
     void insertReport(List<BytedanceXingTuReport> reports);
-    @DS("selectdb")
+    @DS("sd_group_commit")
     void insertReportSD(List<BytedanceXingTuReport> reports);
 
     void insertReportDaily(List<BytedanceXingTuReportDaily> dailyReports);
-    @DS("selectdb")
+    @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

@@ -1,11 +1,19 @@
 package cn.com.ctop.job.bytedance.data.mapper;
 
+import com.alibaba.fastjson.JSONObject;
+import com.baomidou.dynamic.datasource.annotation.DS;
 import org.apache.ibatis.annotations.Param;
 
+import java.util.List;
+
 public interface CPSReportMapper {
 
-    boolean cleanCPSPayDailyReports(@Param("nowDate") Integer nowDate);
+    boolean cleanCPSPayDailyReports(@Param("nowDate") String nowDate);
+
+    @DS("selectdb")
+    List<JSONObject> queryCPSIaaDaily(@Param("nowDate") Integer nowDate, @Param("dateStr") String dateStr);
+
+    boolean cleanCPSIaaDailyReportsDS(@Param("list") List<JSONObject> list);
+
 
-    //TODO 修改为从selectdb读取
-    boolean cleanCPSIaaDailyReports(@Param("nowDate") Integer nowDate);
 }

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

@@ -7,7 +7,6 @@ import org.apache.ibatis.annotations.Param;
 import java.util.List;
 import java.util.Map;
 
-//TODO 修改为从selectdb读取
 
 @DS("selectdb")
 public interface ReportsAdPushMapper {

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/AdminAdvertiserListMapper.xml

@@ -47,8 +47,8 @@
                     WHERE stat_time_day >= 20250601
                       AND stat_time_day <= 20250603
                     GROUP BY advertiser_id) t1
-                       LEFT JOIN deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
-                       LEFT JOIN deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id) t
+                       LEFT JOIN tidb.deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
+                       LEFT JOIN tidb.deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id) t
         WHERE t.advertiserId IS NOT NULL
 
     </select>

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
@@ -377,7 +377,6 @@
         SELECT DISTINCT advertiser_id
         FROM media_api.${db}
         where stat_time_day = #{date}
-        order by create_time
     </select>
 
     <select id="queryMediaAdverIds" resultType="Long">
@@ -422,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=",">
@@ -433,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/BytedanceProductReportMapper.xml

@@ -11,8 +11,8 @@
             </if>
             group by 1
         ) t1
-        LEFT JOIN deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
-        LEFT JOIN deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id
+        LEFT JOIN tidb.deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
+        LEFT JOIN tidb.deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id
     </select>
 
     <insert id="replaceProductReportDailies">

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

@@ -20,8 +20,8 @@
               FROM media_api.bytedance_promoter_report_hour
               WHERE stat_time_day = date_format(CURDATE() - INTERVAL 1 DAY, '%Y%m%d')
               GROUP BY advertiser_id) t1
-                 LEFT JOIN deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
-                 LEFT JOIN deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id
+                 LEFT JOIN tidb.deepthink.bytedance_advertiser_list t2 ON t1.advertiser_id = t2.advertiser_id
+                 LEFT JOIN tidb.deepthink.bytedance_auth_list t3 ON t2.admin_advertiser_id = t3.admin_advertiser_id
     </select>
     <select id="getAllList" resultType="com.alibaba.fastjson.JSONObject">
 

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

@@ -207,9 +207,9 @@
     <select id="queryXingTuActiveTask" resultType="com.alibaba.fastjson.JSONObject">
         select t.advertiser_id,challenge_task_id task_id,access_token
         from media_api.bytedance_xingtu_task_list t
-        join deepthink.bytedance_xingtu_advertiser_list x
+        join tidb.deepthink.bytedance_xingtu_advertiser_list x
         on t.advertiser_id = x.advertiser_id
-        join deepthink.bytedance_auth_list a
+        join tidb.deepthink.bytedance_auth_list a
         on x.admin_advertiser_id = a.admin_advertiser_id
         where (t.om_task_status is null or t.om_task_status not in (6,7,8))
         <if test="advertiserId != null">
@@ -220,9 +220,9 @@
     <select id="queryXingTuReportTask" resultType="com.alibaba.fastjson.JSONObject">
         select t.advertiser_id,challenge_task_id task_id,access_token
         from media_api.bytedance_xingtu_task_list t
-        join deepthink.bytedance_xingtu_advertiser_list x
+        join tidb.deepthink.bytedance_xingtu_advertiser_list x
         on t.advertiser_id = x.advertiser_id
-        join deepthink.bytedance_auth_list a
+        join tidb.deepthink.bytedance_auth_list a
         on x.admin_advertiser_id = a.admin_advertiser_id
         where (t.om_task_status not in (1,2,4) or t.om_task_status is null)
         and (t.end_time is null or t.end_time >= unix_timestamp(date_add(now(),Interval -65 day)))

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,35 +163,8 @@
         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,
+    <select id="queryCPSIaaDaily" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT IFNULL(t.date, t0.date) AS 'date', tt.official_account_id,
                tt.responsibility_person_id,
                tt.app_id,
                tt.id,
@@ -215,55 +186,101 @@
                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
+        FROM (SELECT t0.id, t0.responsibility_person_id, t0.official_account_id, t2.app_id
+              FROM tidb.thirtpart.ly_link t0
+                       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 = #{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
+                                                      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', ROUND( IFNULL(
+                        sum(IF(SUBSTR(dying_time, 1, 10) = SUBSTR(event_time, 1, 10), cost / 100000, 0)),
+                        0.00), 2 ) 'new_iaa_amount', count( DISTINCT uid ) 'iaa_users', sum( 1 ) 'iaa_counts', ROUND( IFNULL(sum(cost / 100000), 0.00), 2 ) 'iaa_amount'
+                            FROM media_api.bytedance_webhook_ma_ecpm t
+                                     LEFT JOIN tidb.thirtpart.ly_link t0 ON t.link_id = t0.id
+                            WHERE t.stat_date = #{nowDate}
+                            GROUP BY t.link_id, t.stat_date
+                            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,
+                                   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
+                            FROM tidb.thirtpart.ly_user_activation_info t
+                                     LEFT JOIN tidb.thirtpart.ly_link t0 ON t.link_id = t0.id
+                            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,
+                                   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 tidb.thirtpart.cps_link_data_report
+                            WHERE stat_date = #{nowDate}) t3 ON t3.link_id = tt.id
+        WHERE t0.iaa_amount IS NOT NULL
+    </select>
+
+    <insert id="cleanCPSIaaDailyReportsDS">
+        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)
+        values
+        <foreach collection="list" item="item" separator=",">
+            (
+            #{item.date},
+            #{item.official_account_id},
+            #{item.responsibility_person_id},
+            #{item.app_id},
+            #{item.id},
+            #{item.new_users},
+            #{item.new_pay_users},
+            #{item.new_pay_counts},
+            #{item.new_order_counts},
+            #{item.new_pay_amount},
+            #{item.new_xn_pay_amount},
+            #{item.pay_amount},
+            #{item.xn_pay_amount},
+            #{item.pay_users},
+            #{item.pay_counts},
+            #{item.order_counts},
+            #{item.new_iaa_users},
+            #{item.new_iaa_counts},
+            #{item.new_iaa_amount},
+            #{item.iaa_users},
+            #{item.iaa_counts},
+            #{item.iaa_amount},
+            #{item.active_users}
+            )
+        </foreach>
     </insert>
 
 </mapper>

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

@@ -10,8 +10,11 @@
 
 
     <select id="refreshList" resultType="com.alibaba.fastjson.JSONObject">
-        select tb1.task_id, tb1.advertiser_id, tb1.promoter_id, tb1.link_uuid, tb4.expire_time, tb2.media_type, tb3.app_plat_type,
-        tb3.movie_id as video_id, tb5.id as video_detail_id, tb1.promoter_uri, tb2.app_id, tb6.id as applet_id
+        select tb1.task_id, tb1.advertiser_id, tb1.promoter_id, tb1.link_uuid, tb4.expire_time,
+        tb2.media_type, tb3.app_plat_type,
+        tb3.movie_id as video_id, tb5.id as video_detail_id,
+        tb1.promoter_uri, tb2.app_id, tb6.id as applet_id,
+        tb5.platform_config, tb5.sort, tb7.album_id
         from
         (select *
         from deepthink.bytedance_advertiser_task_link
@@ -26,6 +29,8 @@
         on tb3.movie_id = tb5.video_id and tb3.entry_chapter_count = tb5.sort
         left join thirtpart.ly_applet tb6
         on tb2.app_id = tb6.app_id
+        left join thirtpart.ly_third_videos tb7
+        on tb3.movie_id = tb7.video_id and platform=2
         where tb4.expire_time &lt;= ${expireTime}
     </select>
 </mapper>

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

@@ -449,7 +449,7 @@
         on a.customer_user_name2=b.nick_name
         left join tidb.deepthink.sys_dept c
         on b.dept_id=c.dept_id
-        where stat_datetime=${date} and b.dept_id=${deptId}
+        where stat_datetime=${date} and b.dept_id=${deptId} and b.status=0 and b.del_flag=0
     </select>
     <select id="allBusinessPushData" resultType="java.util.Map">
         select

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

@@ -1,31 +1,33 @@
 package cn.com.ctop.job.bytedance.data.service.impl;
 
-import cn.com.ctop.job.bytedance.data.entity.BytedanceAdvertiserReportDaily;
-import cn.com.ctop.job.bytedance.data.entity.BytedanceAdvertiserReportHour;
-import cn.com.ctop.job.bytedance.data.entity.BytedanceErrorLog;
+import cn.com.ctop.job.bytedance.data.entity.*;
 import cn.com.ctop.job.bytedance.data.mapper.BytedanceAccountReportMapper;
 import cn.com.ctop.job.bytedance.data.service.IBytedanceAccountReportService;
 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;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.scheduling.annotation.EnableScheduling;
 import org.springframework.stereotype.Service;
 
-import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 
 @Service
 @Slf4j
+@EnableScheduling
 public class BytedanceAccountReportServiceImpl implements IBytedanceAccountReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceAccountReportMapper accountReportMapper;
 
+
     @Override
     public void getBytedanceRetryAccountHourReport(Long errorId, Long advertiserId, String accessToken, String startDate, String endDate, Integer page, Integer isSend) {
         try {
@@ -68,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);
+
                         }
                     }
                 }
@@ -78,15 +83,11 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
-                try {
-                    accountReportMapper.replaceAccountReportHourSD(hours);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportHourSD",e);
-                }
+                //accountReportMapper.replaceAccountReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
-                    Thread.sleep(200L);
+                    Thread.sleep(100L);
                 } catch (InterruptedException e) {
                     throw new RuntimeException(e);
                 }
@@ -143,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);
                         }
                     }
                 }
@@ -153,11 +156,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                try {
-                    accountReportMapper.replaceAccountReportDailySD(dailyList);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportDailySD",e);
-                }
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -170,7 +169,7 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                 log.info("账户ID：{}，日期:{},广告时报（账户粒度）数据请求成功", advertiserId, startDate);
                 accountReportMapper.deleteInfoByErrorId(errorId);
                 if (isSend == 1) {
-                    AlertUtils.sendMessage(advertiserId, startDate, endDate, 1, "hour", 1,"account");
+                    AlertUtils.sendMessage(advertiserId, startDate, endDate, 1, "hour", 1, "account");
                 }
             }
         } catch (Exception e) {
@@ -228,6 +227,8 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                             advertiserReportHour.setStatTimeDay(statTimeDay);
                             advertiserReportHour.setHour(hour);
                             hours.add(advertiserReportHour);
+
+                            dorisLoadManager.add("media_api","bytedance_advertiser_report_hour", advertiserReportHour);
                         }
                     } else {
                         log.info("账户id:{}，日期:{}广告时报（账户粒度）请求结果为空", advertiserId, startDate);
@@ -243,17 +244,14 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                     errorInfo.setEndDate(endDate);
                     errorInfo.setDataDimension("account");
                     errorInfo.setDataType("hour");
+                    errorInfo.setMsg(result.getString("message"));
                     accountReportMapper.replaceErrorInfo(errorInfo);
                 }
                 return;
             }
             if (!Check.isNull(hours)) {
                 accountReportMapper.replaceAccountReportHour(hours);
-                try {
-                    accountReportMapper.replaceAccountReportHourSD(hours);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportHourSD",e);
-                }
+                //accountReportMapper.replaceAccountReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -318,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);
@@ -333,17 +333,14 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
                     errorInfo.setEndDate(endDate);
                     errorInfo.setDataDimension("account");
                     errorInfo.setDataType("daily");
+                    errorInfo.setMsg(result.getString("message"));
                     accountReportMapper.replaceErrorInfo(errorInfo);
                 }
                 return;
             }
             if (!Check.isNull(dailyList)) {
                 accountReportMapper.replaceAccountReportDaily(dailyList);
-                try {
-                    accountReportMapper.replaceAccountReportDailySD(dailyList);
-                }catch (Exception e) {
-                    log.error("replaceAccountReportDailySD",e);
-                }
+                //accountReportMapper.replaceAccountReportDailySD(dailyList);
             }
             if (dailyList.size() >= 100) {
                 try {
@@ -351,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);
             }
@@ -359,4 +356,5 @@ public class BytedanceAccountReportServiceImpl implements IBytedanceAccountRepor
             e.printStackTrace();
         }
     }
+
 }

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;
@@ -12,10 +13,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业务层处理
@@ -32,6 +30,9 @@ import java.util.Map;
 @Service
 public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedanceCdpProjectMapper bytedanceCdpProjectMapper;
 
     @Autowired
@@ -53,7 +54,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
     }
 
     @Override
-    public List<Long> queryPageAdvertiserIds() {
+    public List<JSONObject> queryPageAdvertiserIds() {
         return bytedanceCdpProjectMapper.queryPageAdvertiserIds();
     }
 
@@ -112,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);
                     }
@@ -204,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);
                     }
@@ -329,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);
                     }
@@ -397,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 {
@@ -438,14 +434,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 +450,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 +460,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);
@@ -503,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);
                     }
@@ -518,26 +521,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 +548,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 +563,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)) {
@@ -579,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);
                     }
@@ -594,14 +601,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);
         }
     }
 
@@ -680,7 +687,11 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                             object.put("mini_program_id", wechatMiniProgramSpec.getString("mini_program_id"));
                             object.put("mini_program_path", wechatMiniProgramSpec.getString("mini_program_path"));
                         }
-                        JSONObject officialSpec = pageSpec.getJSONObject("wechat_canvas_spec");
+                        JSONObject wechatCanvasSpec = pageSpec.getJSONObject("wechat_canvas_spec");
+                        if (!Check.isNull(wechatCanvasSpec)) {
+                            object.put("page_id", wechatCanvasSpec.getLong("page_id"));
+                        }
+                        JSONObject officialSpec = pageSpec.getJSONObject("official_spec");
                         if (!Check.isNull(officialSpec)) {
                             object.put("page_id", officialSpec.getLong("page_id"));
                         }

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/BytedanceProductUpdateServiceImpl.java

@@ -35,7 +35,7 @@ public class BytedanceProductUpdateServiceImpl implements IBytedanceProductUpdat
     public void run() throws Exception {
         Set<String> errors = new HashSet<>();
         log.info("付费商品库更新中...");
-        iapProductUpdate(errors);
+//        iapProductUpdate(errors);
         log.info("付费商品库更新完成");
         log.info("免费商品库更新中...");
 //        iaaProductUpdate(errors);
@@ -104,7 +104,7 @@ public class BytedanceProductUpdateServiceImpl implements IBytedanceProductUpdat
         List<JSONObject> iapPlatformList = bytedanceProductUpdateMapper.getIAPPlatformList();
         List<JSONObject> iaaPlatformList = bytedanceProductUpdateMapper.getIAAPlatformList();
         initNew(iaaPlatformList);
-        initNew(iapPlatformList);
+//        initNew(iapPlatformList);
         authMovieResource();
     }
 

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

@@ -8,6 +8,7 @@ import cn.com.ctop.job.bytedance.utils.FeishuRobotUtil;
 import com.alibaba.fastjson.JSON;
 import com.alibaba.fastjson.JSONObject;
 import io.netty.util.internal.StringUtil;
+import lombok.extern.slf4j.Slf4j;
 import org.apache.commons.lang.StringUtils;
 import org.springframework.data.redis.core.RedisTemplate;
 import org.springframework.stereotype.Service;
@@ -19,6 +20,7 @@ import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
+@Slf4j
 @Service
 public class BytedancePromoteRefreshServiceImpl implements IBytedancePromoteRefreshService {
     @Resource
@@ -134,6 +136,20 @@ public class BytedancePromoteRefreshServiceImpl implements IBytedancePromoteRefr
                 }
                 String token = String.valueOf(tokenobj);
 
+                String serialNo = linkInfo.getString("sort");
+                String albumId = linkInfo.getString("album_id");
+                String episodeId = "";
+                try {
+                    String platform_config = linkInfo.getString("platform_config");
+                    if (!Check.isNull(platform_config)){
+                        JSONObject platformConfigJson = JSON.parseObject(platform_config);
+                        if (!platformConfigJson.containsKey("douyinyun")){
+                        }
+                        episodeId = platformConfigJson.getJSONObject("douyinyun").getString("episode_id");
+                    }
+                }catch (Exception e){
+                    log.error("解析平台配置失败: " + e.getMessage(), e);
+                }
                 //douyin
                 String path = promoterUri.split("\\?")[0];
                 String url = "https://open.douyin.com/api/apps/v1/url_link/generate/";
@@ -141,8 +157,12 @@ public class BytedancePromoteRefreshServiceImpl implements IBytedancePromoteRefr
                 query.put("link_id", String.valueOf(linkId));
                 query.put("video_id", String.valueOf(videoId));
                 query.put("video_detail_id", String.valueOf(videoDetailId));
-                query.put("type", String.valueOf(mediaType));
+                query.put("type", "1");
                 query.put("source", "link");
+                query.put("serialNo", serialNo);
+                query.put("app_id", appId);
+                query.put("tt_album_id", albumId);
+                query.put("tt_episode_id", episodeId);
                 Map<String, Object> params = new HashMap<>();
                 params.put("app_id", appId);
                 params.put("app_name", "douyin");

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

@@ -9,21 +9,27 @@ 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;
 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 java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
+import javax.annotation.PreDestroy;
+import java.text.DecimalFormat;
+import java.util.*;
 
 @Service
 @Slf4j
+@EnableScheduling
 public class BytedancePromoterReportServiceImpl implements IBytedancePromoterReportService {
     @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
 
     @Override
@@ -81,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);
                         }
                     }
                 }
@@ -91,11 +99,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                try {
-                    promoterReportMapper.replacePromoterReportHourSD(hours);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //promoterReportMapper.replacePromoterReportHourSD(hours);
             }
             if (hours.size() >= 100) {
                 try {
@@ -165,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);
                         }
                     }
                 }
@@ -175,11 +180,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                try {
-                    promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
@@ -262,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);
                         }
                     }
                 }
@@ -281,11 +284,7 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
             }
             if (!Check.isNull(hours)) {
                 promoterReportMapper.replacePromoterReportHour(hours);
-                try {
-                    promoterReportMapper.replacePromoterReportHourSD(hours);
-                }catch (Exception e) {
-                    e.printStackTrace();
-                }
+                //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);
                         }
                     }
                 }
@@ -371,25 +372,17 @@ public class BytedancePromoterReportServiceImpl implements IBytedancePromoterRep
                     errorInfo.setEndDate(endDate);
                     errorInfo.setDataDimension("promoter");
                     errorInfo.setDataType("daily");
+                    errorInfo.setMsg(result.getString("message"));
                     promoterReportMapper.replaceErrorInfo(errorInfo);
                 }
                 return;
             }
             if (!Check.isNull(promoterReportDailies)) {
                 promoterReportMapper.replacePromoterReportDailies(promoterReportDailies);
-                try {
-                    promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
-                }catch (Exception e) {
-                    log.error("",e);
-                }
+                //promoterReportMapper.replacePromoterReportDailiesSD(promoterReportDailies);
                 log.info("账户ID：{}，广告日报（最细粒度）数据请求成功", advertiserId);
             }
             if (promoterReportDailies.size() >= 100) {
-                try {
-                    Thread.sleep(200L);
-                } catch (InterruptedException e) {
-                    throw new RuntimeException(e);
-                }
                 getBytedancePromoterDaily(advertiserId, accessToken, startDate, endDate, page + 1, 1);
             } else {
                 log.info("账户ID：{}，开始日期:{},结束日期:{},广告日报（最细粒度）数据请求成功", advertiserId, startDate, endDate);

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/CPSReportServiceImpl.java

@@ -13,14 +13,14 @@ 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.cleanCPSIaaDailyReports(nowDate))
-                    log.info("CPS数据清洗完成-:{}", nowDate);
+                if (cpsReportMapper.cleanCPSIaaDailyReportsDS(cpsReportMapper.queryCPSIaaDaily(Integer.valueOf(nowDate), nowDate)))
+                    log.info("CPS数据清洗完成:{}", nowDate);
         } catch (Exception e) {
             e.printStackTrace();
-            log.error("CPS数据清洗完成");
+            log.error("CPS数据清洗异常");
         }
     }
 }

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

@@ -163,7 +163,6 @@ public class TotalRankPushServiceImpl implements ITotalRankPushService {
 
     private void sendToFeishu(String token, String content, String receive_id) {
 
-        System.out.println(content);
         JSONObject request = new JSONObject();
         request.put("receive_id", receive_id);
         request.put("msg_type", "interactive");

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

@@ -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-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/BytedancePromoterReportJob.java

@@ -34,7 +34,7 @@ public class BytedancePromoterReportJob {
     @Autowired
     private IChannelService channelService;
 
-    private static ExecutorService dailyExecutorService = Executors.newFixedThreadPool(4);
+    private static ExecutorService dailyExecutorService = Executors.newFixedThreadPool(10);
     private static ExecutorService hourExecutorService = Executors.newFixedThreadPool(4);
 
     private static ExecutorService dailyRetrExecutorService = Executors.newFixedThreadPool(2);

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

@@ -54,7 +54,8 @@ public class BytedanceReportRetryJob {
     private static ExecutorService productDailyRetrExecutor = Executors.newFixedThreadPool(2);
 
     private static ExecutorService flowDailyRetrExecutor = Executors.newFixedThreadPool(2);
-
+    private static ExecutorService changduExecutorService = Executors.newFixedThreadPool(1);
+    private static ExecutorService xingTuExecutorService = Executors.newFixedThreadPool(2);
 
     @XxlJob("retryReport")
     public void retryReport() throws Exception {
@@ -144,7 +145,6 @@ public class BytedanceReportRetryJob {
 
         List<JSONObject> changDuRetry = channelService.getChangDuRetry();
         if (!Check.isNull(changDuRetry)) {
-            ExecutorService executorService = Executors.newFixedThreadPool(1);
             for (int i = 0; i < changDuRetry.size(); i++) {
                 JSONObject jsonObject = changDuRetry.get(i);
                 Long id = jsonObject.getLong("id");
@@ -158,7 +158,7 @@ public class BytedanceReportRetryJob {
                 String appId = jsonObject.getString("channelAppId");
                 String appName = jsonObject.getString("channelAppName");
                 if ("changdu_order".equals(dataDimension)) {
-                    executorService.submit(new Runnable() {
+                    changduExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
                             channelService.getOrderList(startDate, endDate, appId, appName, channelAccountId, secretKey, 1, id, isSend);
@@ -172,7 +172,6 @@ public class BytedanceReportRetryJob {
 
         List<JSONObject> maEcpmRetryJob = miniAppService.getMaEcpmRetryJob();
         if (!Check.isNull(maEcpmRetryJob)) {
-            ExecutorService executorService = Executors.newFixedThreadPool(1);
             for (int i = 0; i < maEcpmRetryJob.size(); i++) {
                 JSONObject jsonObject = maEcpmRetryJob.get(i);
                 Long id = jsonObject.getLong("id");
@@ -182,7 +181,7 @@ public class BytedanceReportRetryJob {
                 Integer isSend = jsonObject.getInteger("isSend");
                 if (dataDimension != null && dataDimension.startsWith("ma_ecpm:")) {
                     String appId = dataDimension.split(":")[1];
-                    executorService.submit(new Runnable() {
+                    changduExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
                             miniAppService.getMaEcpm(startDate, endDate, appId, 1, id, isSend);
@@ -196,7 +195,6 @@ public class BytedanceReportRetryJob {
 
         List<JSONObject> xingTuRetryJob = bytedanceXingTuService.getXingTuRetryJob();
         if (!Check.isNull(xingTuRetryJob)) {
-            ExecutorService executorService = Executors.newFixedThreadPool(2);
             for (int i = 0; i < xingTuRetryJob.size(); i++) {
                 JSONObject jsonObject = xingTuRetryJob.get(i);
                 Long id = jsonObject.getLong("id");
@@ -206,24 +204,24 @@ 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)) {
-                    executorService.submit(new Runnable() {
+                    xingTuExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
                             bytedanceXingTuService.getXingTuTaskList(advertiserId, accessToken, startDate, endDate, 1, id, null, isSend);
                         }
                     });
                 } else if ("xingtu_task_info".equalsIgnoreCase(dataDimension)) {
-                    executorService.submit(new Runnable() {
+                    Long taskId = jsonObject.getJSONObject("msg").getLong("taskId");
+                    xingTuExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
                             bytedanceXingTuService.getXingTuTaskInfo(advertiserId, taskId, accessToken, id, null, isSend);
                         }
                     });
                 } else if ("xingtu_task_report".equalsIgnoreCase(dataDimension)) {
-                    executorService.submit(new Runnable() {
+                    Long taskId = jsonObject.getJSONObject("msg").getLong("taskId");
+                    xingTuExecutorService.submit(new Runnable() {
                         @Override
                         public void run() {
                             bytedanceXingTuService.getXingTuTaskReport(advertiserId, taskId, accessToken, 1, id, null, isSend);

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

@@ -17,6 +17,7 @@ import java.util.concurrent.Executors;
 @Slf4j
 @Component
 public class BytedanceXingTuJob {
+    private static final ExecutorService executorService = Executors.newFixedThreadPool(4);
     @Autowired
     private IBytedanceXingTuService bytedanceXingTuService;
 
@@ -37,7 +38,6 @@ public class BytedanceXingTuJob {
 
     @XxlJob("xingTuTaskInfo")
     public void taskInfo() {
-        ExecutorService executorService = Executors.newFixedThreadPool(4);
         List<JSONObject> xingTuTasks = bytedanceXingTuService.getXingTuActiveTask(null);
         for (JSONObject taskJson : xingTuTasks) {
             Long advId = taskJson.getLong("advertiser_id");
@@ -55,7 +55,6 @@ public class BytedanceXingTuJob {
 
     @XxlJob("xingTuTaskReport")
     public void taskReport() {
-        ExecutorService executorService = Executors.newFixedThreadPool(4);
         List<JSONObject> xingTuTasks = bytedanceXingTuService.getXingTuReportTask(null);
         for (JSONObject taskJson : xingTuTasks) {
             Long advId = taskJson.getLong("advertiser_id");

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/handler/PromoteLinkRefreshJob.java

@@ -43,6 +43,7 @@ public class PromoteLinkRefreshJob {
                 try {
                     JSONObject response = bytedancePromoteRefreshService.refreshLink(obj);
                     if (response.getInteger("code") != 0) {
+                        log.error("刷新链接失败: " + response.getString("msg"));
                         return response.getString("msg");
                     }
                 } catch (Exception e) {
@@ -58,23 +59,23 @@ public class PromoteLinkRefreshJob {
             failMessages.add(failMessage);
         }
 
-        StringBuilder errorMsg = new StringBuilder();
-        for (Future<String> future : failMessages) {
-            try {
-                String msg = future.get();
-                if (!Check.isNull(msg)) {
-                    errorMsg.append(msg).append("\n");
-                }
-            } catch (Exception e) {
-                log.error("获取刷新链接结果时发生异常: " + e.getMessage(), e);
-            }
-        }
-        if (errorMsg.length() > 0) {
-            errorMsg.insert(0, "**批量刷新推广链任务报错**\n");
-            FeishuRobotUtil.sendTextToWebhook(errorMsg.toString(), HOOK_URL);
-        } else {
-            log.info("批量刷新推广链接任务完成，无错误信息。");
-        }
+//        StringBuilder errorMsg = new StringBuilder();
+//        for (Future<String> future : failMessages) {
+//            try {
+//                String msg = future.get();
+//                if (!Check.isNull(msg)) {
+//                    errorMsg.append(msg).append("\n");
+//                }
+//            } catch (Exception e) {
+//                log.error("获取刷新链接结果时发生异常: " + e.getMessage(), e);
+//            }
+//        }
+//        if (errorMsg.length() > 0) {
+//            errorMsg.insert(0, "**批量刷新推广链任务报错**\n");
+//            FeishuRobotUtil.sendTextToWebhook(errorMsg.toString(), HOOK_URL);
+//        } else {
+//            log.info("批量刷新推广链接任务完成，无错误信息。");
+//        }
     }
 
 

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

@@ -0,0 +1,49 @@
+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 DorisConfig {
+    @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/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/DorisMultiTableStreamLoadManager.java

@@ -0,0 +1,52 @@
+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;
+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(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");
+        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

@@ -1,6 +1,7 @@
 server:
   port: 8085
   ip: 127.0.0.1
+  shutdown: graceful   #IMMEDIATE立即停机（默认值）、GRACEFUL
   servlet:
     context-path: /
     compression:
@@ -84,11 +85,17 @@ 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:
-          url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false
+          url: jdbc:mysql://selectdb-cn-vym4at1qw04.selectdbfe.rds.aliyuncs.com:9030/media_api?characterEncoding=UTF-8&useUnicode=true&useSSL=false&sessionVariables=enable_unique_key_partial_update=true,enable_insert_strict=false
+          username: data
+          password: hr@2025..
+          driver-class-name: com.mysql.jdbc.Driver
+        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
           password: hr@2025..
           driver-class-name: com.mysql.jdbc.Driver
@@ -127,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: