diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java index 7f651b6a..bc4b8ef9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java @@ -97,6 +97,13 @@ public class CollectDataService { public static final String MODULE_TYPE = "COLLECT_DATA"; + /** + * task-50 上线初期写入端（CollectDataResultItemBatchWriter）曾误用小写 + * "collectdata"，遗留 5386 行存量数据；读取端查询统一 in(MODULE_TYPE, LEGACY_MODULE_TYPE) + * 兼容，否则这些任务的明细行在生成结果文件时查不出来。 + */ + private static final String LEGACY_MODULE_TYPE = "collectdata"; + public TaskProgressLightBatchVo progressLight(List taskIds) { return taskProgressLightAssembler.assemble(MODULE_TYPE, null, taskIds); } @@ -937,7 +944,7 @@ public class CollectDataService { private List loadFinalRows(Long taskId) { List rows = taskResultItemMapper.selectList(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, taskId) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) .orderByAsc(TaskResultItemEntity::getId)); if (rows == null || rows.isEmpty()) { return new ArrayList<>(); @@ -1258,7 +1265,7 @@ public class CollectDataService { .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)); taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, task.getId()) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE)); + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)); deleteTransientTaskPayloads( taskChunkMapper.selectList(new LambdaQueryWrapper() .select(TaskChunkEntity::getPayloadJson) @@ -1267,7 +1274,7 @@ public class CollectDataService { taskResultItemMapper.selectList(new LambdaQueryWrapper() .select(TaskResultItemEntity::getPayloadJson) .eq(TaskResultItemEntity::getTaskId, task.getId()) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE))); + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE))); taskFileJobService.deleteTaskJobs(task.getId(), MODULE_TYPE); fileTaskMapper.deleteById(task.getId()); } @@ -1280,11 +1287,11 @@ public class CollectDataService { List resultItems = taskResultItemMapper.selectList(new LambdaQueryWrapper() .select(TaskResultItemEntity::getPayloadJson) .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) .eq(TaskResultItemEntity::getResultId, row.getId())); taskResultItemMapper.delete(new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, row.getTaskId()) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) .eq(TaskResultItemEntity::getResultId, row.getId())); // 与 deleteTask 一致：先删 DB 行再物理删对象，保证行删除与对象删除一致。 deleteResultItemPayloads(resultItems); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java index 873ed1e8..9c3bdee2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriter.java @@ -27,7 +27,12 @@ import java.util.Objects; @Component public class CollectDataResultItemBatchWriter { - private static final String MODULE_TYPE = "collectdata"; + /** 必须与 CollectDataService.MODULE_TYPE 一致（大写 COLLECT_DATA）。 + * 曾因写端误用小写 "collectdata"，导致 task-50 之后所有采集任务的明细行 + * 在生成结果文件时查不出来（biz_task_result_item.module_type 大小写分叉）。 */ + private static final String MODULE_TYPE = "COLLECT_DATA"; + /** 历史存量小写值，查询现有行时兼容，避免旧任务重试重复插入（与 CollectDataService.LEGACY_MODULE_TYPE 对应）。 */ + private static final String LEGACY_MODULE_TYPE = "collectdata"; private static final int DEFAULT_BATCH_SIZE = 100; private final TaskResultItemMapper taskResultItemMapper; @@ -63,10 +68,12 @@ public class CollectDataResultItemBatchWriter { } String scopeHash = sha256(scopeKey); // 一次性取回本 scope 现有行，构建 item_key → 现有行 映射（hash 相等即跳过）。 + // 现有行查询兼容历史小写 "collectdata"（见 MODULE_TYPE 注释），避免旧任务重试 + // 在同一唯一键 (task_id, module_type, scope_hash, item_key) 下重复插入。 List existingList = taskResultItemMapper.selectList( new LambdaQueryWrapper() .eq(TaskResultItemEntity::getTaskId, taskId) - .eq(TaskResultItemEntity::getModuleType, MODULE_TYPE) + .in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE) .eq(TaskResultItemEntity::getScopeHash, scopeHash)); Map existingByKey = new HashMap<>(); for (TaskResultItemEntity existing : existingList) { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java index 8badb61e..f617786a 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataResultItemBatchWriterTest.java @@ -72,6 +72,45 @@ class CollectDataResultItemBatchWriterTest { assertEquals(1L, entities.get(0).getTaskId(), "task_id 保留"); } + @Test + void test_module_type_matches_reader_constant() { + // 回归：写入端 module_type 必须与 CollectDataService.MODULE_TYPE（读取端） + // 写同一大小写值，否则生成结果文件时 loadFinalRows 查不出来（曾用小写 + // "collectdata" 导致 task-50 之后所有采集任务明细行读不到）。 + when(taskResultItemMapper.selectList(any())).thenReturn(List.of()); + when(taskResultItemMapper.upsertBatch(anyList())).thenReturn(1); + + writer.upsertAccepted(1L, 2L, "task:1", 0, + List.of(row("B000000001", "Nike")), "rustfs:detail/abc"); + + ArgumentCaptor captor = ArgumentCaptor.forClass(List.class); + verify(taskResultItemMapper).upsertBatch(captor.capture()); + TaskResultItemEntity entity = (TaskResultItemEntity) captor.getValue().get(0); + assertEquals("COLLECT_DATA", entity.getModuleType(), "写入 module_type 须为读取端大写常量"); + } + + @Test + void test_legacy_lowercase_rows_are_found_on_retry() { + // 回归：历史小写 "collectdata" 存量行在旧任务重提时也要被现有行查询 + // 命中（按大小写兼容匹配），避免同一唯一键重复插入。 + String refJson = codec.encodeRef(0, 0, "rustfs:detail/x"); + TaskResultItemEntity legacy = new TaskResultItemEntity(); + legacy.setId(100L); + legacy.setModuleType("collectdata"); + legacy.setItemKey("asin:B000000001"); + legacy.setPayloadJson(refJson); + legacy.setPayloadHash(sha256(refJson)); + when(taskResultItemMapper.selectList(any())).thenReturn(List.of(legacy)); + + CollectDataResultItemBatchWriter.UpsertCounts counts = + writer.upsertAccepted(1L, 2L, "task:1", 0, + List.of(row("B000000001", "Nike")), "rustfs:detail/x"); + + assertEquals(0, counts.insertedOrUpdated(), "小写存量行命中 → 全部跳过"); + assertEquals(1, counts.skipped(), "1 行跳过"); + verify(taskResultItemMapper, never()).upsertBatch(anyList()); + } + @Test void test_task_050_task_normal_multiple_items() { // 批量场景：75 行（8 批次），顺序稳定不丢失，每批数量正确。