cps IAA、IAP消耗数据清洗

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

@@ -0,0 +1,10 @@
+package cn.com.ctop.job.bytedance.data.mapper;
+
+import org.apache.ibatis.annotations.Param;
+
+public interface CPSReportMapper {
+
+    boolean cleanCPSPayDailyReports(@Param("nowDate") Integer nowDate);
+
+    boolean cleanCPSIaaDailyReports(@Param("nowDate") Integer nowDate);
+}

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

@@ -0,0 +1,140 @@
+<?xml version="1.0" encoding="UTF-8" ?>
+<!DOCTYPE mapper PUBLIC "-//mybatis.org//DTD Mapper 3.0//EN" "http://mybatis.org/dtd/mybatis-3-mapper.dtd" >
+<mapper namespace="cn.com.ctop.job.bytedance.data.mapper.CPSReportMapper">
+
+    <insert id="cleanCPSPayDailyReports">
+        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
+        )
+        SELECT t1.consume_date,
+               t0.official_account_id,
+               t0.responsibility_person_id,
+               t2.app_id,
+               t0.id,
+               IFNULL(t3.new_users, 0) as 'new_users',
+                count(DISTINCT case when t1.consume_date = DATE_FORMAT(t1.activation_time, '%Y%m%d') then t1.uid end ) as 'new_pay_users',
+                SUM(if(t1.consume_date = DATE_FORMAT(t1.activation_time, '%Y%m%d') AND t1.status = 1, 1,0)) as 'new_pay_counts',
+                SUM(if(t1.consume_date = DATE_FORMAT(t1.activation_time, '%Y%m%d'), 1,0)) as 'new_order_counts',
+                IFNULL(ROUND(SUM(if(t1.consume_date = DATE_FORMAT(t1.activation_time, '%Y%m%d'), t1.total_fee,0)),2),0.00) as 'new_pay_amount',
+                IFNULL(ROUND(SUM(IF(t1.consume_date = DATE_FORMAT(t1.activation_time, '%Y%m%d') and (t1.pay_merchant = 15 or pay_type = 20),total_fee,0)),2),0.00) as 'new_xn_pay_amount',
+                IFNULL(ROUND(SUM(t1.total_fee),2),0.00) as 'pay_amount',
+                IFNULL(ROUND(SUM(IF(t1.pay_merchant = 15 or pay_type = 20, total_fee, 0)),2),0.00) as 'xn_pay_amount',
+                count(DISTINCT t1.uid) as 'pay_users',
+                SUM(IF(t1.status = 1, 1, 0)) as 'pay_counts',
+                count(t1.order_no) as 'order_counts'
+        FROM thirtpart.ly_link t0
+                 LEFT JOIN thirtpart.ly_pay_record t1 ON t0.id = t1.link_id
+                 LEFT JOIN thirtpart.ly_official_accounts t2 ON t0.official_account_id = t2.id
+                 LEFT JOIN(
+                 SELECT
+                     DATE_FORMAT(t.activation_time, '%Y%m%d') AS 'date',
+                     t0.id,
+                     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, t0.official_account_id, t0.responsibility_person_id
+                   ) t3 ON t0.id = t3.id
+                          AND t0.official_account_id = t3.official_account_id
+                          AND t0.responsibility_person_id = t3.responsibility_person_id
+        WHERE consume_date = #{nowDate}
+          AND t0.id in (SELECT id FROM thirtpart.ly_link)
+        GROUP BY t0.id, t0.official_account_id, t0.responsibility_person_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
+                )
+        select
+            t.stat_date,
+            t0.official_account_id,
+            t0.responsibility_person_id,
+            t2.app_id,
+            t0.id,
+            t3.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,
+            order_counts,
+            count(distinct case when SUBSTR(register_time,1,10) = SUBSTR(event_time,1,10) then uid end) 'new_iaa_users',
+                sum(if(SUBSTR(register_time,1,10) = SUBSTR(event_time,1,10) ,1,0)) 'new_iaa_counts',
+                IFNULL(ROUND(sum(if(SUBSTR(register_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 t0.id = t.link_id
+                LEFT JOIN thirtpart.ly_official_accounts t2 ON t0.official_account_id = t2.id
+                LEFT JOIN (
+                SELECT 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
+                FROM thirtpart.cps_link_data_report WHERE
+                    stat_date = #{nowDate}
+            ) t3 ON t3.link_id = t.link_id and t0.official_account_id = t3.official_account_id and t0.responsibility_person_id = t3.responsibility_person_id
+        where t.stat_date = #{nowDate}
+          AND t.link_id in (
+            SELECT id FROM thirtpart.ly_link
+        )
+        GROUP BY t0.id,t0.official_account_id,t0.responsibility_person_id
+    </insert>
+
+</mapper>

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

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

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

@@ -0,0 +1,26 @@
+package cn.com.ctop.job.bytedance.data.service.impl;
+
+import cn.com.ctop.job.bytedance.data.mapper.CPSReportMapper;
+import cn.com.ctop.job.bytedance.data.service.ICPSReportService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Service;
+
+@Service
+@Slf4j
+public class CPSReportServiceImpl implements ICPSReportService {
+    @Autowired
+    private CPSReportMapper cpsReportMapper;
+
+    @Override
+    public void cleanCPSDailyReports(Integer nowDate) {
+        try {
+            if (cpsReportMapper.cleanCPSPayDailyReports(nowDate))
+                if (cpsReportMapper.cleanCPSIaaDailyReports(nowDate))
+                    log.info("CPS数据清洗完成-:{}", nowDate);
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.error("CPS数据清洗完成");
+        }
+    }
+}

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

@@ -0,0 +1,40 @@
+package cn.com.ctop.job.bytedance.handler;
+
+
+import cn.com.ctop.job.bytedance.data.service.ICPSReportService;
+import cn.com.ctop.job.bytedance.data.utils.DateUtils;
+import com.xxl.job.core.handler.annotation.XxlJob;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.stereotype.Component;
+
+
+/**
+ * thirtpart.cps_link_data_report
+ */
+@Slf4j
+@Component
+public class CPSReportJob {
+    @Autowired
+    private ICPSReportService cpsReportService;
+
+    /**
+     * 清洗CPS报表
+     */
+    @XxlJob("cleanCPSDailyReports")
+    public void cleanCPSDailyReports() throws Exception {
+        log.info("[cleanCPSDailyReports] 任务启动...");
+        try {
+            String nowDate = DateUtils.getNowDate("yyyyMMdd");
+            String yesterday = DateUtils.getNowAddDate("yyyyMMdd", -1);
+
+            cpsReportService.cleanCPSDailyReports(Integer.valueOf(nowDate));
+            Thread.sleep(5000L);
+            cpsReportService.cleanCPSDailyReports(Integer.valueOf(yesterday));
+        } catch (Exception e) {
+            e.printStackTrace();
+            log.error(e.getMessage());
+        }
+    }
+
+}