Merge branch 'master' into zhaotx

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/DianZhongConfig.java

@@ -4,16 +4,28 @@ import org.springframework.boot.context.properties.ConfigurationProperties;
 import org.springframework.context.annotation.Configuration;
 
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
 
 @Configuration
 @ConfigurationProperties(prefix = "dianzhong.client")
 public class DianZhongConfig {
     private static List<DZConfig> dzConfig;
+    private static final Map<String, String> tokenMap = new ConcurrentHashMap<>();
 
     public List<DZConfig> getDzConfig() {
         return dzConfig;
     }
 
+    public String getToken(String clientId) {
+        if (tokenMap.isEmpty()) {
+            for (DZConfig config : dzConfig) {
+                tokenMap.put(config.getClientId(), config.getToken());
+            }
+        }
+        return tokenMap.get(clientId);
+    }
+
     public void setDzConfig(List<DZConfig> dzConfig) {
         DianZhongConfig.dzConfig = dzConfig;
     }
@@ -21,6 +33,7 @@ public class DianZhongConfig {
     public static class DZConfig {
         private String clientId;
         private String token;
+        private String type;
 
         public String getClientId() {
             return clientId;
@@ -37,5 +50,13 @@ public class DianZhongConfig {
         public void setToken(String token) {
             this.token = token;
         }
+
+        public String getType() {
+            return type;
+        }
+
+        public void setType(String type) {
+            this.type = type;
+        }
     }
 }

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/RedisCacheConfig.java

@@ -0,0 +1,120 @@
+package cn.com.ctop.job.bytedance.config;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.annotation.PropertyAccessor;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.SerializationFeature;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.fasterxml.jackson.datatype.jsr310.deser.LocalDateTimeDeserializer;
+import com.fasterxml.jackson.datatype.jsr310.ser.LocalDateTimeSerializer;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.RequiredArgsConstructor;
+import org.springframework.cache.CacheManager;
+import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.data.redis.cache.RedisCacheConfiguration;
+import org.springframework.data.redis.cache.RedisCacheManager;
+import org.springframework.data.redis.cache.RedisCacheWriter;
+import org.springframework.data.redis.connection.RedisConnectionFactory;
+import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
+import org.springframework.data.redis.serializer.RedisSerializationContext;
+import org.springframework.data.redis.serializer.StringRedisSerializer;
+
+import java.text.SimpleDateFormat;
+import java.time.Duration;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.*;
+
+@RequiredArgsConstructor
+@Configuration
+@EnableCaching
+public class RedisCacheConfig {
+
+    private static final long DEFAULT_DURATION = 5 * 60L;
+
+    private final static String DATE_TIME_FORMATTER = "yyyy-MM-dd HH:mm:ss.SSSXXX";
+
+    public static final List<CacheSpec> CACHE_SPEC_LIST = new ArrayList<>();
+
+
+    @Bean
+    public CacheManager cacheManager(RedisConnectionFactory factory) {
+        //对象的序列化
+        ObjectMapper objectMapper = getNewInstance();
+        objectMapper.enableDefaultTyping(ObjectMapper.DefaultTyping.NON_FINAL, JsonTypeInfo.As.PROPERTY);
+        Jackson2JsonRedisSerializer<Object> jackson2JsonRedisSerializer = new Jackson2JsonRedisSerializer<>(Object.class);
+        jackson2JsonRedisSerializer.setObjectMapper(objectMapper);
+
+        // 全局redis缓存过期时间，默认60秒
+        RedisCacheConfiguration defaultConfig = RedisCacheConfiguration.defaultCacheConfig()
+                .computePrefixWith(name -> name + ":")
+                .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(StringRedisSerializer.UTF_8))
+                .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
+                .entryTtl(Duration.ofSeconds(DEFAULT_DURATION))
+                .disableCachingNullValues();
+
+        Map<String, RedisCacheConfiguration> redisCacheConfigMap = new HashMap<>();
+        // 缓存N秒
+        CACHE_SPEC_LIST.add(new CacheSpec("getBytedanceAdvertiserByAd", 12 * 3600L));
+        CACHE_SPEC_LIST.forEach(cache -> {
+            long duration = DEFAULT_DURATION;
+            if (cache.getExpireTtl() != null && cache.getExpireTtl() > 0) {
+                duration = cache.getExpireTtl();
+            }
+            RedisCacheConfiguration config = RedisCacheConfiguration.defaultCacheConfig()
+                    // 默认name + 两个:,改为一个:
+                    .computePrefixWith(name -> name + ":")
+                    .serializeKeysWith(RedisSerializationContext.SerializationPair.fromSerializer(StringRedisSerializer.UTF_8))
+                    .serializeValuesWith(RedisSerializationContext.SerializationPair.fromSerializer(jackson2JsonRedisSerializer))
+                    .entryTtl(Duration.ofSeconds(duration));
+            redisCacheConfigMap.put(cache.getCacheName(), config);
+        });
+
+        return RedisCacheManager.builder(RedisCacheWriter.nonLockingRedisCacheWriter(factory))
+                .cacheDefaults(defaultConfig)
+                .withInitialCacheConfigurations(redisCacheConfigMap)
+                .build();
+    }
+
+    @Data
+    @AllArgsConstructor
+    public class CacheSpec {
+        private String cacheName;
+        private Long expireTtl;
+
+        /**
+         * default ttl 60s
+         *
+         * @param cacheName
+         */
+        public CacheSpec(String cacheName) {
+            this(cacheName, DEFAULT_DURATION);
+        }
+    }
+
+    private ObjectMapper getNewInstance() {
+        ObjectMapper objectMapper = new ObjectMapper();
+        objectMapper.setVisibility(PropertyAccessor.ALL, JsonAutoDetect.Visibility.ANY);
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        objectMapper.configure(SerializationFeature.WRITE_ENUMS_USING_TO_STRING, true);
+        objectMapper.configure(DeserializationFeature.READ_ENUMS_USING_TO_STRING, true);
+        objectMapper.configure(JsonParser.Feature.ALLOW_SINGLE_QUOTES, true);
+        objectMapper.disable(SerializationFeature.FAIL_ON_EMPTY_BEANS); // 忽略映射属性
+        //objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        objectMapper.setDateFormat(new SimpleDateFormat(DATE_TIME_FORMATTER));
+        // 序列化 LocalDateTime
+        JavaTimeModule timeModule = new JavaTimeModule();
+        DateTimeFormatter dateTimeFormatter = DateTimeFormatter.ofPattern("yyyy_MM_dd_HH_mm_ss", Locale.CHINA);
+        timeModule.addSerializer(LocalDateTime.class, new LocalDateTimeSerializer(dateTimeFormatter));
+        timeModule.addDeserializer(LocalDateTime.class, new LocalDateTimeDeserializer(dateTimeFormatter));
+        objectMapper.registerModule(timeModule);
+
+        return objectMapper;
+    }
+}

job-bytedance/src/main/java/cn/com/ctop/job/bytedance/config/RestTemplateConfig.java

@@ -0,0 +1,61 @@
+package cn.com.ctop.job.bytedance.config;
+
+import org.apache.http.NoHttpResponseException;
+import org.apache.http.client.HttpRequestRetryHandler;
+import org.apache.http.impl.client.CloseableHttpClient;
+import org.apache.http.impl.client.HttpClientBuilder;
+import org.apache.http.protocol.HttpContext;
+import org.springframework.context.annotation.Bean;
+import org.springframework.http.client.HttpComponentsClientHttpRequestFactory;
+import org.springframework.stereotype.Component;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.io.InterruptedIOException;
+import java.util.concurrent.TimeUnit;
+
+@Component
+public class RestTemplateConfig {
+
+
+    @Bean(name = "downloadFileRestTemplate")
+    public RestTemplate restTemplate(){
+        HttpComponentsClientHttpRequestFactory httpRequestFactory = new HttpComponentsClientHttpRequestFactory();
+        httpRequestFactory.setConnectionRequestTimeout(50000);
+        httpRequestFactory.setConnectTimeout(5000);
+        httpRequestFactory.setReadTimeout(300000);
+
+        CloseableHttpClient httpClient = HttpClientBuilder.create()
+                .setConnectionTimeToLive(300, TimeUnit.SECONDS)
+                .setRetryHandler(new HttpRequestRetryHandler() {
+                    @Override
+                    public boolean retryRequest(IOException exception, int executionCount, HttpContext context) {
+                        if (executionCount >= 3) {
+                            // 最多重试3次
+                            return false;
+                        } else if (exception instanceof NoHttpResponseException) {
+                            // 对于服务器未响应的情况进行重试
+                            requestWait();
+                            return true;
+                        } else if (exception instanceof InterruptedIOException) {
+                            // 对于连接中断的情况进行重试
+                            requestWait();
+                            return true;
+                        }
+                        return false;
+                    }
+                }).build();
+
+        httpRequestFactory.setHttpClient(httpClient);
+
+        return new RestTemplate(httpRequestFactory);
+    }
+
+    public void requestWait() {
+        try {
+            Thread.sleep(1000);
+        } catch (InterruptedException e) {
+            e.printStackTrace();
+        }
+    }
+}

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

@@ -1,12 +1,8 @@
 package cn.com.ctop.job.bytedance.controller;
 
 import cn.com.ctop.job.bytedance.config.DianZhongConfig;
-import cn.com.ctop.job.bytedance.data.service.IBytedancePromoterReportService;
-import cn.com.ctop.job.bytedance.data.service.IChannelService;
 import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
-import cn.com.ctop.job.bytedance.data.utils.Check;
 import cn.com.ctop.job.bytedance.data.utils.DateUtils;
-import com.alibaba.fastjson.JSONObject;
 import io.swagger.annotations.Api;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -36,17 +32,20 @@ public class DianZhongController {
                             @RequestParam(required = false, defaultValue = "false") Boolean allDay) {
         List<DianZhongConfig.DZConfig> dzConfig = dianZhongConfig.getDzConfig();
         for (DianZhongConfig.DZConfig config : dzConfig) {
+            if ("iaa".equals(config.getType())) {
+                continue;
+            }
             String clientId = config.getClientId();
             String token = config.getToken();
             String queryStartTime = DateUtils.getQueryStartTime(statDate, "yyyy-MM-dd HH:mm:ss", allDay);
             String queryEndTime = DateUtils.getQueryEndTime(statDate, "yyyy-MM-dd HH:mm:ss", allDay);
-            System.out.println(type);
-            System.out.println(clientId);
-            System.out.println(token);
-            System.out.println(queryStartTime);
-            System.out.println(queryEndTime);
-            //dianZhongService.getDownloadUrl(type, clientId, token, statDate,queryStartTime,queryEndTime);
+            dianZhongService.getDownloadUrl(type, clientId, token, queryStartTime, queryEndTime, null, false);
         }
     }
 
+    @GetMapping("/getUrlData")
+    public void urlData(String statDate) {
+        dianZhongService.getUrlData(statDate);
+    }
+
 }

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

@@ -1,8 +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;
+import java.util.Map;
+
 public interface DianZhongMapper {
     void insertDownloadTask(@Param("item") JSONObject jsonObject);
+
+    List<JSONObject> getDZRetry();
+
+    List<JSONObject> getTaskUrl(@Param("statDate") String statDate);
+
+    void updateTaskDownloaded(@Param("taskId") String taskId);
+
+    @DS("selectdb")
+    Map<String, String> getBytedanceAdvertiserByAd(@Param("adId") String adId, @Param("statTimeTay") Integer statTimeTay);
 }

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

@@ -3,13 +3,40 @@
 <mapper namespace="cn.com.ctop.job.bytedance.data.mapper.DianZhongMapper">
 
     <insert id="insertDownloadTask">
-        replace into media_api.dianzhong_data_download_url
-        (stat_date,task_id,`type`,client_id) values
+        insert into media_api.dianzhong_data_download_url
+        (stat_date,task_id,`type`,client_id,start_time,end_time) values
         (
         #{item.stat_date},
         #{item.task_id},
         #{item.type},
-        #{item.client_id}
+        #{item.client_id},
+        #{item.start_time},
+        #{item.end_time}
         )
+        on duplicate key update
+        stat_date= VALUES(stat_date),`type`= VALUES(`type`),client_id= VALUES(client_id),
+        start_time= VALUES(start_time),end_time= VALUES(end_time)
     </insert>
+    <update id="updateTaskDownloaded">
+        update media_api.dianzhong_data_download_url
+        set downloaded = 1 where task_id = #{taskId}
+    </update>
+
+    <select id="getTaskUrl" resultType="com.alibaba.fastjson.JSONObject">
+        select task_id,url,type,client_id
+        from media_api.dianzhong_data_download_url
+        where stat_date= ${statDate} and downloaded = 0 and url is not null and url != '-1'
+    </select>
+
+    <select id="getDZRetry" resultType="com.alibaba.fastjson.JSONObject">
+        SELECT t1.is_send AS 'isSend', t1.id AS 'id', t1.advertiser_id AS 'advertiserId', t1.start_date AS 'startDate', t1.end_date AS 'endDate', t1.data_dimension AS 'dataDimension', t1.data_type AS 'dataType'
+        FROM bytedance_error_log t1
+        where data_dimension like 'dianzhong%'
+    </select>
+
+    <select id="getBytedanceAdvertiserByAd" resultType="java.util.Map">
+        SELECT cast(advertiser_id as char) advertiser_id,cast(cdp_project_id as char) a_id from media_api.bytedance_promoter_report_hour
+        where cdp_promotion_id =  ${adId} and stat_time_day >= #{statTimeTay} limit 1
+    </select>
+
 </mapper>

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

@@ -0,0 +1,8 @@
+package cn.com.ctop.job.bytedance.data.service;
+
+
+import java.util.Map;
+
+public interface ICacheService {
+    Map<String, String> getBytedanceAdvertiserByAd(String adId, Integer date);
+}

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

@@ -1,6 +1,12 @@
 package cn.com.ctop.job.bytedance.data.service;
 
+import com.alibaba.fastjson.JSONObject;
+
+import java.util.List;
+
 public interface IDianZhongService {
     void getDownloadUrl(String dataType, String clientId, String token,
-                        String statDay, String startTime, String endTime);
+                        String startTime, String endTime,Long errId,boolean isLog);
+    List<JSONObject> getDZRetry();
+    void getUrlData(String statDate);
 }

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

@@ -115,6 +115,7 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                     JSONArray array = data.getJSONArray("list");
                     for (int i = 0; i < array.size(); i++) {
                         JSONObject jsonObject = array.getJSONObject(i);
+                        jsonObject.put("promotion_list", jsonObject.toJSONString());
                         checkPromotionInfo(jsonObject, advertiserId);
                         dorisLoadManager.add("media_api", "bytedance_promotion_list", jsonObject);
                     }
@@ -155,6 +156,10 @@ public class BytedanceCdpProjectServiceImpl implements IBytedanceCdpProjectServi
                 entity.put("url", miniProgramInfo.getString("url"));
             }
         }
+        JSONObject nativeSetting = entity.getJSONObject("native_setting");
+        if (!Check.isNull(nativeSetting)) {
+            entity.put("aweme_id", nativeSetting.getString("aweme_id"));
+        }
     }
 
     @Override

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

@@ -0,0 +1,23 @@
+package cn.com.ctop.job.bytedance.data.service.impl;
+
+
+import cn.com.ctop.job.bytedance.data.mapper.DianZhongMapper;
+import cn.com.ctop.job.bytedance.data.service.ICacheService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cache.annotation.Cacheable;
+import org.springframework.stereotype.Service;
+
+import java.util.Map;
+
+@Service
+@Slf4j
+public class CacheServiceImpl implements ICacheService {
+    @Autowired
+    private DianZhongMapper dianZhongMapper;
+
+    @Cacheable(value = "getBytedanceAdvertiserByAd", key = "#adId", unless = "#result == null")
+    public Map<String, String> getBytedanceAdvertiserByAd(String adId, Integer date) {
+        return dianZhongMapper.getBytedanceAdvertiserByAd(adId, date);
+    }
+}

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

@@ -3,26 +3,73 @@ package cn.com.ctop.job.bytedance.data.service.impl;
 import cn.com.ctop.job.bytedance.data.entity.BytedanceErrorLog;
 import cn.com.ctop.job.bytedance.data.mapper.BytedancePromoterReportMapper;
 import cn.com.ctop.job.bytedance.data.mapper.DianZhongMapper;
+import cn.com.ctop.job.bytedance.data.service.ICacheService;
 import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
+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 cn.hutool.core.date.DateUtil;
+import cn.hutool.core.util.NumberUtil;
 import cn.hutool.crypto.SecureUtil;
 import com.alibaba.fastjson.JSONObject;
 import lombok.extern.slf4j.Slf4j;
+import org.apache.commons.lang3.StringUtils;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Service;
+import org.springframework.web.client.ResponseExtractor;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.BufferedReader;
+import java.io.InputStreamReader;
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
 
 @Service
 @Slf4j
 public class DianZhongServiceImpl implements IDianZhongService {
     @Autowired
+    private ICacheService cacheService;
+
+    @Autowired
+    private DorisMultiTableStreamLoadManager dorisLoadManager;
+
+    @Autowired
     private BytedancePromoterReportMapper promoterReportMapper;
 
     @Autowired
+    private RestTemplate downloadFileRestTemplate;
+
+    @Autowired
     private DianZhongMapper dianZhongMapper;
 
+    private static final Map<Integer, Integer> domain2Platform = new HashMap<Integer, Integer>() {{
+        put(14, 2);//抖小
+        put(16, 1);//微小
+        put(17, 3);//快小
+    }};
+
+    private static final Map<String, String> clientDomain2AppId = new HashMap<String, String>() {{
+        put("10003129" + "14", "dzdyIAP");//抖小
+    }};
+
+    private static final Map<String, String> media2Id = new HashMap<String, String>() {{
+        put("adq", "3");//腾讯广点通
+        put("dy", "1");//字节巨量
+        put("ks", "2");//快手-磁力
+    }};
+
+    public List<JSONObject> getDZRetry() {
+        return dianZhongMapper.getDZRetry();
+    }
+
+    /*
+    获取点众IAP数据下载url
+     */
     public void getDownloadUrl(String dataType, String clientId, String token,
-                                String statDay, String startTime, String endTime) {
+                               String startTime, String endTime, Long errId, boolean isLog) {
         long l = System.currentTimeMillis();
         JSONObject param = new JSONObject()
                 .fluentPut("clientId", clientId)
@@ -30,47 +77,250 @@ public class DianZhongServiceImpl implements IDianZhongService {
                 .fluentPut("edate", String.valueOf(DateUtil.parse(endTime, "yyyy-MM-dd HH:mm:ss").getTime() / 1000))
                 .fluentPut("timestamp", l)
                 .fluentPut("signKey", getSignKey(clientId, token, l));
+        if ("user".equals(dataType)) {
+            param.put("queryField", "loginTime");
+        }
         try {
             String url = "https://routine.wqxsw.com/flames/channel/query/" + dataType;
             String response = HttpUtils.httpPostRequest(url, param, null);
             JSONObject responseJson = JSONObject.parseObject(response);
-            System.out.println(response);
             if ("0".equals(responseJson.getString("retCode"))) {
                 String taskId = responseJson.getString("data");
                 JSONObject jsonObject = new JSONObject()
-                        .fluentPut("stat_date", statDay.replace("-", ""))
+                        .fluentPut("stat_date", DateUtils.transformFormat(endTime, "yyyy-MM-dd HH:mm:ss", "yyyyMMdd"))
                         .fluentPut("task_id", taskId)
                         .fluentPut("type", dataType)
-                        .fluentPut("client_id", clientId);
+                        .fluentPut("client_id", clientId)
+                        .fluentPut("start_time", startTime)
+                        .fluentPut("end_time", endTime);
                 dianZhongMapper.insertDownloadTask(jsonObject);
-                log.info("获取点众数据下载地址成功 {} {} {}~{} ", dataType, clientId, startTime, endTime);
+                log.info("获取点众{}数据下载地址成功 {} {}~{} taskId:{}", dataType, clientId, startTime, endTime, taskId);
+
+                if (errId != null) {
+                    promoterReportMapper.deleteInfoByErrorId(errId);
+                }
             } else {
-                log.error("获取点众数据下载地址错误 {} {} {}~{} msg:{}", dataType, clientId, startTime, endTime, responseJson.getString("retMsg"));
+                log.error("获取点众{}数据下载地址错误 {} {}~{} msg:{}", dataType, clientId, startTime, endTime, responseJson.getString("retMsg"));
+                if (isLog) {
+                    promoterReportMapper.replaceErrorInfo(
+                            BytedanceErrorLog.builder()
+                                    .msg(responseJson.getString("retMsg"))
+                                    .advertiserId(Long.valueOf(clientId))
+                                    .dataDimension("dianzhong_url")
+                                    .dataType(dataType)
+                                    .startDate(startTime)
+                                    .endDate(endTime)
+                                    .build());
+                }
+            }
+        } catch (Exception e) {
+            log.error("获取点众{}数据下载地址错误 {} {}~{} ", dataType, clientId, startTime, endTime, e);
+            if (isLog) {
                 promoterReportMapper.replaceErrorInfo(
                         BytedanceErrorLog.builder()
-                                .msg(responseJson.getString("retMsg"))
+                                .msg(e.getMessage())
                                 .advertiserId(Long.valueOf(clientId))
-                                .dataDimension("dianzhong")
+                                .dataDimension("dianzhong_url")
                                 .dataType(dataType)
                                 .startDate(startTime)
                                 .endDate(endTime)
                                 .build());
             }
-        } catch (Exception e) {
-            log.error("获取点众数据下载地址错误 {} {} {}~{} ", dataType, clientId, startTime, endTime, e);
-            promoterReportMapper.replaceErrorInfo(
-                    BytedanceErrorLog.builder()
-                            .msg(e.getMessage())
-                            .advertiserId(Long.valueOf(clientId))
-                            .dataDimension("dianzhong")
-                            .dataType(dataType)
-                            .startDate(startTime)
-                            .endDate(endTime)
-                            .build());
         }
     }
 
-    private static String getSignKey(String clientId, String token, long timeStamp) {
+    /*
+    下载点众IAP数据
+     */
+    public void getUrlData(String statDate) {
+        List<JSONObject> tasks = dianZhongMapper.getTaskUrl(statDate.replace("-", ""));
+        if (!Check.isNull(tasks)) {
+            for (JSONObject task : tasks) {
+                String taskId = task.getString("task_id");
+                String url = task.getString("url");
+                String type = task.getString("type");
+                String clientId = task.getString("client_id");
+
+                downloadFileRestTemplate.execute(url, org.springframework.http.HttpMethod.GET, null,
+                        (ResponseExtractor<Void>) response -> {
+                            try (BufferedReader reader = new BufferedReader(
+                                    new InputStreamReader(response.getBody(), StandardCharsets.UTF_8))) {
+                                String line;
+                                while ((line = reader.readLine()) != null) {
+                                    if ("user".equals(type)) {
+                                        JSONObject dzUser = getDZUser(line, clientId);
+                                        dorisLoadManager.add("hr_playlet", "user_event", dzUser);
+                                        JSONObject dzLogin = getDZLogin(line, clientId);
+                                        dorisLoadManager.add("hr_playlet", "server_event", dzLogin);
+                                    } else if ("order".equals(type)) {
+                                        JSONObject dzOrder = getDZOrder(line, clientId);
+                                        dorisLoadManager.add("hr_playlet", "server_event", dzOrder);
+                                    }
+                                }
+                            }
+                            return null;
+                        });
+                dianZhongMapper.updateTaskDownloaded(taskId);
+            }
+        }
+    }
+
+    private String getSignKey(String clientId, String token, long timeStamp) {
         return SecureUtil.md5(clientId + token + timeStamp).toLowerCase();
     }
+
+    private JSONObject getDZUser(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject userJson = new JSONObject();
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long ctime = jsonObject.getLong("ctime");
+        String statDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
+        String registerTime = DateUtils.formatDate(ctime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        userJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+        userJson.put("uid", jsonObject.getString("userId"));
+        userJson.put("stat_date", statDate);
+        userJson.put("register_time", registerTime);
+        userJson.put("created_at", ctime);
+        userJson.put("open_id", jsonObject.getString("openId"));
+        userJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        userJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        userJson.put("channel_id", jsonObject.getString("channelId"));
+        userJson.put("link_id", jsonObject.getString("referralId"));
+        userJson.put("expiration_time", jsonObject.getString("vipEndDate"));
+        userJson.put("pay_amount", jsonObject.getString("payAmount"));
+        userJson.put("pay_count", jsonObject.getString("payCount"));
+        userJson.put("last_login_time", jsonObject.getString("loginTime"));
+        userJson.put("ip", jsonObject.getString("ip"));
+        userJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        userJson.put("person_id", jsonObject.getString("kdrId"));
+        userJson.put("a_id", jsonObject.getString("projectId"));
+        userJson.put("ad_id", adId);
+        userJson.put("activation_time", activationTime);
+        userJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+
+        if (NumberUtil.isLong(adId) && !"0".equals(adId)) {
+            String anotherDay = DateUtils.getAnotherDay("yyyy-MM-dd", registerTime.substring(0, 10), -15);
+            Map<String, String> ad2Advertiser = cacheService.getBytedanceAdvertiserByAd(adId, DateUtils.getDateInteger(anotherDay));
+            if (!Check.isNullMap(ad2Advertiser)) {
+                String advertiserId = ad2Advertiser.get("advertiser_id");
+                userJson.put("advertiser_id", advertiserId);
+            }
+        }
+
+        return userJson;
+    }
+
+    private JSONObject getDZLogin(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject loginJson = new JSONObject();
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long loginTime = jsonObject.getLong("loginTime");
+        String statDate = DateUtils.formatDate(loginTime * 1000, "yyyyMMdd");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        String userId = jsonObject.getString("userId");
+        loginJson.put("stat_date", statDate);
+        loginJson.put("event", "LOGIN");
+        loginJson.put("event_pk", SecureUtil.md5(String.format("dianzhong_login_%s_%s", userId, loginTime)));
+        loginJson.put("event_tag", "dianzhong_login");
+        loginJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+
+        loginJson.put("uid", userId);
+        loginJson.put("open_id", jsonObject.getString("openId"));
+        loginJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        loginJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        loginJson.put("channel_id", jsonObject.getString("channelId"));
+        loginJson.put("link_id", jsonObject.getString("referralId"));
+        loginJson.put("register_date", jsonObject.getString("registerDate"));
+        loginJson.put("expiration_time", jsonObject.getString("vipEndDate"));
+        loginJson.put("last_login_time", jsonObject.getString("loginTime"));
+        loginJson.put("created_at", loginTime);
+        loginJson.put("ip", jsonObject.getString("ip"));
+        loginJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        loginJson.put("person_id", jsonObject.getString("kdrId"));
+        loginJson.put("a_id", jsonObject.getString("projectId"));
+        loginJson.put("ad_id", adId);
+        loginJson.put("activation_time", activationTime);
+        loginJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+
+        if (NumberUtil.isLong(adId) && !"0".equals(adId)) {
+            String anotherDay = DateUtils.getAnotherDay("yyyy-MM-dd", DateUtils.formatDate(loginTime * 1000, "yyyy-MM-dd"), -15);
+            Map<String, String> ad2Advertiser = cacheService.getBytedanceAdvertiserByAd(adId, DateUtils.getDateInteger(anotherDay));
+            if (!Check.isNullMap(ad2Advertiser)) {
+                String advertiserId = ad2Advertiser.get("advertiser_id");
+                loginJson.put("advertiser_id", advertiserId);
+            }
+        }
+        return loginJson;
+    }
+
+    private JSONObject getDZOrder(String line, String clientId) {
+        JSONObject jsonObject = JSONObject.parseObject(line);
+        JSONObject orderJson = new JSONObject();
+        String orderId = jsonObject.getString("id");
+        int domain = jsonObject.getIntValue("domain");
+        String appId = clientId + domain;
+        Long ctime = jsonObject.getLong("ctime");
+        String statDate = DateUtils.formatDate(ctime * 1000, "yyyyMMdd");
+        String adId = jsonObject.getString("adid");
+        String promotionId = jsonObject.getString("promotionId");
+        adId = StringUtils.isNotBlank(promotionId) ? promotionId : adId;
+        Long dyeTime = jsonObject.getLong("dyeTime");
+        String activationTime = DateUtils.formatDate(dyeTime * 1000, "yyyy-MM-dd HH:mm:ss");
+        String fromMedia = jsonObject.getString("from");
+        orderJson.put("stat_date", statDate);
+        orderJson.put("event", "PAY");
+        orderJson.put("event_pk", SecureUtil.md5("dianzhong_order_id:" + orderId));
+        orderJson.put("event_tag", "dianzhong_order");
+        orderJson.put("app_id", clientDomain2AppId.getOrDefault(appId, appId));
+        orderJson.put("order_no", orderId);
+        orderJson.put("platform_order_no", jsonObject.getString("outTradeNo"));
+        orderJson.put("total_fee", jsonObject.getString("discount"));
+        orderJson.put("type", jsonObject.getString("type"));
+        orderJson.put("status", jsonObject.getString("statusNotify"));
+        orderJson.put("created_at", ctime);
+        orderJson.put("pay_time", jsonObject.getLong("finishTime"));
+        orderJson.put("uid", jsonObject.getString("userId"));
+        orderJson.put("channel_id", jsonObject.getString("channelId"));
+        orderJson.put("platform", domain2Platform.getOrDefault(domain, domain));
+        orderJson.put("video_id", jsonObject.getString("sourceInfo"));
+        orderJson.put("video_detail_id", jsonObject.getString("chapterId"));
+        orderJson.put("video_name", jsonObject.getString("sourceDesc"));
+        orderJson.put("register_date", jsonObject.getString("registerDate"));
+        orderJson.put("open_id", jsonObject.getString("openId"));
+        orderJson.put("mobile_system", jsonObject.getIntValue("os") == 0 ? "Android" : "ios");
+        orderJson.put("link_id", jsonObject.getString("referralId"));
+        orderJson.put("a_id", jsonObject.getString("projectId"));
+        orderJson.put("ad_id", adId);
+        orderJson.put("person_id", jsonObject.getString("fromDrId") != null ? jsonObject.getString("fromDrId") : jsonObject.getString("kdrId"));
+        orderJson.put("scene", jsonObject.getString("scene"));
+        orderJson.put("official_account_open_id", jsonObject.getString("thirdWxId"));
+        orderJson.put("activation_time", activationTime);
+        orderJson.put("pay_type", jsonObject.getString("orderSubType"));
+        orderJson.put("media_id", media2Id.getOrDefault(fromMedia, fromMedia));
+        orderJson.put("money_benefit", jsonObject.getDouble("moneyBenefit"));
+        orderJson.put("params", jsonObject.toJSONString());
+
+        if (NumberUtil.isLong(adId) && !"0".equals(adId)) {
+            String anotherDay = DateUtils.getAnotherDay("yyyy-MM-dd", DateUtils.formatDate(ctime * 1000, "yyyy-MM-dd"), -15);
+            Map<String, String> ad2Advertiser = cacheService.getBytedanceAdvertiserByAd(adId, DateUtils.getDateInteger(anotherDay));
+            if (!Check.isNullMap(ad2Advertiser)) {
+                String advertiserId = ad2Advertiser.get("advertiser_id");
+                orderJson.put("advertiser_id", advertiserId);
+            }
+        }
+        return orderJson;
+    }
 }

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

@@ -221,7 +221,11 @@ public class DateUtils {
     }
 
     public static String formatDate(Long timestamp, String formatPattern) {
-        return formatDate(new Date(timestamp), formatPattern);
+        try {
+            return formatDate(new Date(timestamp), formatPattern);
+        }catch (Exception e) {
+            return null;
+        }
     }
 
     public static String formatDate(Date date, String formatPattern) {

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

@@ -1,6 +1,7 @@
 package cn.com.ctop.job.bytedance.handler;
 
 
+import cn.com.ctop.job.bytedance.config.DianZhongConfig;
 import cn.com.ctop.job.bytedance.data.enums.AccountTypeEnum;
 import cn.com.ctop.job.bytedance.data.service.*;
 import cn.com.ctop.job.bytedance.data.utils.Check;
@@ -10,6 +11,7 @@ import lombok.extern.slf4j.Slf4j;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 
+import javax.annotation.Resource;
 import java.util.List;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -47,6 +49,12 @@ public class BytedanceReportRetryJob {
     @Autowired
     private IBytedanceMaterialReportService bytedanceMaterialReportService;
 
+    @Autowired
+    private IDianZhongService dianZhongService;
+
+    @Resource
+    private DianZhongConfig dianZhongConfig;
+
 
     private static ExecutorService hourAccountRetrExecutorService = Executors.newFixedThreadPool(2);
     private static ExecutorService dailyAccountRetrExecutorService = Executors.newFixedThreadPool(2);
@@ -245,6 +253,32 @@ public class BytedanceReportRetryJob {
             log.info("巨量星图重试列表为空");
         }
 
+        List<JSONObject> dzRetry = dianZhongService.getDZRetry();
+        if (!Check.isNull(dzRetry)) {
+            for (int i = 0; i < dzRetry.size(); i++) {
+                JSONObject jsonObject = dzRetry.get(i);
+                Long id = jsonObject.getLong("id");
+                String clientId = jsonObject.getString("advertiserId");
+                String startDate = jsonObject.getString("startDate");
+                String endDate = jsonObject.getString("endDate");
+                String dataDimension = jsonObject.getString("dataDimension");
+                String dataType = jsonObject.getString("dataType");
+                String token = dianZhongConfig.getToken(clientId);
+                if ("dianzhong_url".equals(dataDimension)) {
+                    changduExecutorService.submit(new Runnable() {
+                        @Override
+                        public void run() {
+                            if (!Check.isNull(token)) {
+                                dianZhongService.getDownloadUrl(dataType, clientId, token, startDate, endDate, id, false);
+                            }
+                        }
+                    });
+                }
+            }
+        } else {
+            log.info("点众重试列表为空");
+        }
+
     }
 
 

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

@@ -0,0 +1,55 @@
+package cn.com.ctop.job.bytedance.handler;
+
+
+import cn.com.ctop.job.bytedance.config.DianZhongConfig;
+import cn.com.ctop.job.bytedance.data.service.IDianZhongService;
+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;
+
+import javax.annotation.Resource;
+import java.util.Calendar;
+import java.util.Date;
+import java.util.List;
+
+@Slf4j
+@Component
+public class DianZhongJob {
+    @Autowired
+    private IDianZhongService dianZhongService;
+
+    @Resource
+    private DianZhongConfig dianZhongConfig;
+
+    /**
+     * 获取点众IAP数据url
+     */
+    @XxlJob("dzDownloadUrl")
+    public void dzDownloadUrl() {
+        List<DianZhongConfig.DZConfig> dzConfig = dianZhongConfig.getDzConfig();
+        for (DianZhongConfig.DZConfig config : dzConfig) {
+            if ("iaa".equals(config.getType())) {
+                continue;
+            }
+            String statDate = DateUtils.dateAddByType(new Date(), -10, Calendar.MINUTE, "yyyy-MM-dd");
+            String clientId = config.getClientId();
+            String token = config.getToken();
+            String queryStartTime = DateUtils.getQueryStartTime(statDate, "yyyy-MM-dd HH:mm:ss", false);
+            String queryEndTime = DateUtils.getQueryEndTime(statDate, "yyyy-MM-dd HH:mm:ss", false);
+            dianZhongService.getDownloadUrl("user", clientId, token, queryStartTime, queryEndTime, null, true);
+            dianZhongService.getDownloadUrl("order", clientId, token, queryStartTime, queryEndTime, null, true);
+        }
+    }
+
+    /**
+     * 点众IAP数据下载
+     */
+    @XxlJob("dzUrlData")
+    public void dzUrlData() {
+        String statDate = DateUtils.dateAddByType(new Date(), -10, Calendar.MINUTE, "yyyy-MM-dd");
+        dianZhongService.getUrlData(statDate);
+    }
+
+}

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

@@ -101,4 +101,22 @@ public class ToolsSiteJob {
 
     }
 
+    private final String  KEY = "bb3648b0b93f2f82135a6dac38dc418f808346a434";
+
+    @XxlJob("regularUpdatesState")
+    public void regularUpdatesState() throws Exception {
+        log.info("执行更新短剧上下架状态...");
+        //更新短剧上下架状态	当前域名为测试域名TODO
+        String url = "http://test.api.cms.artlinkworld.cn/videoAppletShow/regularUpdatesState";
+        HttpUtils.httpGet(String.format("%s?id=%s", url, KEY), null, null);
+    }
+
+    @XxlJob("regularUpdateVideoInfo")
+    public void regularUpdateVideoInfo() throws Exception {
+        log.info("定期更新短剧审核信息...");
+        //定期更新短剧审核信息 当前域名为测试域名TODO
+        String url = "http://test.api.cms.artlinkworld.cn/thirtPartVideo/regularUpdateVideoInfo";
+        HttpUtils.httpGet(String.format("%s?platform=1&id=%s", url, KEY), null, null);
+        HttpUtils.httpGet(String.format("%s?platform=2&id=%s", url, KEY), null, null);
+    }
 }

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

@@ -5,11 +5,11 @@ 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;
+import java.util.concurrent.TimeUnit;
 
 @Slf4j
 public class DorisMultiTableStreamLoadManager {
@@ -48,5 +48,15 @@ public class DorisMultiTableStreamLoadManager {
         log.info("Shutting down DorisMultiTableStreamLoadManager");
         bufferManagers.values().forEach(TableBufferManager::flushNow);
         uploaderExecutor.shutdown();
+        try {
+            if (!uploaderExecutor.awaitTermination(60, TimeUnit.SECONDS)) {
+                log.info("doris stream load 任务未完成，强制关闭线程池...");
+                uploaderExecutor.shutdownNow();
+            }
+        } catch (InterruptedException e) {
+            uploaderExecutor.shutdownNow();
+            Thread.currentThread().interrupt();
+        }
+        log.info("doris stream load 线程池已关闭");
     }
 }

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

@@ -103,7 +103,7 @@ spring:
   # redis 配置
   redis:
     # 地址
-    host: r-2ze77hczbrb3boqn16pd.redis.rds.aliyuncs.com
+    host: r-2ze77hczbrb3boqn16.redis.rds.aliyuncs.com
     # 端口，默认为6379
     port: 6379
     # 数据库索引
@@ -125,7 +125,7 @@ spring:
 
   redis2:
     database: 0
-    host: r-2zelacjrljt205nsb1pd.redis.rds.aliyuncs.com
+    host: r-2zelacjrljt205nsb1.redis.rds.aliyuncs.com
     password: Huanrui@2025!@
     pool:
       max-active: 10
@@ -195,4 +195,9 @@ dianzhong:
     dzConfig[0]:
       clientId: 10003129
       token: ckogrgCE4PEDxT1oL5
+      type: iap
+    dzConfig[1]:
+      clientId: 10004501
+      token: HgmmEQM4ddcLQ8mn5i
+      type: iaa