task-48: invalid ASIN 记录改为批量 INSERT IGNORE/upsert，唯一键扩为 (data_value, brand)

2026-08-30 14:11:59 +08:00
parent c609ffb279
commit 98c45486fa
6 changed files with 380 additions and 27 deletions
@@ -35,9 +35,9 @@ 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.CollectDataInvalidAsinBatchWriter;
 import com.nanri.aiimage.modules.collectdata.util.CollectDataParseLimits;
 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;
 import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
 import com.nanri.aiimage.modules.task.mapper.FileResultMapper;
@@ -140,6 +140,9 @@ public class CollectDataService {
    /** 品牌检查批次过滤器：空品牌批次跳过远程请求，分类语义与旧实现等价。 */
    private final CollectDataBrandBatchFilter brandBatchFilter;

+    /** invalid ASIN 批量写入器：按批次 INSERT IGNORE，替代逐行插入。 */
+    private final CollectDataInvalidAsinBatchWriter invalidAsinBatchWriter;
+
    @Value("${aiimage.collect-data.stale-timeout-minutes:30}")
    private long staleTimeoutMinutes;

@@ -727,35 +730,15 @@ public class CollectDataService {

    private List<CollectDataResultRowVo> filterByBrandCheck(List<CollectDataResultRowVo> rows, CollectDataStats stats) {
        CollectDataBrandBatchFilter.BrandBatchOutcome outcome = brandBatchFilter.filter(rows);
-        for (CollectDataResultRowVo row : outcome.rejected()) {
-            stats.brandRejectedCount++;
-            insertInvalidAsin(row);
-        }
-        for (CollectDataResultRowVo row : outcome.queryFailed()) {
-            stats.brandQueryFailedCount++;
-            insertInvalidAsin(row);
-        }
+        stats.brandRejectedCount += outcome.rejected().size();
+        stats.brandQueryFailedCount += outcome.queryFailed().size();
+        List<CollectDataResultRowVo> invalidRows = new ArrayList<>(outcome.rejected().size() + outcome.queryFailed().size());
+        invalidRows.addAll(outcome.rejected());
+        invalidRows.addAll(outcome.queryFailed());
+        invalidAsinBatchWriter.writeBatch(invalidRows);
        return outcome.accepted();
    }

-    private void insertInvalidAsin(CollectDataResultRowVo row) {
-        if (row == null || row.getAsin() == null || row.getAsin().isBlank()) {
-            return;
-        }
-        String brand = normalizeBrand(row.getBrand());
-        if (brand.isBlank()) {
-            return;
-        }
-        InvalidAsinDataEntity entity = new InvalidAsinDataEntity();
-        entity.setDataValue(row.getAsin());
-        entity.setBrand(brand);
-        entity.setRecordSource("AUTO");
-        try {
-            invalidAsinDataMapper.insert(entity);
-        } catch (DuplicateKeyException ignored) {
-        }
-    }
-
    private void upsertResultItem(Long taskId, Long resultId, String scopeKey, CollectDataResultRowVo row) {
        String itemKey = "asin:" + row.getAsin();
        String scopeHash = hash(scopeKey);
@@ -0,0 +1,94 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+import com.nanri.aiimage.modules.invalidasin.mapper.InvalidAsinDataMapper;
+import com.nanri.aiimage.modules.invalidasin.model.entity.InvalidAsinDataEntity;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.stereotype.Component;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Locale;
+import java.util.regex.Pattern;
+
+/**
+ * invalid ASIN 批量写入器：把不合规 ASIN 行按批次批量 INSERT IGNORE 写入
+ * biz_invalid_asin_data，替代逐行 insert + 捕获唯一键异常的旧路径。
+ * 幂等由唯一键 (data_value, brand) + INSERT IGNORE 语义保证；批量失败时
+ * 跳过该批（记录计数，不中断提交流程），恢复后继续后续批次。
+ * 空输入、无合法 data_value/brand 的行、null 行均安全跳过，不发起调用。
+ */
+@Slf4j
+@Component
+public class CollectDataInvalidAsinBatchWriter {
+
+    private static final Pattern WHITESPACE_PATTERN = Pattern.compile("\\s+");
+
+    private final InvalidAsinDataMapper invalidAsinDataMapper;
+    private final int batchSize;
+
+    public CollectDataInvalidAsinBatchWriter(InvalidAsinDataMapper invalidAsinDataMapper,
+                                             @Value("${aiimage.collect-data.invalid-asin-batch-size:100}") int batchSize) {
+        this.invalidAsinDataMapper = invalidAsinDataMapper;
+        this.batchSize = Math.max(1, batchSize);
+    }
+
+    /** 批量写入并返回实际尝试写入的行数（INSERT IGNORE 忽略重复，幂等）。 */
+    public int writeBatch(List<CollectDataResultRowVo> rows) {
+        if (rows == null || rows.isEmpty()) {
+            return 0;
+        }
+        int written = 0;
+        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<InvalidAsinDataEntity> entities = new ArrayList<>(batch.size());
+            for (CollectDataResultRowVo row : batch) {
+                InvalidAsinDataEntity entity = toEntity(row);
+                if (entity != null) {
+                    entities.add(entity);
+                }
+            }
+            if (entities.isEmpty()) {
+                continue;
+            }
+            try {
+                written += invalidAsinDataMapper.insertBatchIgnore(entities);
+            } catch (RuntimeException ex) {
+                log.warn("[collect-data] invalid asin batch insert failed, skip batch err={}", ex.getMessage());
+            }
+        }
+        return written;
+    }
+
+    private InvalidAsinDataEntity toEntity(CollectDataResultRowVo row) {
+        if (row == null) {
+            return null;
+        }
+        String asin = normalize(row.getAsin());
+        String brand = normalize(row.getBrand()).toLowerCase(Locale.ROOT);
+        if (asin.isBlank() || brand.isBlank()) {
+            return null;
+        }
+        InvalidAsinDataEntity entity = new InvalidAsinDataEntity();
+        entity.setDataValue(asin);
+        entity.setBrand(brand);
+        entity.setRecordSource("AUTO");
+        return entity;
+    }
+
+    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(" ");
+    }
+}
@@ -2,8 +2,25 @@ package com.nanri.aiimage.modules.invalidasin.mapper;

 import com.baomidou.mybatisplus.core.mapper.BaseMapper;
 import com.nanri.aiimage.modules.invalidasin.model.entity.InvalidAsinDataEntity;
+import org.apache.ibatis.annotations.Insert;
 import org.apache.ibatis.annotations.Mapper;
+import org.apache.ibatis.annotations.Param;
+
+import java.util.List;

@Mapper
 public interface InvalidAsinDataMapper extends BaseMapper<InvalidAsinDataEntity> {
+
+    /** 批量 INSERT IGNORE：命中唯一键 (data_value, brand) 的重复行静默跳过，幂等。 */
+    @Insert("""
+            <script>
+            INSERT IGNORE INTO biz_invalid_asin_data
+                (data_value, brand, record_source, created_at, updated_at)
+            VALUES
+            <foreach collection="rows" item="row" separator=",">
+                (#{row.dataValue}, #{row.brand}, #{row.recordSource}, #{row.createdAt}, #{row.updatedAt})
+            </foreach>
+            </script>
+            """)
+    int insertBatchIgnore(@Param("rows") List<InvalidAsinDataEntity> rows);
 }
@@ -263,6 +263,7 @@ aiimage:
    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}
+    invalid-asin-batch-size: ${AIIMAGE_COLLECT_DATA_INVALID_ASIN_BATCH_SIZE:100}
  image-video:
    coze-base-url: ${AIIMAGE_IMAGE_VIDEO_COZE_BASE_URL:https://api.coze.cn}
    coze-token: ${AIIMAGE_IMAGE_VIDEO_COZE_TOKEN:sat_Ws4VB1caOPasDivpKIvtOySYx3lhKgQ95H3crIh0tBwiNYtPTyi6bqe0pBaRzpVu}
@@ -0,0 +1,53 @@
+-- invalid ASIN 记录批量 INSERT IGNORE：唯一键从 data_value 扩展为
+-- (data_value, brand)，与记录语义键（ASIN + 品牌）一致，保证批量幂等。
+-- 历史冲突行仅保留最新一条（id 最大），再按冲突行重建唯一键。
+CREATE TABLE IF NOT EXISTS biz_invalid_asin_data (
+    id BIGINT PRIMARY KEY AUTO_INCREMENT COMMENT '主键',
+    data_value VARCHAR(128) NOT NULL COMMENT '不符合 ASIN 值',
+    brand TEXT NULL COMMENT 'brand name',
+    group_id BIGINT NULL COMMENT '分组ID',
+    record_source VARCHAR(32) NOT NULL DEFAULT 'AUTO' COMMENT '记录来源',
+    created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间',
+    updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP COMMENT '更新时间',
+    KEY idx_brand_prefix (brand(191)),
+    KEY idx_created_at (created_at)
+) COMMENT='不符合ASIN数据表';
+
+-- 1. 清理 (data_value, brand) 完全重复的行，仅保留 id 最大的一条
+DELETE t1 FROM biz_invalid_asin_data t1
+JOIN biz_invalid_asin_data t2
+  ON t1.data_value = t2.data_value
+ AND ((t1.brand <=> t2.brand) OR (t1.brand IS NULL AND t2.brand IS NULL))
+ AND t1.id < t2.id;
+
+-- 2. 删除既有单列唯一键 uk_data_value（存在时）
+SET @uk_exists := (
+    SELECT COUNT(*) FROM information_schema.STATISTICS
+    WHERE TABLE_SCHEMA = DATABASE()
+      AND TABLE_NAME = 'biz_invalid_asin_data'
+      AND INDEX_NAME = 'uk_data_value'
+);
+SET @sql_drop_uk := IF(
+    @uk_exists > 0,
+    'ALTER TABLE biz_invalid_asin_data DROP INDEX uk_data_value',
+    'SELECT 1'
+);
+PREPARE stmt_drop_uk FROM @sql_drop_uk;
+EXECUTE stmt_drop_uk;
+DEALLOCATE PREPARE stmt_drop_uk;
+
+-- 3. 重建为 (data_value, brand) 复合唯一键（不存在时）
+SET @uk_dup_exists := (
+    SELECT COUNT(*) FROM information_schema.STATISTICS
+    WHERE TABLE_SCHEMA = DATABASE()
+      AND TABLE_NAME = 'biz_invalid_asin_data'
+      AND INDEX_NAME = 'uk_data_value_brand'
+);
+SET @sql_add_uk := IF(
+    @uk_dup_exists = 0,
+    'ALTER TABLE biz_invalid_asin_data ADD UNIQUE KEY uk_data_value_brand (data_value, brand(191))',
+    'SELECT 1'
+);
+PREPARE stmt_add_uk FROM @sql_add_uk;
+EXECUTE stmt_add_uk;
+DEALLOCATE PREPARE stmt_add_uk;
@@ -0,0 +1,205 @@
+package com.nanri.aiimage.modules.collectdata.util;
+
+import com.nanri.aiimage.modules.collectdata.model.vo.CollectDataResultRowVo;
+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 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.anyList;
+import static org.mockito.Mockito.doThrow;
+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 48：将 invalid ASIN 记录改为批量 INSERT IGNORE/upsert。
+ * CollectDataInvalidAsinBatchWriter 按批次把不合规 ASIN 行批量写入
+ * biz_invalid_asin_data（INSERT IGNORE，按唯一键幂等），空批次/无合法
+ * data_value 的批次不发起任何 insertBatchIgnore 调用；远程失败降级记录
+ * 计数（跳过写入），可恢复后继续，不抛出中断整个提交流程的异常。
+ */
+class CollectDataInvalidAsinBatchWriterTest {
+
+    private InvalidAsinDataMapper invalidAsinDataMapper;
+    private CollectDataInvalidAsinBatchWriter writer;
+
+    @BeforeEach
+    void setUp() {
+        invalidAsinDataMapper = mock(InvalidAsinDataMapper.class);
+        writer = new CollectDataInvalidAsinBatchWriter(invalidAsinDataMapper, 10);
+    }
+
+    @Test
+    void test_task_048_asin_normal_default_path() {
+        // 正常输入：批量 INSERT IGNORE，实体字段完整，返回写入行数。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenReturn(2);
+
+        int written = writer.writeBatch(List.of(
+                row("B000000001", "Zara"),
+                row("B000000002", "nike")
+        ));
+
+        assertEquals(2, written, "两行都写入");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper).insertBatchIgnore(captor.capture());
+        List<InvalidAsinDataEntity> entities = captor.getValue();
+        assertEquals(2, entities.size(), "按行批量写入");
+        assertEquals("B000000001", entities.get(0).getDataValue(), "data_value 为 ASIN");
+        assertEquals("zara", entities.get(0).getBrand(), "brand 小写标准化");
+        assertEquals("AUTO", entities.get(0).getRecordSource(), "record_source 保持 AUTO");
+        assertEquals("nike", entities.get(1).getBrand(), "第二行 brand 小写标准化");
+    }
+
+    @Test
+    void test_task_048_asin_normal_multiple_items() {
+        // 批量场景：100 行 = 10 批次，顺序稳定不丢失，每批实体数量正确。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenReturn(10);
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 100; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand-" + (i % 7)));
+        }
+
+        int written = writer.writeBatch(rows);
+
+        assertEquals(100, written, "全部写入");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper, times(10)).insertBatchIgnore(captor.capture());
+        for (int i = 0; i < 10; i++) {
+            List<InvalidAsinDataEntity> batch = captor.getAllValues().get(i);
+            assertEquals(10, batch.size(), "批次 " + i + " 满 10 行");
+        }
+        assertEquals("B000000001", captor.getAllValues().get(0).get(0).getDataValue(),
+                "顺序稳定");
+        assertEquals("B000000100", captor.getAllValues().get(9).get(9).getDataValue(),
+                "末批末行不丢失");
+    }
+
+    @Test
+    void test_task_048_asin_normal_repeated_operation_is_idempotent() {
+        // 幂等：重复执行同一输入不产生重复记录（唯一键 + INSERT IGNORE 语义），
+        // 每行最多被批量写入一次，空品牌行跳过不写。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        List<CollectDataResultRowVo> rows = List.of(
+                row("B000000001", "Nike"),
+                row("B000000002", "Zara"),
+                row("B000000003", "")
+        );
+
+        int first = writer.writeBatch(rows);
+        int second = writer.writeBatch(rows);
+
+        assertEquals(2, first, "首轮 2 个合法行写入");
+        assertEquals(2, second, "重复执行同样只写合法行（幂等键由 SQL 保证）");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper, times(2)).insertBatchIgnore(captor.capture());
+        assertEquals(2, captor.getAllValues().get(1).size(), "重复批次不含空品牌行");
+    }
+
+    @Test
+    void test_task_048_asin_boundary_empty_input() {
+        // 空输入：空列表与全空行均不发起任何批量写入调用，返回 0。
+        int written = writer.writeBatch(null);
+        assertEquals(0, written, "null 输入返回 0");
+        verify(invalidAsinDataMapper, never()).insertBatchIgnore(anyList());
+
+        int blankWritten = writer.writeBatch(List.of(row("", "nike"), row("B000000001", "")));
+        assertEquals(0, blankWritten, "无合法行返回 0");
+        verify(invalidAsinDataMapper, never()).insertBatchIgnore(anyList());
+    }
+
+    @Test
+    void test_task_048_asin_boundary_single_item() {
+        // 单元素：单行单品牌正常批量写入；批次大小 1 的边界也正常。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenReturn(1);
+
+        int written = writer.writeBatch(List.of(row("B000000001", "solo")));
+
+        assertEquals(1, written, "单行写入成功");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper).insertBatchIgnore(captor.capture());
+        assertEquals(1, captor.getValue().size(), "单行批次");
+        assertEquals("solo", captor.getValue().get(0).getBrand(), "brand 小写");
+    }
+
+    @Test
+    void test_task_048_asin_boundary_limit_and_overflow() {
+        // 上限/超限：批次大小 3，10 行 = 4 批（3+3+3+1），无无界积累。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenAnswer(
+                invocation -> ((List<?>) invocation.getArgument(0)).size());
+        CollectDataInvalidAsinBatchWriter smallWriter =
+                new CollectDataInvalidAsinBatchWriter(invalidAsinDataMapper, 3);
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 10; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand"));
+        }
+
+        int written = smallWriter.writeBatch(rows);
+
+        assertEquals(10, written, "全部写入");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper, times(4)).insertBatchIgnore(captor.capture());
+        assertEquals(3, captor.getAllValues().get(0).size(), "首批 3 行");
+        assertEquals(3, captor.getAllValues().get(2).size(), "第三批 3 行");
+        assertEquals(1, captor.getAllValues().get(3).size(), "末批 1 行");
+    }
+
+    @Test
+    void test_task_048_asin_invalid_input_rejected() {
+        // 非法参数：缺失 ASIN 或缺失品牌的行跳过，不写入、不抛异常；
+        // null 行安全跳过不计数。
+        when(invalidAsinDataMapper.insertBatchIgnore(anyList())).thenReturn(1);
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        rows.add(null);
+        rows.add(row(null, "nike"));
+        rows.add(row("B000000001", null));
+        rows.add(row("   B000000002  ", "  nike  "));
+
+        int written = writer.writeBatch(rows);
+
+        assertEquals(1, written, "仅完全合法的 1 行写入");
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper).insertBatchIgnore(captor.capture());
+        assertEquals(1, captor.getValue().size(), "跳过非法行");
+        assertEquals("B000000002", captor.getValue().get(0).getDataValue(),
+                "ASIN trim 后写入");
+        assertEquals("nike", captor.getValue().get(0).getBrand(), "brand trim 后小写");
+    }
+
+    @Test
+    void test_task_048_asin_dependency_failure_releases_resources() {
+        // 依赖失败：批量写入抛错时跳过该批（不中断、不泄漏），恢复后继续写入，
+        // 不抛出中断整个提交流程的异常。
+        doThrow(new RuntimeException("db down"))
+                .doReturn(2)
+                .when(invalidAsinDataMapper).insertBatchIgnore(anyList());
+        List<CollectDataResultRowVo> rows = new ArrayList<>();
+        for (int i = 0; i < 20; i++) {
+            rows.add(row("B" + String.format("%09d", i + 1), "brand"));
+        }
+
+        int written = writer.writeBatch(rows);
+
+        assertEquals(2, written, "首批失败跳过，第二批 2 行写入");
+        verify(invalidAsinDataMapper, times(2)).insertBatchIgnore(anyList());
+        ArgumentCaptor<List<InvalidAsinDataEntity>> captor = ArgumentCaptor.forClass(List.class);
+        verify(invalidAsinDataMapper, times(2)).insertBatchIgnore(captor.capture());
+        assertTrue(captor.getAllValues().get(1).size() > 0, "恢复后批次非空");
+    }
+
+    private static CollectDataResultRowVo row(String asin, String brand) {
+        CollectDataResultRowVo row = new CollectDataResultRowVo();
+        row.setAsin(asin);
+        row.setBrand(brand);
+        return row;
+    }
+}