视频重新同步更新

api-module/src/main/java/com/tzld/piaoquan/api/dao/mapper/contentplatform/ext/ContentPlatformPlanMapperExt.java

@@ -82,4 +82,6 @@ public interface ContentPlatformPlanMapperExt {
     void batchInsertContentPlatformVideoDatastat(@Param("records") List<ContentPlatformVideoDataStat> saveList);
 
     void batchInsertContentPlatformVideoDatastatAgg(@Param("records") List<ContentPlatformVideoDataStatAgg> saveAggList);
+
+    void updateOtherVideoStatus(@Param("dt") String dt, @Param("videoIdList") List<Long> videoIdList, @Param("now") Long now);
 }

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

@@ -58,12 +58,8 @@ public class ContentPlatformVideoJob {
             aggDt = param;
         }
         List<String> dtList = DateUtil.getBeforeDays(aggDt, null, videoAggDays);
-        // 轮询查询大数据获取最近14天视频
+        // 轮询查询大数据获取最近 videoAggDays 天视频
         for (String dt : dtList) {
-            long videoCount = getVideoCount(dt);
-            if (videoCount > 0) {
-                continue;
-            }
             String sql = String.format("SELECT * FROM loghubods.wecom_cooperation_video_candidate_pool WHERE dt=%s;", dt);
             List<Record> dataList = OdpsUtil.getOdpsData(sql);
             Long now = System.currentTimeMillis();
@@ -102,7 +98,18 @@ public class ContentPlatformVideoJob {
                 }
                 // save
                 if (CollectionUtils.isNotEmpty(saveList)) {
-                    planMapperExt.batchInsertContentPlatformVideo(saveList);
+                    List<ContentPlatformVideo> videoList = getVideoList(dt);
+                    if (CollectionUtils.isNotEmpty(videoList)) {
+                        List<Long> existVideoIds = videoList.stream().map(ContentPlatformVideo::getVideoId).collect(Collectors.toList());
+                        List<Long> videoIdList = saveList.stream().map(ContentPlatformVideo::getVideoId).collect(Collectors.toList());
+                        planMapperExt.updateOtherVideoStatus(dt, videoIdList, now);
+                        saveList = saveList.stream().filter(item -> !existVideoIds.contains(item.getVideoId())).collect(Collectors.toList());
+                        if (CollectionUtils.isNotEmpty(saveList)) {
+                            planMapperExt.batchInsertContentPlatformVideo(saveList);
+                        }
+                    } else {
+                        planMapperExt.batchInsertContentPlatformVideo(saveList);
+                    }
                 }
             }
         }
@@ -155,6 +162,12 @@ public class ContentPlatformVideoJob {
         return videoMapper.countByExample(example);
     }
 
+    private List<ContentPlatformVideo> getVideoList(String dt) {
+        ContentPlatformVideoExample example = new ContentPlatformVideoExample();
+        example.createCriteria().andDtEqualTo(dt);
+        return videoMapper.selectByExample(example);
+    }
+
     private long getVideoAggCount(String dt) {
         ContentPlatformVideoAggExample example = new ContentPlatformVideoAggExample();
         example.createCriteria().andDtEqualTo(dt);

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

@@ -302,4 +302,14 @@
         </foreach>
     </insert>
 
+    <update id="updateOtherVideoStatus">
+        update content_platform_video
+        set status = 0,
+            update_timestamp = #{now}
+        where dt = #{dt} and video_id not in
+        <foreach collection="videoIdList" item="item" open="(" close=")" separator=",">
+            #{item}
+        </foreach>
+    </update>
+
 </mapper>