Merge #52 into master from pingkang

Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

* pingkang: (4 commits)
  点众
  点众
  点众
  Merge branch 'master' of codeup.aliyun.com:60c06368dbbed34fe0f0c263/job/hr_job-...

Signed-off-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>
Merged-by: 1129979389@qq.com <accounts_6361eaad7b63b44dd7833e63@mail.teambition.com>

CR-link: https://codeup.aliyun.com/60c06368dbbed34fe0f0c263/job/hr_job-bytedance/change/52

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/data/mapper/DianZhongMapper.java

@@ -1,9 +1,11 @@
 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);
@@ -13,4 +15,7 @@ public interface DianZhongMapper {
     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

@@ -29,9 +29,14 @@
     </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', t3.access_token AS 'accessToken'
+        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/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,12 +3,14 @@ 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;
@@ -29,6 +31,9 @@ import java.util.Map;
 @Slf4j
 public class DianZhongServiceImpl implements IDianZhongService {
     @Autowired
+    private ICacheService cacheService;
+
+    @Autowired
     private DorisMultiTableStreamLoadManager dorisLoadManager;
 
     @Autowired
@@ -160,11 +165,11 @@ public class DianZhongServiceImpl implements IDianZhongService {
         }
     }
 
-    private static String getSignKey(String clientId, String token, long timeStamp) {
+    private String getSignKey(String clientId, String token, long timeStamp) {
         return SecureUtil.md5(clientId + token + timeStamp).toLowerCase();
     }
 
-    private static JSONObject getDZUser(String line, String clientId) {
+    private JSONObject getDZUser(String line, String clientId) {
         JSONObject jsonObject = JSONObject.parseObject(line);
         JSONObject userJson = new JSONObject();
         int domain = jsonObject.getIntValue("domain");
@@ -199,10 +204,20 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 static JSONObject getDZLogin(String line, String clientId) {
+    private JSONObject getDZLogin(String line, String clientId) {
         JSONObject jsonObject = JSONObject.parseObject(line);
         JSONObject loginJson = new JSONObject();
         int domain = jsonObject.getIntValue("domain");
@@ -239,10 +254,19 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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 static JSONObject getDZOrder(String line, String clientId) {
+    private JSONObject getDZOrder(String line, String clientId) {
         JSONObject jsonObject = JSONObject.parseObject(line);
         JSONObject orderJson = new JSONObject();
         String orderId = jsonObject.getString("id");
@@ -288,6 +312,15 @@ public class DianZhongServiceImpl implements IDianZhongService {
         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/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