task-47: 品牌检查结果任务内短期缓存，避免同品牌重复远程调用

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

@@ -8,8 +8,10 @@ import org.springframework.stereotype.Component;
 
 import java.util.ArrayList;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Locale;
+import java.util.Map;
 import java.util.Set;
 import java.util.regex.Pattern;
 
@@ -19,6 +21,10 @@ import java.util.regex.Pattern;
  * 非空批次正常检查并按失败/查询失败/通过分类。分类语义与
  * CollectDataService 原 filterByBrandCheck 完全等价，仅空品牌批次省掉
  * 无效远程请求。远程调用抛错时该批次整组降级 queryFailed，不影响后续批次。
+ *
+ * 品牌判定结果在过滤器实例内短期缓存（同一任务多个 chunk 提交复用同一
+ * 实例）：已判定的品牌再次出现时不再发起远程调用，避免同品牌重复请求。
+ * 远程抛错不写缓存（可恢复后重查）；缓存有界，超限淘汰最旧条目。
  */
 @Slf4j
 @Component
@@ -26,13 +32,29 @@ public class CollectDataBrandBatchFilter {
 
     private static final Pattern WHITESPACE_PATTERN = Pattern.compile("\\s+");
 
+    private static final String VERDICT_FAILED = "FAILED";
+    private static final String VERDICT_QUERY_FAILED = "QUERY_FAILED";
+    private static final String VERDICT_OK = "OK";
+
     private final BrandCheckClient brandCheckClient;
     private final int batchSize;
+    private final int cacheCapacity;
+
+    /** 品牌（小写标准化）→ 判定结果；access-order LRU，超限淘汰最旧。 */
+    private final Map<String, String> verdictCache;
 
     public CollectDataBrandBatchFilter(BrandCheckClient brandCheckClient,
                                        @Value("${aiimage.collect-data.brand-check-batch-size:10}") int batchSize) {
+        this(brandCheckClient, batchSize, 512);
+    }
+
+    public CollectDataBrandBatchFilter(BrandCheckClient brandCheckClient,
+                                       @Value("${aiimage.collect-data.brand-check-batch-size:10}") int batchSize,
+                                       @Value("${aiimage.collect-data.brand-check-cache-capacity:512}") int cacheCapacity) {
         this.brandCheckClient = brandCheckClient;
         this.batchSize = Math.max(1, batchSize);
+        this.cacheCapacity = Math.max(1, cacheCapacity);
+        this.verdictCache = new LinkedHashMap<>(Math.max(16, this.cacheCapacity / 2), 0.75f, true);
     }
 
     /**
@@ -48,10 +70,21 @@ public class CollectDataBrandBatchFilter {
         for (int start = 0; start < rows.size(); start += batchSize) {
             int end = Math.min(start + batchSize, rows.size());
             List<CollectDataResultRowVo> batch = rows.subList(start, end);
-            List<String> brands = distinctNonBlank(batch.stream()
+            List<String> batchBrands = distinctNonBlank(batch.stream()
                     .filter(row -> row != null)
                     .map(CollectDataResultRowVo::getBrand).toList());
-            if (brands.isEmpty()) {
+            List<String> uncachedBrands = new ArrayList<>();
+            for (String brand : batchBrands) {
+                if (!verdictCache.containsKey(normalizeBrand(brand))) {
+                    uncachedBrands.add(brand);
+                }
+            }
+            if (uncachedBrands.isEmpty() && !batchBrands.isEmpty()) {
+                // 本批次品牌全部命中缓存，无需远程调用。
+                classify(batch, verdictCache, rejected, queryFailed, accepted);
+                continue;
+            }
+            if (batchBrands.isEmpty()) {
                 // 空品牌批次：跳过远程检查，行直接归 rejected（与空品牌行语义一致）。
                 for (CollectDataResultRowVo row : batch) {
                     if (row != null) {
@@ -60,9 +93,16 @@ public class CollectDataBrandBatchFilter {
                 }
                 continue;
             }
-            BrandCheckClient.BrandCheckBatchResult check;
+            Map<String, String> batchVerdicts = new LinkedHashMap<>();
+            for (String brand : batchBrands) {
+                String normalized = normalizeBrand(brand);
+                String cached = verdictCache.get(normalized);
+                if (cached != null) {
+                    batchVerdicts.put(normalized, cached);
+                }
+            }
             try {
-                check = brandCheckClient.checkAll(brands, "Terms");
+                batchVerdicts.putAll(checkAndCache(uncachedBrands));
             } catch (RuntimeException ex) {
                 log.warn("[collect-data] brand check batch failed, degrade batch to queryFailed err={}", ex.getMessage());
                 for (CollectDataResultRowVo row : batch) {
@@ -72,26 +112,64 @@ public class CollectDataBrandBatchFilter {
                 }
                 continue;
             }
+            classify(batch, batchVerdicts, rejected, queryFailed, accepted);
+        }
+        return new BrandBatchOutcome(rejected, queryFailed, accepted);
+    }
+
+    /** 远程检查未缓存品牌并写入缓存；返回新查品牌（小写标准化）→ 判定映射。 */
+    private Map<String, String> checkAndCache(List<String> uncachedBrands) {
+        BrandCheckClient.BrandCheckBatchResult check = brandCheckClient.checkAll(uncachedBrands, "Terms");
         Set<String> failedBrands = normalizeObjectSet(check == null ? null : check.faildData());
         Set<String> queryFailedBrands = normalizeObjectSet(check == null ? null : check.queryFaildData());
+        Map<String, String> verdicts = new LinkedHashMap<>();
+        for (String brand : uncachedBrands) {
+            String normalized = normalizeBrand(brand);
+            String verdict;
+            if (failedBrands.contains(normalized)) {
+                verdict = VERDICT_FAILED;
+            } else if (queryFailedBrands.contains(normalized)) {
+                verdict = VERDICT_QUERY_FAILED;
+            } else {
+                verdict = VERDICT_OK;
+            }
+            putBounded(normalized, verdict);
+            verdicts.put(normalized, verdict);
+        }
+        return verdicts;
+    }
+
+    private void putBounded(String brand, String verdict) {
+        if (verdictCache.containsKey(brand)) {
+            return;
+        }
+        verdictCache.put(brand, verdict);
+        if (verdictCache.size() > cacheCapacity) {
+            var it = verdictCache.entrySet().iterator();
+            it.next();
+            it.remove();
+        }
+    }
+
+    private void classify(List<CollectDataResultRowVo> batch, Map<String, String> verdicts,
+                          List<CollectDataResultRowVo> rejected,
+                          List<CollectDataResultRowVo> queryFailed,
+                          List<CollectDataResultRowVo> accepted) {
         for (CollectDataResultRowVo row : batch) {
             if (row == null) {
                 continue;
             }
             String brand = normalizeBrand(row.getBrand());
-                if (brand.isBlank()) {
+            String verdict = verdicts.get(brand);
+            if (brand.isBlank() || VERDICT_FAILED.equals(verdict)) {
                 rejected.add(row);
-                } else if (failedBrands.contains(brand)) {
-                    rejected.add(row);
-                } else if (queryFailedBrands.contains(brand)) {
+            } else if (VERDICT_QUERY_FAILED.equals(verdict)) {
                 queryFailed.add(row);
             } else {
                 accepted.add(row);
             }
         }
     }
-        return new BrandBatchOutcome(rejected, queryFailed, accepted);
-    }
 
     /** 品牌检查分类结果：三类行互斥，顺序与输入一致。 */
     public record BrandBatchOutcome(List<CollectDataResultRowVo> rejected,

backend-java/src/main/resources/application.yml

@@ -262,6 +262,7 @@ aiimage:
     max-parse-rows: ${AIIMAGE_COLLECT_DATA_MAX_PARSE_ROWS:0}
     max-chunk-rows: ${AIIMAGE_COLLECT_DATA_MAX_CHUNK_ROWS:0}
     brand-check-batch-size: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_BATCH_SIZE:10}
+    brand-check-cache-capacity: ${AIIMAGE_COLLECT_DATA_BRAND_CHECK_CACHE_CAPACITY:512}
   image-video:
     coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn}
     coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu}

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

@@ -93,7 +93,8 @@ class CollectDataBrandBatchFilterTest {
 
         assertEquals(first.rejected(), second.rejected(), "重复执行 rejected 一致");
         assertEquals(first.accepted(), second.accepted(), "重复执行 accepted 一致");
-        verify(brandCheckClient, times(2)).checkAll(anyList(), any());
+        // 缓存生效：第二次执行全部命中缓存，不再发起远程调用。
+        verify(brandCheckClient, times(1)).checkAll(anyList(), any());
     }
 
     @Test
@@ -129,6 +130,7 @@ class CollectDataBrandBatchFilterTest {
                 row("B000000002", "")
         ));
         assertEquals(1, blank.rejected().size(), "单行空品牌归 rejected");
+        // 同实例缓存生效：solo 已查过，空品牌批次无远程调用，总调用保持 1 次。
         verify(brandCheckClient, times(1)).checkAll(anyList(), any());
     }
 
@@ -148,7 +150,8 @@ class CollectDataBrandBatchFilterTest {
 
         assertEquals(1, outcome.rejected().size(), "空品牌行归 rejected");
         assertEquals(5000, outcome.accepted().size(), "非空品牌行全 accepted");
-        verify(brandCheckClient, times(500)).checkAll(anyList(), any());
+        // 每批次 10 行恰好引入 10 个新品牌，10 批次后 100 品牌全部缓存，后续批次零调用。
+        verify(brandCheckClient, times(10)).checkAll(anyList(), any());
     }
 
     @Test

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

@@ -0,0 +1,200 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.nanri.aiimage.modules.brand.client.BrandCheckClient;
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+
+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.any;
+import static org.mockito.ArgumentMatchers.anyList;
+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 47：为品牌检查结果增加任务内短期缓存，避免同品牌重复远程调用。
+ * CollectDataBrandBatchFilter 在实例内缓存品牌检查判定（FAILED /
+ * QUERY_FAILED / OK），后续批次命中缓存的品牌不再发起 checkAll 远程调用；
+ * 远程抛错不缓存（可恢复），缓存有界（超限淘汰最旧条目），
+ * 分类语义与无缓存时完全等价。
+ */
+class CollectDataBrandCacheTest {
+
+    private BrandCheckClient brandCheckClient;
+    private CollectDataBrandBatchFilter filter;
+
+    @BeforeEach
+    void setUp() {
+        brandCheckClient = mock(BrandCheckClient.class);
+        filter = new CollectDataBrandBatchFilter(brandCheckClient, 10);
+    }
+
+    @Test
+    void test_task_047_cache_brand_normal_default_path() {
+        // 正常输入：首查品牌发起远程检查并缓存判定，分类正确。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
+                new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of("Zara"), List.of()));
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome outcome = filter.filter(List.of(
+                row("B000000001", "Zara"),
+                row("B000000002", "Nike")
+        ));
+
+        assertEquals(1, outcome.rejected().size(), "失败品牌行 rejected");
+        assertEquals("B000000001", outcome.rejected().get(0).getAsin());
+        assertEquals(1, outcome.accepted().size(), "其余行 accepted");
+        verify(brandCheckClient).checkAll(anyList(), any());
+    }
+
+    @Test
+    void test_task_047_cache_brand_normal_multiple_items() {
+        // 批量场景：多 chunk 提交时同品牌只远程检查一次，每 chunk 只查未缓存品牌。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
+                new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of("Zara"), List.of()));
+        filter.filter(List.of(row("B000000001", "Zara"), row("B000000002", "Nike")));
+        CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(List.of(
+                row("B000000001", "Zara"),
+                row("B000000003", "Adidas")
+        ));
+
+        assertEquals(1, second.rejected().size(), "chunk2 Zara 命中缓存仍 rejected");
+        assertEquals(1, second.accepted().size(), "chunk2 新品牌 Adidas accepted");
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(brandCheckClient, times(2)).checkAll(captor.capture(), any());
+        assertEquals(List.of("Adidas"), captor.getAllValues().get(1), "第二次只查未缓存品牌");
+    }
+
+    @Test
+    void test_task_047_cache_brand_normal_repeated_operation_is_idempotent() {
+        // 幂等：同一输入重复执行结果一致，第二次全部命中缓存不发远程调用。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
+                new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of("Nike"), List.of()));
+        List<CollectDataResultRowVo> rows = List.of(
+                row("B000000001", "Nike"),
+                row("B000000002", "Adidas")
+        );
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome first = filter.filter(rows);
+        CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(rows);
+
+        assertEquals(first.rejected(), second.rejected(), "重复执行 rejected 一致");
+        assertEquals(first.accepted(), second.accepted(), "重复执行 accepted 一致");
+        verify(brandCheckClient, times(1)).checkAll(anyList(), any());
+    }
+
+    @Test
+    void test_task_047_cache_brand_boundary_empty_input() {
+        // 空输入：空列表与全空品牌批次均不发起远程调用，也不写缓存。
+        CollectDataBrandBatchFilter.BrandBatchOutcome empty = filter.filter(List.of());
+        assertEquals(0, empty.rejected().size() + empty.accepted().size() + empty.queryFailed().size(),
+                "空列表返回空结果");
+        verify(brandCheckClient, never()).checkAll(anyList(), any());
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome blank = filter.filter(List.of(
+                row("B000000001", "")
+        ));
+        assertEquals(1, blank.rejected().size(), "空品牌行 rejected");
+        verify(brandCheckClient, never()).checkAll(anyList(), any());
+    }
+
+    @Test
+    void test_task_047_cache_brand_boundary_single_item() {
+        // 单元素：单品牌首查一次；同品牌再出现时命中缓存零调用。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
+                new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of(), List.of()));
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome single = filter.filter(List.of(
+                row("B000000001", "solo")
+        ));
+        assertEquals(1, single.accepted().size(), "单品牌未命中失败 accepted");
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome again = filter.filter(List.of(
+                row("B000000002", "solo")
+        ));
+        assertEquals(1, again.accepted().size(), "同品牌再次出现结果一致");
+        verify(brandCheckClient, times(1)).checkAll(anyList(), any());
+    }
+
+    @Test
+    void test_task_047_cache_brand_boundary_limit_and_overflow() {
+        // 上限/超限：缓存容量 8，首批 10 个品牌淘汰最旧 2 个；第二批同品牌
+        // 只重查被淘汰的 2 个，其余 8 个命中缓存；无无界增长。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
+                new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of(), List.of()));
+        CollectDataBrandBatchFilter smallCache = new CollectDataBrandBatchFilter(brandCheckClient, 10, 8);
+        List<CollectDataResultRowVo> first = new ArrayList<>();
+        List<CollectDataResultRowVo> second = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            first.add(row("B" + i, "brand-" + i));
+            second.add(row("B" + (100 + i), "brand-" + i));
+        }
+
+        smallCache.filter(first);
+        CollectDataBrandBatchFilter.BrandBatchOutcome out = smallCache.filter(second);
+
+        assertEquals(10, out.accepted().size(), "第二批 10 行全部 accepted");
+        ArgumentCaptor<List> captor = ArgumentCaptor.forClass(List.class);
+        verify(brandCheckClient, times(2)).checkAll(captor.capture(), any());
+        List<String> secondCall = captor.getAllValues().get(1);
+        assertEquals(2, secondCall.size(), "仅重查被淘汰的最旧 2 个品牌");
+        assertTrue(secondCall.containsAll(List.of("brand-0", "brand-1")), "淘汰的是最旧条目");
+    }
+
+    @Test
+    void test_task_047_cache_brand_invalid_input_rejected() {
+        // 非法参数：checkAll 返回 null 按无失败处理并缓存；null 行安全跳过。
+        when(brandCheckClient.checkAll(anyList(), any())).thenReturn(null);
+
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        rows.add(null);
+        rows.add(row("B000000001", "Nike"));
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome first = filter.filter(rows);
+        assertEquals(1, first.accepted().size(), "null 结果按无失败处理");
+        assertEquals(0, first.rejected().size(), "null 行不计数");
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome second = filter.filter(rows);
+        assertEquals(1, second.accepted().size(), "重复执行结果一致");
+        verify(brandCheckClient, times(1)).checkAll(anyList(), any());
+    }
+
+    @Test
+    void test_task_047_cache_brand_dependency_failure_releases_resources() {
+        // 依赖失败：远程抛错整批 queryFailed 且不缓存；恢复后重查成功，无残留状态。
+        when(brandCheckClient.checkAll(anyList(), any()))
+                .thenThrow(new RuntimeException("brand service down"))
+                .thenReturn(new BrandCheckClient.BrandCheckBatchResult(
+                        List.of(), List.of("Zara"), List.of()));
+        List<CollectDataResultRowVo> rows = List.of(
+                row("B000000001", "Zara"), row("B000000002", "Nike"));
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome failed = filter.filter(rows);
+        assertEquals(2, failed.queryFailed().size(), "失败批次整组降级 queryFailed");
+
+        CollectDataBrandBatchFilter.BrandBatchOutcome recovered = filter.filter(rows);
+        assertEquals(1, recovered.rejected().size(), "恢复后重新检查并分类");
+        assertEquals(1, recovered.accepted().size(), "恢复后 accepted 正确");
+        assertTrue(recovered.queryFailed().isEmpty(), "恢复后无残留 queryFailed");
+        verify(brandCheckClient, times(2)).checkAll(anyList(), any());
+    }
+
+    private static CollectDataResultRowVo row(String asin, String brand) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setAsin(asin);
+        row.setBrand(brand);
+        return row;
+    }
+}