内容池交互 sync job

api-module/src/main/java/com/tzld/piaoquan/api/controller/contentplatform/ContentPlatformPlanController.java

@@ -124,6 +124,22 @@ public class ContentPlatformPlanController {
         return CommonResponse.success();
     }
 
+    @ApiOperation(value = "同步猜你喜欢视频", hidden = true)
+    @JwtIgnore
+    @GetMapping("/job/syncRecommendedForYouVideoJob")
+    public CommonResponse<Void> syncRecommendedForYouVideoJob(String dateStr) {
+        videoJob.syncRecommendedForYouVideoJob(dateStr);
+        return CommonResponse.success();
+    }
+
+    @ApiOperation(value = "同步视频标签", hidden = true)
+    @JwtIgnore
+    @GetMapping("/job/syncVideoTagJob")
+    public CommonResponse<Void> syncVideoTagJob(String dateStr) {
+        videoJob.syncVideoTagJob(dateStr);
+        return CommonResponse.success();
+    }
+
     @ApiOperation(value = "检查视频状态", hidden = true)
     @JwtIgnore
     @GetMapping("/job/checkContentPlatformVideoStatusJob")

api-module/src/main/java/com/tzld/piaoquan/api/job/contentplatform/ContentPlatformVideoJob.java

@@ -193,6 +193,17 @@ public class ContentPlatformVideoJob {
         planMapperExt.batchInsertContentPlatformVideoAgg(saveAggList);
     }
 
+    @XxlJob("syncRecommendedForYouVideoJob")
+    public ReturnT<String> syncRecommendedForYouVideoJob(String param) {
+        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
+        if (StringUtils.hasText(param)) {
+            dt = param;
+        }
+        List<ContentPlatformVideoAgg> existVideoList = getVideoAggList(dt, VideoSourceEnum.RANK_HOT.getVal());
+        syncRecommendedForYouVideo(dt, existVideoList);
+        return ReturnT.SUCCESS;
+    }
+
     private void syncRecommendedForYouVideo(String dt, List<ContentPlatformVideoAgg> existVideoList) {
         String sql = String.format("SELECT dt, account_id, vid, merge_two_category, title, video_path, base_score, sim_score " +
                 "FROM loghubods.loghubods.new_content_platform_video_recall_i2i_pq WHERE dt=%s;", dt);
@@ -201,9 +212,9 @@ public class ContentPlatformVideoJob {
         if (CollectionUtils.isNotEmpty(dataList)) {
             List<ContentPlatformVideoAgg> saveList = new ArrayList<>();
             List<ContentPlatformVideoAccountRel> saveAccountRelList = new ArrayList<>();
+            List<Long> existVideoIds = new ArrayList<>();
             for (Record record : dataList) {
                 // 保存视频
-                ContentPlatformVideoAgg item = new ContentPlatformVideoAgg();
                 Long accountId = Long.parseLong((String) record.get(1));
                 Long videoId = Long.parseLong((String) record.get(2));
                 String category = (String) record.get(3);
@@ -211,17 +222,21 @@ public class ContentPlatformVideoJob {
                 String videoUrl = (String) record.get(5);
                 Double score = Double.parseDouble((String) record.get(6));
                 Double simScore = Double.parseDouble((String) record.get(7));
-                item.setDt(dt);
-                item.setLastDt(dt);
-                item.setVideoId(videoId);
-                item.setCategory(category);
-                item.setTitle(title);
-                item.setVideo(videoUrl);
-                item.setScore(score);
-                item.setSource(VideoSourceEnum.ACCOUNT_RECOMMEND.getVal());
-                item.setStatus(VideoStatusEnum.NORMAL.getVal());
-                item.setCreateTimestamp(now);
-                saveList.add(item);
+                if (!existVideoIds.contains(videoId)) {
+                    ContentPlatformVideoAgg item = new ContentPlatformVideoAgg();
+                    item.setDt(dt);
+                    item.setLastDt(dt);
+                    item.setVideoId(videoId);
+                    item.setCategory(category);
+                    item.setTitle(title);
+                    item.setVideo(videoUrl);
+                    item.setScore(score);
+                    item.setSource(VideoSourceEnum.ACCOUNT_RECOMMEND.getVal());
+                    item.setStatus(VideoStatusEnum.NORMAL.getVal());
+                    item.setCreateTimestamp(now);
+                    saveList.add(item);
+                    existVideoIds.add(videoId);
+                }
                 // 保存账号关系
                 ContentPlatformVideoAccountRel accountRel = new ContentPlatformVideoAccountRel();
                 accountRel.setDt(dt);
@@ -231,9 +246,10 @@ public class ContentPlatformVideoJob {
                 accountRel.setCreateTimestamp(now);
                 saveAccountRelList.add(accountRel);
             }
-            List<Long> existVideoIds = existVideoList.stream().map(ContentPlatformVideoAgg::getVideoId).collect(Collectors.toList());
+            existVideoIds = existVideoList.stream().map(ContentPlatformVideoAgg::getVideoId).collect(Collectors.toList());
             if (CollectionUtils.isNotEmpty(existVideoIds)) {
-                saveList.removeIf(o -> existVideoIds.contains(o.getVideoId()));
+                List<Long> finalExistVideoIds = existVideoIds;
+                saveList.removeIf(o -> finalExistVideoIds.contains(o.getVideoId()));
             }
             if (CollectionUtils.isNotEmpty(saveList)) {
                 // 获取视频封面
@@ -266,6 +282,16 @@ public class ContentPlatformVideoJob {
         }
     }
 
+    @XxlJob("syncVideoTagJob")
+    public ReturnT<String> syncVideoTagJob(String param) {
+        String dt = DateUtil.getBeforeDayDateString("yyyyMMdd");
+        if (StringUtils.hasText(param)) {
+            dt = param;
+        }
+        syncVideoTag(dt);
+        return ReturnT.SUCCESS;
+    }
+
     private void syncVideoTag(String dt) {
         String sql = String.format("SELECT dt, vid, platform, type, channel, account, tag " +
                 "FROM loghubods.loghubods.new_content_platform_video_display_tags WHERE dt=%s;", dt);
@@ -278,14 +304,17 @@ public class ContentPlatformVideoJob {
                 String platform = (String) record.get(2);
                 String type = (String) record.get(3);
                 String channel = (String) record.get(4);
-                Long accountId = Long.parseLong((String) record.get(5));
-                Integer tag = Integer.valueOf((String) record.get(6));
+                String accountObj = (String) record.get(5);
+                Long accountId = "\\N".equals(accountObj) ? null : Long.parseLong(accountObj);
+                String tagObj = (String) record.get(6);
+                Integer tag = "\\N".equals(tagObj) ? null : Integer.valueOf(tagObj);
+
                 ContentPlatformVideoTag videoTag = new ContentPlatformVideoTag();
                 videoTag.setDt(dt);
                 videoTag.setVideoId(videoId);
-                videoTag.setPlatform(platform);
-                videoTag.setType(type);
-                videoTag.setChannel(channel);
+                videoTag.setPlatform("\\N".equals(platform) ? null : platform);
+                videoTag.setType("\\N".equals(type) ? null : type);
+                videoTag.setChannel("\\N".equals(channel) ? null : channel);
                 videoTag.setAccountId(accountId);
                 videoTag.setTag(tag);
                 videoTag.setCreateTimestamp(now);
@@ -321,6 +350,12 @@ public class ContentPlatformVideoJob {
         return videoAggMapper.countByExample(example);
     }
 
+    private List<ContentPlatformVideoAgg> getVideoAggList(String dt, Integer source) {
+        ContentPlatformVideoAggExample example = new ContentPlatformVideoAggExample();
+        example.createCriteria().andDtEqualTo(dt).andSourceEqualTo(source);
+        return videoAggMapper.selectByExample(example);
+    }
+
     @XxlJob("checkContentPlatformVideoStatusJob")
     public ReturnT<String> checkContentPlatformVideoStatusJob(String param) {
         String dt = planMapperExt.getVideoMaxDt();

api-module/src/main/resources/mapper/contentplatform/ext/ContentPlatformPlanMapperExt.xml

@@ -224,11 +224,12 @@
     </insert>
 
     <insert id="batchInsertContentPlatformVideoAgg">
-        insert into content_platform_video_agg (dt, video_id, category, title, cover, video, score, create_timestamp)
+        insert into content_platform_video_agg (dt, last_dt, source, video_id, category, title, cover,
+        video, score, create_timestamp)
         values
         <foreach collection="records" item="item" separator=",">
-            (#{item.dt}, #{item.videoId}, #{item.category}, #{item.title}, #{item.cover}, #{item.video}, #{item.score},
-            #{item.createTimestamp})
+            (#{item.dt}, #{item.lastDt}, #{item.source}, #{item.videoId}, #{item.category}, #{item.title},
+            #{item.cover}, #{item.video}, #{item.score}, #{item.createTimestamp})
         </foreach>
     </insert>