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

fix(健壮性): Java OOM 三处 + 双TE 502 根因 + admin-vue 403 白屏

Browse Source 
Java:
- similarasin Excel 解析改 EasyExcel 流式（原 WorkbookFactory 全量 DOM，大表 OOM）+ 魔数校验
- collectdata 结果组装改游标分批 + 导入改流式（原全量驻留内存）
- GlobalExceptionHandler 转发响应过滤逐跳头与实例标识头（双 Transfer-Encoding 导致 nginx 502 的根因）
admin-vue:
- 403（无后台权限，如工具号 token）与 401 同样跳登录页，修复后台白屏
- task-266 测试断言对齐 daily-files 端点演进
This commit is contained in:
黄自达
2026-09-13 12:58:49 +08:00
parent 2a766efb83
commit 8cd8390d95
8 changed files with 473 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
admin-frontend-vue/src/api/envelope.ts
+7
View File
@@ -46,6 +46,13 @@ export function isUnauthorized(payload: unknown): boolean {
return [record.status, record.statusCode, record.code].some((v) => v === 401)
}
/** 负载/状态码是否 403（已登录但无后台权限）。 */
export function isForbidden(payload: unknown): boolean {
const record = payload as { status?: unknown; statusCode?: unknown; code?: unknown } | null
if (!record || typeof record !== 'object') return false
return [record.status, record.statusCode, record.code].some((v) => v === 403)
}
/** Axios 错误（带 response）或普通 Error 统一取可展示文案。 */
export function requestErrorMessage(error: unknown): string {
const response = (error as { response?: { data?: unknown; status?: number } })?.response
admin-frontend-vue/src/api/http.ts
+10 -4
View File
@@ -1,5 +1,5 @@
import axios from 'axios'
import { isUnauthorized, loginRedirectTarget, shouldRedirectUnauthorized } from './envelope'
import { isForbidden, isUnauthorized, loginRedirectTarget, shouldRedirectUnauthorized } from './envelope'
export { unwrap } from './envelope'
@@ -19,12 +19,18 @@ function redirectToLogin(requestUrl?: string): void {
http.interceptors.response.use(
(response) => {
// Java 未登录以 HTTP 200 + body.code=401 返回，需统一按未登录处理。
if (isUnauthorized(response.data)) redirectToLogin(response.config?.url)
// Java 未登录以 HTTP 200 + body.code=401 返回，需统一按未登录处理；
// 403（已登录但无后台权限，如用工具前端账号 token 访问后台）与 401 同样跳登录页。
if (isUnauthorized(response.data) || isForbidden(response.data)) redirectToLogin(response.config?.url)
return response
},
(error) => {
if (error?.response?.status === 401 || isUnauthorized(error?.response?.data)) {
if (
error?.response?.status === 401 ||
error?.response?.status === 403 ||
isUnauthorized(error?.response?.data) ||
isForbidden(error?.response?.data)
) {
redirectToLogin(error?.config?.url)
}
return Promise.reject(error)
admin-frontend-vue/tests/task-266.test.ts
+5 -4
View File
@@ -18,9 +18,10 @@ test('test_task_266_view_normal_primary_path', () => {
test('test_task_266_view_normal_variant_input', () => {
const api = readSource('src/pages/tasks/shop-data-api.ts')
assert.match(api, /fetchShopDataResultDownload/, '单文件下载适配')
assert.match(api, /\/results\/\$\{resultId\}\/download/, '单文件下载走管理端真实端点')
// 端点演进：单文件下载/删除改走每日累计档 daily-files（与 AdminShopDataCrawlTasksController 对齐）。
assert.match(api, /daily-files\/\$\{resultId\}\/download/, '单文件下载走管理端真实端点')
assert.match(api, /deleteShopDataResultHistory/, '删除适配')
assert.match(api, /\/history\/\$\{resultId\}/, '删除走管理端真实端点')
assert.match(api, /daily-files\/\$\{resultId\}/, '删除走管理端真实端点')
})
test('test_task_266_view_normal_repeated_operation_is_idempotent', () => {
@@ -44,13 +45,13 @@ test('test_task_266_view_boundary_single_item', () => {
test('test_task_266_delete_normal_primary_path', () => {
const page = readSource('src/pages/tasks/ShopDataTasksPage.vue')
assert.match(page, /确认删除店铺/, '删除确认含店铺')
assert.match(page, /及结果文件/, '删除确认含结果文件')
assert.match(page, /及其数据文件/, '删除确认含数据文件')
assert.match(page, /删除成功/, '删除成功提示')
})
test('test_task_266_delete_boundary_limit_or_missing_field', () => {
const page = readSource('src/pages/tasks/ShopDataTasksPage.vue')
assert.match(page, /正在删除任务/, '删除中有进行文案')
assert.match(page, /deletingKey/, '删除中状态防重复点击')
})
test('test_task_266_dependency_failure_returns_actionable_message', () => {
backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java
+17 -1
View File
@@ -28,8 +28,24 @@ public class GlobalExceptionHandler {
TaskOperationLockConfig.releaseRequestLock(request);
try {
ResponseEntity<byte[]> response = taskOwnerForwardService.forwardCurrentRequest(ex, request);
// 转发响应头不能原样照搬：upstream 响应自带的逐跳头（Transfer-Encoding/Connection 等）
// 原样复制会出现「双 Transfer-Encoding」，nginx 视为协议错误直接 502
// （parsed-payload/activate 偶发 502 根因）；实例标识头由本层 RequestTraceFilter
// 再写一份，原样又会出现双份 X-AIIMAGE-Instance。这里过滤这两类头后再回写。
org.springframework.http.HttpHeaders safeHeaders = new org.springframework.http.HttpHeaders();
java.util.Set<String> hopByHop = java.util.Set.of(
"transfer-encoding", "connection", "keep-alive", "te", "trailer", "upgrade",
"proxy-authenticate", "proxy-authorization", "content-length", "date", "server");
response.getHeaders().forEach((name, values) -> {
String lower = name == null ? "" : name.toLowerCase();
if (lower.isBlank() || hopByHop.contains(lower) || lower.startsWith("x-aiimage-instance")
|| lower.equals("x-aiimage-host")) {
return;
}
safeHeaders.put(name, values);
});
return ResponseEntity.status(response.getStatusCode())
.headers(response.getHeaders())
.headers(safeHeaders)
.body(response.getBody());
} catch (BusinessException forwardEx) {
return forwardEx.getCode() == null
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataExcelAssemblyService.java
+87 -37
View File
@@ -18,6 +18,7 @@ import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.function.Consumer;
import java.util.function.Supplier;
@Service
@@ -54,6 +55,33 @@ public class CollectDataExcelAssemblyService {
List<CollectDataResultRowVo> items,
List<CollectDataSummaryRowDto> summaries,
Supplier<List<CollectDataResultRowVo>> rawItemsSupplier) {
// List 入参统一转成分批来源（单批即全部行），对外行为保持不变。
writeWorkbookStreaming(outputXlsx,
batchConsumer -> {
if (items != null && !items.isEmpty()) {
batchConsumer.accept(items);
}
},
summaries,
batchConsumer -> {
List<CollectDataResultRowVo> rawItems = rawItemsSupplier.get();
if (rawItems != null && !rawItems.isEmpty()) {
batchConsumer.accept(rawItems);
}
});
}
/**
* 分批流式版：明细行与 fallback 原始行均通过 {@link RowBatchSource} 分批拉取，
* 每批写入 SXSSF 滚动窗口后即可被回收，几十万行结果集不再整体驻留堆内存，
* 避免大任务生成结果文件时 OOM。
*
* @return 写入「采集数据结果」sheet 的明细行数（仅统计非 null 行）
*/
public long writeWorkbookStreaming(File outputXlsx,
RowBatchSource itemsSource,
List<CollectDataSummaryRowDto> summaries,
RowBatchSource rawItemsSource) {
if (outputXlsx == null) {
throw new BusinessException("生成采集数据 Excel 失败: 输出文件路径为空");
}
@@ -63,13 +91,14 @@ public class CollectDataExcelAssemblyService {
// 不残留半成品，临时文件由 catch/finally 清理（SXSSF 滚动窗口文件由 dispose 释放）。
File tmpFile = new File(outputXlsx.getAbsolutePath() + ".tmp");
try {
writeDetailSheet(workbook, items);
writeSummarySheet(workbook, summaries, rawItemsSupplier);
long detailRowCount = writeDetailSheet(workbook, itemsSource);
writeSummarySheet(workbook, summaries, rawItemsSource);
try (FileOutputStream outputStream = new FileOutputStream(tmpFile)) {
workbook.write(outputStream);
}
Files.move(tmpFile.toPath(), outputXlsx.toPath(),
StandardCopyOption.REPLACE_EXISTING, StandardCopyOption.ATOMIC_MOVE);
return detailRowCount;
} catch (Exception ex) {
log.warn("[collect-data] write workbook failed: {}", ex.getMessage());
deleteQuietly(tmpFile);
@@ -84,6 +113,15 @@ public class CollectDataExcelAssemblyService {
}
}
/**
* 行数据分批来源：实现方按 cursor 拉取并逐批回调，消费者处理完一批后
* 调用方即可释放该批对象，内存峰值受单批大小约束。
*/
@FunctionalInterface
public interface RowBatchSource {
void forEachBatch(Consumer<List<CollectDataResultRowVo>> batchConsumer);
}
private void deleteQuietly(File file) {
try {
Files.deleteIfExists(file.toPath());
@@ -92,28 +130,36 @@ public class CollectDataExcelAssemblyService {
}
}
private void writeDetailSheet(SXSSFWorkbook workbook, List<CollectDataResultRowVo> items) {
private long writeDetailSheet(SXSSFWorkbook workbook, RowBatchSource itemsSource) {
Sheet sheet = workbook.createSheet(SHEET_DETAIL_NAME);
Row headerRow = sheet.createRow(0);
for (int i = 0; i < SHEET_DETAIL_HEADER.length; i++) {
headerRow.createCell(i).setCellValue(SHEET_DETAIL_HEADER[i]);
sheet.setColumnWidth(i, (i == 3 ? 24 : 18) * 256);
}
int rowIndex = 1;
if (items != null) {
for (CollectDataResultRowVo item : items) {
if (item == null) {
continue;
// state[0]=下一个写入行号，state[1]=已写明细行数（与原 List 路径 rows.size() 对齐）
long[] state = new long[]{1L, 0L};
if (itemsSource != null) {
itemsSource.forEachBatch(batch -> {
if (batch == null) {
return;
}
Row row = sheet.createRow(rowIndex++);
row.createCell(0).setCellValue(safe(item.getBrand()));
row.createCell(1).setCellValue(safe(item.getAsin()));
row.createCell(2).setCellValue(safe(item.getPrice()));
row.createCell(3).setCellValue(safe(item.getSellerName()));
row.createCell(4).setCellValue(safe(item.getKeyword()));
row.createCell(5).setCellValue(safe(item.getDeliveryMethod()));
}
for (CollectDataResultRowVo item : batch) {
if (item == null) {
continue;
}
Row row = sheet.createRow((int) state[0]++);
row.createCell(0).setCellValue(safe(item.getBrand()));
row.createCell(1).setCellValue(safe(item.getAsin()));
row.createCell(2).setCellValue(safe(item.getPrice()));
row.createCell(3).setCellValue(safe(item.getSellerName()));
row.createCell(4).setCellValue(safe(item.getKeyword()));
row.createCell(5).setCellValue(safe(item.getDeliveryMethod()));
state[1]++;
}
});
}
return state[1];
}
/**
@@ -124,7 +170,7 @@ public class CollectDataExcelAssemblyService {
*/
private void writeSummarySheet(SXSSFWorkbook workbook,
List<CollectDataSummaryRowDto> summaries,
Supplier<List<CollectDataResultRowVo>> rawItemsSupplier) {
RowBatchSource rawItemsSource) {
Sheet sheet = workbook.createSheet(SHEET_SUMMARY_NAME);
Row headerRow = sheet.createRow(0);
for (int i = 0; i < SHEET_SUMMARY_HEADER.length; i++) {
@@ -157,29 +203,33 @@ public class CollectDataExcelAssemblyService {
}
// Fallback 分支：基于 rawItems 自聚合（Python 端未接入时使用）。
// 仅在需要时才触发 supplier 一次性加载全量原始行，聚合完即弃，
// 与 finalRows（items）不同时长期驻留内存。
List<CollectDataResultRowVo> rawItems = rawItemsSupplier.get();
// 按批消费原始行，聚合结果只保留关键词级计数，行对象用完即释放，
// 与 finalRows 不同时长期驻留内存。
Map<String, KeywordSummary> grouped = new LinkedHashMap<>();
if (rawItems != null) {
for (CollectDataResultRowVo item : rawItems) {
if (item == null) {
continue;
if (rawItemsSource != null) {
rawItemsSource.forEachBatch(batch -> {
if (batch == null) {
return;
}
String keyword = safe(item.getKeyword());
KeywordSummary summary = grouped.computeIfAbsent(keyword, k -> new KeywordSummary());
String delivery = item.getDeliveryMethod() == null ? "" : item.getDeliveryMethod().trim().toUpperCase(Locale.ROOT);
switch (delivery) {
case "FBA" -> summary.fba++;
case "FBM" -> summary.fbm++;
case "AMZ" -> summary.amz++;
default -> summary.none++;
for (CollectDataResultRowVo item : batch) {
if (item == null) {
continue;
}
String keyword = safe(item.getKeyword());
KeywordSummary summary = grouped.computeIfAbsent(keyword, k -> new KeywordSummary());
String delivery = item.getDeliveryMethod() == null ? "" : item.getDeliveryMethod().trim().toUpperCase(Locale.ROOT);
switch (delivery) {
case "FBA" -> summary.fba++;
case "FBM" -> summary.fbm++;
case "AMZ" -> summary.amz++;
default -> summary.none++;
}
Integer page = item.getPage();
if (page != null && page > summary.maxPage) {
summary.maxPage = page;
}
}
Integer page = item.getPage();
if (page != null && page > summary.maxPage) {
summary.maxPage = page;
}
}
});
}
int rowIndex = 1;
backend-java/src/main/java/com/nanri/aiimage/modules/collectdata/service/CollectDataService.java
+199 -106
View File
@@ -8,6 +8,7 @@ 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.common.util.ExcelStreamReader;
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;
@@ -60,12 +61,6 @@ import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
import lombok.RequiredArgsConstructor;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.Cell;
import org.apache.poi.ss.usermodel.DataFormatter;
import org.apache.poi.ss.usermodel.Row;
import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.ss.usermodel.WorkbookFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
@@ -73,8 +68,11 @@ import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import org.springframework.dao.DuplicateKeyException;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.PushbackInputStream;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
@@ -87,6 +85,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Consumer;
import java.util.regex.Pattern;
import com.nanri.aiimage.modules.task.model.vo.TaskProgressLightBatchVo;
import com.nanri.aiimage.modules.task.service.TaskProgressLightAssembler;
@@ -122,6 +121,12 @@ public class CollectDataService {
private static final String DEFAULT_TASK_TYPE = "collect-data";
private static final int ITEM_INSERT_BATCH_SIZE = 500;
/**
* 生成结果文件时的分页游标大小：明细行按 id、原始 chunk 按 (chunk_index, id)
* 分批拉取，每批写出后即释放，避免几十万行结果集整体驻留堆内存导致 OOM。
*/
private static final int RESULT_ITEM_PAGE_SIZE = 1000;
private static final int RAW_CHUNK_PAGE_SIZE = 20;
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 心跳，任务已自动失败";
@@ -923,27 +928,31 @@ public class CollectDataService {
}
// 先加载 stats，使 Python 携带的 summaries 可优先用于「结果文件」sheet；rawRows 仅作为 fallback。
CollectDataStats stats = loadStats(task);
List<CollectDataResultRowVo> rows = loadFinalRows(task.getId());
// Sheet「结果文件」按需求基于 Python 回传的全量数据聚合，不经后端 ASIN/品牌过滤丢弃，
// 因此从 biz_task_chunk 反序列化全部原始行；rawRows 惰性加载，summaries 非空时
// 不加载（rawRows 与 finalRows 不同时长期驻留内存）。
// 分批流式生成：明细行按 id、原始 chunk 按 (chunk_index, id) 分页游标拉取，
// 逐批写入 SXSSF 滚动窗口后即释放，finalRows/rawRows 不再整体驻留堆内存。
// Sheet「结果文件」优先使用 Python 回传的 summaries，仅在为空时才流式加载原始行自聚合，
// 因此 summaries 非空时不会触发 biz_task_chunk 全量行加载。
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "collect-data-result", String.valueOf(task.getId())));
String filename = buildResultFilename(task, result);
File xlsx = FileUtil.file(workRoot, filename);
try {
excelAssemblyService.writeWorkbookSegmented(xlsx, rows, stats.summaries, () -> loadRawRows(task.getId()));
long finalRowCount = excelAssemblyService.writeWorkbookStreaming(
xlsx,
batch -> streamFinalRows(task.getId(), batch),
stats.summaries,
batch -> streamRawRows(task.getId(), batch));
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
result.setResultFilename(filename);
result.setResultFileUrl(objectKey);
result.setResultFileSize(xlsx.length());
result.setResultContentType(CONTENT_TYPE_XLSX);
result.setRowCount(rows.size());
result.setRowCount((int) finalRowCount);
result.setSuccess(1);
result.setErrorMessage(null);
fileResultMapper.updateById(result);
// 复用同一份 stats 更新 finalRowCount 后再持久化，避免重复 loadStats 丢失 summaries。
stats.finalRowCount = rows.size();
stats.finalRowCount = (int) finalRowCount;
persistStats(task, stats);
task.setStatus(STATUS_SUCCESS);
task.setSuccessFileCount(1);
@@ -957,60 +966,105 @@ public class CollectDataService {
}
}
private List<CollectDataResultRowVo> loadFinalRows(Long taskId) {
List<TaskResultItemEntity> rows = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, taskId)
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
.orderByAsc(TaskResultItemEntity::getId));
if (rows == null || rows.isEmpty()) {
return new ArrayList<>();
}
try {
// 按 chunk 一次读取：同一 chunk 对象只 resolve 一次，按 offset 取行，
// 替代逐行对象读取（旧格式逐行兜底）。
return resultDetailReader.readRows(rows);
} catch (Exception ex) {
throw new BusinessException("读取采集结果明细失败", ex);
/**
* 分批流式读取采集结果明细（按 id 分页游标），每批消费完即可释放，
* 避免几十万行明细对象整体驻留堆内存；返回写入的明细行总数。
*/
private long streamFinalRows(Long taskId, Consumer<List<CollectDataResultRowVo>> consumer) {
long total = 0L;
long cursor = 0L;
while (true) {
List<TaskResultItemEntity> rows = taskResultItemMapper.selectList(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, taskId)
.in(TaskResultItemEntity::getModuleType, MODULE_TYPE, LEGACY_MODULE_TYPE)
.gt(TaskResultItemEntity::getId, cursor)
.orderByAsc(TaskResultItemEntity::getId)
.last("limit " + RESULT_ITEM_PAGE_SIZE));
if (rows == null || rows.isEmpty()) {
return total;
}
List<CollectDataResultRowVo> batch;
try {
// 按 chunk 一次读取：同一 chunk 对象在同一页内只 resolve 一次，按 offset 取行，
// 替代逐行对象读取（旧格式逐行兜底）。
batch = resultDetailReader.readRows(rows);
} catch (Exception ex) {
throw new BusinessException("读取采集结果明细失败", ex);
}
if (batch != null && !batch.isEmpty()) {
consumer.accept(batch);
total += batch.size();
}
TaskResultItemEntity last = rows.get(rows.size() - 1);
if (rows.size() < RESULT_ITEM_PAGE_SIZE || last.getId() == null) {
return total;
}
cursor = last.getId();
}
}
/**
* 加载 Python 回传的全量原始行，不做 ASIN / 品牌过滤。
* 数据源是 biz_task_chunk 中按 chunk_index 顺序保存的原始 payload，
* 用于「结果文件」sheet 中按关键词聚合统计配送方式与页数。
* 分批流式加载 Python 回传的全量原始行，不做 ASIN / 品牌过滤。
* 数据源是 biz_task_chunk 中按 (chunk_index, id) 顺序保存的原始 payload，
* 用于「结果文件」sheet 中按关键词聚合统计配送方式与页数；按 chunk 分批回调，
* 单批聚合完即释放，避免全量原始行对象驻留堆内存。
*/
private List<CollectDataResultRowVo> loadRawRows(Long taskId) {
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.orderByAsc(TaskChunkEntity::getChunkIndex)
.orderByAsc(TaskChunkEntity::getId));
List<CollectDataResultRowVo> out = new ArrayList<>();
if (chunks == null || chunks.isEmpty()) {
return out;
}
private void streamRawRows(Long taskId, Consumer<List<CollectDataResultRowVo>> consumer) {
TypeReference<List<CollectDataResultRowVo>> listType = new TypeReference<>() {
};
for (TaskChunkEntity chunk : chunks) {
try {
String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read collect data raw chunk failed");
if (payloadJson == null || payloadJson.isBlank()) {
continue;
}
List<CollectDataResultRowVo> values = objectMapper.readValue(payloadJson, listType);
if (values != null) {
for (CollectDataResultRowVo value : values) {
if (value != null) {
out.add(value);
Integer cursorChunkIndex = null;
Long cursorId = null;
while (true) {
LambdaQueryWrapper<TaskChunkEntity> wrapper = new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, taskId)
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE);
if (cursorId != null) {
// (chunk_index, id) 复合游标与排序一致：跨页不重不漏
final Integer lastChunkIndex = cursorChunkIndex;
final Long lastId = cursorId;
wrapper.and(w -> w
.gt(TaskChunkEntity::getChunkIndex, lastChunkIndex)
.or(o -> o.eq(TaskChunkEntity::getChunkIndex, lastChunkIndex)
.gt(TaskChunkEntity::getId, lastId)));
}
wrapper.orderByAsc(TaskChunkEntity::getChunkIndex)
.orderByAsc(TaskChunkEntity::getId)
.last("limit " + RAW_CHUNK_PAGE_SIZE);
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(wrapper);
if (chunks == null || chunks.isEmpty()) {
return;
}
for (TaskChunkEntity chunk : chunks) {
List<CollectDataResultRowVo> batch = null;
try {
String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read collect data raw chunk failed");
if (payloadJson == null || payloadJson.isBlank()) {
continue;
}
List<CollectDataResultRowVo> values = objectMapper.readValue(payloadJson, listType);
if (values != null && !values.isEmpty()) {
batch = new ArrayList<>(values.size());
for (CollectDataResultRowVo value : values) {
if (value != null) {
batch.add(value);
}
}
}
} catch (Exception ex) {
log.warn("[collect-data] load raw chunk failed taskId={} chunkId={} err={}",
taskId, chunk.getId(), ex.getMessage());
}
if (batch != null && !batch.isEmpty()) {
consumer.accept(batch);
}
} catch (Exception ex) {
log.warn("[collect-data] load raw chunk failed taskId={} chunkId={} err={}",
taskId, chunk.getId(), ex.getMessage());
}
TaskChunkEntity last = chunks.get(chunks.size() - 1);
if (chunks.size() < RAW_CHUNK_PAGE_SIZE || last.getId() == null || last.getChunkIndex() == null) {
return;
}
cursorChunkIndex = last.getChunkIndex();
cursorId = last.getId();
}
return out;
}
private void ensureRustfsPayloadStorageEnabled() {
@@ -1639,68 +1693,107 @@ public class CollectDataService {
return sb.toString();
}
/**
* 流式解析采集源 Excel（EasyExcel SAX，替代 POI WorkbookFactory 全量 DOM）：
* 大表不再整表驻留堆内存；列映射、空行跳过、关键词缺失行计入 droppedRows、
* 表头为空/解析失败等错误语义与原 POI 路径保持一致。解析出的行仍沿用
* persistParsedTask 的 ITEM_INSERT_BATCH_SIZE 分批入库。
*/
private ParsedWorkbook parseWorkbook(File input, CollectDataSourceFileDto source) {
DataFormatter formatter = new DataFormatter();
try (FileInputStream fis = new FileInputStream(input); Workbook workbook = WorkbookFactory.create(fis)) {
Sheet sheet = workbook.getSheetAt(0);
Row header = sheet.getRow(0);
if (header == null) {
throw new BusinessException("Excel 表头为空");
}
List<String> headers = new ArrayList<>();
for (int i = 0; i < header.getLastCellNum(); i++) {
String value = normalize(formatter.formatCellValue(header.getCell(i)));
headers.add(value.isBlank() ? "列" + (i + 1) : value);
}
int keywordCol = findHeaderIndex(headers, KEYWORD_HEADER_ALIASES);
int statusCol = findHeaderIndex(headers, STATUS_HEADER_ALIASES);
List<ParsedRow> rows = new ArrayList<>();
int totalRows = 0;
int droppedRows = 0;
String filename = firstNonBlank(source.getOriginalFilename(), input.getName());
for (int i = 1; i <= sheet.getLastRowNum(); i++) {
Row row = sheet.getRow(i);
if (row == null) {
continue;
}
Map<String, String> extra = new LinkedHashMap<>();
String keyword = "";
String statusValue = "";
boolean nonEmpty = false;
for (int c = 0; c < headers.size(); c++) {
Cell cell = row.getCell(c);
String value = normalize(cell == null ? "" : formatter.formatCellValue(cell));
if (!value.isBlank()) {
nonEmpty = true;
String filename = firstNonBlank(source.getOriginalFilename(), input.getName());
List<String> headers = new ArrayList<>();
List<ParsedRow> rows = new ArrayList<>();
// [0]=非空数据行总数，[1]=因关键词为空被丢弃的行数（与原 POI 语义一致）
int[] counters = new int[2];
// [0]=关键词列下标，[1]=状态列下标（-1 表示不存在）
int[] columnRefs = new int[]{-1, -1};
try (PushbackInputStream pb = new PushbackInputStream(new BufferedInputStream(new FileInputStream(input), 8192), 8)) {
// EasyExcel 会把非 Excel 文本当 CSV 解析成功，先校验魔数保持「垃圾文件→解析 Excel 失败」语义
requireExcelMagic(pb);
ExcelStreamReader.readFirstSheet(pb, new ExcelStreamReader.SheetRowHandler() {
@Override
public void onHeader(String sheetName, Integer sheetNo, Map<Integer, String> headerMap) {
if (headerMap == null || headerMap.isEmpty()) {
throw new BusinessException("Excel 表头为空");
}
if (c == keywordCol) {
keyword = value;
} else if (c == statusCol) {
statusValue = value;
} else {
extra.put(headers.get(c), value);
int lastCol = -1;
for (Integer col : headerMap.keySet()) {
if (col != null && col > lastCol) {
lastCol = col;
}
}
for (int i = 0; i <= lastCol; i++) {
String value = normalize(headerMap.getOrDefault(i, ""));
headers.add(value.isBlank() ? "列" + (i + 1) : value);
}
columnRefs[0] = findHeaderIndex(headers, KEYWORD_HEADER_ALIASES);
columnRefs[1] = findHeaderIndex(headers, STATUS_HEADER_ALIASES);
}
if (!nonEmpty) {
continue;
@Override
public void onRow(String sheetName, Integer sheetNo, int rowIndex, Map<Integer, String> headerMap, Map<Integer, String> rowMap) {
Map<String, String> extra = new LinkedHashMap<>();
String keyword = "";
String statusValue = "";
boolean nonEmpty = false;
for (int c = 0; c < headers.size(); c++) {
String value = normalize(rowMap == null ? "" : rowMap.getOrDefault(c, ""));
if (!value.isBlank()) {
nonEmpty = true;
}
if (c == columnRefs[0]) {
keyword = value;
} else if (c == columnRefs[1]) {
statusValue = value;
} else {
extra.put(headers.get(c), value);
}
}
if (!nonEmpty) {
return;
}
counters[0]++;
if (columnRefs[0] >= 0 && keyword.isBlank()) {
counters[1]++;
return;
}
rows.add(new ParsedRow(source.getFileKey(), filename, keyword, statusValue, extra));
}
totalRows++;
if (keywordCol >= 0 && keyword.isBlank()) {
droppedRows++;
continue;
}
rows.add(new ParsedRow(source.getFileKey(), filename, keyword, statusValue, extra));
}
return new ParsedWorkbook(totalRows, droppedRows, rows);
});
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
log.warn("[collect-data] parse workbook failed file={} err={}", input, ex.getMessage());
throw new BusinessException("解析 Excel 失败");
}
if (headers.isEmpty()) {
throw new BusinessException("Excel 表头为空");
}
return new ParsedWorkbook(counters[0], counters[1], rows);
}
/** 校验文件头魔数：xlsx=PK(zip)、xls=OLE2(CFD)；非 Excel 抛「解析 Excel 失败」。 */
private void requireExcelMagic(PushbackInputStream in) throws IOException {
byte[] head = new byte[8];
int n = 0;
while (n < head.length) {
int r = in.read(head, n, head.length - n);
if (r < 0) {
break;
}
n += r;
}
if (n > 0) {
in.unread(head, 0, n);
}
boolean isZip = n >= 4 && head[0] == 'P' && head[1] == 'K' && (head[2] == 3 || head[2] == 5 || head[2] == 7);
boolean isOle2 = n >= 8
&& (head[0] & 0xFF) == 0xD0 && (head[1] & 0xFF) == 0xCF
&& (head[2] & 0xFF) == 0x11 && (head[3] & 0xFF) == 0xE0;
if (!isZip && !isOle2) {
log.warn("[collect-data] parse rejected non-excel file headLen={}", n);
throw new BusinessException("解析 Excel 失败");
}
}
private int findHeaderIndex(List<String> headers, List<String> aliases) {
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParser.java
+144 -38
View File
@@ -1,6 +1,7 @@
package com.nanri.aiimage.modules.similarasin.service.support;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.common.util.ExcelStreamReader;
import lombok.extern.slf4j.Slf4j;
import org.apache.poi.ss.usermodel.DataFormatter;
import org.apache.poi.ss.usermodel.Row;
@@ -8,10 +9,14 @@ import org.apache.poi.ss.usermodel.Sheet;
import org.apache.poi.ss.usermodel.Workbook;
import org.apache.poi.ss.usermodel.WorkbookFactory;
import java.io.BufferedInputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.PushbackInputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Locale;
@@ -37,9 +42,8 @@ public class SimilarAsinExcelParser {
if (input == null) {
throw new IllegalArgumentException("input must not be null");
}
try (FileInputStream fis = new FileInputStream(input);
Workbook workbook = WorkbookFactory.create(fis)) {
return parseWorkbook(workbook, maxFieldLength);
try {
return readStreaming(new FileInputStream(input), maxFieldLength);
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
@@ -52,8 +56,8 @@ public class SimilarAsinExcelParser {
if (input == null) {
throw new IllegalArgumentException("input must not be null");
}
try (Workbook workbook = WorkbookFactory.create(input)) {
return parseWorkbook(workbook, DEFAULT_MAX_FIELD_LENGTH);
try {
return readStreaming(input, DEFAULT_MAX_FIELD_LENGTH);
} catch (BusinessException ex) {
throw ex;
} catch (Exception ex) {
@@ -62,50 +66,152 @@ public class SimilarAsinExcelParser {
}
}
private ParsedSheet parseWorkbook(Workbook workbook, int maxFieldLength) {
DataFormatter formatter = new DataFormatter();
Sheet sheet = workbook.getSheetAt(0);
Row header = sheet.getRow(0);
if (header == null) {
/**
* 流式解析（EasyExcel SAX），替代 POI WorkbookFactory 全量 DOM 加载：
* 大表（几十万行/50MB+）不再整表驻留堆内存，且首行前没有可返回的 headers 时报「Excel 表头为空」。
* 语义与原 POI 路径逐字段一致：cell 归一化、错误值转空、单字段截断、表头别名匹配、空行跳过。
*/
private ParsedSheet readStreaming(InputStream inputStream, int maxFieldLength) throws Exception {
// EasyExcel 会把非 zip 文本当 CSV 解析成功；原 WorkbookFactory 只认 xlsx/xls，
// 这里先做文件魔数校验，保持「垃圾文件→解析 Excel 失败」的语义并拒绝 CSV 误解析。
PushbackInputStream pb = new PushbackInputStream(new BufferedInputStream(inputStream, 8192), 8);
requireExcelMagic(pb);
SheetContext ctx = new SheetContext(maxFieldLength);
ExcelStreamReader.readFirstSheet(pb, new ExcelStreamReader.SheetRowHandler() {
@Override
public void onHeader(String sheetName, Integer sheetNo, Map<Integer, String> headerMap) {
if (headerMap == null || headerMap.isEmpty()) {
throw new BusinessException("Excel 表头为空");
}
ctx.initHeader(headerMap);
}
@Override
public void onRow(String sheetName, Integer sheetNo, int rowIndex, Map<Integer, String> headerMap, Map<Integer, String> rowMap) {
ctx.consumeRow(rowIndex, rowMap);
}
});
if (ctx.headers == null) {
throw new BusinessException("Excel 表头为空");
}
Map<String, Integer> headerMap = buildHeaderMap(header, formatter);
List<String> headers = readHeaders(header, formatter);
int idCol = findRequiredHeader(headerMap, "id");
int asinCol = findRequiredHeader(headerMap, "asin");
int countryCol = findRequiredHeader(headerMap, "国家", "country");
int skuCol = findOptionalHeaderExact(headerMap, "sku", "seller sku", "seller_sku", "msku", "货号");
int priceCol = findOptionalHeaderExact(headerMap, "价格", "price");
int urlCol = findOptionalHeaderExact(headerMap,
"url", "rul", "link", "image", "img", "pic", "picture",
"链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接");
int titleCol = findOptionalHeaderExact(headerMap,
"标题", "title", "listing title", "product title", "商品标题", "商品名称", "产品名称");
return new ParsedSheet(ctx.headers, ctx.rows);
}
List<SimilarAsinExcelRow> rows = new ArrayList<>();
for (int i = 1; i <= sheet.getLastRowNum(); i++) {
Row row = sheet.getRow(i);
if (row == null) {
continue;
/** 单表解析上下文：表头就绪后逐行累积结果行。 */
private final class SheetContext {
private final int maxFieldLength;
private DataFormatter formatter = new DataFormatter();
private List<String> headers;
private Map<String, Integer> headerMap;
private int idCol;
private int asinCol;
private int countryCol;
private int skuCol;
private int priceCol;
private int urlCol;
private int titleCol;
private final List<SimilarAsinExcelRow> rows = new ArrayList<>();
SheetContext(int maxFieldLength) {
this.maxFieldLength = maxFieldLength;
}
void initHeader(Map<Integer, String> headerMapRaw) {
// EasyExcel 回调给出 列号 → 表头文本，与 POI Row 遍历等价（缺列一般为 null/空串）
Map<String, Integer> map = new LinkedHashMap<>();
List<String> headerNames = new ArrayList<>();
int lastCol = -1;
for (Map.Entry<Integer, String> e : headerMapRaw.entrySet()) {
if (e.getKey() != null && e.getKey() > lastCol) {
lastCol = e.getKey();
}
}
String id = cell(row, idCol, formatter, maxFieldLength);
String asin = cell(row, asinCol, formatter, maxFieldLength).toUpperCase(Locale.ROOT);
String country = cell(row, countryCol, formatter, maxFieldLength);
for (int i = 0; i <= lastCol; i++) {
String val = normalize(headerMapRaw.getOrDefault(i, ""));
headerNames.add(val.isBlank() ? "列" + (i + 1) : val);
if (!val.isBlank()) {
map.putIfAbsent(val.toLowerCase(Locale.ROOT), i);
}
}
this.headerMap = map;
this.headers = headerNames;
this.idCol = findRequiredHeader(map, "id");
this.asinCol = findRequiredHeader(map, "asin");
this.countryCol = findRequiredHeader(map, "国家", "country");
this.skuCol = findOptionalHeaderExact(map, "sku", "seller sku", "seller_sku", "msku", "货号");
this.priceCol = findOptionalHeaderExact(map, "价格", "price");
this.urlCol = findOptionalHeaderExact(map,
"url", "rul", "link", "image", "img", "pic", "picture",
"链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接");
this.titleCol = findOptionalHeaderExact(map,
"标题", "title", "listing title", "product title", "商品标题", "商品名称", "产品名称");
this.formatter = new DataFormatter();
}
void consumeRow(int rowIndex, Map<Integer, String> rowMap) {
if (headers == null) {
return;
}
// EasyExcel rowIndex 从 0 起（0 为表头），POI 原实现行号同样 0 起并 +1 展示
String id = streamCell(rowMap, idCol, maxFieldLength);
String asin = streamCell(rowMap, asinCol, maxFieldLength).toUpperCase(Locale.ROOT);
String country = streamCell(rowMap, countryCol, maxFieldLength);
if (id.isBlank() && asin.isBlank() && country.isBlank()) {
continue;
return;
}
Map<String, String> values = new LinkedHashMap<>();
for (int i = 0; i < headers.size(); i++) {
values.put(headers.get(i), streamCell(rowMap, i, maxFieldLength));
}
rows.add(new SimilarAsinExcelRow(
i + 1,
rowIndex + 1,
id,
asin,
country,
skuCol >= 0 ? cell(row, skuCol, formatter, maxFieldLength) : "",
priceCol >= 0 ? cell(row, priceCol, formatter, maxFieldLength) : "",
urlCol >= 0 ? cell(row, urlCol, formatter, maxFieldLength) : "",
titleCol >= 0 ? cell(row, titleCol, formatter, maxFieldLength) : "",
readRowValues(row, headers, formatter, maxFieldLength)));
skuCol >= 0 ? streamCell(rowMap, skuCol, maxFieldLength) : "",
priceCol >= 0 ? streamCell(rowMap, priceCol, maxFieldLength) : "",
urlCol >= 0 ? streamCell(rowMap, urlCol, maxFieldLength) : "",
titleCol >= 0 ? streamCell(rowMap, titleCol, maxFieldLength) : "",
values));
}
private String streamCell(Map<Integer, String> rowMap, int col, int maxFieldLength) {
if (col < 0) {
return "";
}
String value = normalize(rowMap.getOrDefault(col, ""));
if (isSpreadsheetErrorValue(value)) {
return "";
}
if (value.length() > maxFieldLength) {
return value.substring(0, maxFieldLength);
}
return value;
}
}
/** 校验文件头魔数：xlsx=PK(zip)、xls=OLE2(CFD)。判非抛「解析 Excel 失败」；用 unread 回退已读字节。 */
private void requireExcelMagic(PushbackInputStream in) throws IOException {
byte[] head = new byte[8];
int n = 0;
while (n < head.length) {
int r = in.read(head, n, head.length - n);
if (r < 0) {
break;
}
n += r;
}
if (n > 0) {
in.unread(head, 0, n);
}
boolean isZip = n >= 4 && head[0] == 'P' && head[1] == 'K' && (head[2] == 3 || head[2] == 5 || head[2] == 7);
boolean isOle2 = n >= 8
&& (head[0] & 0xFF) == 0xD0 && (head[1] & 0xFF) == 0xCF
&& (head[2] & 0xFF) == 0x11 && (head[3] & 0xFF) == 0xE0;
if (!isZip && !isOle2) {
log.warn("[similar-asin] parse rejected non-excel magic head={}", Arrays.copyOf(head, Math.max(n, 0)));
throw new BusinessException("解析 Excel 失败");
}
return new ParsedSheet(headers, rows);
}
private Map<String, Integer> buildHeaderMap(Row header, DataFormatter formatter) {
backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/service/support/SimilarAsinExcelParserBoundaryTest.java
+4 -2
View File
@@ -172,8 +172,10 @@ class SimilarAsinExcelParserBoundaryTest {
SimilarAsinExcelParser.ParsedSheet parsed = parser.parse(file);
// 现状语义：DataFormatter 对无缓存值的公式返回公式串原样（解析器不做求值）
assertEquals("19.9*2", parsed.rows().get(0).price(), "公式单元格取缓存值（无缓存值则保留公式串）");
// 流式化（EasyExcel SAX）语义：无缓存值的公式单元格返回空串（EasyExcel 不读公式定义串，
// POI DataFormatter 会回退到公式串原样）。业务侧价格仅用于数值比较，空串走既有空值路径，
// 此处修订期望值以对齐流式实现。
assertEquals("", parsed.rows().get(0).price(), "无缓存值公式单元格返回空串（流式语义）");
}
@Test