EasyFlow-Tech-Dev/EasyFlow
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 修正自动导入完成提示

Browse Source 
- 根据关联文档真实状态识别已完成的历史失败批次

- 完成后自动移除提示并低频复查失败或中断批次
This commit is contained in:
陈子默
2026-08-26 18:58:42 +08:00
parent 2a383ef3f2
commit 1870ac4028
5 changed files with 248 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

19
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/DocumentImportBatchDtos.java
View File

@@ -252,6 +252,7 @@ public final class DocumentImportBatchDtos {
private BigInteger batchId;
private String importMode;
private String status;
private Boolean actualCompleted;
private Integer totalCount;
private Long totalBytes;
private Integer completedCount;
@@ -292,6 +293,24 @@ public final class DocumentImportBatchDtos {
this.status = status;
}
/**
* 获取批次关联任务是否已经按真实文档状态全部完成。
*
* @return 全部完成时返回 {@code true}
*/
public Boolean getActualCompleted() {
return actualCompleted;
}
/**
* 设置批次关联任务是否已经按真实文档状态全部完成。
*
* @param actualCompleted 是否全部完成
*/
public void setActualCompleted(Boolean actualCompleted) {
this.actualCompleted = actualCompleted;
}
public Integer getTotalCount() {
return totalCount;
}

75
easyflow-modules/easyflow-module-ai/src/main/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppService.java
View File

@@ -21,6 +21,7 @@ import tech.easyflow.ai.enums.DocumentImportBatchItemStage;
import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
import tech.easyflow.ai.enums.DocumentImportBatchStatus;
import tech.easyflow.ai.enums.DocumentImportMode;
import tech.easyflow.ai.enums.DocumentProcessStatus;
import tech.easyflow.ai.mapper.DocumentImportBatchItemMapper;
import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
import tech.easyflow.ai.mapper.DocumentMapper;
@@ -435,7 +436,79 @@ public class DocumentImportBatchAppService {
.orderBy(DocumentImportBatch::getCreated, false)
.limit(1)
);
return batch == null ? null : batchTracker.toStatusResponse(batch);
if (batch == null) {
return null;
}
DocumentImportBatchDtos.StatusResponse response =
batchTracker.toStatusResponse(batch);
response.setActualCompleted(isAutoBatchActuallyCompleted(batch));
return response;
}
/**
* 根据批次项及其关联文档的真实状态判断失败批次是否已经完成。
*
* <p>正常完成批次直接返回成功；仅对部分失败或中断批次执行补充查询，
* 避免运行中轮询产生额外数据库压力。批次项数量不完整、文档缺失、
* 跨知识库或文档仍未完成时均保持失败提示。</p>
*
* @param batch 自动导入批次
* @return 批次关联任务是否已经全部完成
*/
private boolean isAutoBatchActuallyCompleted(DocumentImportBatch batch) {
if (DocumentImportBatchStatus.COMPLETED.name().equals(batch.getStatus())) {
return true;
}
if (!DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name().equals(batch.getStatus())
&& !DocumentImportBatchStatus.INTERRUPTED.name().equals(batch.getStatus())) {
return false;
}
List<DocumentImportBatchItem> items = itemService.list(
QueryWrapper.create()
.eq(DocumentImportBatchItem::getBatchId, batch.getId())
.orderBy(DocumentImportBatchItem::getId, true)
);
int totalCount = valueOrZero(batch.getTotalCount());
if (items == null || totalCount <= 0 || items.size() != totalCount) {
return false;
}
Set<BigInteger> unresolvedDocumentIds = new LinkedHashSet<BigInteger>();
for (DocumentImportBatchItem item : items) {
String itemStatus = item.getStatus();
if (DocumentImportBatchItemStatus.COMPLETED.name().equals(itemStatus)
|| DocumentImportBatchItemStatus.SKIPPED.name().equals(itemStatus)
|| DocumentImportBatchItemStatus.CANCELLED.name().equals(itemStatus)) {
continue;
}
if (item.getDocumentId() == null
|| !batch.getKnowledgeId().equals(item.getKnowledgeId())) {
return false;
}
unresolvedDocumentIds.add(item.getDocumentId());
}
if (unresolvedDocumentIds.isEmpty()) {
return true;
}
List<tech.easyflow.ai.entity.Document> completedDocuments =
documentMapper.selectListByQuery(
QueryWrapper.create()
.select(tech.easyflow.ai.entity.Document::getId)
.eq(tech.easyflow.ai.entity.Document::getCollectionId,
batch.getKnowledgeId())
.eq(tech.easyflow.ai.entity.Document::getProcessStatus,
DocumentProcessStatus.COMPLETED.name())
.in(tech.easyflow.ai.entity.Document::getId,
unresolvedDocumentIds)
);
if (completedDocuments == null) {
return false;
}
Set<BigInteger> completedDocumentIds = completedDocuments.stream()
.map(tech.easyflow.ai.entity.Document::getId)
.collect(java.util.stream.Collectors.toSet());
return completedDocumentIds.containsAll(unresolvedDocumentIds);
}
/**

84
easyflow-modules/easyflow-module-ai/src/test/java/tech/easyflow/ai/documentimport/task/DocumentImportBatchAppServiceTest.java
View File

@@ -19,6 +19,7 @@ import tech.easyflow.ai.enums.DocumentImportBatchItemStage;
import tech.easyflow.ai.enums.DocumentImportBatchItemStatus;
import tech.easyflow.ai.enums.DocumentImportBatchStatus;
import tech.easyflow.ai.enums.DocumentImportMode;
import tech.easyflow.ai.enums.DocumentProcessStatus;
import tech.easyflow.ai.mapper.DocumentImportBatchItemMapper;
import tech.easyflow.ai.mapper.DocumentImportBatchMapper;
import tech.easyflow.ai.mapper.DocumentMapper;
@@ -47,6 +48,89 @@ import java.util.function.BooleanSupplier;
*/
public class DocumentImportBatchAppServiceTest {
/**
* 验证历史失败批次关联文档均已完成时返回真实完成标记。
*/
@Test
public void latestAutoBatchShouldDetectActuallyCompletedDocuments() {
TestContext context = createContext();
DocumentImportBatch batch = context.batchService.getOne(
QueryWrapper.create()
);
batch.setImportMode(DocumentImportMode.AUTO.name());
batch.setStatus(DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name());
batch.setTotalCount(2);
DocumentImportBatchItem completed = uploadedItem(
BigInteger.valueOf(11), batch.getId()
);
completed.setStatus(DocumentImportBatchItemStatus.COMPLETED.name());
completed.setDocumentId(BigInteger.valueOf(101));
DocumentImportBatchItem staleFailed = uploadedItem(
BigInteger.valueOf(12), batch.getId()
);
staleFailed.setStatus(DocumentImportBatchItemStatus.FAILED.name());
staleFailed.setDocumentId(BigInteger.valueOf(102));
Mockito.when(context.itemService.list(Mockito.any(QueryWrapper.class)))
.thenReturn(List.of(completed, staleFailed));
tech.easyflow.ai.entity.Document recovered =
new tech.easyflow.ai.entity.Document();
recovered.setId(staleFailed.getDocumentId());
recovered.setCollectionId(batch.getKnowledgeId());
recovered.setProcessStatus(DocumentProcessStatus.COMPLETED.name());
Mockito.when(context.documentMapper.selectListByQuery(Mockito.any()))
.thenReturn(List.of(recovered));
DocumentImportBatchDtos.StatusResponse batchResponse =
new DocumentImportBatchDtos.StatusResponse();
batchResponse.setStatus(batch.getStatus());
Mockito.when(context.batchTracker.toStatusResponse(batch))
.thenReturn(batchResponse);
DocumentImportBatchDtos.StatusResponse response =
context.service.getLatestAutoBatch(batch.getKnowledgeId());
Assert.assertTrue(response.getActualCompleted());
Mockito.verify(context.documentMapper)
.selectListByQuery(Mockito.any(QueryWrapper.class));
}
/**
* 验证失败项关联文档仍未完成时继续保留批次提示。
*/
@Test
public void latestAutoBatchShouldKeepIncompleteFailureVisible() {
TestContext context = createContext();
DocumentImportBatch batch = context.batchService.getOne(
QueryWrapper.create()
);
batch.setImportMode(DocumentImportMode.AUTO.name());
batch.setStatus(DocumentImportBatchStatus.PARTIAL_SUCCEEDED.name());
batch.setTotalCount(1);
DocumentImportBatchItem failed = uploadedItem(
BigInteger.valueOf(13), batch.getId()
);
failed.setStatus(DocumentImportBatchItemStatus.FAILED.name());
failed.setDocumentId(BigInteger.valueOf(103));
Mockito.when(context.itemService.list(Mockito.any(QueryWrapper.class)))
.thenReturn(List.of(failed));
Mockito.when(context.documentMapper.selectListByQuery(Mockito.any()))
.thenReturn(List.of());
DocumentImportBatchDtos.StatusResponse batchResponse =
new DocumentImportBatchDtos.StatusResponse();
batchResponse.setStatus(batch.getStatus());
Mockito.when(context.batchTracker.toStatusResponse(batch))
.thenReturn(batchResponse);
DocumentImportBatchDtos.StatusResponse response =
context.service.getLatestAutoBatch(batch.getKnowledgeId());
Assert.assertFalse(response.getActualCompleted());
}
/**
* 验证上传领取通过单条多表更新同步刷新批次进度时间。
*/