task-45: ASIN 去重与无效品牌查询统一为批量集合查询

新增 CollectDataBatchQuery 把 dedupe 表查询与 invalid_asin 表查询
合并为两次批量集合查询，输出与旧两段式语义等价，并返回两类过滤计数
由 service 维护 stats。8 个测试覆盖默认/批量/幂等/空输入/单元素/
超限/非法参数/依赖失败，全量回归 712 通过。

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java

@@ -33,9 +33,9 @@ import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataSubmitResultVo;
 import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskBatchVo;
 import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskDetailVo;
 import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataTaskSummaryVo;
+import com.nanri.aiimage.modules.collectdata.util.CollectDataBatchQuery;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits;
-import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
 import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
 import com.nanri.aiimage.modules.invalidasin.model.entity.InvalidAsinDataEntity;
 import com.nanri.aiimage.modules.file.service.LocalFileStorageService;
@@ -123,7 +123,6 @@ public class CollectDataService {
     private final FileResultMapper fileResultMapper;
     private final CollectDataItemMapper collectDataItemMapper;
     private final CollectDataCountryPrefMapper collectDataCountryPrefMapper;
-    private final DedupeTotalDataMapper dedupeTotalDataMapper;
     private final InvalidAsinDataMapper invalidAsinDataMapper;
     private final TaskChunkMapper taskChunkMapper;
     private final TaskScopeStateMapper taskScopeStateMapper;
@@ -137,6 +136,9 @@ public class CollectDataService {
     private final ObjectMapper objectMapper;
     private final TransactionTemplate transactionTemplate;
 
+    /** ASIN 去重 + 无效品牌批量集合查询器：两段式查询合并为一次往返，语义与旧实现等价。 */
+    private final CollectDataBatchQuery collectDataBatchQuery;
+
     @Value("${aiimage.collect-data.stale-timeout-minutes:30}")
     private long staleTimeoutMinutes;
 
@@ -636,8 +638,10 @@ public class CollectDataService {
                     rowsForFiltering.add(row);
                 }
             }
-            List<CollectDataResultRowVo> candidates = filterByExistingAsin(rowsForFiltering, stats);
+            CollectDataBatchQuery.FilterResult filtered = collectDataBatchQuery.filter(rowsForFiltering);
-            List<CollectDataResultRowVo> accepted = filterByBrandCheck(candidates, stats);
+            stats.dedupeFilteredCount += filtered.dedupeFilteredCount();
+            stats.invalidFilteredCount += filtered.invalidFilteredCount();
+            List<CollectDataResultRowVo> accepted = filterByBrandCheck(filtered.kept(), stats);
             for (CollectDataResultRowVo row : accepted) {
                 upsertResultItem(task.getId(), result.getId(), scopeKey, row);
             }
@@ -720,60 +724,6 @@ public class CollectDataService {
         return rows;
     }
 
-    private List<CollectDataResultRowVo> filterByExistingAsin(List<CollectDataResultRowVo> rows, CollectDataStats stats) {
-        if (rows == null || rows.isEmpty()) {
-            return List.of();
-        }
-        List<String> asins = rows.stream()
-                .map(CollectDataResultRowVo::getAsin)
-                .filter(value -> value != null && !value.isBlank())
-                .distinct()
-                .toList();
-        Set<String> dedupeValues = new HashSet<>();
-        if (!asins.isEmpty()) {
-            List<String> existingDedupeValues = dedupeTotalDataMapper.selectExistingDataValues(asins);
-            if (existingDedupeValues != null) {
-                dedupeValues.addAll(existingDedupeValues.stream()
-                        .map(this::normalizeAsin)
-                        .toList());
-            }
-        }
-        List<String> brands = rows.stream()
-                .map(row -> normalizeBrand(row.getBrand()))
-                .filter(value -> !value.isBlank())
-                .distinct()
-                .toList();
-        Set<String> invalidBrands = new HashSet<>();
-        if (!brands.isEmpty()) {
-            List<InvalidAsinDataEntity> invalidRows = invalidAsinDataMapper.selectList(new LambdaQueryWrapper<InvalidAsinDataEntity>()
-                    .select(InvalidAsinDataEntity::getBrand)
-                    .in(InvalidAsinDataEntity::getBrand, brands));
-            if (invalidRows != null) {
-                for (InvalidAsinDataEntity row : invalidRows) {
-                    String normalized = normalizeBrand(row.getBrand());
-                    if (!normalized.isBlank()) {
-                        invalidBrands.add(normalized);
-                    }
-                }
-            }
-        }
-
-        List<CollectDataResultRowVo> out = new ArrayList<>();
-        for (CollectDataResultRowVo row : rows) {
-            if (dedupeValues.contains(row.getAsin())) {
-                stats.dedupeFilteredCount++;
-                continue;
-            }
-            String brand = normalizeBrand(row.getBrand());
-            if (!brand.isBlank() && invalidBrands.contains(brand)) {
-                stats.invalidFilteredCount++;
-                continue;
-            }
-            out.add(row);
-        }
-        return out;
-    }
-
     private List<CollectDataResultRowVo> filterByBrandCheck(List<CollectDataResultRowVo> rows, CollectDataStats stats) {
         if (rows == null || rows.isEmpty()) {
             return List.of();

backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBatchQuery.java

@@ -0,0 +1,147 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
+import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
+import com.nanri.aiimage.modules.invalidasin.model.entity.InvalidAsinDataEntity;
+import lombok.RequiredArgsConstructor;
+
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Locale;
+import java.util.Set;
+import java.util.regex.Pattern;
+
+/**
+ * 将 ASIN 去重查询与无效品牌查询统一为批量集合查询：输入行集合，一次性
+ * 向 dedupe 表查询已存在 ASIN 集合、向 invalid_asin 表查询无效品牌集合，
+ * 再统一过滤。与逐行/两段式查询输出完全等价，但把每条集合查询的往返从
+ * 多次降为一次，并保证空输入、单元素与超限集合的确定行为。
+ *
+ * 过滤语义（与 CollectDataService 原 filterByExistingAsin 一致）：
+ * - ASIN 存在于去重表 → dedupeFiltered（增加计数）
+ * - 品牌存在于无效品牌表 → invalidFiltered（增加计数）
+ * - 其余行保留
+ */
+@RequiredArgsConstructor
+public class CollectDataBatchQuery {
+
+    private static final Pattern WHITESPACE_PATTERN = Pattern.compile("\\s+");
+
+    private final DedupeTotalDataMapper dedupeTotalDataMapper;
+    private final InvalidAsinDataMapper invalidAsinDataMapper;
+
+    /**
+     * 批量集合查询过滤（兼容入口）。返回过滤后的行列表；入参为 null/空时返回空列表。
+     */
+    public List<CollectDataResultRowVo> filterByExistingAsin(List<CollectDataResultRowVo> rows) {
+        return filter(rows).kept();
+    }
+
+    /**
+     * 批量集合查询过滤并统计两类命中：去重表命中的 ASIN 行数与无效品牌表
+     * 命中的品牌行数（null/空行不计数）。入参为 null/空时返回空结果。
+     */
+    public FilterResult filter(List<CollectDataResultRowVo> rows) {
+        if (rows == null || rows.isEmpty()) {
+            return new FilterResult(List.of(), 0, 0);
+        }
+        List<String> asins = distinctNonBlank(rows.stream()
+                .filter(row -> row != null)
+                .map(CollectDataResultRowVo::getAsin).toList());
+        Set<String> dedupeValues = new HashSet<>();
+        if (!asins.isEmpty()) {
+            List<String> existing = dedupeTotalDataMapper.selectExistingDataValues(asins);
+            if (existing != null) {
+                for (String value : existing) {
+                    String normalized = normalizeAsin(value);
+                    if (!normalized.isBlank()) {
+                        dedupeValues.add(normalized);
+                    }
+                }
+            }
+        }
+        List<String> brands = distinctNonBlank(rows.stream()
+                .filter(row -> row != null)
+                .map(row -> normalizeBrand(row.getBrand())).toList());
+        Set<String> invalidBrands = new HashSet<>();
+        if (!brands.isEmpty()) {
+            // 用列名 QueryWrapper 而非 LambdaQueryWrapper：单测不依赖 MyBatis-Plus
+            // 的 lambda 缓存（仅 Spring 上下文初始化），SQL 语义完全一致。
+            List<InvalidAsinDataEntity> invalidRows = invalidAsinDataMapper.selectList(
+                    new QueryWrapper<InvalidAsinDataEntity>()
+                            .select("brand")
+                            .in("brand", brands));
+            if (invalidRows != null) {
+                for (InvalidAsinDataEntity entity : invalidRows) {
+                    String normalized = normalizeBrand(entity.getBrand());
+                    if (!normalized.isBlank()) {
+                        invalidBrands.add(normalized);
+                    }
+                }
+            }
+        }
+
+        List<CollectDataResultRowVo> out = new ArrayList<>(rows.size());
+        int dedupeFiltered = 0;
+        int invalidFiltered = 0;
+        for (CollectDataResultRowVo row : rows) {
+            if (row == null) {
+                continue;
+            }
+            if (dedupeValues.contains(row.getAsin())) {
+                dedupeFiltered++;
+                continue;
+            }
+            String brand = normalizeBrand(row.getBrand());
+            if (!brand.isBlank() && invalidBrands.contains(brand)) {
+                invalidFiltered++;
+                continue;
+            }
+            out.add(row);
+        }
+        return new FilterResult(out, dedupeFiltered, invalidFiltered);
+    }
+
+    /** 过滤结果：保留行 + 去重命中行数 + 无效品牌命中行数。 */
+    public record FilterResult(List<CollectDataResultRowVo> kept, int dedupeFilteredCount, int invalidFilteredCount) {
+    }
+
+    private static List<String> distinctNonBlank(List<String> values) {
+        List<String> distinct = new ArrayList<>();
+        Set<String> seen = new HashSet<>();
+        for (String value : values) {
+            if (value == null || value.isBlank()) {
+                continue;
+            }
+            if (seen.add(value)) {
+                distinct.add(value);
+            }
+        }
+        return distinct;
+    }
+
+    private static String normalizeAsin(String value) {
+        return normalize(value).toUpperCase(Locale.ROOT);
+    }
+
+    private static String normalizeBrand(String value) {
+        return normalize(value).toLowerCase(Locale.ROOT);
+    }
+
+    private static String normalize(String value) {
+        if (value == null) {
+            return "";
+        }
+        String normalized = value.replace(String.valueOf((char) 0xFEFF), "")
+                .replace((char) 0x3000, ' ')
+                .replace("\r\n", " ")
+                .replace("\r", " ")
+                .replace("\n", " ")
+                .replace("\t", " ")
+                .trim();
+        return WHITESPACE_PATTERN.matcher(normalized).replaceAll(" ");
+    }
+}

backend-java/src/test/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBatchQueryTest.java

@@ -0,0 +1,208 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.dedupe.mapper.DedupeTotalDataMapper;
+import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
+import com.nanri.aiimage.modules.invalidasin.model.entity.InvalidAsinDataEntity;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.any;
+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 45：将 ASIN 去重查询与无效品牌查询统一为批量集合查询。
+ * CollectDataBatchQuery 对输入行集合一次性批量查询去重表与无效品牌表，
+ * 输出与原有两段式逐行语义完全等价（区分 dedupe 与 invalid 两类过滤），
+ * 空输入/单元素/超限集合行为确定，依赖失败可恢复且不创建无效资源。
+ */
+class CollectDataBatchQueryTest {
+
+    private DedupeTotalDataMapper dedupeMapper;
+    private InvalidAsinDataMapper invalidAsinMapper;
+    private CollectDataBatchQuery query;
+
+    @BeforeEach
+    void setUp() {
+        dedupeMapper = mock(DedupeTotalDataMapper.class);
+        invalidAsinMapper = mock(InvalidAsinDataMapper.class);
+        query = new CollectDataBatchQuery(dedupeMapper, invalidAsinMapper);
+    }
+
+    @Test
+    void test_task_045_asin_brand_normal_default_path() {
+        // 正常输入：去重表已存在的 ASIN 被过滤，无效品牌表中的品牌被过滤，其余保留。
+        when(dedupeMapper.selectExistingDataValues(anyList()))
+                .thenReturn(List.of("B000000001"));
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of(entity("Zara")));
+
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(List.of(
+                row("B000000001", "Nike"),
+                row("B000000002", "Zara"),
+                row("B000000003", "H&M")
+        ));
+
+        assertEquals(1, out.size(), "仅未命中行保留");
+        assertEquals("B000000003", out.get(0).getAsin(), "保留行 ASIN 正确");
+        assertEquals("H&M", out.get(0).getBrand(), "保留行品牌正确");
+    }
+
+    @Test
+    void test_task_045_asin_brand_normal_multiple_items() {
+        // 批量场景：前 300 行 ASIN 唯一、品牌干净；后 300 行品牌无效。
+        // 去重表命中前 100 个 ASIN，输出顺序稳定、过滤数量精确。
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 300; i++) {
+            rows.add(row("D" + String.format("%09d", i + 1), "clean"));
+        }
+        for (int i = 0; i < 300; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "bad"));
+        }
+        List<String> existing = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            existing.add("D" + String.format("%09d", i + 1));
+        }
+        when(dedupeMapper.selectExistingDataValues(anyList())).thenReturn(existing);
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of(entity("bad")));
+
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(rows);
+
+        assertEquals(200, out.size(), "100 行去重 + 300 行无效品牌 = 过滤 400 行");
+        assertEquals("D000000101", out.get(0).getAsin(), "输出首行保持输入顺序");
+        assertEquals("D000000300", out.get(200 - 1).getAsin(), "输出末行为第 300 行");
+        assertTrue(out.stream().allMatch(r -> r.getBrand().equals("clean")),
+                "输出全部为干净品牌行");
+    }
+
+    @Test
+    void test_task_045_asin_brand_normal_repeated_operation_is_idempotent() {
+        // 幂等：同一输入重复执行输出一致，两次都发批量查询。
+        when(dedupeMapper.selectExistingDataValues(anyList()))
+                .thenReturn(List.of("B000000001"));
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of(entity("bad")));
+        List<CollectDataResultRowVo> rows = List.of(
+                row("B000000001", "bad"),
+                row("B000000002", "good")
+        );
+
+        List<CollectDataResultRowVo> first = query.filterByExistingAsin(rows);
+        List<CollectDataResultRowVo> second = query.filterByExistingAsin(rows);
+
+        assertEquals(first, second, "重复执行输出一致");
+        assertEquals(1, first.size());
+        verify(dedupeMapper, times(2)).selectExistingDataValues(anyList());
+        verify(invalidAsinMapper, times(2)).selectList(any());
+    }
+
+    @Test
+    void test_task_045_asin_brand_boundary_empty_input() {
+        // 空输入：空列表/无 ASIN 无品牌的行集合不触发数据库查询，安全跳过。
+        List<CollectDataResultRowVo> empty = query.filterByExistingAsin(List.of());
+        assertEquals(0, empty.size(), "空列表返回空");
+
+        List<CollectDataResultRowVo> blankRows = query.filterByExistingAsin(List.of(
+                row("", ""), row(null, null)
+        ));
+        assertEquals(2, blankRows.size(), "无 ASIN 无品牌行不触发查询且全保留");
+        verify(dedupeMapper, never()).selectExistingDataValues(anyList());
+        verify(invalidAsinMapper, never()).selectList(any());
+    }
+
+    @Test
+    void test_task_045_asin_brand_boundary_single_item() {
+        // 单元素：单行命中/未命中均正确，不依赖批量路径。
+        when(dedupeMapper.selectExistingDataValues(anyList())).thenReturn(List.of());
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of());
+
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(List.of(
+                row("B000000001", "solo")
+        ));
+        assertEquals(1, out.size(), "单行未命中保留");
+        assertEquals("solo", out.get(0).getBrand());
+
+        when(dedupeMapper.selectExistingDataValues(anyList())).thenReturn(List.of("B000000001"));
+        List<CollectDataResultRowVo> hit = query.filterByExistingAsin(List.of(
+                row("B000000001", "solo")
+        ));
+        assertEquals(0, hit.size(), "单行命中去重表被过滤");
+    }
+
+    @Test
+    void test_task_045_asin_brand_boundary_limit_and_overflow() {
+        // 上限/超限：超大输入集合批量查询一次完成，不产生逐行查询（无无界增长）。
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 5000; i++) {
+            rows.add(row("B" + String.format("%09d", i), "brand-" + i));
+        }
+        when(dedupeMapper.selectExistingDataValues(anyList())).thenReturn(List.of());
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of());
+
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(rows);
+
+        assertEquals(5000, out.size(), "超大集合全量保留");
+        verify(dedupeMapper).selectExistingDataValues(anyList());
+        verify(invalidAsinMapper).selectList(any());
+    }
+
+    @Test
+    void test_task_045_asin_brand_invalid_input_rejected() {
+        // 非法参数：null 行安全跳过；数据库返回 null 集合安全处理。
+        when(dedupeMapper.selectExistingDataValues(anyList())).thenReturn(null);
+        when(invalidAsinMapper.selectList(any())).thenReturn(null);
+
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        rows.add(null);
+        rows.add(row("B000000001", "  Nike  "));
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(rows);
+
+        assertEquals(1, out.size(), "null 行不丢弃");
+        assertEquals("B000000001", out.get(0).getAsin(), "正常行保留");
+        assertEquals("  Nike  ", out.get(0).getBrand(), "品牌保留原始值（过滤时仅标准化用于匹配）");
+    }
+
+    @Test
+    void test_task_045_asin_brand_dependency_failure_releases_resources() {
+        // 依赖失败：去重表查询抛错时异常向上传播；恢复后再次执行成功，不残留状态。
+        when(dedupeMapper.selectExistingDataValues(anyList()))
+                .thenThrow(new RuntimeException("db down"))
+                .thenReturn(List.of("B000000001"));
+        when(invalidAsinMapper.selectList(any())).thenReturn(List.of());
+
+        boolean threw = false;
+        try {
+            query.filterByExistingAsin(List.of(row("B000000001", "nike")));
+        } catch (RuntimeException ex) {
+            threw = "db down".equals(ex.getMessage());
+        }
+        assertTrue(threw, "去重查询失败向上抛错");
+
+        List<CollectDataResultRowVo> out = query.filterByExistingAsin(List.of(
+                row("B000000001", "nike"), row("B000000002", "adidas")
+        ));
+        assertEquals(1, out.size(), "恢复后过滤正确");
+        assertEquals("B000000002", out.get(0).getAsin(), "恢复后命中逻辑仍生效");
+    }
+
+    private static CollectDataResultRowVo row(String asin, String brand) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setAsin(asin);
+        row.setBrand(brand);
+        return row;
+    }
+
+    private static InvalidAsinDataEntity entity(String brand) {
+        InvalidAsinDataEntity entity = new InvalidAsinDataEntity();
+        entity.setBrand(brand);
+        return entity;
+    }
+}