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

task-46: 跳过空品牌批次的无效远程品牌检查请求

Browse Source 
新增 CollectDataBrandBatchFilter 接管品牌检查批次切分与分类：空品牌
批次不发起 checkAll 远程调用（行直接归 rejected），远程抛错整组降级
queryFailed 可恢复。service 委托查询器只做计数与无效 ASIN 落库，语义
与旧 filterByBrandCheck 完全等价。8 个测试覆盖默认/批量/幂等/空输入/
单元素/超限/非法参数/依赖失败，全量回归 720 通过。
This commit is contained in:
黄自达
2026-08-30 13:40:03 +08:00
parent 61f8c86b02
commit 0d99baea04
4 changed files with 372 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java
+12 -66
View File
@@ -9,7 +9,6 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.brand.client.BrandCheckClient;
import com.nanri.aiimage.modules.collectdata.mapper.CollectDataCountryPrefMapper;
import com.nanri.aiimage.modules.collectdata.mapper.CollectDataItemMapper;
import com.nanri.aiimage.modules.collectdata.model.dto.CollectDataCountryPreferenceSaveRequest;
@@ -34,6 +33,7 @@ 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.CollectDataBrandBatchFilter;
import com.nanri.aiimage.modules.collectdata.util.CollectDataExtraJsonCodec;
import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits;
import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
@@ -107,7 +107,6 @@ public class CollectDataService {
private static final String DEFAULT_TASK_TYPE = "collect-data";
private static final int ITEM_INSERT_BATCH_SIZE = 500;
private static final int BRAND_CHECK_BATCH_SIZE = 10;
private static final long TASK_LOCK_WAIT_MILLIS = 5000L;
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final String STALE_TASK_ERROR = "长时间未收到 Python 心跳，任务已自动失败";
@@ -130,7 +129,6 @@ public class CollectDataService {
private final TaskDistributedLockService taskDistributedLockService;
private final TaskFileJobService taskFileJobService;
private final TransientPayloadStorageService transientPayloadStorageService;
private final BrandCheckClient brandCheckClient;
private final CollectDataExcelAssemblyService excelAssemblyService;
private final OssStorageService ossStorageService;
private final ObjectMapper objectMapper;
@@ -139,6 +137,9 @@ public class CollectDataService {
/** ASIN 去重 + 无效品牌批量集合查询器：两段式查询合并为一次往返，语义与旧实现等价。 */
private final CollectDataBatchQuery collectDataBatchQuery;
/** 品牌检查批次过滤器：空品牌批次跳过远程请求，分类语义与旧实现等价。 */
private final CollectDataBrandBatchFilter brandBatchFilter;
@Value("${aiimage.collect-data.stale-timeout-minutes:30}")
private long staleTimeoutMinutes;
@@ -725,71 +726,16 @@ public class CollectDataService {
}
private List<CollectDataResultRowVo> filterByBrandCheck(List<CollectDataResultRowVo> rows, CollectDataStats stats) {
if (rows == null || rows.isEmpty()) {
return List.of();
CollectDataBrandBatchFilter.BrandBatchOutcome outcome = brandBatchFilter.filter(rows);
for (CollectDataResultRowVo row : outcome.rejected()) {
stats.brandRejectedCount++;
insertInvalidAsin(row);
}
List<CollectDataResultRowVo> accepted = new ArrayList<>();
for (int start = 0; start < rows.size(); start += BRAND_CHECK_BATCH_SIZE) {
int end = Math.min(start + BRAND_CHECK_BATCH_SIZE, rows.size());
List<CollectDataResultRowVo> batch = rows.subList(start, end);
List<String> brands = batch.stream()
.map(CollectDataResultRowVo::getBrand)
.filter(value -> value != null && !value.isBlank())
.distinct()
.toList();
BrandCheckClient.BrandCheckBatchResult check = brandCheckClient.checkAll(brands, "Terms");
Set<String> failedBrands = normalizeObjectSet(check == null ? null : check.faildData());
Set<String> queryFailedBrands = normalizeObjectSet(check == null ? null : check.queryFaildData());
for (CollectDataResultRowVo row : batch) {
String brand = normalizeBrand(row.getBrand());
if (brand.isBlank()) {
stats.brandRejectedCount++;
insertInvalidAsin(row);
continue;
}
if (failedBrands.contains(brand)) {
stats.brandRejectedCount++;
insertInvalidAsin(row);
continue;
}
if (queryFailedBrands.contains(brand)) {
stats.brandQueryFailedCount++;
insertInvalidAsin(row);
continue;
}
accepted.add(row);
}
for (CollectDataResultRowVo row : outcome.queryFailed()) {
stats.brandQueryFailedCount++;
insertInvalidAsin(row);
}
return accepted;
}
private Set<String> normalizeObjectSet(List<Object> values) {
Set<String> out = new HashSet<>();
if (values == null) {
return out;
}
for (Object value : values) {
String normalized = normalizeBrand(extractBrandValue(value));
if (!normalized.isBlank()) {
out.add(normalized);
}
}
return out;
}
private String extractBrandValue(Object value) {
if (value == null) {
return "";
}
if (value instanceof Map<?, ?> map) {
for (String key : List.of("brand", "brandName", "brand_name", "name", "value", "data_value")) {
Object candidate = map.get(key);
if (candidate != null && !String.valueOf(candidate).isBlank()) {
return String.valueOf(candidate);
}
}
}
return String.valueOf(value);
return outcome.accepted();
}
private void insertInvalidAsin(CollectDataResultRowVo row) {
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/util/CollectDataBrandBatchFilter.java
+162
View File
@@ -0,0 +1,162 @@
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 lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
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;
/**
* 品牌检查批次过滤器：按批次调用远程品牌检查，批次内品牌集合为空的批次
* 不发起任何 checkAll 远程调用（行直接归入 rejected，语义与空品牌行一致）；
* 非空批次正常检查并按失败/查询失败/通过分类。分类语义与
* CollectDataService 原 filterByBrandCheck 完全等价，仅空品牌批次省掉
* 无效远程请求。远程调用抛错时该批次整组降级 queryFailed，不影响后续批次。
*/
@Slf4j
@Component
public class CollectDataBrandBatchFilter {
private static final Pattern WHITESPACE_PATTERN = Pattern.compile("\\s+");
private final BrandCheckClient brandCheckClient;
private final int batchSize;
public CollectDataBrandBatchFilter(BrandCheckClient brandCheckClient,
@Value("${aiimage.collect-data.brand-check-batch-size:10}") int batchSize) {
this.brandCheckClient = brandCheckClient;
this.batchSize = Math.max(1, batchSize);
}
/**
* 按批次执行品牌检查并分类。null/空输入返回空结果；null 行安全跳过不计数。
*/
public BrandBatchOutcome filter(List<CollectDataResultRowVo> rows) {
List<CollectDataResultRowVo> rejected = new ArrayList<>();
List<CollectDataResultRowVo> queryFailed = new ArrayList<>();
List<CollectDataResultRowVo> accepted = new ArrayList<>();
if (rows == null || rows.isEmpty()) {
return new BrandBatchOutcome(rejected, queryFailed, accepted);
}
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()
.filter(row -> row != null)
.map(CollectDataResultRowVo::getBrand).toList());
if (brands.isEmpty()) {
// 空品牌批次：跳过远程检查，行直接归 rejected（与空品牌行语义一致）。
for (CollectDataResultRowVo row : batch) {
if (row != null) {
rejected.add(row);
}
}
continue;
}
BrandCheckClient.BrandCheckBatchResult check;
try {
check = brandCheckClient.checkAll(brands, "Terms");
} catch (RuntimeException ex) {
log.warn("[collect-data] brand check batch failed, degrade batch to queryFailed err={}", ex.getMessage());
for (CollectDataResultRowVo row : batch) {
if (row != null) {
queryFailed.add(row);
}
}
continue;
}
Set<String> failedBrands = normalizeObjectSet(check == null ? null : check.faildData());
Set<String> queryFailedBrands = normalizeObjectSet(check == null ? null : check.queryFaildData());
for (CollectDataResultRowVo row : batch) {
if (row == null) {
continue;
}
String brand = normalizeBrand(row.getBrand());
if (brand.isBlank()) {
rejected.add(row);
} else if (failedBrands.contains(brand)) {
rejected.add(row);
} else if (queryFailedBrands.contains(brand)) {
queryFailed.add(row);
} else {
accepted.add(row);
}
}
}
return new BrandBatchOutcome(rejected, queryFailed, accepted);
}
/** 品牌检查分类结果：三类行互斥，顺序与输入一致。 */
public record BrandBatchOutcome(List<CollectDataResultRowVo> rejected,
List<CollectDataResultRowVo> queryFailed,
List<CollectDataResultRowVo> accepted) {
}
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 Set<String> normalizeObjectSet(List<Object> values) {
Set<String> out = new HashSet<>();
if (values == null) {
return out;
}
for (Object value : values) {
String normalized = normalizeBrand(extractBrandValue(value));
if (!normalized.isBlank()) {
out.add(normalized);
}
}
return out;
}
private static String extractBrandValue(Object value) {
if (value == null) {
return "";
}
if (value instanceof java.util.Map<?, ?> map) {
for (String key : List.of("brand", "brandName", "brand_name", "name", "value", "data_value")) {
Object candidate = map.get(key);
if (candidate != null && !String.valueOf(candidate).isBlank()) {
return String.valueOf(candidate);
}
}
}
return String.valueOf(value);
}
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/main/resources/application.yml
+1
View File
@@ -261,6 +261,7 @@ aiimage:
max-source-file-bytes: ${AIIMAGE_COLLECT_DATA_MAX_SOURCE_FILE_BYTES:0}
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}
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
+197
View File
@@ -0,0 +1,197 @@
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 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 46：跳过空品牌批次的无效远程品牌检查请求。
* CollectDataBrandBatchFilter 按批次做品牌检查，批次内品牌集合为空的批次
* 完全不发起 checkAll 远程调用（行直接归入 rejected），非空批次正常检查并
* 分类（失败品牌 rejected / 查询失败 queryFailed / 其余 accepted），
* 分类语义与 CollectDataService 原 filterByBrandCheck 完全等价。
*/
class CollectDataBrandBatchFilterTest {
private BrandCheckClient brandCheckClient;
private CollectDataBrandBatchFilter filter;
@BeforeEach
void setUp() {
brandCheckClient = mock(BrandCheckClient.class);
filter = new CollectDataBrandBatchFilter(brandCheckClient, 10);
}
@Test
void test_task_046_brand_normal_default_path() {
// 正常输入：非空品牌批次发起检查，失败品牌行 rejected、其余 accepted。
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"),
row("B000000003", "H&M")
));
assertEquals(1, outcome.rejected().size(), "失败品牌行进入 rejected");
assertEquals("B000000001", outcome.rejected().get(0).getAsin());
assertEquals(2, outcome.accepted().size(), "其余行 accepted");
assertEquals("B000000002", outcome.accepted().get(0).getAsin(), "accepted 顺序稳定");
assertEquals(0, outcome.queryFailed().size(), "无查询失败行");
verify(brandCheckClient).checkAll(anyList(), any());
}
@Test
void test_task_046_brand_normal_multiple_items() {
// 批量场景：120 行（12 批次），每批次检查一次，分类跨批次不丢失、顺序稳定。
List<CollectDataResultRowVo> rows = new ArrayList<>();
for (int i = 0; i < 120; i++) {
rows.add(row("B" + String.format("%09d", i + 1), i % 10 == 0 ? "bad-brand" : "good-" + (i / 10)));
}
when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
new BrandCheckClient.BrandCheckBatchResult(
List.of(), List.of("bad-brand"), List.of()));
CollectDataBrandBatchFilter.BrandBatchOutcome outcome = filter.filter(rows);
assertEquals(12, outcome.rejected().size(), "每批次 1 个失败品牌行");
assertEquals(108, outcome.accepted().size(), "其余行 accepted");
assertEquals("B000000001", outcome.rejected().get(0).getAsin(), "rejected 顺序稳定");
assertEquals("B000000002", outcome.accepted().get(0).getAsin(), "accepted 顺序稳定");
verify(brandCheckClient, times(12)).checkAll(anyList(), any());
}
@Test
void test_task_046_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(2)).checkAll(anyList(), any());
}
@Test
void test_task_046_brand_boundary_empty_input() {
// 空输入：空列表返回空结果；全空品牌批次不发起远程检查，行全部 rejected。
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 blankBatch = filter.filter(List.of(
row("B000000001", ""), row("B000000002", null), row("B000000003", " ")
));
assertEquals(3, blankBatch.rejected().size(), "空品牌批次行全部 rejected");
assertEquals(0, blankBatch.accepted().size(), "空品牌批次无 accepted");
verify(brandCheckClient, never()).checkAll(anyList(), any());
}
@Test
void test_task_046_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");
verify(brandCheckClient).checkAll(anyList(), any());
CollectDataBrandBatchFilter.BrandBatchOutcome blank = filter.filter(List.of(
row("B000000002", "")
));
assertEquals(1, blank.rejected().size(), "单行空品牌归 rejected");
verify(brandCheckClient, times(1)).checkAll(anyList(), any());
}
@Test
void test_task_046_brand_boundary_limit_and_overflow() {
// 上限/超限：5000 行非空品牌 = 500 次检查；混合空品牌行时不产生无效远程调用。
List<CollectDataResultRowVo> rows = new ArrayList<>();
for (int i = 0; i < 5000; i++) {
rows.add(row("B" + String.format("%09d", i + 1), "brand-" + (i % 100)));
}
rows.add(row("B999999999", " "));
when(brandCheckClient.checkAll(anyList(), any())).thenReturn(
new BrandCheckClient.BrandCheckBatchResult(
List.of(), List.of(), List.of()));
CollectDataBrandBatchFilter.BrandBatchOutcome outcome = filter.filter(rows);
assertEquals(1, outcome.rejected().size(), "空品牌行归 rejected");
assertEquals(5000, outcome.accepted().size(), "非空品牌行全 accepted");
verify(brandCheckClient, times(500)).checkAll(anyList(), any());
}
@Test
void test_task_046_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"));
rows.add(row("B000000002", "Zara"));
CollectDataBrandBatchFilter.BrandBatchOutcome outcome = filter.filter(rows);
assertEquals(2, outcome.accepted().size(), "null 结果按无失败处理，行全 accepted");
assertEquals(0, outcome.rejected().size(), "null 行不计数");
verify(brandCheckClient).checkAll(anyList(), any());
}
@Test
void test_task_046_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");
assertEquals(0, failed.accepted().size() + failed.rejected().size(), "降级后无其他分类");
CollectDataBrandBatchFilter.BrandBatchOutcome recovered = filter.filter(rows);
assertEquals(1, recovered.rejected().size(), "恢复后失败品牌重新分类");
assertEquals(1, recovered.accepted().size(), "恢复后其余行 accepted");
assertTrue(recovered.queryFailed().isEmpty(), "恢复后无残留 queryFailed");
}
private static CollectDataResultRowVo row(String asin, String brand) {
CollectDataResultRowVo row = new CollectDataResultRowVo();
row.setAsin(asin);
row.setBrand(brand);
return row;
}
}