cps_link_data_report 清洗切换数据源

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

@@ -1,11 +1,21 @@
 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);
 
-    //TODO 修改为从selectdb读取
     boolean cleanCPSIaaDailyReports(@Param("nowDate") Integer nowDate);
+
+    @DS("selectdb")
+    List<JSONObject> queryCPSIaaDaily(@Param("nowDate") Integer nowDate);
+
+    boolean cleanCPSIaaDailyReportsDS(@Param("list") List<JSONObject> list);
+
+
 }

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

@@ -266,4 +266,123 @@
         GROUP BY tt.id
     </insert>
 
+    <select id="queryCPSIaaDaily" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT IFNULL(t.date, t0.date) AS 'date', tt.official_account_id,
+               tt.responsibility_person_id,
+               tt.app_id,
+               tt.id,
+               t1.new_users,
+               t3.new_pay_users,
+               t3.new_pay_counts,
+               t3.new_order_counts,
+               t3.new_pay_amount,
+               t3.new_xn_pay_amount,
+               t3.pay_amount,
+               t3.xn_pay_amount,
+               t3.pay_users,
+               t3.pay_counts,
+               t3.order_counts,
+               t0.new_iaa_users,
+               t0.new_iaa_counts,
+               t0.new_iaa_amount,
+               t0.iaa_users,
+               t0.iaa_counts,
+               t0.iaa_amount,
+               t.active_users
+        FROM (SELECT t0.id, t0.responsibility_person_id, t0.official_account_id, t2.app_id
+              FROM 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 = #{nowDate} 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,
+                                   t0.responsibility_person_id,
+                                   count(DISTINCT t.uid) AS 'new_users'
+                            FROM tidb.thirtpart.ly_user_activation_info t
+                                     LEFT JOIN tidb.thirtpart.ly_link t0 ON t.link_id = t0.id
+                            WHERE DATE_FORMAT(t.activation_time, '%Y%m%d') = #{nowDate}
+                            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/service/impl/CPSReportServiceImpl.java

@@ -16,11 +16,11 @@ public class CPSReportServiceImpl implements ICPSReportService {
     public void cleanCPSDailyReports(Integer nowDate) {
         try {
             if (cpsReportMapper.cleanCPSPayDailyReports(nowDate))
-                if (cpsReportMapper.cleanCPSIaaDailyReports(nowDate))
-                    log.info("CPS数据清洗完成-:{}", nowDate);
+                if (cpsReportMapper.cleanCPSIaaDailyReportsDS(cpsReportMapper.queryCPSIaaDaily(nowDate)))
+                    log.info("CPS数据清洗完成:{}", nowDate);
         } catch (Exception e) {
             e.printStackTrace();
-            log.error("CPS数据清洗完成");
+            log.error("CPS数据清洗异常");
         }
     }
 }