update

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/PipelineService.java

@@ -10,13 +10,6 @@ public interface PipelineService {
 
     void execute(ContentAnalyseParam param);
 
-//    /**
-//     * 执行流程
-//     *
-//     * @param dto
-//     */
-//    void execute(ContentUnderstandDTO dto);
-
     /**
      * 执行任务
      * @param contentDTO

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/ContentServiceImpl.java

@@ -9,7 +9,6 @@ import com.tzld.piaoquan.content.understanding.dao.mapper.CuPipelineMapper;
 import com.tzld.piaoquan.content.understanding.dao.mapper.CuTaskMapper;
 import com.tzld.piaoquan.content.understanding.model.dto.ContentDTO;
 import com.tzld.piaoquan.content.understanding.model.dto.TaskResultDTO;
-import com.tzld.piaoquan.content.understanding.model.dto.VideoResultDTO;
 import com.tzld.piaoquan.content.understanding.model.param.*;
 import com.tzld.piaoquan.content.understanding.model.po.CuPipeline;
 import com.tzld.piaoquan.content.understanding.model.po.CuPipelineExample;

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/DeepSeekGenerateContentAction.java

@@ -0,0 +1,74 @@
+package com.tzld.piaoquan.content.understanding.service.impl;
+
+import com.alibaba.fastjson.JSON;
+import com.tzld.piaoquan.content.understanding.common.base.CommonResponse;
+import com.tzld.piaoquan.content.understanding.common.enums.ContentTypeEnum;
+import com.tzld.piaoquan.content.understanding.model.param.ActionParam;
+import com.tzld.piaoquan.content.understanding.model.param.GeminiParam;
+import com.tzld.piaoquan.content.understanding.service.Action;
+import com.tzld.piaoquan.content.understanding.util.HttpClientUtil;
+import com.tzld.piaoquan.content.understanding.util.HttpPoolClient;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Service;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.concurrent.ThreadLocalRandom;
+
+/**
+ * @author supeng
+ */
+@Slf4j
+@Service(value = "geminiGenerateContentAction")
+public class DeepSeekGenerateContentAction implements Action {
+
+    @Value("${ai.deepSeekApi.url:http://ai-api-internal.piaoquantv.com/aigc-server/deepseek/native/chat/completions}")
+    private String deepSeekApiUrl;
+
+    @Value("#{'${deepseek.apikey.list}'.split(',')}")
+    private List<String> apiKeyList;
+
+    private static final HttpPoolClient httpPoolClient = HttpClientUtil.create(3000, 900000, 50, 200, 0, 3000);
+
+    private static final String MODEL = "gemini-2.0-flash";
+    private static final Double TEMPERATURE = 1.0;
+
+    @Override
+    public String execute(ActionParam param) {
+        Integer type = param.getType();
+        //类型校验
+        if (Objects.isNull(type)) {
+
+        }
+        ThreadLocalRandom random = ThreadLocalRandom.current();
+        int index = random.nextInt(apiKeyList.size());
+        String apiKey = apiKeyList.get(index);
+        GeminiParam geminiParam = new GeminiParam();
+        geminiParam.setType(type);
+        geminiParam.setPrompt(param.getPrompt());
+        if (Objects.equals(ContentTypeEnum.COVER.getValue(), type) || Objects.equals(ContentTypeEnum.VIDEO.getValue(), type)
+                || Objects.equals(ContentTypeEnum.AUDIO.getValue(), type) || Objects.equals(ContentTypeEnum.SRT.getValue(), type)
+                || Objects.equals(ContentTypeEnum.VTT.getValue(), type)) {
+            geminiParam.setMediaUrl(param.getInput());
+        }
+        geminiParam.setApiKey(apiKey);
+        geminiParam.setModel(MODEL);
+        geminiParam.setTemperature(TEMPERATURE);
+        log.info("geminiGenerateContentAction geminiParam = {}", JSON.toJSONString(geminiParam));
+        Optional<String> optionalS = httpPoolClient.postJson(deepSeekApiUrl, JSON.toJSONString(geminiParam));
+        log.info("geminiGenerateContentAction optionalS = {}", optionalS);
+        if (optionalS.isPresent()) {
+            CommonResponse<Map<String, Object>> commonResponse = JSON.parseObject(optionalS.get(), CommonResponse.class);
+            if (commonResponse.isSuccess() && Objects.nonNull(commonResponse.getData()) && Objects.nonNull(commonResponse.getData().get("result"))) {
+                String content = commonResponse.getData().get("result").toString();
+                //过滤一些特殊格式
+                content = content.replace("```json", "").replace("```", "");
+                return content;
+            }
+        }
+        return null;
+    }
+}

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/GeminiGenerateContentAction.java

@@ -31,7 +31,7 @@ public class GeminiGenerateContentAction implements Action {
     @Value("#{'${gemini.apikey.list}'.split(',')}")
     private List<String> apiKeyList;
 
-    private static final HttpPoolClient httpPoolClient = HttpClientUtil.create(3000, 600000, 50, 200, 0, 3000);
+    private static final HttpPoolClient httpPoolClient = HttpClientUtil.create(3000, 1200000, 50, 200, 0, 3000);
 
     private static final String MODEL = "gemini-2.0-flash";
     private static final Double TEMPERATURE = 1.0;

content-understanding-core/src/main/java/com/tzld/piaoquan/content/understanding/service/impl/PipelineServiceImpl.java

@@ -76,40 +76,6 @@ public class PipelineServiceImpl implements PipelineService {
         executeTreeNodeBFS(root, dto);
     }
 
-//    @Override
-//    public void execute(ContentUnderstandDTO dto) {
-//        // 1.数据准备
-//        Long pipelineId = dto.getPipelineId();
-//        // 1.新增任务
-//        String taskId = UUID.randomUUID().toString().replace("-", "") + System.currentTimeMillis();
-//        CuTask cuTask = new CuTask();
-//        cuTask.setTaskId(taskId);
-//        cuTask.setPipelineId(pipelineId);
-////        cuTask.setInput();
-//        int insertCount = cuTaskMapper.insertSelective(cuTask);
-//        // 2.获取pipeline配置
-//        Integer contentType = dto.getContentType();
-//        CuPipelineStepExample example = new CuPipelineStepExample();
-//        example.createCriteria().andPipelineIdEqualTo(pipelineId);
-//        List<CuPipelineStep> stepList = cuPipelineStepMapper.selectByExample(example);
-//        if (Objects.isNull(stepList) || stepList.isEmpty()) {
-//            throw new CommonException(ExceptionEnum.CONFIG_ERROR, "无配置 pipelineId：" + pipelineId);
-//        }
-//        // 3.构建tree结构
-//        TreeNode root = buildTree(stepList);
-//        root.setType(contentType);
-//        String content = getContent(dto, contentType);
-//        if (Objects.isNull(content)) {
-//            throw new CommonException(ExceptionEnum.DATA_ERROR, "数据异常content：" + content);
-//        }
-//        root.setInput(content);
-//        ContentAnalyseDTO dto1 = new ContentAnalyseDTO();
-//        dto1.setVideoId(dto.getVideoId());
-//        dto1.setContent(content);
-//        // 4.按照步骤执行pipeline每一步动作
-//        executeTreeNodeBFS(root, dto1);
-//    }
-
     @Override
     public String executeTask(ContentDTO contentDTO) {
         // 1.数据准备
@@ -145,28 +111,10 @@ public class PipelineServiceImpl implements PipelineService {
         dto1.setVideoId(contentDTO.getVideoId());
         dto1.setContent(content);
         // 4.按照步骤执行pipeline每一步动作
-        executeTreeNodeBFS(root, dto1);
+        executeTaskTreeNodeBFS(root, dto1);
         return taskId;
     }
 
-//    private String getContent(ContentUnderstandDTO dto, Integer contentType) {
-//        String content = null;
-//        if (Objects.equals(contentType, ContentTypeEnum.TITLE.getValue())) {
-//            content = dto.getTitle();
-//        } else if (Objects.equals(contentType, ContentTypeEnum.COVER.getValue())) {
-//            content = dto.getCoverUrl();
-//        } else if (Objects.equals(contentType, ContentTypeEnum.VIDEO.getValue())) {
-//            content = dto.getVideoUrl();
-//        } else if (Objects.equals(contentType, ContentTypeEnum.INTRODUCTION.getValue())) {
-//            content = dto.getIntroduction();
-//        } else if (Objects.equals(contentType, ContentTypeEnum.AUDIO.getValue())) {
-//            content = dto.getAudioUrl();
-//        } else if (Objects.equals(contentType, ContentTypeEnum.SRT_VTT.getValue())) {
-//            content = dto.getSrt();
-//        }
-//        return content;
-//    }
-
     public TreeNode buildTree(List<CuPipelineStep> stepList) {
         Map<Long, TreeNode> nodeMap = new HashMap<>();
         TreeNode root = null;
@@ -219,6 +167,49 @@ public class PipelineServiceImpl implements PipelineService {
             String input = currentNode.getInput() == null ? "" : currentNode.getInput();
             // 执行action
             String result = executeStep(step, input, type);
+            if (Objects.isNull(currentNode.getChildren()) || currentNode.getChildren().isEmpty()) {
+                if (Objects.nonNull(result)) {
+                    //异常不阻塞其他分支执行
+                    try {
+                        Map<String, Object> resultMap = JSON.parseObject(result, new TypeReference<Map<String, Object>>() {
+                        });
+                        resultMap.put("video_id", videoId);
+                        mongoTemplate.insert(resultMap, COLLECTION_NAME);
+                    } catch (Exception e) {
+                        log.error("step execute error step = {}, result = {}", e, JSON.toJSONString(step), result);
+                    }
+                }
+                continue;
+            }
+            for (TreeNode child : currentNode.getChildren()) {
+                child.setInput(result);
+                queue.offer(child);
+            }
+        }
+    }
+
+    /**
+     * 遍历执行
+     *
+     * @param root 根节点
+     * @param dto  数据
+     */
+    public void executeTaskTreeNodeBFS(TreeNode root, ContentAnalyseDTO dto) {
+        if (Objects.isNull(root) || Objects.isNull(dto) || Objects.isNull(dto.getVideoId())) {
+            return;
+        }
+        Queue<TreeNode> queue = new LinkedList<>();
+        queue.offer(root);
+
+        Long videoId = dto.getVideoId();
+
+        while (!queue.isEmpty()) {
+            TreeNode currentNode = queue.poll();
+            CuPipelineStep step = currentNode.getStep();
+            Integer type = currentNode.getType();
+            String input = currentNode.getInput() == null ? "" : currentNode.getInput();
+            // 执行action
+            String result = executeTaskStep(step, input, type);
             if (Objects.isNull(currentNode.getChildren()) || currentNode.getChildren().isEmpty()) {
                 if (Objects.nonNull(result)) {
                     //异常不阻塞其他分支执行
@@ -264,4 +255,26 @@ public class PipelineServiceImpl implements PipelineService {
         actionParam.setPrompt(prompt);
         return action.execute(actionParam);
     }
+
+    /**
+     * 执行step 动作
+     *
+     * @param step
+     * @param input
+     * @return
+     */
+    public String executeTaskStep(CuPipelineStep step, String input, Integer type) {
+        CuPrompt cuPrompt = cuPromptMapper.selectByPrimaryKey(step.getPromptId());
+        if (Objects.isNull(cuPrompt) || Objects.isNull(cuPrompt.getId())) {
+            return null;
+        }
+        String prompt = String.format(cuPrompt.getPrompt(), input);
+
+        Action action = actionMap.get(step.getAction());
+        ActionParam actionParam = new ActionParam();
+        actionParam.setType(type);
+        actionParam.setInput(input);
+        actionParam.setPrompt(prompt);
+        return action.execute(actionParam);
+    }
 }