定时任务 结构梳理

core/src/main/java/com/tzld/videoVector/dao/mapper/pgVector/ext/VideoVectorMapperExt.java

@@ -78,24 +78,6 @@ public interface VideoVectorMapperExt {
                                         @Param("queryVector") String queryVector,
                                         @Param("topN") int topN);
 
-    /**
-     * 分页查询指定 configCode 下的记录（不过滤 text 是否为空，只要有数据就返回）
-     * @param configCode 配置编码
-     * @param offset     偏移量
-     * @param limit      每页数量
-     * @return VideoVector 列表（包含 videoId、pointIndex）
-     */
-    List<VideoVector> selectByConfigCodePaged(@Param("configCode") String configCode,
-                                              @Param("offset") int offset,
-                                              @Param("limit") int limit);
-
-    /**
-     * 批量更新 text 字段
-     * @param id   主键ID
-     * @param text 文本内容
-     */
-    int updateTextById(@Param("id") Long id, @Param("text") String text);
-
     /**
      * 根据 text_hash 和 configCode 查询已存在的 embedding（取第一条匹配记录）
      * 用于相同文本复用已有的向量化结果，避免重复调用 embedding API

core/src/main/java/com/tzld/videoVector/job/DataMigrationJob.java

@@ -264,69 +264,90 @@ public class DataMigrationJob {
 
             // 2. 逐个配置迁移
             for (String configCode : configCodes) {
-                log.info("开始迁移配置 {} 的向量数据", configCode);
+                int[] counts = migrateRedisVectorsForConfig(configCode);
+                totalSuccess += counts[0];
+                totalSkip += counts[1];
+                totalFail += counts[2];
+            }
+
+            log.info("Redis 视频向量迁移完成，总成功: {}, 总跳过: {}, 总失败: {}",
+                    totalSuccess, totalSkip, totalFail);
+            return ReturnT.SUCCESS;
+        } catch (Exception e) {
+            log.error("Redis 向量迁移失败: {}", e.getMessage(), e);
+            return new ReturnT<>(ReturnT.FAIL_CODE, "迁移失败: " + e.getMessage());
+        }
+    }
 
-                // 从 Redis Set 获取所有 videoId
-                String idsKey = VectorConstants.VECTOR_KEY_PREFIX + configCode + ":ids";
-                Set<Object> idMembers = redisTemplate.opsForSet().members(idsKey);
+    /**
+     * 迁移单个 configCode 下的 Redis 向量到 PG
+     *
+     * @return int[]{成功数, 跳过数, 失败数}
+     */
+    private int[] migrateRedisVectorsForConfig(String configCode) {
+        log.info("开始迁移配置 {} 的向量数据", configCode);
 
-                if (idMembers == null || idMembers.isEmpty()) {
-                    log.info("配置 {} 在 Redis 中无 videoId 索引，跳过", configCode);
-                    continue;
-                }
+        String idsKey = VectorConstants.VECTOR_KEY_PREFIX + configCode + ":ids";
+        Set<Object> idMembers = redisTemplate.opsForSet().members(idsKey);
 
-                log.info("配置 {} 共有 {} 个 videoId 待迁移", configCode, idMembers.size());
+        if (idMembers == null || idMembers.isEmpty()) {
+            log.info("配置 {} 在 Redis 中无 videoId 索引，跳过", configCode);
+            return new int[]{0, 0, 0};
+        }
 
-                // Redis value 可能是 Integer/Long/String，统一转为 String
-                List<String> idList = new ArrayList<>();
-                for (Object member : idMembers) {
-                    idList.add(String.valueOf(member));
-                }
+        log.info("配置 {} 共有 {} 个 videoId 待迁移", configCode, idMembers.size());
 
-                // 分批处理
-                for (int i = 0; i < idList.size(); i += BATCH_SIZE) {
-                    int end = Math.min(i + BATCH_SIZE, idList.size());
-                    List<String> batch = idList.subList(i, end);
-
-                    for (String idStr : batch) {
-                        try {
-                            Long videoId = Long.parseLong(idStr);
-                            String vectorKey = VectorConstants.VECTOR_KEY_PREFIX + configCode + ":" + videoId;
-                            String vectorJson = stringRedisTemplate.opsForValue().get(vectorKey);
-
-                            if (!StringUtils.hasText(vectorJson)) {
-                                totalSkip++;
-                                continue;
-                            }
-
-                            // Redis 中存储的是归一化后的 JSON 数组 "[0.1,0.2,...]"
-                            // pgvector 接受此格式
-                            videoVectorMapperExt.upsertVector(videoId, configCode, 0, vectorJson, null, null);
-                            totalSuccess++;
-                        } catch (NumberFormatException e) {
-                            log.warn("非法 videoId: {}", idStr);
-                            totalSkip++;
-                        } catch (Exception e) {
-                            if (e.getMessage() != null && e.getMessage().contains("duplicate key")) {
-                                totalSkip++;
-                            } else {
-                                totalFail++;
-                                log.error("迁移 Redis 向量失败，configCode={}, videoId={}, error={}",
-                                        configCode, idStr, e.getMessage());
-                            }
-                        }
-                    }
+        List<String> idList = new ArrayList<>();
+        for (Object member : idMembers) {
+            idList.add(String.valueOf(member));
+        }
 
-                    log.info("配置 {} 进度: {}/{}", configCode, Math.min(end, idList.size()), idList.size());
-                }
+        int success = 0, skip = 0, fail = 0;
+
+        for (int i = 0; i < idList.size(); i += BATCH_SIZE) {
+            int end = Math.min(i + BATCH_SIZE, idList.size());
+            List<String> batch = idList.subList(i, end);
+
+            for (String idStr : batch) {
+                int result = migrateSingleRedisVector(configCode, idStr);
+                if (result == 1) success++;
+                else if (result == 0) skip++;
+                else fail++;
             }
 
-            log.info("Redis 视频向量迁移完成，总成功: {}, 总跳过: {}, 总失败: {}",
-                    totalSuccess, totalSkip, totalFail);
-            return ReturnT.SUCCESS;
+            log.info("配置 {} 进度: {}/{}", configCode, end, idList.size());
+        }
+
+        return new int[]{success, skip, fail};
+    }
+
+    /**
+     * 迁移单个 videoId 的 Redis 向量到 PG
+     *
+     * @return 1=成功, 0=跳过, -1=失败
+     */
+    private int migrateSingleRedisVector(String configCode, String idStr) {
+        try {
+            Long videoId = Long.parseLong(idStr);
+            String vectorKey = VectorConstants.VECTOR_KEY_PREFIX + configCode + ":" + videoId;
+            String vectorJson = stringRedisTemplate.opsForValue().get(vectorKey);
+
+            if (!StringUtils.hasText(vectorJson)) {
+                return 0;
+            }
+
+            videoVectorMapperExt.upsertVector(videoId, configCode, 0, vectorJson, null, null);
+            return 1;
+        } catch (NumberFormatException e) {
+            log.warn("非法 videoId: {}", idStr);
+            return 0;
         } catch (Exception e) {
-            log.error("Redis 向量迁移失败: {}", e.getMessage(), e);
-            return new ReturnT<>(ReturnT.FAIL_CODE, "迁移失败: " + e.getMessage());
+            if (e.getMessage() != null && e.getMessage().contains("duplicate key")) {
+                return 0;
+            }
+            log.error("迁移 Redis 向量失败，configCode={}, videoId={}, error={}",
+                    configCode, idStr, e.getMessage());
+            return -1;
         }
     }
 

core/src/main/java/com/tzld/videoVector/job/MaterialDeconstructCheckJob.java

@@ -64,63 +64,20 @@ public class MaterialDeconstructCheckJob {
 
                 for (DeconstructContent content : contents) {
                     totalChecked++;
-                    try {
-                        String taskId = content.getTaskId();
-                        if (!StringUtils.hasText(taskId)) {
-                            log.warn("contentId={} 的 taskId 为空，跳过", content.getId());
-                            continue;
+                    int result = checkAndUpdateSingleContent(content);
+                    if (result == 1) {
+                        totalCompleted++;
+                        try {
+                            vectorizeService.vectorizeContent(content);
+                            totalVectorized++;
+                            log.info("解构完成并向量化成功，contentId={}, taskId={}",
+                                    content.getId(), content.getTaskId());
+                        } catch (Exception e) {
+                            log.error("向量化失败，contentId={}, taskId={}, error={}",
+                                    content.getId(), content.getTaskId(), e.getMessage(), e);
                         }
-
-                        // 调用解构 API 查询结果
-                        DeconstructResult result = deconstructService.getDeconstructResult(taskId);
-                        if (result == null) {
-                            log.warn("查询解构结果返回空，taskId={}", taskId);
-                            continue;
-                        }
-
-                        // 更新状态
-                        Short newStatus = result.getStatus() != null
-                                ? result.getStatus().shortValue() : content.getStatus();
-
-                        // 状态未变化则跳过更新
-                        if (newStatus.equals(content.getStatus())) {
-                            log.debug("解构状态未变化，taskId={}, status={}", taskId, newStatus);
-                            continue;
-                        }
-
-                        content.setStatus(newStatus);
-                        content.setResultJson(result.getResult());
-                        content.setFailureReason(result.getReason());
-                        content.setPointUrl(result.getPointUrl());
-                        content.setWeightUrl(result.getWeightUrl());
-                        content.setPatternUrl(result.getPatternUrl());
-                        content.setUpdateTime(new Date());
-                        deconstructContentMapper.updateByPrimaryKeySelective(content);
-
-                        log.info("更新解构状态，contentId={}, taskId={}, status={} -> {}",
-                                content.getId(), taskId, content.getStatus(), newStatus);
-
-                        // 解构成功，触发向量化
-                        if (newStatus == 2) {
-                            totalCompleted++;
-                            try {
-                                vectorizeService.vectorizeContent(content);
-                                totalVectorized++;
-                                log.info("解构完成并向量化成功，contentId={}, taskId={}",
-                                        content.getId(), taskId);
-                            } catch (Exception e) {
-                                log.error("向量化失败，contentId={}, taskId={}, error={}",
-                                        content.getId(), taskId, e.getMessage(), e);
-                            }
-                        } else if (newStatus == 3) {
-                            totalFailed++;
-                            log.warn("解构失败，contentId={}, taskId={}, reason={}",
-                                    content.getId(), taskId, result.getReason());
-                        }
-
-                    } catch (Exception e) {
-                        log.error("处理解构结果失败，contentId={}, error={}",
-                                content.getId(), e.getMessage(), e);
+                    } else if (result == -1) {
+                        totalFailed++;
                     }
                 }
 
@@ -154,4 +111,58 @@ public class MaterialDeconstructCheckJob {
         example.setOrderByClause("id ASC LIMIT " + pageSize + " OFFSET " + (pageNum * pageSize));
         return deconstructContentMapper.selectByExample(example);
     }
+
+    /**
+     * 检查并更新单条解构内容的状态
+     *
+     * @return 1=解构成功, -1=解构失败, 0=未完成或未变化
+     */
+    private int checkAndUpdateSingleContent(DeconstructContent content) {
+        try {
+            String taskId = content.getTaskId();
+            if (!StringUtils.hasText(taskId)) {
+                log.warn("contentId={} 的 taskId 为空，跳过", content.getId());
+                return 0;
+            }
+
+            DeconstructResult result = deconstructService.getDeconstructResult(taskId);
+            if (result == null) {
+                log.warn("查询解构结果返回空，taskId={}", taskId);
+                return 0;
+            }
+
+            Short newStatus = result.getStatus() != null
+                    ? result.getStatus().shortValue() : content.getStatus();
+
+            if (newStatus.equals(content.getStatus())) {
+                log.debug("解构状态未变化，taskId={}, status={}", taskId, newStatus);
+                return 0;
+            }
+
+            content.setStatus(newStatus);
+            content.setResultJson(result.getResult());
+            content.setFailureReason(result.getReason());
+            content.setPointUrl(result.getPointUrl());
+            content.setWeightUrl(result.getWeightUrl());
+            content.setPatternUrl(result.getPatternUrl());
+            content.setUpdateTime(new Date());
+            deconstructContentMapper.updateByPrimaryKeySelective(content);
+
+            log.info("更新解构状态，contentId={}, taskId={}, status={} -> {}",
+                    content.getId(), taskId, content.getStatus(), newStatus);
+
+            if (newStatus == 2) {
+                return 1;
+            } else if (newStatus == 3) {
+                log.warn("解构失败，contentId={}, taskId={}, reason={}",
+                        content.getId(), taskId, result.getReason());
+                return -1;
+            }
+            return 0;
+        } catch (Exception e) {
+            log.error("处理解构结果失败，contentId={}, error={}",
+                    content.getId(), e.getMessage(), e);
+            return 0;
+        }
+    }
 }

core/src/main/java/com/tzld/videoVector/job/VideoDetailSyncJob.java

@@ -115,65 +115,89 @@ public class VideoDetailSyncJob {
             for (int i = 0; i < allVideoIds.size(); i += ODPS_BATCH_SIZE) {
                 int end = Math.min(i + ODPS_BATCH_SIZE, allVideoIds.size());
                 List<Long> batchIds = allVideoIds.subList(i, end);
+                processBatchVideoDetail(batchIds, dtMin, dtMax, totalSuccess, totalFail, i, end);
+            }
 
-                try {
-                    String idsStr = batchIds.stream()
-                            .map(String::valueOf)
-                            .collect(Collectors.joining(","));
+            log.info("视频基础信息同步任务完成，总成功: {}, 总失败: {}", totalSuccess.get(), totalFail.get());
+            return ReturnT.SUCCESS;
+        } catch (Exception e) {
+            log.error("视频基础信息同步任务执行失败: {}", e.getMessage(), e);
+            return new ReturnT<>(ReturnT.FAIL_CODE, "任务执行失败: " + e.getMessage());
+        }
+    }
 
-                    // 2.1 查询维度字段（按日期取最大那天的数据）
-                    Map<Long, JSONObject> dimensionMap = new HashMap<>();
-                    String dimensionSql = buildDimensionSql(idsStr, dtMin, dtMax);
-                    OdpsUtil.getOdpsDataStream(dimensionSql, record -> {
-                        try {
-                            Long videoId = record.getBigint("视频id");
-                            if (videoId != null) {
-                                dimensionMap.put(videoId, buildVideoDetail(record));
-                            }
-                        } catch (Exception e) {
-                            log.error("处理维度字段记录失败: {}", e.getMessage());
-                        }
-                    });
-                    log.info("批次 {}-{} 维度查询完成，获取 {} 条记录", i, end, dimensionMap.size());
+    /**
+     * 处理单批次视频详情的 ODPS 查询与 Redis 写入
+     */
+    private void processBatchVideoDetail(List<Long> batchIds, String dtMin, String dtMax,
+                                         AtomicInteger totalSuccess, AtomicInteger totalFail,
+                                         int batchStart, int batchEnd) {
+        try {
+            String idsStr = batchIds.stream()
+                    .map(String::valueOf)
+                    .collect(Collectors.joining(","));
 
-                    // 2.2 查询聚合指标（只按视频id分组）
-                    String metricsSql = buildMetricsSql(idsStr, dtMin, dtMax);
-                    long processed = OdpsUtil.getOdpsDataStream(metricsSql, record -> {
-                        try {
-                            Long videoId = record.getBigint("视频id");
-                            if (videoId == null) {
-                                return;
-                            }
-                            // 合并维度字段 + 指标数据
-                            JSONObject detail = dimensionMap.getOrDefault(videoId, new JSONObject());
-                            JSONObject metrics = buildVideoDetail(record);
-                            if (metrics != null) {
-                                detail.putAll(metrics);
-                            }
-                            if (!detail.isEmpty()) {
-                                String redisKey = VectorConstants.VIDEO_DETAIL_KEY_PREFIX + videoId;
-                                redisUtils.set(redisKey, detail.toJSONString(),
-                                        VectorConstants.VIDEO_DETAIL_EXPIRE_SECONDS);
-                                totalSuccess.incrementAndGet();
-                            }
-                        } catch (Exception e) {
-                            log.error("处理指标记录失败: {}", e.getMessage());
-                            totalFail.incrementAndGet();
-                        }
-                    });
+            // 1. 查询维度字段（按日期取最大那天的数据）
+            Map<Long, JSONObject> dimensionMap = queryDimensionData(idsStr, dtMin, dtMax);
+            log.info("批次 {}-{} 维度查询完成，获取 {} 条记录", batchStart, batchEnd, dimensionMap.size());
 
-                    log.info("批次 {}-{} 指标查询完成，处理 {} 条记录", i, end, processed);
-                } catch (Exception e) {
-                    log.error("批次 {}-{} 查询ODPS失败: {}", i, end, e.getMessage(), e);
-                    totalFail.addAndGet(batchIds.size());
+            // 2. 查询聚合指标并合并写入 Redis
+            String metricsSql = buildMetricsSql(idsStr, dtMin, dtMax);
+            long processed = OdpsUtil.getOdpsDataStream(metricsSql, record ->
+                    mergeMetricsAndWriteRedis(record, dimensionMap, totalSuccess, totalFail)
+            );
+
+            log.info("批次 {}-{} 指标查询完成，处理 {} 条记录", batchStart, batchEnd, processed);
+        } catch (Exception e) {
+            log.error("批次 {}-{} 查询ODPS失败: {}", batchStart, batchEnd, e.getMessage(), e);
+            totalFail.addAndGet(batchIds.size());
+        }
+    }
+
+    /**
+     * 查询维度字段数据
+     */
+    private Map<Long, JSONObject> queryDimensionData(String idsStr, String dtMin, String dtMax) {
+        Map<Long, JSONObject> dimensionMap = new HashMap<>();
+        String dimensionSql = buildDimensionSql(idsStr, dtMin, dtMax);
+        OdpsUtil.getOdpsDataStream(dimensionSql, record -> {
+            try {
+                Long videoId = record.getBigint("视频id");
+                if (videoId != null) {
+                    dimensionMap.put(videoId, buildVideoDetail(record));
                 }
+            } catch (Exception e) {
+                log.error("处理维度字段记录失败: {}", e.getMessage());
             }
+        });
+        return dimensionMap;
+    }
 
-            log.info("视频基础信息同步任务完成，总成功: {}, 总失败: {}", totalSuccess.get(), totalFail.get());
-            return ReturnT.SUCCESS;
+    /**
+     * 合并指标数据与维度数据，写入 Redis
+     */
+    private void mergeMetricsAndWriteRedis(com.aliyun.odps.data.Record record,
+                                           Map<Long, JSONObject> dimensionMap,
+                                           AtomicInteger totalSuccess, AtomicInteger totalFail) {
+        try {
+            Long videoId = record.getBigint("视频id");
+            if (videoId == null) {
+                return;
+            }
+            JSONObject detail = dimensionMap.getOrDefault(videoId, new JSONObject());
+            JSONObject metrics = buildVideoDetail(record);
+            if (metrics != null) {
+                detail.putAll(metrics);
+            }
+            if (!detail.isEmpty()) {
+                String redisKey = VectorConstants.VIDEO_DETAIL_KEY_PREFIX + videoId;
+                redisUtils.set(redisKey, detail.toJSONString(),
+                        VectorConstants.VIDEO_DETAIL_EXPIRE_SECONDS);
+                totalSuccess.incrementAndGet();
+            }
         } catch (Exception e) {
-            log.error("视频基础信息同步任务执行失败: {}", e.getMessage(), e);
-            return new ReturnT<>(ReturnT.FAIL_CODE, "任务执行失败: " + e.getMessage());
+            log.error("处理指标记录失败: {}", e.getMessage());
+            totalFail.incrementAndGet();
         }
     }
 

core/src/main/java/com/tzld/videoVector/job/VideoVectorJob.java

@@ -32,6 +32,7 @@ import javax.annotation.Resource;
 import java.util.*;
 import java.util.concurrent.*;
 import java.util.concurrent.atomic.AtomicInteger;
+import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
 
@@ -122,105 +123,11 @@ public class VideoVectorJob {
                 ExecutorService configExecutor = Executors.newFixedThreadPool(configs.size());
                 List<Future<?>> configFutures = new ArrayList<>();
                 for (DeconstructVectorConfig config : configs) {
-                    configFutures.add(configExecutor.submit(() -> {
-                        String configCode = config.getConfigCode();
-                        try {
-                            // 4.1 查询哪些 videoId 在该配置下已有向量（数据库层已做 DISTINCT video_id）
-                            Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
-
-                            // 4.1.1 检查已存在记录中 text 为空的，删除后重新向量化
-                            if (!existingVideoIds.isEmpty()) {
-                                Set<Long> emptyTextIds = vectorStoreService.findVideoIdsWithEmptyText(configCode, existingVideoIds);
-                                if (!emptyTextIds.isEmpty()) {
-                                    log.info("配置 {} 下发现 {} 个 text 为空的记录，删除后重新向量化: {}", configCode, emptyTextIds.size(), emptyTextIds);
-                                    vectorStoreService.deleteBatch(configCode, emptyTextIds);
-                                    existingVideoIds.removeAll(emptyTextIds);
-                                }
-                            }
-
-                            // 4.2 过滤出需要处理的 videoId（排除已有向量的）
-                            List<Long> needProcessIds = auditPassedIds.stream()
-                                    .filter(id -> !existingVideoIds.contains(id))
-                                    .collect(Collectors.toList());
-
-                            if (needProcessIds.isEmpty()) {
-                                log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
-                                return;
-                            }
-                            log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
-
-                            // 4.3 流式查询 raw_result，Semaphore 控制并发在途数（不阻塞流式读取，避免OOM）
-                            ExecutorService embedExecutor = Executors.newFixedThreadPool(VectorConstants.EMBEDDING_PARALLELISM);
-                            Semaphore inFlightLimiter = new Semaphore(VectorConstants.MAX_EMBEDDING_IN_FLIGHT);
-                            List<Future<?>> futures = new ArrayList<>();
-
-                            try {
-                                // 分批查询，防止 IN 子句 ID 过多导致 ODPS SQL 超长
-                                for (List<Long> batchIds : Lists.partition(needProcessIds, VectorConstants.ODPS_IN_BATCH_SIZE)) {
-                                    String idsStr = batchIds.stream().map(String::valueOf).collect(Collectors.joining(","));
-                                    String sql = String.format(
-                                            "SELECT content_id, raw_result FROM videoods.content_profile " +
-                                                    "WHERE status = 3 AND is_deleted = 0 AND content_id IN (%s);", idsStr);
-
-                                    OdpsUtil.getOdpsDataStream(sql, record -> {
-                                        Long videoId = Long.valueOf(record.getString(0));
-                                        String rawResult = record.getString(1);
-                                        if (videoId == null || !StringUtils.hasText(rawResult)) {
-                                            return;
-                                        }
-                                        try {
-                                            inFlightLimiter.acquire();
-                                        } catch (InterruptedException e) {
-                                            Thread.currentThread().interrupt();
-                                            return;
-                                        }
-                                        futures.add(embedExecutor.submit(() -> {
-                                            try {
-                                                List<String> texts = extractTextsFromRawResult(rawResult, config);
-                                                if (CollectionUtils.isEmpty(texts)) {
-                                                    log.debug("videoId={} 配置 {} 未提取到文本，跳过", videoId, configCode);
-                                                    totalFailCount.incrementAndGet();
-                                                    return;
-                                                }
-                                                int storeCount = vectorizeAndStore(config, videoId, texts);
-                                                if (storeCount > 0) {
-                                                    totalSuccessCount.incrementAndGet();
-                                                } else {
-                                                    totalFailCount.incrementAndGet();
-                                                }
-                                            } catch (Exception e) {
-                                                log.error("处理 videoId={} 配置 {} 时发生异常: {}", videoId, configCode, e.getMessage(), e);
-                                                totalFailCount.incrementAndGet();
-                                            } finally {
-                                                inFlightLimiter.release();
-                                            }
-                                        }));
-                                    });
-                                }
-                            } finally {
-                                for (Future<?> future : futures) {
-                                    try {
-                                        future.get(30, TimeUnit.MINUTES);
-                                    } catch (Exception e) {
-                                        log.error("embedding 并发任务等待异常: {}", e.getMessage(), e);
-                                    }
-                                }
-                                embedExecutor.shutdown();
-                            }
-                        } catch (Exception e) {
-                            log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
-                        }
-                    }));
-                }
-                // 等待所有配置处理完成
-                for (Future<?> future : configFutures) {
-                    try {
-                        future.get(10, TimeUnit.MINUTES);
-                    } catch (Exception e) {
-                        log.error("配置并发任务等待异常: {}", e.getMessage(), e);
-                    }
+                    configFutures.add(configExecutor.submit(() ->
+                            processConfigForRawResult(config, auditPassedIds, totalSuccessCount, totalFailCount)
+                    ));
                 }
-                configExecutor.shutdown();
+                awaitAndShutdown(configFutures, configExecutor, 10, "配置并发");
 
                 // 如果查询到的数据少于 PAGE_SIZE，说明已经是最后一页
                 if (videoIds.size() < VectorConstants.PAGE_SIZE) {
@@ -238,6 +145,129 @@ public class VideoVectorJob {
     }
 
 
+    /**
+     * 处理单个配置下的 raw_result 视频向量化
+     * 包含：空text检查清理、过滤已有向量、流式查询ODPS并并发embedding
+     */
+    private void processConfigForRawResult(DeconstructVectorConfig config, List<Long> auditPassedIds,
+                                           AtomicInteger totalSuccessCount, AtomicInteger totalFailCount) {
+        String configCode = config.getConfigCode();
+        try {
+            // 1. 查询已有向量并清理空text记录
+            Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
+            cleanEmptyTextRecords(configCode, existingVideoIds);
+
+            // 2. 过滤出需要处理的 videoId
+            List<Long> needProcessIds = auditPassedIds.stream()
+                    .filter(id -> !existingVideoIds.contains(id))
+                    .collect(Collectors.toList());
+            if (needProcessIds.isEmpty()) {
+                log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
+                return;
+            }
+            log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
+
+            // 3. 流式查询并并发embedding
+            ExecutorService embedExecutor = Executors.newFixedThreadPool(VectorConstants.EMBEDDING_PARALLELISM);
+            Semaphore inFlightLimiter = new Semaphore(VectorConstants.MAX_EMBEDDING_IN_FLIGHT);
+            List<Future<?>> futures = new ArrayList<>();
+
+            try {
+                for (List<Long> batchIds : Lists.partition(needProcessIds, VectorConstants.ODPS_IN_BATCH_SIZE)) {
+                    String idsStr = batchIds.stream().map(String::valueOf).collect(Collectors.joining(","));
+                    String sql = String.format(
+                            "SELECT content_id, raw_result FROM videoods.content_profile " +
+                                    "WHERE status = 3 AND is_deleted = 0 AND content_id IN (%s);", idsStr);
+
+                    OdpsUtil.getOdpsDataStream(sql, record ->
+                            submitEmbeddingTask(record, config, embedExecutor, inFlightLimiter, futures,
+                                    totalSuccessCount, totalFailCount, "raw_result")
+                    );
+                }
+            } finally {
+                awaitAndShutdown(futures, embedExecutor, 30, "embedding");
+            }
+        } catch (Exception e) {
+            log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
+        }
+    }
+
+    /**
+     * 提交单条记录的 embedding 任务到线程池
+     * 使用 Semaphore 控制在途并发数
+     */
+    private void submitEmbeddingTask(com.aliyun.odps.data.Record record, DeconstructVectorConfig config,
+                                     ExecutorService executor, Semaphore inFlightLimiter,
+                                     List<Future<?>> futures, AtomicInteger successCount,
+                                     AtomicInteger failCount, String dataType) {
+        Long videoId = Long.valueOf(record.getString(0));
+        String rawData = record.getString(1);
+        if (videoId == null || !StringUtils.hasText(rawData)) {
+            return;
+        }
+        try {
+            inFlightLimiter.acquire();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return;
+        }
+        String configCode = config.getConfigCode();
+        futures.add(executor.submit(() -> {
+            try {
+                List<String> texts = "result_log".equals(dataType)
+                        ? extractTextsFromResultLogData(rawData, config)
+                        : extractTextsFromRawResult(rawData, config);
+                if (CollectionUtils.isEmpty(texts)) {
+                    log.debug("videoId={} 配置 {} 未提取到文本，跳过", videoId, configCode);
+                    failCount.incrementAndGet();
+                    return;
+                }
+                int storeCount = vectorizeAndStore(config, videoId, texts);
+                if (storeCount > 0) {
+                    successCount.incrementAndGet();
+                } else {
+                    failCount.incrementAndGet();
+                }
+            } catch (Exception e) {
+                log.error("处理 videoId={} 配置 {} 时发生异常: {}", videoId, configCode, e.getMessage(), e);
+                failCount.incrementAndGet();
+            } finally {
+                inFlightLimiter.release();
+            }
+        }));
+    }
+
+    /**
+     * 检查已存在记录中 text 为空的，删除后重新向量化
+     * 会从 existingVideoIds 中移除被清理的ID
+     */
+    private void cleanEmptyTextRecords(String configCode, Set<Long> existingVideoIds) {
+        if (existingVideoIds.isEmpty()) {
+            return;
+        }
+        Set<Long> emptyTextIds = vectorStoreService.findVideoIdsWithEmptyText(configCode, existingVideoIds);
+        if (!emptyTextIds.isEmpty()) {
+            log.info("配置 {} 下发现 {} 个 text 为空的记录，删除后重新向量化: {}", configCode, emptyTextIds.size(), emptyTextIds);
+            vectorStoreService.deleteBatch(configCode, emptyTextIds);
+            existingVideoIds.removeAll(emptyTextIds);
+        }
+    }
+
+    /**
+     * 等待所有 Future 完成并关闭 ExecutorService
+     */
+    private void awaitAndShutdown(List<Future<?>> futures, ExecutorService executor,
+                                  long timeoutMinutes, String taskDesc) {
+        for (Future<?> future : futures) {
+            try {
+                future.get(timeoutMinutes, TimeUnit.MINUTES);
+            } catch (Exception e) {
+                log.error("{} 并发任务等待异常: {}", taskDesc, e.getMessage(), e);
+            }
+        }
+        executor.shutdown();
+    }
+
     /**
      * 根据配置从 raw_result 中提取文本
      * 当配置了 extract_rule 时，启用置信度过滤逻辑
@@ -560,95 +590,12 @@ public class VideoVectorJob {
             ExecutorService configExecutor = Executors.newFixedThreadPool(configs.size());
             List<Future<?>> configFutures = new ArrayList<>();
             for (DeconstructVectorConfig config : configs) {
-                configFutures.add(configExecutor.submit(() -> {
-                    String configCode = config.getConfigCode();
-                    try {
-                        // 5.1 查询该配置下已有向量的 videoId，排除已处理过的（数据库层已做 DISTINCT video_id）
-                        Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
-
-                        // 5.1.1 检查已存在记录中 text 为空的，删除后重新向量化
-                        if (!existingVideoIds.isEmpty()) {
-                            Set<Long> emptyTextIds = vectorStoreService.findVideoIdsWithEmptyText(configCode, existingVideoIds);
-                            if (!emptyTextIds.isEmpty()) {
-                                log.info("配置 {} 下发现 {} 个 text 为空的记录，删除后重新向量化: {}", configCode, emptyTextIds.size(), emptyTextIds);
-                                vectorStoreService.deleteBatch(configCode, emptyTextIds);
-                                existingVideoIds.removeAll(emptyTextIds);
-                            }
-                        }
-
-                        List<Long> needProcessIds = auditPassedIds.stream()
-                                .filter(id -> !existingVideoIds.contains(id))
-                                .collect(Collectors.toList());
-                        if (needProcessIds.isEmpty()) {
-                            log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
-                            return;
-                        }
-                        log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
-
-                        // 5.2 并发调用 detail 接口，提取选题并向量化存储
-                        ExecutorService detailExecutor = Executors.newFixedThreadPool(VectorConstants.AIGC_DETAIL_PARALLELISM);
-                        List<Future<?>> detailFutures = new ArrayList<>();
-                        for (Long videoId : needProcessIds) {
-                            detailFutures.add(detailExecutor.submit(() -> {
-                                try {
-                                    Long taskInstanceId = videoIdToTaskInstanceId.get(videoId);
-                                    if (taskInstanceId == null) {
-                                        log.warn("videoId={} 无对应 taskInstanceId，跳过", videoId);
-                                        totalFailCount.incrementAndGet();
-                                        return;
-                                    }
-
-                                    // 调用 detail 接口获取 dataContent（解构详情）
-                                    JSONObject dataContent = aigcApiService.getTaskCallbackDetail(taskInstanceId);
-                                    if (dataContent == null) {
-                                        log.warn("videoId={} taskInstanceId={} 获取 dataContent 失败，跳过", videoId, taskInstanceId);
-                                        totalFailCount.incrementAndGet();
-                                        return;
-                                    }
-
-                                    // 从 dataContent 中提取文本（支持置信度过滤）
-                                    List<String> texts = extractTextsFromDataContent(dataContent, config);
-                                    if (CollectionUtils.isEmpty(texts)) {
-                                        log.debug("videoId={} 配置 {} 未提取到选题文本，跳过", videoId, configCode);
-                                        totalFailCount.incrementAndGet();
-                                        return;
-                                    }
-
-                                    // 向量化并存储（多点模式返回成功数>0即为成功）
-                                    int storeCount = vectorizeAndStore(config, videoId, texts);
-                                    if (storeCount > 0) {
-                                        totalSuccessCount.incrementAndGet();
-                                    } else {
-                                        totalFailCount.incrementAndGet();
-                                    }
-                                } catch (Exception e) {
-                                    log.error("处理 videoId={} 配置 {} 时发生异常: {}", videoId, configCode, e.getMessage(), e);
-                                    totalFailCount.incrementAndGet();
-                                }
-                            }));
-                        }
-                        for (Future<?> future : detailFutures) {
-                            try {
-                                future.get(30, TimeUnit.MINUTES);
-                            } catch (Exception e) {
-                                log.error("AIGC detail 并发任务等待异常: {}", e.getMessage(), e);
-                            }
-                        }
-                        detailExecutor.shutdown();
-                    } catch (Exception e) {
-                        log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
-                    }
-                }));
-            }
-            // 等待所有配置处理完成
-            for (Future<?> future : configFutures) {
-                try {
-                    future.get(30, TimeUnit.MINUTES);
-                } catch (Exception e) {
-                    log.error("配置并发任务等待异常: {}", e.getMessage(), e);
-                }
+                configFutures.add(configExecutor.submit(() ->
+                        processConfigForAigc(config, auditPassedIds, videoIdToTaskInstanceId,
+                                totalSuccessCount, totalFailCount)
+                ));
             }
-            configExecutor.shutdown();
+            awaitAndShutdown(configFutures, configExecutor, 30, "配置并发");
 
             log.info("AIGC 来源视频向量化任务完成，总成功: {}, 总失败: {}", totalSuccessCount.get(), totalFailCount.get());
             return ReturnT.SUCCESS;
@@ -658,6 +605,85 @@ public class VideoVectorJob {
         }
     }
 
+    /**
+     * 处理单个配置下的 AIGC 视频向量化
+     * 包含：空text检查清理、过滤已有向量、并发调用detail接口并向量化
+     */
+    private void processConfigForAigc(DeconstructVectorConfig config, List<Long> auditPassedIds,
+                                      Map<Long, Long> videoIdToTaskInstanceId,
+                                      AtomicInteger totalSuccessCount, AtomicInteger totalFailCount) {
+        String configCode = config.getConfigCode();
+        try {
+            // 1. 查询已有向量并清理空text记录
+            Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
+            cleanEmptyTextRecords(configCode, existingVideoIds);
+
+            // 2. 过滤出需要处理的 videoId
+            List<Long> needProcessIds = auditPassedIds.stream()
+                    .filter(id -> !existingVideoIds.contains(id))
+                    .collect(Collectors.toList());
+            if (needProcessIds.isEmpty()) {
+                log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
+                return;
+            }
+            log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
+
+            // 3. 并发调用 detail 接口，提取文本并向量化
+            ExecutorService detailExecutor = Executors.newFixedThreadPool(VectorConstants.AIGC_DETAIL_PARALLELISM);
+            List<Future<?>> detailFutures = new ArrayList<>();
+            for (Long videoId : needProcessIds) {
+                detailFutures.add(detailExecutor.submit(() ->
+                        processAigcSingleVideo(videoId, config, videoIdToTaskInstanceId,
+                                totalSuccessCount, totalFailCount)
+                ));
+            }
+            awaitAndShutdown(detailFutures, detailExecutor, 30, "AIGC detail");
+        } catch (Exception e) {
+            log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
+        }
+    }
+
+    /**
+     * 处理单个 AIGC 视频的向量化
+     */
+    private void processAigcSingleVideo(Long videoId, DeconstructVectorConfig config,
+                                        Map<Long, Long> videoIdToTaskInstanceId,
+                                        AtomicInteger successCount, AtomicInteger failCount) {
+        String configCode = config.getConfigCode();
+        try {
+            Long taskInstanceId = videoIdToTaskInstanceId.get(videoId);
+            if (taskInstanceId == null) {
+                log.warn("videoId={} 无对应 taskInstanceId，跳过", videoId);
+                failCount.incrementAndGet();
+                return;
+            }
+
+            JSONObject dataContent = aigcApiService.getTaskCallbackDetail(taskInstanceId);
+            if (dataContent == null) {
+                log.warn("videoId={} taskInstanceId={} 获取 dataContent 失败，跳过", videoId, taskInstanceId);
+                failCount.incrementAndGet();
+                return;
+            }
+
+            List<String> texts = extractTextsFromDataContent(dataContent, config);
+            if (CollectionUtils.isEmpty(texts)) {
+                log.debug("videoId={} 配置 {} 未提取到选题文本，跳过", videoId, configCode);
+                failCount.incrementAndGet();
+                return;
+            }
+
+            int storeCount = vectorizeAndStore(config, videoId, texts);
+            if (storeCount > 0) {
+                successCount.incrementAndGet();
+            } else {
+                failCount.incrementAndGet();
+            }
+        } catch (Exception e) {
+            log.error("处理 videoId={} 配置 {} 时发生异常: {}", videoId, configCode, e.getMessage(), e);
+            failCount.incrementAndGet();
+        }
+    }
+
     /**
      * 从 dataContent 中提取文本
      * 根据配置的 extract_rule 决定是否进行置信度过滤
@@ -949,103 +975,11 @@ public class VideoVectorJob {
                 ExecutorService configExecutor = Executors.newFixedThreadPool(configs.size());
                 List<Future<?>> configFutures = new ArrayList<>();
                 for (DeconstructVectorConfig config : configs) {
-                    configFutures.add(configExecutor.submit(() -> {
-                        String configCode = config.getConfigCode();
-                        try {
-                            // 5.1 已向量化过滤
-                            Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
-
-                            // 5.1.1 检查已存在记录中 text 为空的，删除后重新向量化
-                            if (!existingVideoIds.isEmpty()) {
-                                Set<Long> emptyTextIds = vectorStoreService.findVideoIdsWithEmptyText(configCode, existingVideoIds);
-                                if (!emptyTextIds.isEmpty()) {
-                                    log.info("配置 {} 下发现 {} 个 text 为空的记录，删除后重新向量化: {}", configCode, emptyTextIds.size(), emptyTextIds);
-                                    vectorStoreService.deleteBatch(configCode, emptyTextIds);
-                                    existingVideoIds.removeAll(emptyTextIds);
-                                }
-                            }
-
-                            List<Long> needProcessIds = auditPassedIds.stream()
-                                    .filter(id -> !existingVideoIds.contains(id))
-                                    .collect(Collectors.toList());
-                            if (needProcessIds.isEmpty()) {
-                                log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
-                                return;
-                            }
-                            log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
-
-                            // 5.2 流式查询 result_log data，Semaphore 控制并发在途数（不阻塞流式读取，避免OOM）
-                            ExecutorService embedExecutor = Executors.newFixedThreadPool(VectorConstants.EMBEDDING_PARALLELISM);
-                            Semaphore inFlightLimiter = new Semaphore(VectorConstants.MAX_EMBEDDING_IN_FLIGHT);
-                            List<Future<?>> futures = new ArrayList<>();
-
-                            try {
-                                // 分批查询，防止 IN 子句 ID 过多导致 ODPS SQL 超长
-                                for (List<Long> batchIds : Lists.partition(needProcessIds, VectorConstants.ODPS_IN_BATCH_SIZE)) {
-                                    String idsStr = batchIds.stream().map(String::valueOf).collect(Collectors.joining(","));
-                                    String sql = String.format(
-                                            "SELECT video_id, data FROM loghubods.result_log " +
-                                                    "WHERE video_id IN (%s) AND dt > 20240001;", idsStr);
-
-                                    OdpsUtil.getOdpsDataStream(sql, record -> {
-                                        Long videoId = Long.valueOf(record.getString(0));
-                                        String data = record.getString(1);
-                                        if (videoId == null || !StringUtils.hasText(data)) {
-                                            return;
-                                        }
-                                        try {
-                                            inFlightLimiter.acquire();
-                                        } catch (InterruptedException e) {
-                                            Thread.currentThread().interrupt();
-                                            return;
-                                        }
-                                        futures.add(embedExecutor.submit(() -> {
-                                            try {
-                                                List<String> texts = extractTextsFromResultLogData(data, config);
-                                                if (CollectionUtils.isEmpty(texts)) {
-                                                    log.debug("videoId={} 配置 {} 未提取到文本，跳过", videoId, configCode);
-                                                    totalFailCount.incrementAndGet();
-                                                    return;
-                                                }
-                                                int storeCount = vectorizeAndStore(config, videoId, texts);
-                                                if (storeCount > 0) {
-                                                    totalSuccessCount.incrementAndGet();
-                                                } else {
-                                                    totalFailCount.incrementAndGet();
-                                                }
-                                            } catch (Exception e) {
-                                                log.error("处理 videoId={} 配置 {} 时发生异常: {}", videoId, configCode, e.getMessage(), e);
-                                                totalFailCount.incrementAndGet();
-                                            } finally {
-                                                inFlightLimiter.release();
-                                            }
-                                        }));
-                                    });
-                                }
-                            } finally {
-                                for (Future<?> future : futures) {
-                                    try {
-                                        future.get(30, TimeUnit.MINUTES);
-                                    } catch (Exception e) {
-                                        log.error("embedding 并发任务等待异常: {}", e.getMessage(), e);
-                                    }
-                                }
-                                embedExecutor.shutdown();
-                            }
-                        } catch (Exception e) {
-                            log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
-                        }
-                    }));
+                    configFutures.add(configExecutor.submit(() ->
+                            processConfigForResultLog(config, auditPassedIds, totalSuccessCount, totalFailCount)
+                    ));
                 }
-                // 等待所有配置处理完成
-                for (Future<?> future : configFutures) {
-                    try {
-                        future.get(10, TimeUnit.MINUTES);
-                    } catch (Exception e) {
-                        log.error("配置并发任务等待异常: {}", e.getMessage(), e);
-                    }
-                }
-                configExecutor.shutdown();
+                awaitAndShutdown(configFutures, configExecutor, 10, "配置并发");
 
                 if (videoIds.size() < VectorConstants.PAGE_SIZE) {
                     log.info("第 {} 页数据量 {} 小于 PAGE_SIZE {}，分页查询结束", pageNum, videoIds.size(), VectorConstants.PAGE_SIZE);
@@ -1112,6 +1046,53 @@ public class VideoVectorJob {
         return videoIds;
     }
 
+    /**
+     * 处理单个配置下的 result_log 视频向量化
+     * 包含：空text检查清理、过滤已有向量、流式查询ODPS并并发embedding
+     */
+    private void processConfigForResultLog(DeconstructVectorConfig config, List<Long> auditPassedIds,
+                                           AtomicInteger totalSuccessCount, AtomicInteger totalFailCount) {
+        String configCode = config.getConfigCode();
+        try {
+            // 1. 查询已有向量并清理空text记录
+            Set<Long> existingVideoIds = vectorStoreService.existsByIds(configCode, auditPassedIds);
+            cleanEmptyTextRecords(configCode, existingVideoIds);
+
+            // 2. 过滤出需要处理的 videoId
+            List<Long> needProcessIds = auditPassedIds.stream()
+                    .filter(id -> !existingVideoIds.contains(id))
+                    .collect(Collectors.toList());
+            if (needProcessIds.isEmpty()) {
+                log.debug("配置 {} 下所有视频已有向量，跳过", configCode);
+                return;
+            }
+            log.info("配置 {} 需要处理 {} 个视频", configCode, needProcessIds.size());
+
+            // 3. 流式查询并并发embedding
+            ExecutorService embedExecutor = Executors.newFixedThreadPool(VectorConstants.EMBEDDING_PARALLELISM);
+            Semaphore inFlightLimiter = new Semaphore(VectorConstants.MAX_EMBEDDING_IN_FLIGHT);
+            List<Future<?>> futures = new ArrayList<>();
+
+            try {
+                for (List<Long> batchIds : Lists.partition(needProcessIds, VectorConstants.ODPS_IN_BATCH_SIZE)) {
+                    String idsStr = batchIds.stream().map(String::valueOf).collect(Collectors.joining(","));
+                    String sql = String.format(
+                            "SELECT video_id, data FROM loghubods.result_log " +
+                                    "WHERE video_id IN (%s) AND dt > 20240001;", idsStr);
+
+                    OdpsUtil.getOdpsDataStream(sql, record ->
+                            submitEmbeddingTask(record, config, embedExecutor, inFlightLimiter, futures,
+                                    totalSuccessCount, totalFailCount, "result_log")
+                    );
+                }
+            } finally {
+                awaitAndShutdown(futures, embedExecutor, 30, "embedding");
+            }
+        } catch (Exception e) {
+            log.error("配置 {} 处理异常: {}", configCode, e.getMessage(), e);
+        }
+    }
+
     /**
      * 从 result_log 的 data 字段中提取文本
      * data 结构参考 r.json，直接按 sourcePath 提取单点文本值
@@ -1168,75 +1149,50 @@ public class VideoVectorJob {
     public ReturnT<String> allVideoVectorJob(String param) {
         log.info("开始执行聚合视频向量化任务, param: {}", param);
 
+        // 定义子任务列表：名称 + 执行逻辑
+        List<Map.Entry<String, Supplier<ReturnT<String>>>> subTasks = Arrays.asList(
+                new AbstractMap.SimpleEntry<>("vectorVideoJob", () -> vectorVideoJob(param)),
+                new AbstractMap.SimpleEntry<>("aigcVideoVectorJob", () -> aigcVideoVectorJob(param)),
+                new AbstractMap.SimpleEntry<>("resultLogVideoVectorJob", () -> resultLogVideoVectorJob(param)),
+                new AbstractMap.SimpleEntry<>("videoTitleVectorJob", () -> videoTitleVectorJob.videoTitleVectorJob(param))
+        );
+
         boolean hasFailure = false;
         StringBuilder failMessages = new StringBuilder();
 
-        // 1. 执行 vectorVideoJob
-        try {
-            log.info("===== 开始执行子任务: vectorVideoJob =====");
-            ReturnT<String> result = vectorVideoJob(param);
-            if (result.getCode() != ReturnT.SUCCESS_CODE) {
-                hasFailure = true;
-                failMessages.append("vectorVideoJob失败: ").append(result.getMsg()).append("; ");
-            }
-            log.info("===== 子任务 vectorVideoJob 执行完成, code={} =====", result.getCode());
-        } catch (Exception e) {
-            hasFailure = true;
-            failMessages.append("vectorVideoJob异常: ").append(e.getMessage()).append("; ");
-            log.error("子任务 vectorVideoJob 执行异常: {}", e.getMessage(), e);
+        for (Map.Entry<String, Supplier<ReturnT<String>>> task : subTasks) {
+            hasFailure |= executeSubTask(task.getKey(), task.getValue(), failMessages);
         }
 
-        // 2. 执行 aigcVideoVectorJob
-        try {
-            log.info("===== 开始执行子任务: aigcVideoVectorJob =====");
-            ReturnT<String> result = aigcVideoVectorJob(param);
-            if (result.getCode() != ReturnT.SUCCESS_CODE) {
-                hasFailure = true;
-                failMessages.append("aigcVideoVectorJob失败: ").append(result.getMsg()).append("; ");
-            }
-            log.info("===== 子任务 aigcVideoVectorJob 执行完成, code={} =====", result.getCode());
-        } catch (Exception e) {
-            hasFailure = true;
-            failMessages.append("aigcVideoVectorJob异常: ").append(e.getMessage()).append("; ");
-            log.error("子任务 aigcVideoVectorJob 执行异常: {}", e.getMessage(), e);
+        if (hasFailure) {
+            log.warn("聚合视频向量化任务完成，部分子任务失败: {}", failMessages);
+            return new ReturnT<>(ReturnT.FAIL_CODE, "部分子任务失败: " + failMessages);
         }
 
-        // 3. 执行 resultLogVideoVectorJob
-        try {
-            log.info("===== 开始执行子任务: resultLogVideoVectorJob =====");
-            ReturnT<String> result = resultLogVideoVectorJob(param);
-            if (result.getCode() != ReturnT.SUCCESS_CODE) {
-                hasFailure = true;
-                failMessages.append("resultLogVideoVectorJob失败: ").append(result.getMsg()).append("; ");
-            }
-            log.info("===== 子任务 resultLogVideoVectorJob 执行完成, code={} =====", result.getCode());
-        } catch (Exception e) {
-            hasFailure = true;
-            failMessages.append("resultLogVideoVectorJob异常: ").append(e.getMessage()).append("; ");
-            log.error("子任务 resultLogVideoVectorJob 执行异常: {}", e.getMessage(), e);
-        }
+        log.info("聚合视频向量化任务全部完成");
+        return ReturnT.SUCCESS;
+    }
 
-        // 4. 执行 videoTitleVectorJob
+    /**
+     * 执行单个子任务，捕获异常并记录失败信息
+     *
+     * @return true 表示该子任务失败
+     */
+    private boolean executeSubTask(String taskName, Supplier<ReturnT<String>> taskSupplier,
+                                   StringBuilder failMessages) {
         try {
-            log.info("===== 开始执行子任务: videoTitleVectorJob =====");
-            ReturnT<String> result = videoTitleVectorJob.videoTitleVectorJob(param);
+            log.info("===== 开始执行子任务: {} =====", taskName);
+            ReturnT<String> result = taskSupplier.get();
+            log.info("===== 子任务 {} 执行完成, code={} =====", taskName, result.getCode());
             if (result.getCode() != ReturnT.SUCCESS_CODE) {
-                hasFailure = true;
-                failMessages.append("videoTitleVectorJob失败: ").append(result.getMsg()).append("; ");
+                failMessages.append(taskName).append("失败: ").append(result.getMsg()).append("; ");
+                return true;
             }
-            log.info("===== 子任务 videoTitleVectorJob 执行完成, code={} =====", result.getCode());
         } catch (Exception e) {
-            hasFailure = true;
-            failMessages.append("videoTitleVectorJob异常: ").append(e.getMessage()).append("; ");
-            log.error("子任务 videoTitleVectorJob 执行异常: {}", e.getMessage(), e);
+            failMessages.append(taskName).append("异常: ").append(e.getMessage()).append("; ");
+            log.error("子任务 {} 执行异常: {}", taskName, e.getMessage(), e);
+            return true;
         }
-
-        if (hasFailure) {
-            log.warn("聚合视频向量化任务完成，部分子任务失败: {}", failMessages);
-            return new ReturnT<>(ReturnT.FAIL_CODE, "部分子任务失败: " + failMessages);
-        }
-
-        log.info("聚合视频向量化任务全部完成");
-        return ReturnT.SUCCESS;
+        return false;
     }
 }

core/src/main/java/com/tzld/videoVector/job/VideoVectorTextBackfillJob.java

@@ -1,526 +0,0 @@
-package com.tzld.videoVector.job;
-
-import com.alibaba.fastjson.JSON;
-import com.alibaba.fastjson.JSONArray;
-import com.alibaba.fastjson.JSONObject;
-import com.aliyun.odps.data.Record;
-import com.google.common.collect.Lists;
-import com.tzld.videoVector.api.AigcApiService;
-import com.tzld.videoVector.api.VideoApiService;
-import com.tzld.videoVector.dao.mapper.pgVector.DeconstructVectorConfigMapper;
-import com.tzld.videoVector.dao.mapper.pgVector.ext.VideoVectorMapperExt;
-import com.tzld.videoVector.model.entity.VideoDetail;
-import com.tzld.videoVector.model.po.pgVector.DeconstructVectorConfig;
-import com.tzld.videoVector.model.po.pgVector.DeconstructVectorConfigExample;
-import com.tzld.videoVector.model.po.pgVector.VideoVector;
-import com.tzld.videoVector.util.OdpsUtil;
-import com.tzld.videoVector.util.VectorUtils;
-import com.xxl.job.core.biz.model.ReturnT;
-import com.xxl.job.core.handler.annotation.XxlJob;
-import lombok.extern.slf4j.Slf4j;
-import org.springframework.stereotype.Component;
-import org.springframework.util.CollectionUtils;
-import org.springframework.util.StringUtils;
-
-import javax.annotation.Resource;
-import java.util.*;
-import java.util.concurrent.atomic.AtomicInteger;
-import java.util.stream.Collectors;
-
-/**
- * 历史数据 text 字段补充任务
- * <p>
- * 扫描 video_vectors 中的现有数据，根据其 config_code 对应的配置，
- * 从原始数据源（content_profile / result_log / aigc_deconstruct）中提取文本，
- * 回填到 video_vectors.text 字段。
- * <p>
- * 不检查数据状态，只要有数据就根据 config 提取对应数据。
- */
-@Slf4j
-@Component
-public class VideoVectorTextBackfillJob {
-
-    @Resource
-    private VideoVectorMapperExt videoVectorMapperExt;
-
-    @Resource
-    private DeconstructVectorConfigMapper vectorConfigMapper;
-
-    @Resource
-    private AigcApiService aigcApiService;
-
-    @Resource
-    private VideoApiService videoApiService;
-
-    /** 每批处理数量 */
-    private static final int BATCH_SIZE = 500;
-
-    /**
-     * 历史数据 text 字段补充
-     * 遍历所有启用的配置，对 video_vectors 中已有记录补充 text 字段
-     *
-     * @param param 参数（可选：指定 configCode 只处理特定配置）
-     * @return 执行结果
-     */
-    @XxlJob("videoVectorTextBackfillJob")
-    public ReturnT<String> videoVectorTextBackfillJob(String param) {
-        log.info("开始执行历史数据 text 字段补充任务, param: {}", param);
-
-        try {
-            // 获取所有启用的配置
-            List<DeconstructVectorConfig> configs = getAllEnabledConfigs();
-            if (CollectionUtils.isEmpty(configs)) {
-                log.warn("未找到启用的向量化配置");
-                return ReturnT.SUCCESS;
-            }
-
-            // 如果指定了 configCode，只处理特定配置
-            if (StringUtils.hasText(param)) {
-                configs = configs.stream()
-                        .filter(c -> param.equals(c.getConfigCode()))
-                        .collect(Collectors.toList());
-                if (configs.isEmpty()) {
-                    log.warn("未找到 configCode={} 的启用配置", param);
-                    return ReturnT.SUCCESS;
-                }
-            }
-
-            log.info("共加载 {} 个向量化配置", configs.size());
-
-            int totalUpdateCount = 0;
-            int totalSkipCount = 0;
-            int totalFailCount = 0;
-
-            // 逐个配置处理
-            for (DeconstructVectorConfig config : configs) {
-                String configCode = config.getConfigCode();
-                String sourceField = config.getSourceField();
-                log.info("开始处理配置: configCode={}, sourceField={}", configCode, sourceField);
-
-                int configUpdateCount = 0;
-                int configSkipCount = 0;
-                int configFailCount = 0;
-                int offset = 0;
-
-                while (true) {
-                    // 分页查询该配置下的记录
-                    List<VideoVector> vectors = videoVectorMapperExt.selectByConfigCodePaged(configCode, offset, BATCH_SIZE);
-                    if (CollectionUtils.isEmpty(vectors)) {
-                        log.info("配置 {} 第 {} 偏移无数据，处理结束", configCode, offset);
-                        break;
-                    }
-
-                    // 按 sourceField 类型分发处理
-                    int[] counts = processVectorBatch(vectors, config);
-                    configUpdateCount += counts[0];
-                    configSkipCount += counts[1];
-                    configFailCount += counts[2];
-
-                    if (vectors.size() < BATCH_SIZE) {
-                        break;
-                    }
-                    offset += BATCH_SIZE;
-                }
-
-                log.info("配置 {} 处理完成，更新: {}, 跳过: {}, 失败: {}",
-                        configCode, configUpdateCount, configSkipCount, configFailCount);
-                totalUpdateCount += configUpdateCount;
-                totalSkipCount += configSkipCount;
-                totalFailCount += configFailCount;
-            }
-
-            log.info("历史数据 text 字段补充任务完成，总更新: {}, 总跳过: {}, 总失败: {}",
-                    totalUpdateCount, totalSkipCount, totalFailCount);
-            return ReturnT.SUCCESS;
-        } catch (Exception e) {
-            log.error("历史数据 text 字段补充任务执行失败: {}", e.getMessage(), e);
-            return new ReturnT<>(ReturnT.FAIL_CODE, "任务执行失败: " + e.getMessage());
-        }
-    }
-
-    /**
-     * 处理一批 VideoVector 记录
-     * <p>
-     * 多点模式处理策略：
-     * 原始向量化时 texts[i] 对应 point_index=i，回填时保持相同映射。
-     * 由于置信度过滤可能导致重新提取的文本列表与原始不同，
-     * 当 point_index 超出当前提取的 texts 范围时直接跳过，避免错误匹配。
-     * <p>
-     * 单点模式（point_index=0）：直接取第一个有效文本。
-     *
-     * @return int[]{更新数, 跳过数, 失败数}
-     */
-    private int[] processVectorBatch(List<VideoVector> vectors, DeconstructVectorConfig config) {
-        String sourceField = config.getSourceField();
-        boolean multiPoint = VectorUtils.isMultiPointConfig(config);
-        AtomicInteger updateCount = new AtomicInteger(0);
-        AtomicInteger skipCount = new AtomicInteger(0);
-        AtomicInteger failCount = new AtomicInteger(0);
-
-        // 收集需要处理的 videoId（去重）
-        List<Long> videoIds = vectors.stream()
-                .map(VideoVector::getVideoId)
-                .distinct()
-                .collect(Collectors.toList());
-
-        // 根据 sourceField 获取原始数据
-        Map<Long, String> rawDataMap;
-        switch (sourceField) {
-            case "result_json":
-                rawDataMap = batchQueryVideoRawResults(videoIds);
-                break;
-            case "result_log":
-                rawDataMap = batchQueryResultLogData(videoIds);
-                break;
-            case "aigc_deconstruct":
-                rawDataMap = batchQueryAigcData(videoIds);
-                break;
-            case "video_title":
-                rawDataMap = batchQueryVideoTitles(videoIds);
-                break;
-            default:
-                log.warn("不支持的 sourceField: {}", sourceField);
-                return new int[]{0, vectors.size(), 0};
-        }
-
-        // 缓存每个 videoId 提取的文本列表，避免对同一 videoId 重复解析
-        Map<Long, List<String>> textsCache = new HashMap<>();
-
-        // 遍历每条记录，提取文本并更新
-        for (VideoVector vector : vectors) {
-            try {
-                Long videoId = vector.getVideoId();
-                String rawData = rawDataMap.get(videoId);
-                if (!StringUtils.hasText(rawData)) {
-                    skipCount.incrementAndGet();
-                    continue;
-                }
-
-                // 根据配置提取文本（相同 videoId 复用缓存）
-                List<String> texts = textsCache.computeIfAbsent(videoId,
-                        id -> extractTexts(rawData, config));
-                if (CollectionUtils.isEmpty(texts)) {
-                    skipCount.incrementAndGet();
-                    continue;
-                }
-
-                // 根据 pointIndex 取对应的文本
-                int pointIndex = vector.getPointIndex() != null ? vector.getPointIndex() : 0;
-                String text;
-
-                if (multiPoint) {
-                    // 多点模式：严格按 point_index 映射，超出范围则跳过
-                    if (pointIndex < texts.size()) {
-                        text = texts.get(pointIndex);
-                    } else {
-                        log.debug("videoId={} point_index={} 超出提取文本范围(size={})，跳过",
-                                videoId, pointIndex, texts.size());
-                        skipCount.incrementAndGet();
-                        continue;
-                    }
-                } else {
-                    // 单点模式：取第一个有效文本
-                    text = texts.stream()
-                            .filter(StringUtils::hasText)
-                            .findFirst()
-                            .orElse(null);
-                }
-
-                if (!StringUtils.hasText(text)) {
-                    skipCount.incrementAndGet();
-                    continue;
-                }
-
-                // 截断文本
-                Integer maxLength = config.getMaxLength();
-                if (maxLength != null && maxLength > 0 && text.length() > maxLength) {
-                    text = text.substring(0, maxLength);
-                }
-
-                // 更新 text 字段
-                videoVectorMapperExt.updateTextById(vector.getId(), text);
-                updateCount.incrementAndGet();
-            } catch (Exception e) {
-                log.error("处理 videoVector id={} videoId={} 时异常: {}",
-                        vector.getId(), vector.getVideoId(), e.getMessage());
-                failCount.incrementAndGet();
-            }
-        }
-
-        return new int[]{updateCount.get(), skipCount.get(), failCount.get()};
-    }
-
-    /**
-     * 根据配置从原始数据中提取文本
-     */
-    private List<String> extractTexts(String rawData, DeconstructVectorConfig config) {
-        List<String> texts = new ArrayList<>();
-        try {
-            String sourceField = config.getSourceField();
-
-            // video_title 来源：原始数据就是标题文本，无需 JSON 解析
-            if ("video_title".equals(sourceField)) {
-                if (StringUtils.hasText(rawData)) {
-                    texts.add(rawData);
-                }
-                return texts;
-            }
-
-            JSONObject json = JSON.parseObject(rawData);
-            if (json == null) {
-                return texts;
-            }
-
-            String sourcePath = config.getSourcePath();
-            if (!StringUtils.hasText(sourcePath)) {
-                return texts;
-            }
-
-            String extractRule = config.getExtractRule();
-            if (StringUtils.hasText(extractRule)) {
-                // 多点模式：带置信度过滤
-                texts.addAll(extractTextsWithConfidence(json, sourcePath, extractRule));
-            } else {
-                // 单点模式：直接提取
-                texts.addAll(VectorUtils.extractFromJson(json, sourcePath));
-            }
-        } catch (Exception e) {
-            log.error("解析原始数据失败: {}", e.getMessage());
-        }
-        return texts;
-    }
-
-    /**
-     * 带置信度过滤的文本提取
-     */
-    private List<String> extractTextsWithConfidence(JSONObject json, String sourcePath, String extractRule) {
-        List<String> texts = new ArrayList<>();
-        try {
-            JSONObject rule = JSON.parseObject(extractRule);
-            String textField = rule.getString("text_field");
-            String confidenceField = rule.getString("confidence_field");
-            double confidenceThreshold = rule.getDoubleValue("confidence_threshold");
-
-            if (!StringUtils.hasText(textField) || !StringUtils.hasText(confidenceField)) {
-                return texts;
-            }
-
-            if (sourcePath.endsWith("[*]")) {
-                // 数组路径模式
-                String arrayPath = sourcePath.substring(0, sourcePath.length() - 3);
-                Object arrayObj = navigateToValue(json, arrayPath);
-                if (arrayObj instanceof JSONArray) {
-                    JSONArray array = (JSONArray) arrayObj;
-                    for (int i = 0; i < array.size(); i++) {
-                        JSONObject item = array.getJSONObject(i);
-                        if (item != null && isConfidenceQualified(item, confidenceField, confidenceThreshold)) {
-                            String text = item.getString(textField);
-                            if (StringUtils.hasText(text)) {
-                                texts.add(text);
-                            }
-                        }
-                    }
-                }
-            } else {
-                // 单对象路径模式
-                Object obj = navigateToValue(json, sourcePath);
-                if (obj instanceof JSONObject) {
-                    JSONObject item = (JSONObject) obj;
-                    if (isConfidenceQualified(item, confidenceField, confidenceThreshold)) {
-                        String text = item.getString(textField);
-                        if (StringUtils.hasText(text)) {
-                            texts.add(text);
-                        }
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("置信度提取失败: {}", e.getMessage());
-        }
-        return texts;
-    }
-
-    /**
-     * 根据 JSON 路径导航到目标值
-     */
-    private Object navigateToValue(JSONObject json, String path) {
-        if (json == null || !StringUtils.hasText(path) || !path.startsWith("$.")) {
-            return null;
-        }
-        try {
-            String pathContent = path.substring(2);
-            String[] parts = pathContent.split("\\.");
-            Object current = json;
-            for (String part : parts) {
-                if (current instanceof JSONObject) {
-                    current = ((JSONObject) current).get(part);
-                } else {
-                    return null;
-                }
-            }
-            return current;
-        } catch (Exception e) {
-            return null;
-        }
-    }
-
-    /**
-     * 判断置信度是否满足条件
-     */
-    private boolean isConfidenceQualified(JSONObject item, String confidenceField, double threshold) {
-        Object value = item.get(confidenceField);
-        if (value == null) {
-            return false;
-        }
-        if (value instanceof String) {
-            return "high".equalsIgnoreCase((String) value);
-        }
-        if (value instanceof Number) {
-            return ((Number) value).doubleValue() >= threshold;
-        }
-        return false;
-    }
-
-    // ========================== 数据源查询方法 ==========================
-
-    /**
-     * 批量查询 content_profile 的 raw_result
-     */
-    private Map<Long, String> batchQueryVideoRawResults(List<Long> videoIds) {
-        Map<Long, String> result = new HashMap<>();
-        for (List<Long> partition : Lists.partition(videoIds, 200)) {
-            String idsStr = partition.stream()
-                    .map(String::valueOf)
-                    .collect(Collectors.joining(","));
-            String sql = String.format(
-                    "SELECT content_id, raw_result " +
-                            "FROM videoods.content_profile " +
-                            "WHERE content_id IN (%s);",
-                    idsStr);
-            try {
-                List<Record> records = OdpsUtil.getOdpsData(sql);
-                if (records != null) {
-                    for (Record record : records) {
-                        Long videoId = Long.valueOf(record.getString(0));
-                        String rawResult = record.getString(1);
-                        if (videoId != null && rawResult != null) {
-                            result.put(videoId, rawResult);
-                        }
-                    }
-                }
-            } catch (Exception e) {
-                log.error("查询 content_profile raw_result 失败: {}", e.getMessage());
-            }
-        }
-        return result;
-    }
-
-    /**
-     * 批量查询 result_log 的 data 字段
-     */
-    private Map<Long, String> batchQueryResultLogData(List<Long> videoIds) {
-        Map<Long, String> result = new HashMap<>();
-        for (List<Long> partition : Lists.partition(videoIds, 200)) {
-            String idsStr = partition.stream()
-                    .map(String::valueOf)
-                    .collect(Collectors.joining(","));
-            String sql = String.format(
-                    "SELECT video_id, data " +
-                            "FROM loghubods.result_log " +
-                            "WHERE video_id IN (%s) AND dt > 20240001;",
-                    idsStr);
-            try {
-                List<Record> records = OdpsUtil.getOdpsData(sql);
-                if (records != null) {
-                    for (Record record : records) {
-                        Long videoId = Long.valueOf(record.getString(0));
-                        String data = record.getString(1);
-                        if (videoId != null && data != null) {
-                            result.put(videoId, data);
-                        }
-                    }
-                }
-            } catch (Exception e) {
-                log.error("查询 result_log data 失败: {}", e.getMessage());
-            }
-        }
-        return result;
-    }
-
-    /**
-     * 批量查询 AIGC 解构数据
-     * 通过 AIGC API 获取 dataContent
-     */
-    private Map<Long, String> batchQueryAigcData(List<Long> videoIds) {
-        Map<Long, String> result = new HashMap<>();
-        try {
-            // 获取 AIGC 任务输入列表
-            List<AigcApiService.AigcTaskInput> taskInputList = aigcApiService.getTaskInputList(46);
-            if (CollectionUtils.isEmpty(taskInputList)) {
-                return result;
-            }
-
-            // 构建 videoId -> taskInstanceId 映射
-            Map<Long, Long> videoIdToTaskInstanceId = new HashMap<>();
-            for (AigcApiService.AigcTaskInput input : taskInputList) {
-                try {
-                    Long videoId = Long.parseLong(input.getBizUniqueId());
-                    videoIdToTaskInstanceId.put(videoId, input.getTaskInstanceId());
-                } catch (NumberFormatException e) {
-                    // 忽略格式非法的
-                }
-            }
-
-            // 对需要的 videoId 查询 dataContent
-            Set<Long> targetIds = new HashSet<>(videoIds);
-            for (Map.Entry<Long, Long> entry : videoIdToTaskInstanceId.entrySet()) {
-                Long videoId = entry.getKey();
-                if (!targetIds.contains(videoId)) {
-                    continue;
-                }
-                try {
-                    JSONObject dataContent = aigcApiService.getTaskCallbackDetail(entry.getValue());
-                    if (dataContent != null) {
-                        result.put(videoId, dataContent.toJSONString());
-                    }
-                } catch (Exception e) {
-                    log.error("查询 AIGC dataContent 失败, videoId={}: {}", videoId, e.getMessage());
-                }
-            }
-        } catch (Exception e) {
-            log.error("批量查询 AIGC 数据失败: {}", e.getMessage());
-        }
-        return result;
-    }
-
-    /**
-     * 批量查询视频标题
-     * 通过视频详情 API 获取标题
-     */
-    private Map<Long, String> batchQueryVideoTitles(List<Long> videoIds) {
-        Map<Long, String> result = new HashMap<>();
-        try {
-            Map<Long, VideoDetail> detailMap = videoApiService.getVideoDetail(new HashSet<>(videoIds));
-            if (detailMap != null) {
-                for (Map.Entry<Long, VideoDetail> entry : detailMap.entrySet()) {
-                    if (entry.getValue() != null && StringUtils.hasText(entry.getValue().getTitle())) {
-                        result.put(entry.getKey(), entry.getValue().getTitle());
-                    }
-                }
-            }
-        } catch (Exception e) {
-            log.error("批量查询视频标题失败: {}", e.getMessage());
-        }
-        return result;
-    }
-
-    /**
-     * 获取所有启用的向量化配置
-     */
-    private List<DeconstructVectorConfig> getAllEnabledConfigs() {
-        DeconstructVectorConfigExample example = new DeconstructVectorConfigExample();
-        example.createCriteria().andEnabledEqualTo((short) 1);
-        example.setOrderByClause("priority ASC");
-        return vectorConfigMapper.selectByExample(example);
-    }
-}

core/src/main/resources/mapper/pgVector/ext/VideoVectorMapperExt.xml

@@ -99,21 +99,6 @@
     LIMIT #{topN}
   </select>
 
-  <!-- 分页查询指定 configCode 下的记录 -->
-  <select id="selectByConfigCodePaged" resultMap="VectorWithEmbeddingResultMap">
-    SELECT id, video_id, config_code, point_index, created_at, updated_at
-    FROM video_vectors
-    WHERE config_code = #{configCode}
-    ORDER BY id
-    LIMIT #{limit} OFFSET #{offset}
-  </select>
-
-  <!-- 更新 text 字段 -->
-  <update id="updateTextById">
-    UPDATE video_vectors SET "text" = #{text}, updated_at = NOW()
-    WHERE id = #{id}
-  </update>
-
   <!-- 查询所有不重复的 video_id（跨所有 configCode） -->
   <select id="selectAllDistinctVideoIds" resultType="java.lang.Long">
     SELECT DISTINCT video_id FROM video_vectors

server/src/main/java/com/tzld/videoVector/controller/XxlJobController.java

@@ -20,9 +20,6 @@ public class XxlJobController {
     @Autowired
     private MaterialVectorJob materialVectorJob;
 
-    @Autowired
-    private VideoVectorTextBackfillJob videoVectorTextBackfillJob;
-
     @Autowired
     private VideoDetailSyncJob videoDetailSyncJob;
 
@@ -75,12 +72,6 @@ public class XxlJobController {
         return CommonResponse.success();
     }
 
-    @GetMapping("/videoVectorTextBackfillJob")
-    public CommonResponse<Void> videoVectorTextBackfillJob() {
-        videoVectorTextBackfillJob.videoVectorTextBackfillJob(null);
-        return CommonResponse.success();
-    }
-
     // ==================== 视频详情同步任务 ====================
 
     @GetMapping("/syncVideoDetailJob")