super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

task-38

Browse Source
This commit is contained in:
黄自达
2026-08-29 23:11:59 +08:00
parent a9a72c7bdf
commit 36bed2bdc6
5 changed files with 653 additions and 21 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/controller/ShopDataCrawlTaskController.java
+8 -4
View File
@@ -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<ShopDataCrawlHistoryVo> 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")
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/model/vo/ShopDataCrawlHistoryVo.java
+11 -2
View File
@@ -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<ShopDataCrawlResultItemVo> 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;
}
backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java
+88 -10
View File
@@ -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<FileResultEntity> entities = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.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<FileResultEntity>()
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.eq(FileResultEntity::getUserId, userId));
return count == null ? 0L : count;
}
public ShopDataCrawlTaskBatchVo getTaskProgressBatch(List<Long> taskIds) {
ShopDataCrawlTaskBatchVo batch = new ShopDataCrawlTaskBatchVo();
if (taskIds == null || taskIds.isEmpty()) {
@@ -294,10 +332,18 @@ public class ShopDataCrawlTaskService {
}
Map<Long, FileTaskEntity> taskMap = loadTaskMapByIds(normalizedTaskIds);
List<FileResultEntity> rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
.in(FileResultEntity::getTaskId, normalizedTaskIds)
.orderByAsc(FileResultEntity::getId));
List<FileResultEntity> rows;
try {
rows = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
.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<Long, List<FileResultEntity>> 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<ShopDataCrawlCountryResultDto> mergeSnapshotCountries(List<ShopDataCrawlCountryResultDto> base, List<ShopDataCrawlCountryResultDto> incoming) {
Map<String, ShopDataCrawlCountryResultDto> 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<ShopDataCrawlResultItemVo> snapshots, boolean terminal) {
if (terminal) {
persistSnapshotJson(task, snapshots);
} else if (snapshotCarriesCountryRows(snapshots) && !snapshotJsonHasCountryRows(task)) {
persistSnapshotJson(task, snapshots);
}
fileTaskMapper.updateById(task);
}
private boolean snapshotCarriesCountryRows(List<ShopDataCrawlResultItemVo> 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<ShopDataCrawlResultItemVo> snapshots) {
List<ShopDataCrawlResultItemVo> safe = snapshots == null ? List.of() : snapshots;
taskResultItemService.replaceTaskSnapshots(task.getId(), MODULE_TYPE, safe, new TaskResultItemService.SnapshotKeyResolver() {
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlProgressQueryTest.java
+537
View File
@@ -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<FileResultEntity> dbResultRows = new ArrayList<>();
private final List<FileTaskEntity> dbTasks = new ArrayList<>();
/** 内存 file_job 表：findAssembleJobsByResultIds / findAssembleJob 的仿真数据源。 */
private final List<TaskFileJobEntity> dbFileJobs = new ArrayList<>();
private final Map<Long, FileTaskEntity> 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<FileTaskEntity> 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<FileResultEntity> wrapper = invocation.getArgument(0);
LambdaQueryWrapper<FileResultEntity> query = (LambdaQueryWrapper<FileResultEntity>) 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<Long, ShopDataCrawlResultItemVo> 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<Long> 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<FileResultEntity> wrapper = invocation.getArgument(0);
LambdaQueryWrapper<FileResultEntity> query = (LambdaQueryWrapper<FileResultEntity>) wrapper;
String segment = query.getSqlSegment();
List<FileResultEntity> 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<FileTaskEntity> wrapper = invocation.getArgument(0);
if (wrapper instanceof LambdaQueryWrapper<?> query
&& query.getSqlSegment() != null
&& query.getSqlSegment().contains("IN")) {
java.util.Set<Long> wanted = new java.util.HashSet<>();
for (Object value : query.getParamNameValuePairs().values()) {
if (value instanceof Number number) {
wanted.add(number.longValue());
}
}
List<FileTaskEntity> 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<Long, TaskFileJobEntity> 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);
}
}
backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlScopeMergeTest.java
+9 -5
View File
@@ -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<String, ShopDataCrawlShopPayloadDto> 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));