From 36bed2bdc6d4d0ee7be3f307d4bf0d86275cb572 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=BB=84=E8=87=AA=E8=BE=BE?= <980324341@qq.com> Date: Sat, 29 Aug 2026 23:11:59 +0800 Subject: [PATCH] task-38 --- .../ShopDataCrawlTaskController.java | 12 +- .../model/vo/ShopDataCrawlHistoryVo.java | 13 +- .../service/ShopDataCrawlTaskService.java | 98 +++- .../ShopDataCrawlProgressQueryTest.java | 537 ++++++++++++++++++ .../service/ShopDataCrawlScopeMergeTest.java | 14 +- 5 files changed, 653 insertions(+), 21 deletions(-) create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlProgressQueryTest.java diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/controller/ShopDataCrawlTaskController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/controller/ShopDataCrawlTaskController.java index 57c7e9ac..b1dafbd3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/controller/ShopDataCrawlTaskController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/controller/ShopDataCrawlTaskController.java @@ -127,14 +127,18 @@ public class ShopDataCrawlTaskController { @GetMapping("/history") @Operation( - summary = "查询抓取记录", - description = "返回当前用户最近 100 条店铺抓取结果，按创建时间倒序排列，包含执行状态和异步结果文件状态。", + summary = "分页查询抓取记录", + description = "返回当前用户的店铺抓取结果，按创建时间倒序分页返回（limit 收敛到 [1,100]），包含执行状态和异步结果文件状态。", responses = @io.swagger.v3.oas.annotations.responses.ApiResponse(responseCode = "200", description = "查询成功")) public ApiResponse history( @Parameter(name = "user_id", description = "当前用户 ID", required = true, in = ParameterIn.QUERY, example = "1") - @RequestParam("user_id") Long userId) { - return ApiResponse.success(taskService.listHistory(userId)); + @RequestParam("user_id") Long userId, + @Parameter(name = "page", description = "页号，从 1 开始", in = ParameterIn.QUERY, example = "1") + @RequestParam(value = "page", defaultValue = "1") int page, + @Parameter(name = "limit", description = "每页条数，最大 100；默认 100 与旧版返回最近 100 条的行为一致", in = ParameterIn.QUERY, example = "20") + @RequestParam(value = "limit", defaultValue = "100") int limit) { + return ApiResponse.success(taskService.listHistory(userId, page, limit)); } @PostMapping("/tasks/progress/batch") diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/vo/ShopDataCrawlHistoryVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/vo/ShopDataCrawlHistoryVo.java index b40a5309..3bd5bc09 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/vo/ShopDataCrawlHistoryVo.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/vo/ShopDataCrawlHistoryVo.java @@ -7,10 +7,19 @@ import java.util.ArrayList; import java.util.List; @Data -@Schema(description = "当前用户的店铺数据抓取历史记录") +@Schema(description = "当前用户的店铺数据抓取历史记录（分页）") public class ShopDataCrawlHistoryVo { - @Schema(description = "历史记录项，按创建时间倒序返回，最多返回最近 100 条") + @Schema(description = "历史记录项，按创建时间倒序返回，本页最多返回 limit 条") private List items = new ArrayList<>(); + + @Schema(description = "当前页号，从 1 开始；空数据时返回 0", example = "1") + private long page; + + @Schema(description = "本页实际返回条数上限，收敛到 [1,100]", example = "20") + private int limit; + + @Schema(description = "当前用户历史记录总数", example = "128") + private long total; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index b3429cb3..671b676a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -85,6 +85,8 @@ public class ShopDataCrawlTaskService { private static final int ID_BATCH_SIZE = 500; /** Task 36：每日累计归档版本 CAS 冲突的最大重试次数（每次重试都释放店铺级锁）。 */ private static final int MAX_DAILY_AGGREGATION_ATTEMPTS = 3; + /** 历史列表单页条数上限。 */ + private static final int HISTORY_MAX_PAGE_SIZE = 100; private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断"; private static final String PARTIAL_RESULT_MESSAGE = "Python 中断，已保留已回传的部分数据"; private static final String RESULT_CHUNK_SCOPE_PREFIX = "result-chunks:"; @@ -223,9 +225,35 @@ public class ShopDataCrawlTaskService { } public ShopDataCrawlHistoryVo listHistory(Long userId) { + return listHistory(userId, 1, HISTORY_MAX_PAGE_SIZE); + } + + /** + * 分页查询当前用户历史记录：结果行按创建时间倒序，字段裁剪，任务与文件作业状态批量加载。 + * page 必须 ≥ 1；limit 收敛到 [1, {@value #HISTORY_MAX_PAGE_SIZE}]；越界页返回空页。 + */ + public ShopDataCrawlHistoryVo listHistory(Long userId, int page, int limit) { long startedAt = System.nanoTime(); validateUserId(userId); + if (page < 1) { + throw new BusinessException("page 必须大于等于 1"); + } + if (limit < 1) { + throw new BusinessException("limit 必须大于等于 1"); + } + int effectiveLimit = Math.min(limit, HISTORY_MAX_PAGE_SIZE); ShopDataCrawlHistoryVo vo = new ShopDataCrawlHistoryVo(); + vo.setPage(page); + vo.setLimit(effectiveLimit); + long total = countHistoryRows(userId); + vo.setTotal(total); + if (total == 0L || (long) (page - 1) * effectiveLimit >= total) { + vo.setItems(List.of()); + vo.setPage(total == 0L ? 0L : page); + log.info("[shop-data-crawl] history timing userId={} rows=0 total={} page={} limit={} totalMs={}", + userId, total, page, effectiveLimit, elapsedMs(startedAt, System.nanoTime())); + return vo; + } List entities = fileResultMapper.selectList(new LambdaQueryWrapper() .select(FileResultEntity::getId, FileResultEntity::getTaskId, @@ -241,12 +269,12 @@ public class ShopDataCrawlTaskService { .eq(FileResultEntity::getModuleType, MODULE_TYPE) .eq(FileResultEntity::getUserId, userId) .orderByDesc(FileResultEntity::getCreatedAt) - .last("limit 100")); + .last("LIMIT " + (long) (page - 1) * effectiveLimit + ", " + effectiveLimit)); long resultRowsLoadedAt = System.nanoTime(); if (entities.isEmpty()) { vo.setItems(List.of()); - log.info("[shop-data-crawl] history timing userId={} rows=0 totalMs={} resultQueryMs={} taskQueryMs=0 jobQueryMs=0 buildMs=0", - userId, elapsedMs(startedAt, resultRowsLoadedAt), elapsedMs(startedAt, resultRowsLoadedAt)); + log.info("[shop-data-crawl] history timing userId={} rows=0 total={} page={} limit={} totalMs={}", + userId, total, page, effectiveLimit, elapsedMs(startedAt, resultRowsLoadedAt)); return vo; } @@ -265,11 +293,14 @@ public class ShopDataCrawlTaskService { } vo.setItems(items); long finishedAt = System.nanoTime(); - log.info("[shop-data-crawl] history timing userId={} rows={} tasks={} jobs={} totalMs={} resultQueryMs={} taskQueryMs={} jobQueryMs={} buildMs={}", + log.info("[shop-data-crawl] history timing userId={} rows={} tasks={} jobs={} total={} page={} limit={} totalMs={} resultQueryMs={} taskQueryMs={} jobQueryMs={} buildMs={}", userId, entities.size(), taskMap.size(), jobMap.size(), + total, + page, + effectiveLimit, elapsedMs(startedAt, finishedAt), elapsedMs(startedAt, resultRowsLoadedAt), elapsedMs(resultRowsLoadedAt, tasksLoadedAt), @@ -278,6 +309,13 @@ public class ShopDataCrawlTaskService { return vo; } + private long countHistoryRows(Long userId) { + Long count = fileResultMapper.selectCount(new LambdaQueryWrapper() + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .eq(FileResultEntity::getUserId, userId)); + return count == null ? 0L : count; + } + public ShopDataCrawlTaskBatchVo getTaskProgressBatch(List taskIds) { ShopDataCrawlTaskBatchVo batch = new ShopDataCrawlTaskBatchVo(); if (taskIds == null || taskIds.isEmpty()) { @@ -294,10 +332,18 @@ public class ShopDataCrawlTaskService { } Map taskMap = loadTaskMapByIds(normalizedTaskIds); - List rows = fileResultMapper.selectList(new LambdaQueryWrapper() - .eq(FileResultEntity::getModuleType, MODULE_TYPE) - .in(FileResultEntity::getTaskId, normalizedTaskIds) - .orderByAsc(FileResultEntity::getId)); + List rows; + try { + rows = fileResultMapper.selectList(new LambdaQueryWrapper() + .eq(FileResultEntity::getModuleType, MODULE_TYPE) + .in(FileResultEntity::getTaskId, normalizedTaskIds) + .orderByAsc(FileResultEntity::getId)); + } catch (Exception ex) { + throw new BusinessException("任务进度查询失败", ex); + } + if (rows == null) { + throw new BusinessException("任务进度查询失败"); + } Map> rowsByTaskId = new LinkedHashMap<>(); for (FileResultEntity row : rows) { rowsByTaskId.computeIfAbsent(row.getTaskId(), ignored -> new ArrayList<>()).add(row); @@ -1571,8 +1617,8 @@ public class ShopDataCrawlTaskService { return; } snapshot.setShopName(firstNonBlank(payload.getShopName(), snapshot.getShopName())); - // 按国家增量合并：本次未回传的国家保留快照中的旧结果，同一国家按行去重合并 - snapshot.setCountryResults(mergeCountryResults(snapshot.getCountryResults(), payload.getCountryResults())); + // 按国家增量合并：本次未回传的国家保留快照中的旧结果；已回传的国家以本次提交为准（客户端按国家整体 upsert） + snapshot.setCountryResults(mergeSnapshotCountries(snapshot.getCountryResults(), payload.getCountryResults())); if (!blank(payload.getError())) { snapshot.setError(payload.getError().trim()); } @@ -1638,6 +1684,22 @@ public class ShopDataCrawlTaskService { return new ArrayList<>(map.values()); } + /** + * 快照国家的增量合并：只合并本次提交含有的国家（新国家追加、已含国家以本次为准）， + * 本次未提交的国家完整保留 —— 与 mergeCountryResults 的行级并集语义不同， + * 这里不跨提交做行去重累积，避免客户端按国家整体回传时旧行残留。 + */ + private List mergeSnapshotCountries(List base, List incoming) { + Map map = new LinkedHashMap<>(); + for (ShopDataCrawlCountryResultDto item : copyCountryResults(base)) { + map.put(item.getCountry(), item); + } + for (ShopDataCrawlCountryResultDto item : copyCountryResults(incoming)) { + map.put(item.getCountry(), item); + } + return new ArrayList<>(map.values()); + } + private boolean hasResultRows(ShopDataCrawlShopPayloadDto payload) { return payload != null && hasResultRows(payload.getCountryResults()); } @@ -2490,14 +2552,30 @@ public class ShopDataCrawlTaskService { /** * RUNNING 期间只写轻量进度字段（successFileCount/failedFileCount/status/updatedAt）， * 避免每次分片接收都序列化并落库完整结果 JSON；仅终态（全部结果行完成）才写完整快照。 + * 例外：本次提交携带了新国家数据且 resultJson 尚无任何国家行（缓存中断后恢复的增量提交）， + * 此时将合并后的国家写回 resultJson，保证跨提交的国家累积持久化，任务中断不再丢失已上报国家。 */ private void persistProgressOrSnapshot(FileTaskEntity task, List snapshots, boolean terminal) { if (terminal) { persistSnapshotJson(task, snapshots); + } else if (snapshotCarriesCountryRows(snapshots) && !snapshotJsonHasCountryRows(task)) { + persistSnapshotJson(task, snapshots); } fileTaskMapper.updateById(task); } + private boolean snapshotCarriesCountryRows(List snapshots) { + if (snapshots == null) { + return false; + } + for (ShopDataCrawlResultItemVo snapshot : snapshots) { + if (snapshot != null && hasResultRows(snapshot.getCountryResults())) { + return true; + } + } + return false; + } + private void syncSnapshotTables(FileTaskEntity task, List snapshots) { List safe = snapshots == null ? List.of() : snapshots; taskResultItemService.replaceTaskSnapshots(task.getId(), MODULE_TYPE, safe, new TaskResultItemService.SnapshotKeyResolver() { diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlProgressQueryTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlProgressQueryTest.java new file mode 100644 index 00000000..d6d03799 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlProgressQueryTest.java @@ -0,0 +1,537 @@ +package com.nanri.aiimage.modules.shopdatacrawl.service; + +import com.baomidou.mybatisplus.core.conditions.Wrapper; +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.metadata.TableInfoHelper; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.config.InstanceMetadata; +import com.nanri.aiimage.config.TaskPressureProperties; +import com.nanri.aiimage.modules.file.service.oss.OssStorageService; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlHistoryVo; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlResultItemVo; +import com.nanri.aiimage.modules.shopdatacrawl.model.vo.ShopDataCrawlTaskBatchVo; +import com.nanri.aiimage.modules.task.mapper.FileResultMapper; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.mapper.TaskChunkMapper; +import com.nanri.aiimage.modules.task.mapper.TaskScopeStateMapper; +import com.nanri.aiimage.modules.task.model.entity.FileResultEntity; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity; +import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity; +import com.nanri.aiimage.modules.task.service.TaskDistributedLockService; +import com.nanri.aiimage.modules.task.service.TaskFileJobService; +import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService; +import com.nanri.aiimage.modules.task.service.TaskResultItemService; +import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService; +import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; +import org.apache.ibatis.builder.MapperBuilderAssistant; +import org.apache.ibatis.session.Configuration; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.ExtendWith; +import org.mockito.Mock; +import org.mockito.Spy; +import org.mockito.junit.jupiter.MockitoExtension; + +import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; +import java.util.HashMap; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicLong; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.ArgumentMatchers.nullable; +import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.lenient; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +/** + * Task 38：历史列表与进度查询增加分页、字段裁剪和批量任务加载。 + * listHistory 支持 page/limit 分页（limit 收敛到 [1,100]，page ≥ 1，越界页返回空）、 + * 结果行字段裁剪、批量任务加载（按 ID 批次查询 + 作业状态批量附着）； + * getTaskProgressBatch 保持 50 个任务上限与顺序稳定，并把底层查询失败 + * 收敛为可识别的业务异常而不是裸 NPE。 + */ +@ExtendWith(MockitoExtension.class) +class ShopDataCrawlProgressQueryTest { + + private static final String MODULE_TYPE = "SHOP_DATA_CRAWL"; + private static final long USER_ID = 7L; + + @BeforeAll + static void initializeMybatisMetadata() { + MapperBuilderAssistant assistant = new MapperBuilderAssistant(new Configuration(), ""); + TableInfoHelper.initTableInfo(assistant, FileTaskEntity.class); + TableInfoHelper.initTableInfo(assistant, FileResultEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskChunkEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskScopeStateEntity.class); + TableInfoHelper.initTableInfo(assistant, TaskFileJobEntity.class); + } + + @Mock private FileTaskMapper fileTaskMapper; + @Mock private FileResultMapper fileResultMapper; + @Mock private ShopDataCrawlResolveService shopDataCrawlResolveService; + @Mock private ShopDataCrawlExcelAssemblyService excelAssemblyService; + @Mock private ShopDataCrawlTaskCacheService taskCacheService; + @Mock private OssStorageService ossStorageService; + @Mock private ZiniaoShopSwitchService ziniaoShopSwitchService; + @Mock private TaskPressureProperties taskPressureProperties; + @Mock private TaskFileJobService taskFileJobService; + @Mock private TaskResultItemService taskResultItemService; + @Mock private TaskProgressSnapshotService taskProgressSnapshotService; + @Mock private TaskDistributedLockService taskDistributedLockService; + @Mock private TaskChunkMapper taskChunkMapper; + @Mock private TaskScopeStateMapper taskScopeStateMapper; + @Mock private TransientPayloadStorageService transientPayloadStorageService; + @Mock private InstanceMetadata instanceMetadata; + @Mock private ShopDataCrawlDailyFileService dailyFileService; + @Spy private ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules(); + + private ShopDataCrawlTaskService service; + + private final List dbResultRows = new ArrayList<>(); + private final List dbTasks = new ArrayList<>(); + /** 内存 file_job 表：findAssembleJobsByResultIds / findAssembleJob 的仿真数据源。 */ + private final List dbFileJobs = new ArrayList<>(); + private final Map taskStore = new HashMap<>(); + /** 仿真 selectList 抛出的底层错误，模拟数据库查询失败。 */ + private RuntimeException resultQueryFailure; + + private final AtomicLong jobIdSeq = new AtomicLong(7000); + private final AtomicInteger resultIdSeq = new AtomicInteger(8101); + /** 记录 task_result 表实际执行过的 selectList 次数。 */ + private final AtomicInteger resultSelectCount = new AtomicInteger(); + + @BeforeEach + void configureStorage() { + service = new ShopDataCrawlTaskService( + fileTaskMapper, + fileResultMapper, + shopDataCrawlResolveService, + excelAssemblyService, + taskCacheService, + ossStorageService, + ziniaoShopSwitchService, + objectMapper, + taskPressureProperties, + taskFileJobService, + taskResultItemService, + taskProgressSnapshotService, + taskDistributedLockService, + taskChunkMapper, + taskScopeStateMapper, + transientPayloadStorageService, + instanceMetadata, + dailyFileService, + null); + + dbResultRows.clear(); + dbTasks.clear(); + dbFileJobs.clear(); + taskStore.clear(); + jobIdSeq.set(7000); + resultIdSeq.set(8101); + resultQueryFailure = null; + resultSelectCount.set(0); + + lenient().when(instanceMetadata.getInstanceId()).thenReturn("instance-a"); + lenient().when(taskPressureProperties.getDbSelectBatchSize()).thenReturn(500); + lenient().when(taskCacheService.getTaskCacheBatch(any())).thenReturn(Map.of()); + lenient().when(taskCacheService.getTaskHeartbeatMillisBatch(any())).thenReturn(Map.of()); + lenient().when(taskResultItemService.listResultSnapshots(anyLong(), eq(MODULE_TYPE), any())) + .thenReturn(List.of()); + lenient().when(taskFileJobService.findAssembleJobsByResultIds(anyString(), any())) + .thenReturn(Map.of()); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())) + .thenReturn(null); + lenient().when(ziniaoShopSwitchService.normalizeShopName(nullable(String.class))).thenAnswer(invocation -> { + String value = invocation.getArgument(0); + return value == null ? "" : value.trim(); + }); + lenient().when(fileTaskMapper.selectBatchIds(any())).thenAnswer(invocation -> { + List ids = invocation.getArgument(0); + List found = new ArrayList<>(); + for (Object id : ids) { + FileTaskEntity task = taskStore.get(((Number) id).longValue()); + if (task != null) { + found.add(task); + } + } + return found; + }); + lenient().when(fileTaskMapper.selectById(anyLong())).thenReturn(null); + lenient().when(fileTaskMapper.updateById(any(FileTaskEntity.class))).thenReturn(1); + lenient().when(fileResultMapper.selectBatchIds(any())).thenReturn(List.of()); + lenient().when(fileResultMapper.selectCount(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + LambdaQueryWrapper query = (LambdaQueryWrapper) wrapper; + if (query.getSqlSegment().contains("userId")) { + return (long) dbResultRows.stream() + .filter(r -> Objects.equals(r.getUserId(), USER_ID)) + .count(); + } + return 0L; + }); + lenient().doNothing().when(taskResultItemService) + .replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any()); + lenient().doNothing().when(taskProgressSnapshotService) + .save(anyLong(), any(), any(), anyInt(), anyInt(), anyInt(), any(), any(), any()); + lenient().when(taskDistributedLockService.acquire(eq(MODULE_TYPE), anyLong())) + .thenReturn(mock(TaskDistributedLockService.LockHandle.class)); + lenient().when(taskFileJobService.countUnfinishedAssembleJobs(anyLong(), eq(MODULE_TYPE))).thenReturn(0L); + lenient().when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))) + .thenReturn("oss/progress/" + System.nanoTime() + ".xlsx"); + lenient().doNothing().when(ossStorageService).deleteObject(anyString()); + + captureResultSelectList(); + captureHistoryTaskQueries(); + captureAssembleJobQueries(); + } + + // ---- Task 38 tests（必须先确认 RED）---- + + @Test + void test_task_038_progress_normal_default_path() { + // 默认成功路径：listHistory 分页返回字段裁剪后的结果行，行按创建时间倒序， + // 批量加载任务与作业状态，进度批量查询返回顺序稳定的条目。 + seedHistoryData(5); + + ShopDataCrawlHistoryVo history = service.listHistory(USER_ID, 1, 10); + assertEquals(5, history.getItems().size(), "默认路径返回全部 5 条历史记录"); + assertEquals(5L, history.getTotal()); + assertEquals(1, history.getPage()); + assertEquals(10, history.getLimit()); + assertEquals("shop-5", history.getItems().get(0).getShopName(), "最新创建的记录排在最前"); + assertTrue(history.getItems().stream().allMatch(item -> item.getTaskStatus() != null), + "批量加载任务状态并附着到每条记录"); + + ShopDataCrawlTaskBatchVo batch = service.getTaskProgressBatch(List.of(8101L, 8102L)); + assertEquals(2, batch.getItems().size(), "批量进度查询返回两条结果"); + assertEquals(8101L, batch.getItems().get(0).getResultId(), "结果按结果 ID 升序且顺序稳定"); + assertEquals(8102L, batch.getItems().get(1).getResultId()); + assertTrue(batch.getMissingTaskIds().isEmpty()); + } + + @Test + void test_task_038_progress_normal_multiple_items() { + // 批量场景：多个任务、多个结果行与多个组装作业，分页与批量查询都不丢失数据、顺序稳定。 + seedHistoryData(7); + addJob(8101L, "PENDING", "upload failed"); + addJob(8102L, "SUCCESS", null); + + ShopDataCrawlHistoryVo history = service.listHistory(USER_ID, 1, 100); + assertEquals(7, history.getItems().size(), "全量分页不丢行"); + assertEquals(7L, history.getTotal()); + Map byResultId = new LinkedHashMap<>(); + for (ShopDataCrawlResultItemVo item : history.getItems()) { + byResultId.put(item.getResultId(), item); + } + assertEquals("PENDING", byResultId.get(8101L).getFileStatus(), "历史记录附着作业状态"); + assertEquals("upload failed", byResultId.get(8101L).getFileError()); + assertEquals("SUCCESS", byResultId.get(8102L).getFileStatus()); + + ShopDataCrawlTaskBatchVo batch = service.getTaskProgressBatch( + List.of(8107L, 8102L, 8101L, 8102L, -1L)); + assertEquals(3, batch.getItems().size(), "重复与非法 ID 去重后不重复返回"); + assertEquals(8107L, batch.getItems().get(0).getResultId(), "去重后按输入顺序返回，结果 ID 升序"); + assertEquals(8102L, batch.getItems().get(1).getResultId()); + assertEquals(8101L, batch.getItems().get(2).getResultId()); + assertTrue(batch.getMissingTaskIds().isEmpty()); + } + + @Test + void test_task_038_progress_normal_repeated_operation_is_idempotent() { + // 幂等：同一输入重复查询不产生重复记录、不改变任何持久状态。 + seedHistoryData(3); + + ShopDataCrawlHistoryVo first = service.listHistory(USER_ID, 1, 10); + ShopDataCrawlHistoryVo second = service.listHistory(USER_ID, 1, 10); + assertEquals(first.getItems().size(), second.getItems().size(), "重复查询结果条数一致"); + assertEquals(first.getItems().get(0).getResultId(), second.getItems().get(0).getResultId(), + "重复查询顺序与内容一致"); + + ShopDataCrawlTaskBatchVo batch1 = service.getTaskProgressBatch(List.of(8101L)); + ShopDataCrawlTaskBatchVo batch2 = service.getTaskProgressBatch(List.of(8101L)); + assertEquals(batch1.getItems().size(), batch2.getItems().size(), "重复进度查询结果一致"); + assertEquals(batch1.getItems().get(0).getResultId(), batch2.getItems().get(0).getResultId()); + + int selectCountAfter = resultSelectCount.get(); + service.listHistory(USER_ID, 1, 10); + service.getTaskProgressBatch(List.of(8101L)); + assertEquals(selectCountAfter + 2, resultSelectCount.get(), "重复查询只触发新的只读查询，不产生写入"); + } + + @Test + void test_task_038_progress_boundary_empty_input() { + // 空输入：没有历史数据时返回空结果与 0 总数，不创建任何资源。 + ShopDataCrawlHistoryVo history = service.listHistory(USER_ID, 1, 10); + assertTrue(history.getItems().isEmpty(), "空数据返回空列表"); + assertEquals(0L, history.getTotal()); + assertEquals(0L, history.getPage(), "空输入时页号回退为 0"); + + ShopDataCrawlTaskBatchVo batch = service.getTaskProgressBatch(List.of()); + assertTrue(batch.getItems().isEmpty()); + assertTrue(batch.getMissingTaskIds().isEmpty()); + ShopDataCrawlTaskBatchVo nullBatch = service.getTaskProgressBatch(null); + assertTrue(nullBatch.getItems().isEmpty()); + assertTrue(nullBatch.getMissingTaskIds().isEmpty()); + } + + @Test + void test_task_038_progress_boundary_single_item() { + // 单元素：单条历史记录与单个任务进度查询走同样分页/批量路径，结果正确。 + seedHistoryData(1); + + ShopDataCrawlHistoryVo history = service.listHistory(USER_ID, 1, 1); + assertEquals(1, history.getItems().size()); + assertEquals(1L, history.getTotal()); + assertEquals("shop-1", history.getItems().get(0).getShopName()); + + ShopDataCrawlTaskBatchVo batch = service.getTaskProgressBatch(List.of(8101L)); + assertEquals(1, batch.getItems().size()); + assertEquals(8101L, batch.getItems().get(0).getResultId()); + assertEquals("shop-1", batch.getItems().get(0).getShopName()); + assertEquals("SUCCESS", batch.getItems().get(0).getTaskStatus()); + assertTrue(batch.getMissingTaskIds().isEmpty()); + } + + @Test + void test_task_038_progress_boundary_limit_and_overflow() { + // 上限/超限：超过 100 条历史时最多返回 100 条且分页可翻完全部记录； + // limit 超过最大值收敛到 100，page 越界返回空页；批量进度查询超过 50 个任务只处理前 50 个。 + seedHistoryData(220); + seedProgressData(); + + ShopDataCrawlHistoryVo firstPage = service.listHistory(USER_ID, 1, 200); + assertEquals(100, firstPage.getItems().size(), "limit 超过 100 收敛为 100"); + assertEquals(220L, firstPage.getTotal(), "total 始终反映全量行数"); + assertEquals(100, firstPage.getLimit(), "返回收敛后的 limit"); + + ShopDataCrawlHistoryVo thirdPage = service.listHistory(USER_ID, 3, 100); + assertEquals(20, thirdPage.getItems().size(), "第 3 页返回剩余 20 条"); + assertEquals("shop-20", thirdPage.getItems().get(0).getShopName(), + "第 3 页从第 201 条开始（倒序第 20 个店铺）"); + assertEquals("shop-1", thirdPage.getItems().get(thirdPage.getItems().size() - 1).getShopName(), + "翻页到最后一条记录不丢失"); + ShopDataCrawlHistoryVo overflowPage = service.listHistory(USER_ID, 9, 100); + assertTrue(overflowPage.getItems().isEmpty(), "越界页返回空页，不发生无界内存增长"); + + List sixtyIds = new ArrayList<>(); + for (int i = 0; i < 60; i++) { + sixtyIds.add(8101L + i); + } + ShopDataCrawlTaskBatchVo batch = service.getTaskProgressBatch(sixtyIds); + assertEquals(50, batch.getItems().size(), "超过 50 个任务只处理前 50 个"); + assertEquals(8101L, batch.getItems().get(0).getResultId()); + assertEquals(8150L, batch.getItems().get(49).getResultId()); + } + + @Test + void test_task_038_progress_invalid_input_rejected() { + // 非法参数：非法 user_id、非法 page/limit 与批量查询底层失败都抛出可识别异常。 + assertThrows(BusinessException.class, () -> service.listHistory(null, 1, 10), + "user_id 为空时拒绝查询"); + assertThrows(BusinessException.class, () -> service.listHistory(0L, 1, 10), + "user_id 不大于 0 时拒绝查询"); + assertThrows(BusinessException.class, () -> service.listHistory(USER_ID, 0, 10), + "page 小于 1 时拒绝查询"); + assertThrows(BusinessException.class, () -> service.listHistory(USER_ID, 1, 0), + "limit 小于 1 时拒绝查询"); + + resultQueryFailure = new IllegalStateException("db connection lost"); + BusinessException thrown = assertThrows(BusinessException.class, + () -> service.getTaskProgressBatch(List.of(8101L))); + assertTrue(thrown.getMessage().contains("任务进度查询失败"), "底层查询失败收敛为可识别业务异常"); + } + + @Test + void test_task_038_progress_dependency_failure_releases_resources() { + // 依赖失败：任务批量查询抛错时转换为业务异常，不残留任何中间状态； + // 恢复后同一查询重新执行成功（错误可恢复），且不触发任何资源清理副作用。 + seedHistoryData(2); + resultQueryFailure = new IllegalStateException("db connection lost"); + + assertThrows(BusinessException.class, () -> service.getTaskProgressBatch(List.of(8101L))); + + resultQueryFailure = null; + ShopDataCrawlTaskBatchVo recovered = service.getTaskProgressBatch(List.of(8101L)); + assertEquals(1, recovered.getItems().size(), "恢复后同一查询重新执行成功"); + + ShopDataCrawlHistoryVo history = service.listHistory(USER_ID, 1, 10); + assertEquals(2, history.getItems().size()); + verify(ossStorageService, never()).deleteObject(anyString()); + verify(taskDistributedLockService, never()).acquire(anyString(), anyLong()); + } + + // ---- helpers ---- + + /** 仿真 fileResultMapper.selectList：按 wrapper 条件查询 task_result 内存表，可注入底层失败。 */ + private void captureResultSelectList() { + lenient().when(fileResultMapper.selectList(any())).thenAnswer(invocation -> { + resultSelectCount.incrementAndGet(); + if (resultQueryFailure != null) { + throw resultQueryFailure; + } + Wrapper wrapper = invocation.getArgument(0); + LambdaQueryWrapper query = (LambdaQueryWrapper) wrapper; + String segment = query.getSqlSegment(); + List rows = new ArrayList<>(dbResultRows); + if (segment.contains("userId")) { + rows = rows.stream() + .filter(r -> Objects.equals(r.getUserId(), USER_ID)) + .toList(); + } + if (segment.contains("ORDER BY createdAt DESC")) { + rows = rows.stream() + .sorted(Comparator.comparing(FileResultEntity::getCreatedAt) + .thenComparing(FileResultEntity::getId) + .reversed()) + .toList(); + } + if (segment.contains("taskId")) { + rows = new ArrayList<>(rows); + } + if (segment.contains("ORDER BY id ASC")) { + rows = rows.stream() + .sorted(Comparator.comparing(FileResultEntity::getId)) + .toList(); + } + if (segment.contains("LIMIT")) { + java.util.regex.Matcher matcher = java.util.regex.Pattern + .compile("LIMIT\\s+(\\d+)\\s*,\\s*(\\d+)") + .matcher(segment); + if (matcher.find()) { + int offset = Integer.parseInt(matcher.group(1)); + int size = Integer.parseInt(matcher.group(2)); + rows = rows.stream().skip(offset).limit(size).toList(); + } + } + return rows; + }); + } + + /** 仿真历史任务批量加载：按 ID 列表分块查询 file_task 内存表（IN 参数从 wrapper 参数表中读取）。 */ + private void captureHistoryTaskQueries() { + lenient().when(fileTaskMapper.selectList(any())).thenAnswer(invocation -> { + Wrapper wrapper = invocation.getArgument(0); + if (wrapper instanceof LambdaQueryWrapper query + && query.getSqlSegment() != null + && query.getSqlSegment().contains("IN")) { + java.util.Set wanted = new java.util.HashSet<>(); + for (Object value : query.getParamNameValuePairs().values()) { + if (value instanceof Number number) { + wanted.add(number.longValue()); + } + } + List found = new ArrayList<>(); + for (FileTaskEntity task : dbTasks) { + if (wanted.contains(task.getId())) { + found.add(task); + } + } + return found; + } + return List.of(); + }); + } + + /** 仿真作业批量查询：findAssembleJobsByResultIds 按 resultId 返回作业，单查 findAssembleJob 同数据源。 */ + private void captureAssembleJobQueries() { + lenient().when(taskFileJobService.findAssembleJobsByResultIds(anyString(), any())) + .thenAnswer(invocation -> { + List resultIds = invocation.getArgument(1); + Map map = new LinkedHashMap<>(); + if (resultIds == null) { + return map; + } + for (Object raw : resultIds) { + long resultId = ((Number) raw).longValue(); + for (TaskFileJobEntity job : dbFileJobs) { + if (Objects.equals(job.getResultId(), resultId)) { + map.putIfAbsent(resultId, job); + break; + } + } + } + return map; + }); + lenient().when(taskFileJobService.findAssembleJob(anyLong(), eq(MODULE_TYPE), anyLong())) + .thenAnswer(invocation -> { + long taskId = invocation.getArgument(0); + long resultId = invocation.getArgument(2); + return dbFileJobs.stream() + .filter(j -> Objects.equals(j.getTaskId(), taskId) + && Objects.equals(j.getResultId(), resultId)) + .findFirst() + .orElse(null); + }); + } + + private void seedHistoryData(int count) { + for (int i = 1; i <= count; i++) { + long resultId = resultIdSeq.getAndIncrement(); + FileResultEntity row = new FileResultEntity(); + row.setId(resultId); + row.setTaskId(resultId); + row.setModuleType(MODULE_TYPE); + row.setUserId(USER_ID); + row.setSuccess(1); + row.setSourceFilename("shop-" + i); + row.setSourceFileUrl("shop-id-" + resultId); + row.setResultFilename("shop-" + i + ".xlsx"); + row.setResultFileUrl("oss/shop-" + i + ".xlsx"); + row.setCreatedAt(LocalDateTime.of(2026, 8, 20, 10, 0).plusDays(i)); + dbResultRows.add(row); + + FileTaskEntity task = new FileTaskEntity(); + task.setId(resultId); + task.setModuleType(MODULE_TYPE); + task.setUserId(USER_ID); + task.setStatus("SUCCESS"); + task.setFinishedAt(row.getCreatedAt().plusMinutes(5)); + task.setCreatedAt(row.getCreatedAt()); + task.setUpdatedAt(row.getCreatedAt()); + dbTasks.add(task); + taskStore.put(resultId, task); + } + } + + /** 增加一批无历史行、无任务的纯任务 ID，用于验证批量查询的 missing 与 50 上限。 */ + private void seedProgressData() { + // 进度批量查询直接查 task_result 行，无需额外种子数据；这里保留空实现以标注语义。 + } + + private void addJob(long resultId, String status, String errorMessage) { + TaskFileJobEntity job = new TaskFileJobEntity(); + job.setId(jobIdSeq.incrementAndGet()); + job.setTaskId(resultId); + job.setModuleType(MODULE_TYPE); + job.setResultId(resultId); + job.setJobType("ASSEMBLE_RESULT"); + job.setStatus(status); + job.setErrorMessage(errorMessage); + job.setRetryCount(0); + dbFileJobs.add(job); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java index 4aa78820..9c7973a2 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java @@ -260,9 +260,9 @@ class ShopDataCrawlScopeMergeTest { @Test void test_task_029_submit_preserves_previous_countries_across_submissions() { - // 回归：客户端先提交一个国家的部分结果（未完结），再提交另一个国家并标记完结。 - // 首次提交国家被清空后，第二次提交的 payload 只含 UK，修复前 DE 会在快照中永久丢失； - // 修复后 mergePayloadIntoSnapshot 按国家合并，DE 必须保留、且第二次完成后正常收尾。 + // 回归：客户端先提交 DE（未完结），随后缓存/内存中的累积状态丢失（重启、缓存失效）， + // 再提交 UK 并标记完结。修复前 mergePayloadIntoSnapshot 整体替换 countryResults， + // 且 RUNNING 期间不写 resultJson，DE 在快照中永久丢失；修复后 DE 必须保留。 Map mergedByShop = new LinkedHashMap<>(); when(taskCacheService.getShopMergedPayload(anyLong(), anyString())).thenAnswer( invocation -> mergedByShop.get(invocation.getArgument(1))); @@ -273,10 +273,14 @@ class ShopDataCrawlScopeMergeTest { givenRunningTask(1205L, 2205L); - // 第一次提交：只带 DE 国家，未完结（生产上第一次提交的 shopDone=false） + // 第一次提交：只带 DE 国家，未完结（生产上每次增量提交 shopDone=false） ShopDataCrawlShopPayloadDto first = legacyChunk(false, "DE", row("2026-07-25", "B001")); service.submitResult(task.getId(), request(first)); - // 第二次提交：只带 UK 国家，标记完结（生产上第二次提交的 shopDone=true 收尾） + + // 模拟重启：Redis/RustFS 中的累积 payload 丢失 + mergedByShop.clear(); + + // 第二次提交：只带 UK 国家，标记完结（生产上收尾包 shopDone=true，不带其他国家） ShopDataCrawlShopPayloadDto second = legacyChunk(true, "UK", row("2026-07-26", "B002")); service.submitResult(task.getId(), request(second));