灵狐GEO系统
blame | 历史 | 补丁 | 提交 | 提交对比 | ignore whitespace
guyue
2 天以前 3f95bd04287246a76e113cc8044e1fcbbda2257b 
 src/main/java/com/linghu/controller/CollectController.java


@@ -226,6 +226,7 @@


               }, 1) // 限制并发数为1，确保顺序执行


               .collectList() // 收集所有批次的响应


               .flatMap(responses ->




                       saveKeywordTasks(keywordId, responses) // 保存关联关系


                               .thenReturn(responses) // 返回原始响应


               );


@@ -238,6 +239,7 @@


                    KeywordTask keywordTask = new KeywordTask();


                    keywordTask.setKeyword_id(keywordId);


                    keywordTask.setTask_id(response.getTask_id());


                    keywordTask.setStatus("pending");


                    return keywordTask;


                })


                .collect(Collectors.toList());


@@ -272,11 +274,12 @@


                .bodyToMono(new ParameterizedTypeReference<SearchTaskResponse>() {})


                .flatMap(taskResponse -> {


                    if (taskResponse != null && taskResponse.getTask_id() != null) {




                        // 使用 Reactor 的方式更新数据库


                        return Mono.fromRunnable(() -> {


                                    LambdaUpdateWrapper<Keyword> updateWrapper = new LambdaUpdateWrapper<>();


                                    updateWrapper.eq(Keyword::getKeyword_id, batchRequest.getKeyword_id());


                                    updateWrapper.set(Keyword::getStatus, "Submitted");


                                    updateWrapper.set(Keyword::getStatus, "submitted");


                                    updateWrapper.set(Keyword::getTask_id, taskResponse.getTask_id());


                                    keywordService.update(updateWrapper);


                                }).subscribeOn(Schedulers.boundedElastic()) // 在弹性线程池执行


@@ -342,9 +345,9 @@


                .map(data -> ResponseResult.success(data))


                .onErrorResume(e -> {


                    if (e.getMessage().contains("任务不存在")) {


                        return Mono.just(ResponseResult.error(404, "任务不存在"));


                        return Mono.just(ResponseResult.error(200, "任务不存在"));


                    } else if (e.getMessage().contains("无法取消")) {


                        return Mono.just(ResponseResult.error(400, "任务已完成，无法取消"));


                        return Mono.just(ResponseResult.error(200, "任务已完成，无法取消"));


                    }


                    return Mono.just(ResponseResult.error(500, "取消任务失败: " + e.getMessage()));


                });


@@ -453,20 +456,12 @@


    private Mono<Void> updateQuestionAndReference(TaskResultResponse result) {


        return Mono.fromRunnable(() -> {


            try {


                // 1. 更新关键词状态


                LambdaUpdateWrapper<Keyword> keywordUpdate = new LambdaUpdateWrapper<>();


                keywordUpdate.eq(Keyword::getTask_id, result.getTask_id())


                        .set(Keyword::getStatus, "completed");


                keywordService.update(keywordUpdate);




                // 1. 根据KeywordTask更新关键词状态


                // 查询关键词ID




                LambdaQueryWrapper<KeywordTask> keywordTaskWrapper = new LambdaQueryWrapper<>();


                keywordTaskWrapper.eq(KeywordTask::getTask_id, result.getTask_id());


                KeywordTask keywordTask = keywordTaskService.getOne(keywordTaskWrapper);




//                LambdaQueryWrapper<Keyword> keywordQuery = new LambdaQueryWrapper<>();


//                keywordQuery.eq(Keyword::getTask_id, keywordTask.getTask_id());


                Keyword keyword = keywordService.getById(keywordTask.getKeyword_id());




                if (keyword == null) {


@@ -475,6 +470,8 @@


                    throw new Exception("未找到关联的关键词，task_id: " + result.getTask_id());


//                    return;


                }


                keyword.setStatus("completed");


                keywordService.updateById(keyword);




                // 2. 批量查询所有问题


                LambdaQueryWrapper<Question> queryWrapper = new LambdaQueryWrapper<>();


@@ -536,6 +533,8 @@


                                                    reference.setUrl(ref.getUrl());


                                                    reference.setDomain(ref.getDomain());


                                                    reference.setNum(finalMaxNumByKeywordId);


                                                    reference.setTask_id(result.getTask_id());


                                                    reference.setKeyword_id(keyword.getKeyword_id());


                                                    //域名和平台id映射


                                                    reference.setCreate_time(LocalDateTime.now());


                                                    Platform platform = platformService.getPlatformByDomain(reference.getDomain());


@@ -621,6 +620,7 @@


    }




    @GetMapping("/health")


    @ApiOperation("健康检查")


    public Mono<HealthResponse> checkThirdPartyHealth() {


        return webClient.get()


                .uri(baseUrl + "/health") // 假设第三方健康检查接口路径为/health


@@ -634,6 +634,7 @@


     * 查询服务器资源


     */


    @GetMapping("/server/resource")


    @ApiOperation(value = "查询服务器资源")


    public Mono<ServerResourceResponse> getServerResource() {


        return webClient.get()


                .uri(baseUrl + "/api/v1/system/resources")


@@ -642,4 +643,9 @@


                .onErrorResume(e -> Mono.just(


                        new ServerResourceResponse( e.getMessage())));


    }


    /**


     * 传入orderid查所有关键词id以及关键词下面的所有任务id，轮询所有任务状态，如果状态为completed，则循环调用获取结果接口，处理结果


     */






}