| | |
|---|
| | | private PlatformService platformService; |
|---|
| | | @Autowired |
|---|
| | | private TypeService typeService; |
|---|
| | | @Autowired |
|---|
| | | private UserService userService; |
|---|
| | | @Autowired |
|---|
| | | private OrderService orderService; |
|---|
| | | |
|---|
| | | |
|---|
| | | /* @PostMapping("/search") |
|---|
| | |
|---|
| | | log.warn(errorMsg); |
|---|
| | | return Mono.just(ResponseResult.error(503, errorMsg)); |
|---|
| | | } |
|---|
| | | Integer keywordId = searchTaskRequest.getKeyword_id(); |
|---|
| | | |
|---|
| | | int maxConcurrentUsers = searchTaskRequest.getConfig() != null ? |
|---|
| | | searchTaskRequest.getConfig().getMax_concurrent_users() : 3; |
|---|
| | | List<List<UserDto>> userBatches = splitUsersIntoBatches(searchTaskRequest.getUsers(), maxConcurrentUsers); |
|---|
| | | List<List<UserDto>> userBatches = splitUsersIntoBatches(searchTaskRequest.getUsers(), maxConcurrentUsers,keywordId); |
|---|
| | | |
|---|
| | | // 获取 keywordId |
|---|
| | | Integer keywordId = searchTaskRequest.getKeyword_id(); |
|---|
| | | //分割 |
|---|
| | | |
|---|
| | | |
|---|
| | | return Flux.fromIterable(userBatches) |
|---|
| | | .flatMap(batch -> { |
|---|
| | | SearchTaskRequest batchRequest = new SearchTaskRequest(); |
|---|
| | |
|---|
| | | return Mono.fromRunnable(() -> { |
|---|
| | | boolean success = keywordTaskService.saveOrUpdateBatch(keywordTasks); |
|---|
| | | if (!success) { |
|---|
| | | throw new RuntimeException("保存关键词任务关联失败"); |
|---|
| | | // throw new RuntimeException("保存关键词任务关联失败"); |
|---|
| | | // 添加异常处理 |
|---|
| | | Mono.error( new RuntimeException("保存关键词任务关联失败")); |
|---|
| | | } |
|---|
| | | }) |
|---|
| | | .doFinally(signalType -> log.info("成功保存 {} 个关键词任务关联", keywordTasks.size())) |
|---|
| | |
|---|
| | | updateWrapper.set(Keyword::getStatus, "submitted"); |
|---|
| | | updateWrapper.set(Keyword::getTask_id, taskResponse.getTask_id()); |
|---|
| | | keywordService.update(updateWrapper); |
|---|
| | | //设置轮数 |
|---|
| | | Keyword keyword = keywordService.getById(batchRequest.getKeyword_id()); |
|---|
| | | |
|---|
| | | //设置订单进入采集状态 |
|---|
| | | LambdaUpdateWrapper<Orders> updateOrderWrapper = new LambdaUpdateWrapper<>(); |
|---|
| | | updateOrderWrapper.eq(Orders::getOrder_id, keyword.getOrder_id()) // 确保字段名正确 |
|---|
| | | .set(Orders::getStatus, 2); // 直接设置状态值 |
|---|
| | | |
|---|
| | | boolean success = orderService.update(updateOrderWrapper); |
|---|
| | | log.info("订单状态更新结果: {}", success ? "成功" : "失败"); |
|---|
| | | |
|---|
| | | }).subscribeOn(Schedulers.boundedElastic()) // 在弹性线程池执行 |
|---|
| | | .thenReturn(taskResponse); |
|---|
| | | } |
|---|
| | |
|---|
| | | .onStatus(HttpStatus::isError, response -> response.bodyToMono(TaskCancelResponse.class) |
|---|
| | | .flatMap(errorBody -> Mono.error(new RuntimeException(errorBody.getDetail())))) |
|---|
| | | .bodyToMono(TaskCancelResponse.class) |
|---|
| | | .flatMap(cancelResponse -> { |
|---|
| | | // 更新关键词状态 |
|---|
| | | Mono<Void> updateKeyword = Mono.fromRunnable(() -> { |
|---|
| | | LambdaUpdateWrapper<Keyword> updateWrapper = new LambdaUpdateWrapper<>(); |
|---|
| | | updateWrapper.eq(Keyword::getTask_id, taskId); |
|---|
| | | updateWrapper.set(Keyword::getStatus, "canceled"); // 统一使用"canceled" |
|---|
| | | keywordService.update(updateWrapper); |
|---|
| | | }) |
|---|
| | | .subscribeOn(Schedulers.boundedElastic()) |
|---|
| | | .then(); |
|---|
| | | |
|---|
| | | // 更新关键词任务状态 |
|---|
| | | Mono<Void> updateKeywordTask = Mono.fromRunnable(() -> { |
|---|
| | | LambdaUpdateWrapper<KeywordTask> updateWrapper = new LambdaUpdateWrapper<>(); |
|---|
| | | updateWrapper.eq(KeywordTask::getTask_id, taskId); |
|---|
| | | updateWrapper.set(KeywordTask::getStatus, "canceled"); // 统一使用"canceled" |
|---|
| | | keywordTaskService.update(updateWrapper); |
|---|
| | | }) |
|---|
| | | .subscribeOn(Schedulers.boundedElastic()) |
|---|
| | | .then(); |
|---|
| | | |
|---|
| | | // 并行执行两个更新操作，并在完成后返回cancelResponse |
|---|
| | | return Mono.when(updateKeyword, updateKeywordTask) |
|---|
| | | .thenReturn(cancelResponse); |
|---|
| | | }) |
|---|
| | | .map(data -> ResponseResult.success(data)) |
|---|
| | | .onErrorResume(e -> { |
|---|
| | | if (e.getMessage().contains("任务不存在")) { |
|---|
| | | return Mono.just(ResponseResult.error(200, "任务不存在")); |
|---|
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
|---|
| | | } else if (e.getMessage().contains("无法取消")) { |
|---|
| | | return Mono.just(ResponseResult.error(200, "任务已完成，无法取消")); |
|---|
| | | return Mono.just(ResponseResult.error(200, e.getMessage())); |
|---|
| | | } |
|---|
| | | return Mono.just(ResponseResult.error(500, "取消任务失败: " + e.getMessage())); |
|---|
| | | return Mono.just(ResponseResult.error(500, e.getMessage())); |
|---|
| | | }); |
|---|
| | | } |
|---|
| | | |
|---|
| | |
|---|
| | | }) |
|---|
| | | .onErrorResume(e -> { |
|---|
| | | System.out.println("获取任务结果失败"); |
|---|
| | | TaskResultResponse result = new TaskResultResponse(); |
|---|
| | | if (e.getMessage().contains("登陆失败")){ |
|---|
| | | |
|---|
| | | } |
|---|
| | | TaskResultResponse result = new TaskResultResponse(); |
|---|
| | | result.setDetail("获取任务结果失败: " + e.getMessage()); |
|---|
| | | return Mono.just(result); |
|---|
| | | }); |
|---|
| | |
|---|
| | | private Mono<Void> updateQuestionAndReference(TaskResultResponse result) { |
|---|
| | | return Mono.fromRunnable(() -> { |
|---|
| | | try { |
|---|
| | | //查看每个账号信息的status是否正常 |
|---|
| | | |
|---|
| | | // 1. 根据KeywordTask更新关键词状态 |
|---|
| | | // 查询关键词ID |
|---|
| | | LambdaQueryWrapper<KeywordTask> keywordTaskWrapper = new LambdaQueryWrapper<>(); |
|---|
| | |
|---|
| | | throw new Exception("未找到关联的关键词，task_id: " + result.getTask_id()); |
|---|
| | | // return; |
|---|
| | | } |
|---|
| | | keyword.setStatus("completed"); |
|---|
| | | keywordService.updateById(keyword); |
|---|
| | | LambdaQueryWrapper<KeywordTask> keywordTaskWrapper2 = new LambdaQueryWrapper<>(); |
|---|
| | | keywordTaskWrapper2.eq(KeywordTask::getKeyword_id, keyword.getKeyword_id()); |
|---|
| | | List<KeywordTask> keywordTasks = keywordTaskService.list(keywordTaskWrapper2); |
|---|
| | | |
|---|
| | | // 定义状态优先级：canceled > false > completed |
|---|
| | | String finalStatus = "completed"; // 默认状态为 completed |
|---|
| | | |
|---|
| | | for (KeywordTask task : keywordTasks) { |
|---|
| | | String status = task.getStatus(); |
|---|
| | | if ("canceled".equals(status)) { |
|---|
| | | finalStatus = "canceled"; |
|---|
| | | break; // 遇到 canceled 直接跳出循环，因为优先级最高 |
|---|
| | | } else if ("false".equals(status)) { |
|---|
| | | finalStatus = "false"; |
|---|
| | | // 不跳出循环，继续检查是否存在 canceled |
|---|
| | | } |
|---|
| | | } |
|---|
| | | // 更新关键词状态 |
|---|
| | | if (!finalStatus.equals(keyword.getStatus())) { |
|---|
| | | keyword.setStatus(finalStatus); |
|---|
| | | keywordService.updateById(keyword); |
|---|
| | | String orderId = keyword.getOrder_id(); |
|---|
| | | if (orderId == null || orderId.isEmpty()) { |
|---|
| | | System.out.println("关键词[" + keyword.getKeyword_id() + "]未关联订单，跳过订单状态更新"); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | |
|---|
| | | // 2. 查询该订单下的所有关键词 |
|---|
| | | LambdaQueryWrapper<Keyword> orderKeywordsWrapper = new LambdaQueryWrapper<>(); |
|---|
| | | orderKeywordsWrapper.eq(Keyword::getOrder_id, orderId); |
|---|
| | | List<Keyword> orderKeywords = keywordService.list(orderKeywordsWrapper); |
|---|
| | | |
|---|
| | | if (orderKeywords.isEmpty()) { |
|---|
| | | System.out.println("订单[" + orderId + "]下无关键词，跳过状态更新"); |
|---|
| | | return; |
|---|
| | | } |
|---|
| | | |
|---|
| | | // 3. 检查所有关键词的状态是否均为 completed 或 false |
|---|
| | | boolean allValid = orderKeywords.stream() |
|---|
| | | .allMatch(k -> "completed".equals(k.getStatus()) || "false".equals(k.getStatus())); |
|---|
| | | |
|---|
| | | // 4. 若所有关键词状态均有效，更新订单状态为3 |
|---|
| | | if (allValid) { |
|---|
| | | Orders orders = orderService.getById(orderId); |
|---|
| | | if (orders != null) { |
|---|
| | | orders.setStatus(3); // 假设Orders有Integer类型的status字段 |
|---|
| | | orderService.updateById(orders); |
|---|
| | | System.out.println("订单[" + orderId + "]所有关键词状态符合条件，已更新状态为3"); |
|---|
| | | } else { |
|---|
| | | System.out.println("未找到订单[" + orderId + "]，无法更新状态"); |
|---|
| | | } |
|---|
| | | } |
|---|
| | | } |
|---|
| | | Orders orders = orderService.getById(keyword.getOrder_id()); |
|---|
| | | |
|---|
| | | // 2. 批量查询所有问题 |
|---|
| | | LambdaQueryWrapper<Question> queryWrapper = new LambdaQueryWrapper<>(); |
|---|
