From 5b1ccad40e5213cc52b147e3eaf3b30d11ee89a0 Mon Sep 17 00:00:00 2001 From: supernijia Date: Tue, 11 Aug 2026 14:50:21 +0800 Subject: [PATCH] =?UTF-8?q?=E5=90=8E=E5=8F=B0=E6=9B=B4=E6=96=B0=20=20=20?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1=E4=BC=98=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend-java/docs/baota-start-commands.md | 5 + .../aiimage/config/ImageVideoProperties.java | 2 + .../aiimage/config/SimilarAsinProperties.java | 11 +- .../service/AppearancePatentTaskService.java | 202 +++++--- .../controller/DedupeTotalDataController.java | 9 +- .../dedupe/mapper/DedupeTotalDataMapper.java | 18 + .../service/DedupeTotalDataService.java | 259 +++++++---- .../service/ImageVideoCozeService.java | 60 ++- .../service/ShopDataCrawlTaskService.java | 437 ++++++++++-------- .../SimilarAsinImagePrefetchService.java | 4 +- .../service/SimilarAsinTaskService.java | 413 +++++++++++++++-- .../util/SimilarAsinImageEmbedder.java | 220 ++++++--- .../task/model/dto/TaskHeartbeatRequest.java | 3 + .../task/service/TaskHeartbeatService.java | 32 +- .../service/ZiniaoTransientCacheService.java | 52 +++ .../src/main/resources/application.yml | 2 +- .../service/DedupeTotalDataServiceTest.java | 9 +- .../ShopDataCrawlTaskServiceChunkTest.java | 8 +- ...ShopDataCrawlTaskServiceRetentionTest.java | 123 ++++- .../util/SimilarAsinImageEmbedderTest.java | 77 ++- .../service/TaskHeartbeatServiceTest.java | 6 +- backend/ali_oss.py | 119 ++++- backend/blueprints/admin_api.py | 46 +- backend/tests/test_admin_dedupe_total_data.py | 8 + 24 files changed, 1623 insertions(+), 502 deletions(-) diff --git a/backend-java/docs/baota-start-commands.md b/backend-java/docs/baota-start-commands.md index 8c88c105..9f73e4bf 100644 --- a/backend-java/docs/baota-start-commands.md +++ b/backend-java/docs/baota-start-commands.md @@ -28,6 +28,7 @@ AIIMAGE_OSS_PUBLIC_ENDPOINT=https://oss.aishufu.top AIIMAGE_RESULT_FILE_JOB_MQ_ENABLED=true AIIMAGE_ROCKETMQ_NAME_SERVER=121.196.149.225:9876 AIIMAGE_TRANSIENT_STORAGE_ENDPOINT=http://121.196.149.225:9000 +AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE=2 ``` Recommended startup command: @@ -61,3 +62,7 @@ or: ``` If the instanceId in the log does not match the startup command, the actual process was not started with the expected command. + +For the 4-core production host, set the Java container CPU limit to 2.5 cores in 1Panel +(container resources / CPU limit). Keep `AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE=2`. +The application also clamps this pool to half of the JVM-visible processors as a second guard. diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/ImageVideoProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/ImageVideoProperties.java index abf0d8ad..5a49abbb 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/ImageVideoProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/ImageVideoProperties.java @@ -18,6 +18,8 @@ public class ImageVideoProperties { private String voiceSynthesisWorkflowId = "7652954297530105894"; private int cozeConnectTimeoutMillis = 10000; private int cozeReadTimeoutMillis = 3600000; + /** Maximum Coze response body retained by the image/video client. */ + private int cozeResponseMaxBytes = 2 * 1024 * 1024; private int archiveConnectTimeoutMillis = 10000; private int archiveReadTimeoutMillis = 600000; private int archiveMaxAttempts = 3; diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index 5c0404a9..485193b5 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -61,8 +61,8 @@ public class SimilarAsinProperties { */ private int cozeSubmitMaxRetryCount = 5; - /** 图片下载、解码和缩放共享该池；默认 8，避免批量结果生成占满整机 CPU。 */ - private int imageDownloadPoolSize = 8; + /** 图片下载、解码和缩放共享该池；4 核生产机默认 2，避免图片任务占满整机 CPU。 */ + private int imageDownloadPoolSize = 2; /** * 单张图片下载超时（秒）。 @@ -78,6 +78,13 @@ public class SimilarAsinProperties { /** 多源结果文件组装的单任务硬上限；运行期间由文件任务 heartbeat 保活。 */ private int resultFileTimeoutMinutes = 90; + /** + * Assemble each source workbook from chunk-scoped result maps. Keeping a + * switch makes rollback possible for an existing deployment while the new + * bounded path is observed in production. + */ + private boolean boundedResultAssemblyEnabled = true; + /** * assemble 阶段 taskImageCache 的字节上限。 * 默认 256MB：5000 行 × 3 列 × 平均 100KB = 1.5GB 远超 2GB 堆， diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index f3386e1c..b3c67cef 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -2678,9 +2678,10 @@ public class AppearancePatentTaskService { markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit task missing"); return; } - Map currentRows = loadPersistedResultRows(task.getId()); + Map currentRows = loadPersistedResultRowsForResultRows( + task.getId(), batchRows); List currentBatchRows = batchRows.stream() - .map(row -> currentRows.get(rowKey(row))) + .map(row -> findPersistedResultRow(row, currentRows)) .filter(Objects::nonNull) .toList(); if (currentBatchRows.size() == batchRows.size() @@ -3006,45 +3007,48 @@ public class AppearancePatentTaskService { private void assembleResultWorkbook(FileTaskEntity task, FileResultEntity result) { AppearancePatentParsedPayloadDto parsed = readParsedPayload(task); - Map resultMap = loadPersistedResultRowsWithRetry(task.getId(), parsed.getAllItems().size()); - List receivedRows = filterReceivedParsedRows(parsed.getAllItems(), resultMap); - long resolvedRows = receivedRows.stream() - .filter(row -> findResultRow(row, resultMap) != null) - .count(); - long reasonRows = resultMap.values().stream() - .filter(this::hasReasonFields) - .count(); - log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} receivedRows={} resultRows={} resolvedRows={} reasonRows={}", - task.getId(), parsed.getAllItems().size(), receivedRows.size(), resultMap.size(), resolvedRows, reasonRows); - if (!parsed.getAllItems().isEmpty() && receivedRows.isEmpty()) { + List parsedRows = parsed.getAllItems() == null + ? List.of() : parsed.getAllItems(); + List sourceRows = splitRowsBySourceFile(parsed, parsedRows, result.getSourceFilename()); + if (sourceRows.isEmpty()) { throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件"); } - validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap); - // 公共分组处理：同 baseId 连续行视为一组（如 1、1_1、1_2），组内任一行结论命中 - // "已侵权"/"侵权"（包含匹配，"已侵权"优先），则组内所有行结论统一为该标准值。 - // 单行组跳过。仅修改 conclusion 列，不影响其他列。 - int conclusionPropagated = CozeGroupResultPropagator.propagateByGroup( - receivedRows, - AppearancePatentParsedRowVo::getDisplayId, - row -> findResultRow(row, resultMap), - AppearancePatentResultRowDto::getConclusion, - AppearancePatentResultRowDto::setConclusion, - List.of("已侵权", "侵权"), - "[appearance-patent] taskId=" + task.getId() - ); - if (conclusionPropagated > 0) { - log.info("[appearance-patent] group propagate conclusion taskId={} updatedRows={}", - task.getId(), conclusionPropagated); - } + int receivedRowCount = 0; + long resolvedRows = 0L; + long reasonRows = 0L; + int conclusionPropagated = 0; File outputDir = new File(storageProperties.getLocalTempDir(), "appearance-patent-result"); if (!outputDir.exists() && !outputDir.mkdirs()) { throw new BusinessException("创建结果目录失败"); } - List sourceRows = splitRowsBySourceFile(parsed, receivedRows, result.getSourceFilename()); List workbooks = new ArrayList<>(); File zip = null; try { for (SourceRows item : sourceRows) { + Map resultMap = + loadPersistedResultRowsForRowsWithRetry(task.getId(), item.rows()); + List receivedRows = filterReceivedParsedRows(item.rows(), resultMap); + if (receivedRows.isEmpty()) { + resultMap.clear(); + continue; + } + receivedRowCount += receivedRows.size(); + resolvedRows += receivedRows.stream() + .filter(row -> findResultRow(row, resultMap) != null) + .count(); + reasonRows += resultMap.values().stream() + .filter(this::hasReasonFields) + .count(); + validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap); + conclusionPropagated += CozeGroupResultPropagator.propagateByGroup( + receivedRows, + AppearancePatentParsedRowVo::getDisplayId, + row -> findResultRow(row, resultMap), + AppearancePatentResultRowDto::getConclusion, + AppearancePatentResultRowDto::setConclusion, + List.of("已侵权", "侵权"), + "[appearance-patent] taskId=" + task.getId() + ); String filename = safeFileStem(item.sourceFilename()) + "-result.xlsx"; String tempFilename = safeFileStem(item.sourceFilename()) + "-" + task.getId() @@ -3052,12 +3056,20 @@ public class AppearancePatentTaskService { + "-" + UUID.randomUUID() + "-result.xlsx"; File xlsx = new File(outputDir, tempFilename); - writeResultWorkbook(xlsx, parsed, item.rows(), resultMap); - workbooks.add(new SourceResultWorkbook(xlsx, filename, item.rows().size())); + writeResultWorkbook(xlsx, parsed, receivedRows, resultMap); + workbooks.add(new SourceResultWorkbook(xlsx, filename, receivedRows.size())); + resultMap.clear(); } if (workbooks.isEmpty()) { throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件"); } + log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} receivedRows={} resultRows={} resolvedRows={} reasonRows={}", + task.getId(), parsedRows.size(), receivedRowCount, + workbooks.stream().mapToInt(SourceResultWorkbook::rowCount).sum(), resolvedRows, reasonRows); + if (conclusionPropagated > 0) { + log.info("[appearance-patent] group propagate conclusion taskId={} updatedRows={}", + task.getId(), conclusionPropagated); + } File uploadFile; String filename; @@ -3085,7 +3097,7 @@ public class AppearancePatentTaskService { result.setResultFileUrl(objectKey); result.setResultFileSize(uploadFile.length()); result.setResultContentType(contentType); - result.setRowCount(receivedRows.size()); + result.setRowCount(receivedRowCount); } finally { for (SourceResultWorkbook workbook : workbooks) { if (workbook.file().exists() && !workbook.file().delete()) { @@ -3249,18 +3261,6 @@ public class AppearancePatentTaskService { && hasUsableCozeField(row.getConclusion()); } - private Map loadPersistedResultRows(Long taskId) { - Map result = new LinkedHashMap<>(); - List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() - .eq(TaskChunkEntity::getTaskId, taskId) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) - .orderByAsc(TaskChunkEntity::getChunkIndex)); - for (TaskChunkEntity chunk : chunks) { - result.putAll(readChunkRows(chunk)); - } - return result; - } - private boolean hasPersistedResultRows(Long taskId) { if (taskId == null) { return false; @@ -3271,25 +3271,121 @@ public class AppearancePatentTaskService { return count != null && count > 0; } - private Map loadPersistedResultRowsWithRetry(Long taskId, int expectedParsedRows) { - Map result = loadPersistedResultRows(taskId); - if (expectedParsedRows <= 0 || !result.isEmpty()) { + private Map loadPersistedResultRowsForRowsWithRetry( + Long taskId, List requestedRows) { + Set requestedKeys = resultLookupKeysForParsedRows(requestedRows); + Map result = + loadPersistedResultRowsForKeys(taskId, requestedKeys); + if (requestedKeys.isEmpty() || !result.isEmpty()) { return result; } for (int attempt = 1; attempt <= RESULT_ROWS_READ_RETRY_LIMIT && result.isEmpty(); attempt++) { sleepBeforeResultRowsRetry(attempt); - result = loadPersistedResultRows(taskId); + result = loadPersistedResultRowsForKeys(taskId, requestedKeys); if (!result.isEmpty()) { log.info("[appearance-patent] result rows recovered after retry taskId={} attempt={} rows={}", taskId, attempt, result.size()); - return result; } - log.warn("[appearance-patent] result rows still empty after retry taskId={} attempt={}/{}", - taskId, attempt, RESULT_ROWS_READ_RETRY_LIMIT); } return result; } + private Map loadPersistedResultRowsForResultRows( + Long taskId, List requestedRows) { + Set requestedKeys = resultLookupKeysForResultRows(requestedRows); + return loadPersistedResultRowsForKeys(taskId, requestedKeys); + } + + private Map loadPersistedResultRowsForKeys( + Long taskId, Set requestedKeys) { + Map result = new LinkedHashMap<>(); + if (taskId == null || requestedKeys == null || requestedKeys.isEmpty()) { + return result; + } + Long lastChunkId = null; + while (true) { + LambdaQueryWrapper query = new LambdaQueryWrapper() + .eq(TaskChunkEntity::getTaskId, taskId) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) + .orderByAsc(TaskChunkEntity::getId) + .last("LIMIT 1"); + if (lastChunkId != null) { + query.gt(TaskChunkEntity::getId, lastChunkId); + } + List page = taskChunkMapper.selectList(query); + if (page == null || page.isEmpty()) { + break; + } + TaskChunkEntity chunk = page.getFirst(); + appendRequestedResultRows(result, readChunkRows(chunk).values(), requestedKeys); + lastChunkId = chunk.getId(); + if (lastChunkId == null) { + break; + } + } + return result; + } + + private Set resultLookupKeysForParsedRows(List rows) { + Set keys = new LinkedHashSet<>(); + for (AppearancePatentParsedRowVo row : rows == null ? List.of() : rows) { + String primary = rowKey(row); + if (!primary.isBlank()) { + keys.add(primary); + } + String legacy = legacyRowKey(row); + if (!legacy.isBlank()) { + keys.add(legacy); + } + } + return keys; + } + + private Set resultLookupKeysForResultRows(List rows) { + Set keys = new LinkedHashSet<>(); + for (AppearancePatentResultRowDto row : rows == null ? List.of() : rows) { + String primary = rowKey(row); + if (!primary.isBlank()) { + keys.add(primary); + } + String legacy = legacyRowKey(row); + if (!legacy.isBlank()) { + keys.add(legacy); + } + } + return keys; + } + + private void appendRequestedResultRows(Map sink, + Iterable rows, + Set requestedKeys) { + if (rows == null || requestedKeys == null || requestedKeys.isEmpty()) { + return; + } + for (AppearancePatentResultRowDto row : rows) { + if (row == null) { + continue; + } + String primary = rowKey(row); + String legacy = legacyRowKey(row); + String matchedKey = requestedKeys.contains(primary) + ? primary : requestedKeys.contains(legacy) ? legacy : ""; + if (!matchedKey.isBlank()) { + sink.put(matchedKey, row); + } + } + } + + private AppearancePatentResultRowDto findPersistedResultRow( + AppearancePatentResultRowDto row, + Map resultMap) { + if (row == null || resultMap == null || resultMap.isEmpty()) { + return null; + } + AppearancePatentResultRowDto result = resultMap.get(rowKey(row)); + return result == null ? resultMap.get(legacyRowKey(row)) : result; + } + private void sleepBeforeResultRowsRetry(int attempt) { try { Thread.sleep(RESULT_ROWS_READ_RETRY_DELAY_MS * attempt); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/controller/DedupeTotalDataController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/controller/DedupeTotalDataController.java index 4db63e59..3aaf6e79 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/controller/DedupeTotalDataController.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/controller/DedupeTotalDataController.java @@ -38,6 +38,7 @@ import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RequestParam; import org.springframework.web.bind.annotation.RestController; import org.springframework.web.multipart.MultipartFile; +import org.springframework.web.servlet.mvc.method.annotation.StreamingResponseBody; import java.time.LocalDate; import java.time.LocalDateTime; @@ -91,7 +92,7 @@ public class DedupeTotalDataController { @GetMapping("/export") @Operation(summary = "导出总数据", description = "按上传用户名和创建日期导出当前用户可访问的总数据。") - public ResponseEntity export( + public ResponseEntity export( @Parameter(description = "用户名模糊搜索关键字") @RequestParam(required = false) String username, @Parameter(description = "开始日期（包含）") @RequestParam(required = false) @DateTimeFormat(iso = DateTimeFormat.ISO.DATE) LocalDate startDate, @@ -100,13 +101,13 @@ public class DedupeTotalDataController { @Parameter(description = "分组 ID") @RequestParam(required = false) Long groupId, HttpServletRequest request) { RequestOperator operator = requireDedupeTotalDataAccess(request); - byte[] bytes = dedupeTotalDataService.export(username, startDate, endDate, groupId, operator.id()); String filename = "dedupe-total-data-" + LocalDateTime.now().format(EXPORT_FILENAME_FORMATTER) + ".xlsx"; + StreamingResponseBody body = outputStream -> dedupeTotalDataService.writeExport( + outputStream, username, startDate, endDate, groupId, operator.id()); return ResponseEntity.ok() .header(HttpHeaders.CONTENT_DISPOSITION, DownloadHeaderUtil.contentDisposition(filename)) .contentType(MediaType.parseMediaType("application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")) - .contentLength(bytes.length) - .body(bytes); + .body(body); } @PostMapping diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/mapper/DedupeTotalDataMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/mapper/DedupeTotalDataMapper.java index 6b9f8a68..ebcc8686 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/mapper/DedupeTotalDataMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/mapper/DedupeTotalDataMapper.java @@ -2,6 +2,7 @@ package com.nanri.aiimage.modules.dedupe.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.nanri.aiimage.modules.dedupe.model.entity.DedupeTotalDataEntity; +import org.apache.ibatis.annotations.Insert; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; @@ -19,6 +20,23 @@ public interface DedupeTotalDataMapper extends BaseMapper @SelectProvider(type = SqlProvider.class, method = "selectExistingDataValues") List selectExistingDataValues(@Param("values") List values); + /** + * Import is deliberately a single SQL round trip per batch. The unique + * index remains the final concurrency guard when another request inserts + * the same value between the pre-check and this statement. + */ + @Insert(""" + + """) + int insertBatchIgnore(@Param("rows") List rows); + class SqlProvider { public String selectExistingDataValues(@Param("values") List values) { StringJoiner placeholders = new StringJoiner(", "); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java index 12c82154..abf240cd 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java @@ -36,10 +36,12 @@ import org.springframework.web.server.ResponseStatusException; import java.io.ByteArrayOutputStream; import java.io.File; import java.io.InputStream; +import java.io.OutputStream; import java.nio.file.Files; import java.time.LocalDate; import java.time.LocalDateTime; import java.time.format.DateTimeFormatter; +import java.util.ArrayList; import java.util.Collection; import java.util.HashMap; import java.util.HashSet; @@ -50,12 +52,15 @@ import java.util.Locale; import java.util.Map; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; +import java.util.function.Supplier; @Service @RequiredArgsConstructor public class DedupeTotalDataService { private static final int COMPARE_BATCH_SIZE = 5000; + private static final int IMPORT_BATCH_SIZE = 5000; + private static final int EXPORT_PAGE_SIZE = 2000; private static final long COMPLETED_PROGRESS_RETENTION_MILLIS = 60 * 60 * 1000L; private static final DateTimeFormatter EXPORT_TIME_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"); @@ -130,21 +135,35 @@ public class DedupeTotalDataService { } String safeUsername = username == null ? "" : username.trim(); AccessScope scope = resolveAccessScope(operatorId); - LambdaQueryWrapper query = new LambdaQueryWrapper() - .like(!safeUsername.isEmpty(), DedupeTotalDataEntity::getUploaderUsername, safeUsername) - .ge(startDate != null, DedupeTotalDataEntity::getCreatedAt, - startDate == null ? null : startDate.atStartOfDay()) - .lt(endDate != null, DedupeTotalDataEntity::getCreatedAt, - endDate == null ? null : endDate.plusDays(1).atStartOfDay()) - .orderByDesc(DedupeTotalDataEntity::getId); - applyGroupScope(query, scope, groupId); - List rows = dedupeTotalDataMapper.selectList(query); - return buildExportWorkbook(rows, loadGroupNames(rows)); + ByteArrayOutputStream outputStream = new ByteArrayOutputStream(); + buildExportWorkbook(outputStream, safeUsername, startDate, endDate, groupId, scope); + return outputStream.toByteArray(); } - private byte[] buildExportWorkbook(List rows, Map groupNames) { - try (SXSSFWorkbook workbook = new SXSSFWorkbook(100); - ByteArrayOutputStream outputStream = new ByteArrayOutputStream()) { + public void writeExport(OutputStream outputStream, + String username, + LocalDate startDate, + LocalDate endDate, + Long groupId, + Long operatorId) { + if (outputStream == null) { + throw new BusinessException("导出输出流不能为空"); + } + if (startDate != null && endDate != null && startDate.isAfter(endDate)) { + throw new BusinessException("invalid export date range"); + } + String safeUsername = username == null ? "" : username.trim(); + AccessScope scope = resolveAccessScope(operatorId); + buildExportWorkbook(outputStream, safeUsername, startDate, endDate, groupId, scope); + } + + private void buildExportWorkbook(OutputStream outputStream, + String username, + LocalDate startDate, + LocalDate endDate, + Long groupId, + AccessScope scope) { + try (SXSSFWorkbook workbook = new SXSSFWorkbook(100)) { Sheet sheet = workbook.createSheet("DedupeTotalData"); Row header = sheet.createRow(0); header.createCell(0).setCellValue("ID"); @@ -152,16 +171,38 @@ public class DedupeTotalDataService { header.createCell(2).setCellValue("用户名"); header.createCell(3).setCellValue("分组"); header.createCell(4).setCellValue("创建时间"); - for (int index = 0; index < rows.size(); index++) { - DedupeTotalDataEntity entity = rows.get(index); - Row row = sheet.createRow(index + 1); - row.createCell(0).setCellValue(entity.getId() == null ? "" : String.valueOf(entity.getId())); - row.createCell(1).setCellValue(entity.getDataValue() == null ? "" : entity.getDataValue()); - row.createCell(2).setCellValue(entity.getUploaderUsername() == null ? "" : entity.getUploaderUsername()); - row.createCell(3).setCellValue(entity.getGroupId() == null - ? "" - : groupNames.getOrDefault(entity.getGroupId(), "")); - row.createCell(4).setCellValue(formatExportTime(entity.getCreatedAt())); + Long lastId = null; + int rowIndex = 1; + while (true) { + LambdaQueryWrapper pageQuery = buildExportQuery( + username, startDate, endDate, groupId, scope); + if (lastId != null) { + pageQuery.lt(DedupeTotalDataEntity::getId, lastId); + } + List rows = dedupeTotalDataMapper.selectList( + pageQuery.last("LIMIT " + EXPORT_PAGE_SIZE)); + if (rows == null || rows.isEmpty()) { + break; + } + Map groupNames = loadGroupNames(rows); + for (DedupeTotalDataEntity entity : rows) { + Row row = sheet.createRow(rowIndex++); + row.createCell(0).setCellValue(entity.getId() == null ? "" : String.valueOf(entity.getId())); + row.createCell(1).setCellValue(entity.getDataValue() == null ? "" : entity.getDataValue()); + row.createCell(2).setCellValue(entity.getUploaderUsername() == null ? "" : entity.getUploaderUsername()); + row.createCell(3).setCellValue(entity.getGroupId() == null + ? "" + : groupNames.getOrDefault(entity.getGroupId(), "")); + row.createCell(4).setCellValue(formatExportTime(entity.getCreatedAt())); + } + DedupeTotalDataEntity lastRow = rows.getLast(); + lastId = lastRow == null ? null : lastRow.getId(); + if (lastId == null) { + break; + } + if (rows.size() < EXPORT_PAGE_SIZE) { + break; + } } sheet.setColumnWidth(0, 3600); sheet.setColumnWidth(1, 5200); @@ -170,12 +211,28 @@ public class DedupeTotalDataService { sheet.setColumnWidth(4, 5600); workbook.write(outputStream); workbook.dispose(); - return outputStream.toByteArray(); } catch (Exception ex) { throw new BusinessException("导出数据去重总数据失败"); } } + private LambdaQueryWrapper buildExportQuery(String username, + LocalDate startDate, + LocalDate endDate, + Long groupId, + AccessScope scope) { + String safeUsername = username == null ? "" : username.trim(); + LambdaQueryWrapper query = new LambdaQueryWrapper() + .like(!safeUsername.isEmpty(), DedupeTotalDataEntity::getUploaderUsername, safeUsername) + .ge(startDate != null, DedupeTotalDataEntity::getCreatedAt, + startDate == null ? null : startDate.atStartOfDay()) + .lt(endDate != null, DedupeTotalDataEntity::getCreatedAt, + endDate == null ? null : endDate.plusDays(1).atStartOfDay()) + .orderByDesc(DedupeTotalDataEntity::getId); + applyGroupScope(query, scope, groupId); + return query; + } + private String formatExportTime(LocalDateTime value) { return value == null ? "" : value.format(EXPORT_TIME_FORMATTER); } @@ -481,6 +538,7 @@ public class DedupeTotalDataService { } Set seenInFile = new HashSet<>(); + List pendingValues = new ArrayList<>(IMPORT_BATCH_SIZE); int totalRows = Math.max(sheet.getLastRowNum(), 0); int asinCount = 0; int insertedCount = 0; @@ -498,77 +556,39 @@ public class DedupeTotalDataService { Row row = sheet.getRow(rowNum); if (row == null) { skippedCount++; - if (progress != null) { - progress.setProcessedRows(rowNum); - progress.setAsinCount(asinCount); - progress.setInsertedCount(insertedCount); - progress.setSkippedCount(skippedCount); - } + updateImportProgress(progress, rowNum, asinCount, insertedCount, skippedCount); continue; } String asin = normalizeComparableValueOrBlank(formatter.formatCellValue(row.getCell(asinIndex))); if (asin.isBlank()) { skippedCount++; - if (progress != null) { - progress.setProcessedRows(rowNum); - progress.setAsinCount(asinCount); - progress.setInsertedCount(insertedCount); - progress.setSkippedCount(skippedCount); - } + updateImportProgress(progress, rowNum, asinCount, insertedCount, skippedCount); continue; } asinCount++; - String dataValue = asin; - if (!seenInFile.add(dataValue)) { + if (!seenInFile.add(asin)) { skippedCount++; - if (progress != null) { - progress.setProcessedRows(rowNum); - progress.setAsinCount(asinCount); - progress.setInsertedCount(insertedCount); - progress.setSkippedCount(skippedCount); - } + updateImportProgress(progress, rowNum, asinCount, insertedCount, skippedCount); continue; } - if (existsDataValue(dataValue)) { - skippedCount++; - if (progress != null) { - progress.setProcessedRows(rowNum); - progress.setAsinCount(asinCount); - progress.setInsertedCount(insertedCount); - progress.setSkippedCount(skippedCount); - } - continue; - } - final String pendingDataValue = dataValue; - Boolean inserted; - try { - inserted = newRequiresNewTemplate().execute(status -> { - if (existsDataValue(pendingDataValue)) { - return false; - } - DedupeTotalDataEntity entity = new DedupeTotalDataEntity(); - entity.setDataValue(pendingDataValue); - entity.setGroupId(groupId); - entity.setUploaderUserId(uploaderUserId); - entity.setUploaderUsername(uploaderUsername); - dedupeTotalDataMapper.insert(entity); - return true; - }); - } catch (DuplicateKeyException ignored) { - inserted = false; - } - if (Boolean.TRUE.equals(inserted)) { - insertedCount++; - } else { - skippedCount++; - } - if (progress != null) { - progress.setProcessedRows(rowNum); - progress.setAsinCount(asinCount); - progress.setInsertedCount(insertedCount); - progress.setSkippedCount(skippedCount); + pendingValues.add(asin); + if (pendingValues.size() >= IMPORT_BATCH_SIZE) { + ImportBatchResult batch = insertImportBatch( + pendingValues, groupId, uploaderUserId, uploaderUsername); + insertedCount += batch.insertedCount(); + skippedCount += batch.skippedCount(); + pendingValues.clear(); } + updateImportProgress(progress, rowNum, asinCount, insertedCount, skippedCount); } + if (!pendingValues.isEmpty()) { + ImportBatchResult batch = insertImportBatch( + pendingValues, groupId, uploaderUserId, uploaderUsername); + insertedCount += batch.insertedCount(); + skippedCount += batch.skippedCount(); + pendingValues.clear(); + } + updateImportProgress(progress, totalRows, asinCount, insertedCount, skippedCount); DedupeTotalDataImportVo vo = new DedupeTotalDataImportVo(); vo.setTotalRows(totalRows); @@ -583,6 +603,81 @@ public class DedupeTotalDataService { } } + private ImportBatchResult insertImportBatch(List values, + Long groupId, + Long uploaderUserId, + String uploaderUsername) { + if (values == null || values.isEmpty()) { + return new ImportBatchResult(0, 0); + } + Set existingValues = findExistingComparableValues(values); + List rows = new ArrayList<>(values.size()); + for (String value : values) { + if (value == null || value.isBlank() || existingValues.contains(value)) { + continue; + } + DedupeTotalDataEntity entity = new DedupeTotalDataEntity(); + entity.setDataValue(value); + entity.setGroupId(groupId); + entity.setUploaderUserId(uploaderUserId); + entity.setUploaderUsername(uploaderUsername); + rows.add(entity); + } + if (rows.isEmpty()) { + return new ImportBatchResult(0, values.size()); + } + + int inserted; + try { + inserted = executeInNewTransaction(() -> dedupeTotalDataMapper.insertBatchIgnore(rows)); + } catch (DuplicateKeyException ex) { + // INSERT IGNORE is the normal path. Keep a correctness-preserving + // fallback for deployments whose SQL mode/driver still surfaces a + // duplicate-key error for a multi-row statement. + inserted = 0; + for (DedupeTotalDataEntity row : rows) { + try { + Boolean oneInserted = executeInNewTransaction(() -> { + if (existsDataValue(row.getDataValue())) { + return false; + } + return dedupeTotalDataMapper.insert(row) > 0; + }); + if (Boolean.TRUE.equals(oneInserted)) { + inserted++; + } + } catch (DuplicateKeyException ignored) { + // Another importer won the unique-key race. + } + } + } + return new ImportBatchResult(inserted, values.size() - inserted); + } + + private void updateImportProgress(DedupeTotalDataImportProgressVo progress, + int processedRows, + int asinCount, + int insertedCount, + int skippedCount) { + if (progress == null) { + return; + } + progress.setProcessedRows(processedRows); + progress.setAsinCount(asinCount); + progress.setInsertedCount(insertedCount); + progress.setSkippedCount(skippedCount); + } + + private T executeInNewTransaction(Supplier action) { + if (transactionManager == null) { + return action.get(); + } + return newRequiresNewTemplate().execute(status -> action.get()); + } + + private record ImportBatchResult(int insertedCount, int skippedCount) { + } + private DedupeTotalDataImportVo deleteFromExcelInternal(InputStream inputStream, String filename, DedupeTotalDataImportProgressVo progress, AccessScope scope, Long groupId) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoCozeService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoCozeService.java index 5fb56347..efc62bd7 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoCozeService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/imagevideo/service/ImageVideoCozeService.java @@ -14,7 +14,9 @@ import lombok.extern.slf4j.Slf4j; import org.springframework.stereotype.Service; import org.springframework.web.multipart.MultipartFile; +import java.io.ByteArrayOutputStream; import java.io.File; +import java.io.InputStream; import java.net.URI; import java.net.http.HttpClient; import java.net.http.HttpRequest; @@ -33,6 +35,8 @@ import java.util.Set; @RequiredArgsConstructor public class ImageVideoCozeService { + private static final int RESPONSE_LOG_PREVIEW_CHARS = 4096; + private static final List ORIGINAL_FIELDS = List.of( "recognizedContent", "recognized_content", "originalContent", "original_content", "originContent", "origin_content", "rawContent", "raw_content", "description", "content" @@ -447,7 +451,8 @@ public class ImageVideoCozeService { } catch (BusinessException ex) { throw ex; } catch (Exception ex) { - log.warn("[image-video] parse douyin copy response failed body={}", responseText, ex); + log.warn("[image-video] parse douyin copy response failed bodyPreview={}", + responsePreview(responseText), ex); throw new BusinessException("Coze 返回解析失败: " + ex.getMessage()); } } @@ -588,12 +593,12 @@ public class ImageVideoCozeService { .header("Accept-Charset", StandardCharsets.UTF_8.name()) .POST(HttpRequest.BodyPublishers.ofString(requestBody, StandardCharsets.UTF_8)) .build(); - HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - String responseBody = response.body() == null ? "" : response.body(); + HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofInputStream()); + String responseBody = readResponseBody(response.body()); log.info("[image-video] {} coze response status={} bodyLength={}", label, response.statusCode(), responseBody.length()); - log.info("[image-video] {} coze response detail status={} body={}", - label, response.statusCode(), responseBody); + log.debug("[image-video] {} coze response detail status={} bodyPreview={}", + label, response.statusCode(), responsePreview(responseBody)); if (response.statusCode() < 200 || response.statusCode() >= 300) { throw new BusinessException("Coze 调用失败: HTTP " + response.statusCode()); } @@ -618,12 +623,12 @@ public class ImageVideoCozeService { .header("Accept-Charset", StandardCharsets.UTF_8.name()) .GET() .build(); - HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8)); - String responseBody = response.body() == null ? "" : response.body(); + HttpResponse response = httpClient().send(request, HttpResponse.BodyHandlers.ofInputStream()); + String responseBody = readResponseBody(response.body()); log.info("[image-video] {} coze response status={} bodyLength={}", label, response.statusCode(), responseBody.length()); - log.info("[image-video] {} coze response detail status={} body={}", - label, response.statusCode(), responseBody); + log.debug("[image-video] {} coze response detail status={} bodyPreview={}", + label, response.statusCode(), responsePreview(responseBody)); if (response.statusCode() < 200 || response.statusCode() >= 300) { throw new BusinessException("Coze 调用失败: HTTP " + response.statusCode()); } @@ -638,6 +643,37 @@ public class ImageVideoCozeService { } } + private String readResponseBody(InputStream inputStream) throws java.io.IOException { + if (inputStream == null) { + return ""; + } + int maxBytes = Math.max(1024, properties.getCozeResponseMaxBytes()); + try (InputStream input = inputStream; + ByteArrayOutputStream output = new ByteArrayOutputStream(Math.min(maxBytes, 8192))) { + byte[] buffer = new byte[8192]; + int total = 0; + int read; + while ((read = input.read(buffer)) != -1) { + if (read == 0) { + continue; + } + total += read; + if (total > maxBytes) { + throw new java.io.IOException("Coze response exceeds configured size limit"); + } + output.write(buffer, 0, read); + } + return output.toString(StandardCharsets.UTF_8); + } + } + + private String responsePreview(String responseBody) { + if (responseBody == null || responseBody.length() <= RESPONSE_LOG_PREVIEW_CHARS) { + return responseBody == null ? "" : responseBody; + } + return responseBody.substring(0, RESPONSE_LOG_PREVIEW_CHARS) + "...(truncated)"; + } + private HttpClient httpClient() { HttpClient client = sharedHttpClient; if (client == null) { @@ -706,11 +742,13 @@ public class ImageVideoCozeService { } private String toJsonForLog(Object value) { + String json; try { - return objectMapper.writeValueAsString(maskSecretsForLog(value)); + json = objectMapper.writeValueAsString(maskSecretsForLog(value)); } catch (Exception ex) { - return String.valueOf(value); + json = String.valueOf(value); } + return responsePreview(json); } private Object maskSecretsForLog(Object value) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java index c509ab21..05e16d82 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskService.java @@ -42,6 +42,8 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; import com.nanri.aiimage.modules.ziniao.model.vo.ZiniaoShopMatchResultVo; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; import org.springframework.stereotype.Service; import org.springframework.transaction.annotation.Transactional; import org.springframework.beans.factory.annotation.Value; @@ -49,6 +51,7 @@ import org.springframework.dao.DuplicateKeyException; import org.springframework.scheduling.annotation.Scheduled; import org.springframework.transaction.support.TransactionSynchronization; import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.support.TransactionTemplate; import java.io.File; import java.nio.file.Files; @@ -65,6 +68,7 @@ import java.util.Objects; import java.util.Set; import java.util.TreeMap; import java.util.UUID; +import java.util.function.Supplier; @Service @RequiredArgsConstructor @@ -98,6 +102,7 @@ public class ShopDataCrawlTaskService { private final TransientPayloadStorageService transientPayloadStorageService; private final InstanceMetadata instanceMetadata; private final ShopDataCrawlDailyFileService dailyFileService; + private final PlatformTransactionManager transactionManager; @Value("${aiimage.shop-data-crawl.stale-timeout-minutes:30}") private long staleTimeoutMinutes; @@ -1573,7 +1578,6 @@ public class ShopDataCrawlTaskService { } } - @Transactional public void processResultFileJob(TaskFileJobEntity job) { if (job == null || job.getTaskId() == null) { throw new BusinessException("结果文件任务参数不完整"); @@ -1596,39 +1600,35 @@ public class ShopDataCrawlTaskService { } Map snapshotsByResultId = indexSnapshotByResultId(successItems); LocalDate businessDate = dailyFileService.currentBusinessDate(); - List uploadedObjectKeys = new ArrayList<>(); - List obsoleteObjectKeys = new ArrayList<>(); - try { - for (FileResultEntity row : rows) { - if (!Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) { - continue; - } - ShopDataCrawlResultItemVo snapshot = snapshotsByResultId.get(row.getId()); - if (snapshot == null) { - snapshot = successItems.size() == 1 ? successItems.get(0) : null; - } - if (snapshot == null) { - throw new BusinessException("店铺结果快照不存在，无法生成累计文件"); - } - DailyAggregationResult result = aggregateDailyResult( - task, row, snapshot, businessDate, uploadedObjectKeys); - obsoleteObjectKeys.addAll(result.obsoleteObjectKeys()); + for (FileResultEntity row : rows) { + if (!Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) { + continue; } - updateTaskStatusFromRows(task, rows); - persistSnapshotJson(task, buildSnapshotFromDb(task, rows)); - fileTaskMapper.updateById(task); - } catch (RuntimeException ex) { - registerRollbackObjectCleanup(uploadedObjectKeys); - throw ex; + ShopDataCrawlResultItemVo snapshot = snapshotsByResultId.get(row.getId()); + if (snapshot == null) { + snapshot = successItems.size() == 1 ? successItems.get(0) : null; + } + if (snapshot == null) { + throw new BusinessException("店铺结果快照不存在，无法生成累计文件"); + } + aggregateDailyResult(task, row, snapshot, businessDate); } - registerDailyObjectLifecycle(uploadedObjectKeys, obsoleteObjectKeys); + + // Keep task status persistence short-lived. Workbook/OSS work and the + // snapshot payload upload are completed outside this transaction. + updateTaskStatusFromRows(task, rows); + List finalSnapshots = buildSnapshotFromDb(task, rows); + persistSnapshotJson(task, finalSnapshots); + executeShortTransaction(() -> { + fileTaskMapper.updateById(task); + return null; + }); } private DailyAggregationResult aggregateDailyResult(FileTaskEntity task, FileResultEntity row, ShopDataCrawlResultItemVo snapshot, - LocalDate businessDate, - List uploadedObjectKeys) { + LocalDate businessDate) { Long userId = row.getUserId() != null ? row.getUserId() : task.getUserId(); String shopKey = dailyFileService.shopKey(row); String shopKeyHash = dailyFileService.shopKeyHash(shopKey); @@ -1639,136 +1639,220 @@ public class ShopDataCrawlTaskService { if (lock == null) { throw new BusinessException("店铺当天累计文件正在处理中，请稍后重试"); } + DailyAggregationResult persistedResult = null; try { - ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate(userId, shopKeyHash, businessDate); - ShopDataCrawlDailyFileEntity existingMembership = findExistingDailyMembership(row.getId()); - if (existingMembership != null) { - boolean currentFileOwnsMembership = dailyFile != null - && Objects.equals(dailyFile.getId(), existingMembership.getId()); - if (Objects.equals(existingMembership.getLatestResultId(), row.getId()) - && (dailyFile == null || currentFileOwnsMembership)) { - attachCanonicalResult(row, existingMembership); - } else if (!blank(row.getResultFileUrl())) { - row.setResultFileUrl(null); - row.setResultFileSize(null); - row.setResultContentType(null); - fileResultMapper.updateById(row); - } - return new DailyAggregationResult(List.of()); - } - if (dailyFile != null && dailyFileService.containsResult(dailyFile.getId(), row.getId())) { - if (Objects.equals(dailyFile.getLatestResultId(), row.getId())) { - attachCanonicalResult(row, dailyFile); - } else if (!blank(row.getResultFileUrl())) { - row.setResultFileUrl(null); - row.setResultFileSize(null); - row.setResultContentType(null); - fileResultMapper.updateById(row); - } - return new DailyAggregationResult(List.of()); - } - - List olderFiles = dailyFileService.findOlder(userId, shopKeyHash, businessDate); - Set obsoleteObjectKeys = new HashSet<>(); - collectObjectKey(obsoleteObjectKeys, dailyFile == null ? null : dailyFile.getResultFileUrl()); - for (ShopDataCrawlDailyFileEntity older : olderFiles) { - collectObjectKey(obsoleteObjectKeys, older.getResultFileUrl()); + DailyAggregationPreparation preparation = executeShortTransaction( + () -> prepareDailyAggregation(userId, shopKeyHash, businessDate, row)); + if (preparation.alreadyArchived()) { + return new DailyAggregationResult(List.of(), false); } int addedRowCount = excelAssemblyService.countRows(List.of(snapshot)); - String filename = dailyFile != null && !blank(dailyFile.getResultFilename()) - ? dailyFile.getResultFilename() - : buildTaskWorkbookFilename(task); - File workRoot = FileUtil.mkdir(FileUtil.file( - System.getProperty("java.io.tmpdir"), - "shop-data-crawl-result", - String.valueOf(task.getId()), - "daily-" + UUID.randomUUID())); - File baseXlsx = FileUtil.file(workRoot, "base.xlsx"); - File outputXlsx = FileUtil.file(workRoot, filename); - String objectKey = null; + DailyWorkbookArtifact artifact = assembleDailyWorkbook( + task, snapshot, preparation.dailyFile(), addedRowCount); try { - if (dailyFile != null && !blank(dailyFile.getResultFileUrl())) { - try { - Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(dailyFile.getResultFileUrl())); - } catch (Exception ex) { - throw new BusinessException("读取当天累计文件失败: " + safeMessage(ex)); - } - if (addedRowCount > 0) { - excelAssemblyService.appendWorkbook(baseXlsx, outputXlsx, List.of(snapshot)); - } else { - try { - Files.copy(baseXlsx.toPath(), outputXlsx.toPath()); - } catch (Exception ex) { - throw new BusinessException("复制当天累计文件失败: " + safeMessage(ex)); - } - } - } else { - excelAssemblyService.writeWorkbook(outputXlsx, List.of(snapshot)); + persistedResult = executeShortTransaction(() -> persistDailyAggregation( + task, row, userId, shopKey, shopKeyHash, businessDate, + preparation, artifact, addedRowCount)); + if (persistedResult.discardUploadedObject() && artifact.uploaded()) { + deleteObjectQuietly(artifact.objectKey()); } - if (dailyFile != null && addedRowCount == 0) { - objectKey = dailyFile.getResultFileUrl(); - } else { - objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE); - uploadedObjectKeys.add(objectKey); + return persistedResult; + } catch (RuntimeException ex) { + if (artifact.uploaded()) { + deleteObjectQuietly(artifact.objectKey()); } - - List shopRows = findShopResultRows(userId, row); - clearShopResultPointers(shopRows, row.getId()); - row.setResultFilename(filename); - row.setResultFileUrl(objectKey); - row.setResultFileSize(outputXlsx.length()); - row.setResultContentType(CONTENT_TYPE_XLSX); - row.setRowCount(dailyFile == null - ? addedRowCount - : Math.max(0, Objects.requireNonNullElse(dailyFile.getRowCount(), 0)) + addedRowCount); - fileResultMapper.updateById(row); - - LocalDateTime now = dailyFileService.currentBusinessDateTime(); - if (dailyFile == null) { - dailyFile = new ShopDataCrawlDailyFileEntity(); - dailyFile.setUserId(userId); - dailyFile.setShopKeyHash(shopKeyHash); - dailyFile.setShopKey(shopKey); - dailyFile.setBusinessDate(businessDate); - dailyFile.setVersion(1L); - dailyFile.setCreatedAt(now); - } else { - dailyFile.setVersion(Math.max(0L, Objects.requireNonNullElse(dailyFile.getVersion(), 0L)) + 1L); - } - dailyFile.setLatestTaskId(row.getTaskId()); - dailyFile.setLatestResultId(row.getId()); - dailyFile.setResultFilename(filename); - dailyFile.setResultFileUrl(objectKey); - dailyFile.setResultFileSize(row.getResultFileSize()); - dailyFile.setResultContentType(CONTENT_TYPE_XLSX); - dailyFile.setRowCount(row.getRowCount()); - dailyFile.setLastSuccessAt(now); - dailyFile.setUpdatedAt(now); - if (dailyFile.getId() == null) { - dailyFileService.insert(dailyFile); - } else { - dailyFileService.update(dailyFile); - } - if (!dailyFileService.addMember(dailyFile.getId(), row.getTaskId(), row.getId())) { - throw new BusinessException("结果已归档，请重试文件任务"); - } - - for (ShopDataCrawlDailyFileEntity older : olderFiles) { - dailyFileService.deleteDailyFile(older.getId()); - } - obsoleteObjectKeys.remove(objectKey); - return new DailyAggregationResult(new ArrayList<>(obsoleteObjectKeys)); - } finally { - FileUtil.del(baseXlsx); - FileUtil.del(outputXlsx); - FileUtil.del(workRoot); + throw ex; } } finally { - releaseDailyLockAfterTransaction(lock); + try { + lock.close(); + } finally { + if (persistedResult != null) { + persistedResult.obsoleteObjectKeys().forEach(this::deleteObjectQuietly); + } + } } } + private DailyAggregationPreparation prepareDailyAggregation(Long userId, + String shopKeyHash, + LocalDate businessDate, + FileResultEntity row) { + ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( + userId, shopKeyHash, businessDate); + if (handleExistingDailyMembership(row, dailyFile)) { + return new DailyAggregationPreparation(dailyFile, true); + } + return new DailyAggregationPreparation(dailyFile, false); + } + + private DailyWorkbookArtifact assembleDailyWorkbook(FileTaskEntity task, + ShopDataCrawlResultItemVo snapshot, + ShopDataCrawlDailyFileEntity dailyFile, + int addedRowCount) { + String filename = dailyFile != null && !blank(dailyFile.getResultFilename()) + ? dailyFile.getResultFilename() + : buildTaskWorkbookFilename(task); + String existingObjectKey = dailyFile == null ? null : dailyFile.getResultFileUrl(); + + // A result with no new rows only needs a new database membership. Reusing + // the canonical daily object avoids both a local copy and an OSS round trip. + if (dailyFile != null && addedRowCount == 0 && !blank(existingObjectKey)) { + return new DailyWorkbookArtifact( + existingObjectKey, + Math.max(0L, Objects.requireNonNullElse(dailyFile.getResultFileSize(), 0L)), + false, + filename); + } + + File workRoot = FileUtil.mkdir(FileUtil.file( + System.getProperty("java.io.tmpdir"), + "shop-data-crawl-result", + String.valueOf(task.getId()), + "daily-" + UUID.randomUUID())); + File baseXlsx = FileUtil.file(workRoot, "base.xlsx"); + File outputXlsx = FileUtil.file(workRoot, filename); + try { + if (dailyFile != null && !blank(existingObjectKey)) { + try { + Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(existingObjectKey)); + } catch (Exception ex) { + throw new BusinessException("读取当天累计文件失败: " + safeMessage(ex)); + } + excelAssemblyService.appendWorkbook(baseXlsx, outputXlsx, List.of(snapshot)); + } else { + excelAssemblyService.writeWorkbook(outputXlsx, List.of(snapshot)); + } + String objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE); + if (blank(objectKey)) { + throw new BusinessException("当天累计文件上传后未返回文件地址"); + } + return new DailyWorkbookArtifact(objectKey, outputXlsx.length(), true, filename); + } finally { + FileUtil.del(baseXlsx); + FileUtil.del(outputXlsx); + FileUtil.del(workRoot); + } + } + + private DailyAggregationResult persistDailyAggregation(FileTaskEntity task, + FileResultEntity row, + Long userId, + String shopKey, + String shopKeyHash, + LocalDate businessDate, + DailyAggregationPreparation preparation, + DailyWorkbookArtifact artifact, + int addedRowCount) { + ShopDataCrawlDailyFileEntity dailyFile = dailyFileService.findForUpdate( + userId, shopKeyHash, businessDate); + if (handleExistingDailyMembership(row, dailyFile)) { + return new DailyAggregationResult(List.of(), true); + } + if (!sameDailyFileState(preparation.dailyFile(), dailyFile)) { + throw new BusinessException("当天累计文件状态已变化，请重试文件任务"); + } + + List olderFiles = dailyFileService.findOlder( + userId, shopKeyHash, businessDate); + Set obsoleteObjectKeys = new HashSet<>(); + collectObjectKey(obsoleteObjectKeys, dailyFile == null ? null : dailyFile.getResultFileUrl()); + for (ShopDataCrawlDailyFileEntity older : olderFiles) { + collectObjectKey(obsoleteObjectKeys, older.getResultFileUrl()); + } + + String objectKey = artifact.objectKey(); + String filename = artifact.filename(); + List shopRows = findShopResultRows(userId, row); + clearShopResultPointers(shopRows, row.getId()); + row.setResultFilename(filename); + row.setResultFileUrl(objectKey); + row.setResultFileSize(artifact.fileSize()); + row.setResultContentType(CONTENT_TYPE_XLSX); + row.setRowCount(dailyFile == null + ? addedRowCount + : Math.max(0, Objects.requireNonNullElse(dailyFile.getRowCount(), 0)) + addedRowCount); + fileResultMapper.updateById(row); + + LocalDateTime now = dailyFileService.currentBusinessDateTime(); + if (dailyFile == null) { + dailyFile = new ShopDataCrawlDailyFileEntity(); + dailyFile.setUserId(userId); + dailyFile.setShopKeyHash(shopKeyHash); + dailyFile.setShopKey(shopKey); + dailyFile.setBusinessDate(businessDate); + dailyFile.setVersion(1L); + dailyFile.setCreatedAt(now); + } else { + dailyFile.setVersion(Math.max(0L, Objects.requireNonNullElse(dailyFile.getVersion(), 0L)) + 1L); + } + dailyFile.setLatestTaskId(row.getTaskId()); + dailyFile.setLatestResultId(row.getId()); + dailyFile.setResultFilename(filename); + dailyFile.setResultFileUrl(objectKey); + dailyFile.setResultFileSize(row.getResultFileSize()); + dailyFile.setResultContentType(CONTENT_TYPE_XLSX); + dailyFile.setRowCount(row.getRowCount()); + dailyFile.setLastSuccessAt(now); + dailyFile.setUpdatedAt(now); + if (dailyFile.getId() == null) { + dailyFileService.insert(dailyFile); + } else { + dailyFileService.update(dailyFile); + } + if (!dailyFileService.addMember(dailyFile.getId(), row.getTaskId(), row.getId())) { + throw new BusinessException("结果已归档，请重试文件任务"); + } + + for (ShopDataCrawlDailyFileEntity older : olderFiles) { + dailyFileService.deleteDailyFile(older.getId()); + } + obsoleteObjectKeys.remove(objectKey); + return new DailyAggregationResult(new ArrayList<>(obsoleteObjectKeys), false); + } + + private boolean handleExistingDailyMembership(FileResultEntity row, + ShopDataCrawlDailyFileEntity dailyFile) { + ShopDataCrawlDailyFileEntity existingMembership = findExistingDailyMembership(row.getId()); + if (existingMembership != null) { + boolean currentFileOwnsMembership = dailyFile != null + && Objects.equals(dailyFile.getId(), existingMembership.getId()); + if (Objects.equals(existingMembership.getLatestResultId(), row.getId()) + && (dailyFile == null || currentFileOwnsMembership)) { + attachCanonicalResult(row, existingMembership); + } else if (!blank(row.getResultFileUrl())) { + row.setResultFileUrl(null); + row.setResultFileSize(null); + row.setResultContentType(null); + fileResultMapper.updateById(row); + } + return true; + } + if (dailyFile != null && dailyFileService.containsResult(dailyFile.getId(), row.getId())) { + if (Objects.equals(dailyFile.getLatestResultId(), row.getId())) { + attachCanonicalResult(row, dailyFile); + } else if (!blank(row.getResultFileUrl())) { + row.setResultFileUrl(null); + row.setResultFileSize(null); + row.setResultContentType(null); + fileResultMapper.updateById(row); + } + return true; + } + return false; + } + + private boolean sameDailyFileState(ShopDataCrawlDailyFileEntity expected, + ShopDataCrawlDailyFileEntity actual) { + if (expected == null || actual == null) { + return expected == actual; + } + return Objects.equals(expected.getId(), actual.getId()) + && Objects.equals(expected.getVersion(), actual.getVersion()) + && Objects.equals(expected.getResultFileUrl(), actual.getResultFileUrl()); + } + private ShopDataCrawlDailyFileEntity findExistingDailyMembership(Long resultId) { List members = dailyFileService.findMembersByResultId(resultId); for (ShopDataCrawlDailyMemberEntity member : members == null ? List.of() : members) { @@ -1795,6 +1879,17 @@ public class ShopDataCrawlTaskService { fileResultMapper.updateById(row); } + private T executeShortTransaction(Supplier action) { + // Direct construction is used by the focused unit tests; Spring always + // supplies the transaction manager in the running application. + if (transactionManager == null) { + return action.get(); + } + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return template.execute(status -> action.get()); + } + private List findShopResultRows(Long userId, FileResultEntity sourceRow) { if (userId == null || sourceRow == null) { return List.of(); @@ -1856,29 +1951,6 @@ public class ShopDataCrawlTaskService { target.add(value); } - private void registerDailyObjectLifecycle(List uploadedObjectKeys, List obsoleteObjectKeys) { - Set uploaded = uploadedObjectKeys == null ? Set.of() : new HashSet<>(uploadedObjectKeys); - Set obsolete = obsoleteObjectKeys == null ? new HashSet<>() : new HashSet<>(obsoleteObjectKeys); - obsolete.removeAll(uploaded); - if (!TransactionSynchronizationManager.isSynchronizationActive()) { - obsolete.forEach(this::deleteObjectQuietly); - return; - } - TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { - @Override - public void afterCommit() { - obsolete.forEach(ShopDataCrawlTaskService.this::deleteObjectQuietly); - } - - @Override - public void afterCompletion(int status) { - if (status != STATUS_COMMITTED) { - uploaded.forEach(ShopDataCrawlTaskService.this::deleteObjectQuietly); - } - } - }); - } - private void registerRollbackObjectCleanup(List uploadedObjectKeys) { if (uploadedObjectKeys == null || uploadedObjectKeys.isEmpty()) { return; @@ -1960,22 +2032,6 @@ public class ShopDataCrawlTaskService { } } - private void releaseDailyLockAfterTransaction(TaskDistributedLockService.LockHandle lock) { - if (lock == null) { - return; - } - if (!TransactionSynchronizationManager.isSynchronizationActive()) { - lock.close(); - return; - } - TransactionSynchronizationManager.registerSynchronization(new TransactionSynchronization() { - @Override - public void afterCompletion(int status) { - lock.close(); - } - }); - } - private record DailyLockRequest(Long userId, String shopKey) { } @@ -2021,7 +2077,18 @@ public class ShopDataCrawlTaskService { } } - private record DailyAggregationResult(List obsoleteObjectKeys) { + private record DailyAggregationPreparation(ShopDataCrawlDailyFileEntity dailyFile, + boolean alreadyArchived) { + } + + private record DailyWorkbookArtifact(String objectKey, + long fileSize, + boolean uploaded, + String filename) { + } + + private record DailyAggregationResult(List obsoleteObjectKeys, + boolean discardUploadedObject) { } public void cleanupResultFileJob(TaskFileJobEntity job) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java index afe0b82d..f91e338c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinImagePrefetchService.java @@ -47,8 +47,8 @@ import java.util.concurrent.atomic.AtomicInteger; @Slf4j public class SimilarAsinImagePrefetchService { - /** P2-11：预热线程池容量。预热不要求高吞吐，4 个线程足够；避免和 embed 阶段抢 IO 资源。 */ - private static final int PREFETCH_POOL_SIZE = 4; + /** P2-11：预热线程池容量。与图片处理 CPU 槽位对齐，避免预热路径绕过主池放大并发。 */ + private static final int PREFETCH_POOL_SIZE = 2; /** 排队等待上一个 task future 时的最大等待时间（秒），避免被 hung future 永久卡住。 */ private static final long INFLIGHT_WAIT_SECONDS = 60L; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 797afb59..1c3fd050 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -104,7 +104,6 @@ import java.util.concurrent.ExecutionException; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; import java.util.concurrent.Future; -import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -279,11 +278,32 @@ public class SimilarAsinTaskService { * 4 线程足以覆盖常见 1-4 源文件；sourceRows.size() == 1 时仍走串行降级路径， * 避免对单文件场景引入额外线程开销。 */ - private final ExecutorService assembleExecutor = Executors.newFixedThreadPool(4, namedThreadFactory("similar-asin-assemble")); + /** + * Keep at most two source workbooks in POI/image assembly at once. Each + * workbook has its own SXSSF structures and image spool, so four workers + * multiply the peak even though the task itself is a single job. + */ + private final ExecutorService assembleExecutor = Executors.newFixedThreadPool(2, namedThreadFactory("similar-asin-assemble")); + + /** Reused poll workers avoid creating/destroying a new executor every 30s. */ + private final ExecutorService cozePollPrefetchExecutor = Executors.newFixedThreadPool( + COZE_POLL_PREFETCH_CONCURRENCY, namedThreadFactory("similar-asin-coze-prefetch")); @PreDestroy void shutdownAssembleExecutor() { assembleExecutor.shutdownNow(); + cozePollPrefetchExecutor.shutdownNow(); + } + + /** + * Credential names normally come from configuration, but stale/rotated + * credentials can otherwise leave throttle entries in the process forever. + */ + @Scheduled(fixedDelayString = "${aiimage.similar-asin.coze-submit-throttle-cleanup-delay-ms:600000}") + void cleanupCozeThrottleState() { + long cutoff = System.currentTimeMillis() - Duration.ofHours(1).toMillis(); + lastCozeSubmitAtByCredential.entrySet().removeIf(entry -> + entry.getValue() == null || entry.getValue() < cutoff); } private static ThreadFactory namedThreadFactory(String prefix) { @@ -1898,6 +1918,9 @@ public class SimilarAsinTaskService { if (stateId == null) { continue; } + // The task lock is already held here. The outer wrapper + // would perform an extra selectById for every state before + // entering this method. pollPendingCozeStateLocked(stateId); } } finally { @@ -1932,11 +1955,15 @@ public class SimilarAsinTaskService { } // 先批量加载 state，避免每个预取线程各自查 DB；同时筛掉 claim 之前就不可能 poll 的状态。 List candidates = new ArrayList<>(stateIds.size()); - for (Long stateId : stateIds) { - if (stateId == null) { - continue; - } - TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId); + List refreshedStates = taskScopeStateMapper.selectList( + new LambdaQueryWrapper() + .in(TaskScopeStateEntity::getId, stateIds) + .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE) + .in(TaskScopeStateEntity::getCozeStatus, + List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING))); + for (TaskScopeStateEntity state : refreshedStates == null + ? List.of() + : refreshedStates) { if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) { @@ -1953,56 +1980,42 @@ public class SimilarAsinTaskService { if (candidates.size() <= 1) { return; } - Semaphore concurrency = new Semaphore(Math.max(1, COZE_POLL_PREFETCH_CONCURRENCY)); - // 用临时虚拟线程池：调用结束就 close，不影响主调度线程池的容量。 - ExecutorService prefetchPool = Executors.newThreadPerTaskExecutor(Thread.ofVirtual() - .name("similar-asin-coze-prefetch-", 0) - .factory()); + // 复用固定大小的预取线程池，限制并发并避免每轮调度反复创建线程池。 Map> futures = new LinkedHashMap<>(); - try { - long prefetchStart = System.currentTimeMillis(); - for (TaskScopeStateEntity state : candidates) { - final Long stateId = state.getId(); - final String executeId = state.getCozeExecuteId(); - final CozeBatchContext context = readCozeBatchContext(state); - if (context == null) { - continue; - } - final String credentialName = context.credentialName(); - Future future = prefetchPool.submit(() -> { - concurrency.acquire(); - try { - return cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName)); - } finally { - concurrency.release(); - } - }); - futures.put(stateId, future); + long prefetchStart = System.currentTimeMillis(); + for (TaskScopeStateEntity state : candidates) { + final Long stateId = state.getId(); + final String executeId = state.getCozeExecuteId(); + final CozeBatchContext context = readCozeBatchContext(state); + if (context == null) { + continue; } - for (Map.Entry> entry : futures.entrySet()) { - Long stateId = entry.getKey(); - Future future = entry.getValue(); - try { + final String credentialName = context.credentialName(); + Future future = cozePollPrefetchExecutor.submit( + () -> cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName))); + futures.put(stateId, future); + } + for (Map.Entry> entry : futures.entrySet()) { + Long stateId = entry.getKey(); + Future future = entry.getValue(); + try { SimilarAsinCozeClient.CozePollResponse response = future.get(COZE_POLL_PREFETCH_WAIT_MILLIS, java.util.concurrent.TimeUnit.MILLISECONDS); if (response != null) { cache.cozePollResponseCache.put(stateId, response); } - } catch (java.util.concurrent.TimeoutException timeoutEx) { + } catch (java.util.concurrent.TimeoutException timeoutEx) { future.cancel(true); log.warn("[similar-asin] coze poll prefetch timeout stateId={} waitMs={}", stateId, COZE_POLL_PREFETCH_WAIT_MILLIS); - } catch (Exception ex) { + } catch (Exception ex) { log.warn("[similar-asin] coze poll prefetch failed stateId={} err={}", stateId, firstNonBlank(ex.getMessage(), ex.getClass().getSimpleName())); - } } - log.info("[similar-asin] coze poll prefetch finished candidates={} cached={} costMs={}", - candidates.size(), cache.cozePollResponseCache.size(), - System.currentTimeMillis() - prefetchStart); - } finally { - prefetchPool.shutdownNow(); } + log.info("[similar-asin] coze poll prefetch finished candidates={} cached={} costMs={}", + candidates.size(), cache.cozePollResponseCache.size(), + System.currentTimeMillis() - prefetchStart); } /** @@ -4228,7 +4241,166 @@ public class SimilarAsinTaskService { .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)); } + /** + * Bounded assembly path. Chunk payloads are selected and decoded one at a + * time for the current source file; the result map is released when that + * source workbook finishes. This prevents a multi-source/large-task + * assemble from retaining every chunk JSON and every result DTO together. + */ + private void assembleResultWorkbookBounded(FileTaskEntity task, FileResultEntity result) { + SimilarAsinParsedPayloadDto parsed = readParsedPayload(task); + List sourceRows = splitRowsBySourceFile( + parsed, parsed.getAllItems(), result.getSourceFilename()); + if (sourceRows.isEmpty()) { + throw new BusinessException("Similar ASIN result rows are empty"); + } + + File outputDir = new File(storageProperties.getLocalTempDir(), "similar-asin-result"); + if (!outputDir.exists() && !outputDir.mkdirs()) { + throw new BusinessException("failed to create result directory"); + } + List workbooks = new ArrayList<>(); + List workbookFiles = new ArrayList<>(sourceRows.size()); + File zip = null; + long imageCacheMaxBytes = properties.getImageCacheMaxBytes() > 0 + ? properties.getImageCacheMaxBytes() + : BoundedImageCache.DEFAULT_MAX_BYTES; + BoundedImageCache taskImageCache = new BoundedImageCache(imageCacheMaxBytes); + try { + int timeoutMinutes = Math.max(1, properties.getResultFileTimeoutMinutes()); + long deadlineNanos = System.nanoTime() + TimeUnit.MINUTES.toNanos(timeoutMinutes); + long assembleStart = System.currentTimeMillis(); + List> futures = new ArrayList<>(sourceRows.size()); + for (SourceRows item : sourceRows) { + final SourceRows captured = item; + final String filename = safeFileStem(captured.sourceFilename()) + "-result.xlsx"; + final String tempFilename = safeFileStem(captured.sourceFilename()) + + "-" + task.getId() + + "-" + result.getId() + + "-" + UUID.randomUUID() + + "-result.xlsx"; + final File xlsx = new File(outputDir, tempFilename); + workbookFiles.add(xlsx); + futures.add(assembleExecutor.submit(() -> { + ensureResultAssemblyNotInterrupted(); + Map sourceResultMap = + loadPersistedResultRowsForRowsWithRetry(task.getId(), captured.rows()); + int persistedRows = sourceResultMap.size(); + sourceResultMap.entrySet().removeIf(entry -> !isExportableResultRow(entry.getValue())); + long resolvedRows = captured.rows().stream() + .filter(row -> findResultRow(row, sourceResultMap) != null) + .count(); + int conformPropagated = CozeGroupResultPropagator.propagateByGroup( + captured.rows(), + SimilarAsinParsedRowVo::getDisplayId, + row -> findResultRow(row, sourceResultMap), + SimilarAsinResultRowDto::getIsConform, + SimilarAsinResultRowDto::setIsConform, + List.of("不符合"), + "[similar-asin] taskId=" + task.getId() + ); + writeResultWorkbook(xlsx, captured.rows(), sourceResultMap, taskImageCache); + ensureResultAssemblyNotInterrupted(); + return new SourceResultWorkbook( + xlsx, filename, captured.rows().size(), persistedRows, + sourceResultMap.size(), resolvedRows, conformPropagated); + })); + } + try { + for (Future future : futures) { + long remainingNanos = deadlineNanos - System.nanoTime(); + if (remainingNanos <= 0L) { + throw new TimeoutException("result file deadline reached"); + } + workbooks.add(future.get(remainingNanos, TimeUnit.NANOSECONDS)); + } + } catch (InterruptedException ex) { + cancelResultAssemblyFutures(futures); + Thread.currentThread().interrupt(); + throw new BusinessException("result workbook assembly interrupted", ex); + } catch (TimeoutException ex) { + cancelResultAssemblyFutures(futures); + throw new BusinessException("result workbook assembly timed out", ex); + } catch (CancellationException ex) { + cancelResultAssemblyFutures(futures); + throw new BusinessException("result workbook assembly cancelled", ex); + } catch (ExecutionException ex) { + cancelResultAssemblyFutures(futures); + Throwable cause = ex.getCause(); + if (cause instanceof BusinessException businessException) { + throw businessException; + } + throw new BusinessException("result workbook assembly failed", cause == null ? ex : cause); + } + + int persistedResultRows = workbooks.stream() + .mapToInt(SourceResultWorkbook::persistedRows) + .sum(); + int resultRows = workbooks.stream() + .mapToInt(SourceResultWorkbook::resultRows) + .sum(); + long resolvedRows = workbooks.stream() + .mapToLong(SourceResultWorkbook::resolvedRows) + .sum(); + int conformPropagated = workbooks.stream() + .mapToInt(SourceResultWorkbook::conformPropagated) + .sum(); + log.info("[similar-asin] bounded assemble workbook taskId={} parsedRows={} persistedRows={} resultRows={} resolvedRows={} propagated={} sources={} costMs={}", + task.getId(), parsed.getAllItems().size(), persistedResultRows, resultRows, + resolvedRows, conformPropagated, sourceRows.size(), + System.currentTimeMillis() - assembleStart); + if (resultRows == 0) { + String failureSummary = collectChunkReadFailureSummary(task.getId()); + throw new BusinessException(failureSummary.isBlank() + ? "Similar ASIN result rows are empty; please retry result generation" + : "Similar ASIN result rows are empty (" + failureSummary + ")"); + } + + taskImageCache.clear(); + File uploadFile; + String filename; + String contentType; + if (workbooks.size() == 1) { + SourceResultWorkbook workbook = workbooks.get(0); + uploadFile = workbook.file(); + filename = workbook.filename(); + contentType = CONTENT_TYPE_XLSX; + } else { + filename = safeFileStem(result.getSourceFilename()) + "-result.zip"; + String tempFilename = safeFileStem(result.getSourceFilename()) + + "-" + task.getId() + + "-" + result.getId() + + "-" + UUID.randomUUID() + + "-result.zip"; + zip = new File(outputDir, tempFilename); + packageResultWorkbooksAsZip(zip, workbooks); + uploadFile = zip; + contentType = CONTENT_TYPE_ZIP; + } + String objectKey = ossStorageService.uploadResultFile(uploadFile, MODULE_TYPE); + result.setResultFilename(filename); + result.setResultFileUrl(objectKey); + result.setResultFileSize(uploadFile.length()); + result.setResultContentType(contentType); + result.setRowCount(parsed.getAllItems().size()); + } finally { + taskImageCache.clear(); + for (File workbookFile : workbookFiles) { + if (workbookFile.exists() && !workbookFile.delete()) { + log.warn("[similar-asin] delete temp xlsx failed file={}", workbookFile); + } + } + if (zip != null && zip.exists() && !zip.delete()) { + log.warn("[similar-asin] delete temp zip failed file={}", zip); + } + } + } + private void assembleResultWorkbook(FileTaskEntity task, FileResultEntity result) { + if (properties.isBoundedResultAssemblyEnabled()) { + assembleResultWorkbookBounded(task, result); + return; + } SimilarAsinParsedPayloadDto parsed = readParsedPayload(task); Map resultMap = loadPersistedResultRowsWithRetry(task.getId(), parsed.getAllItems().size()); int persistedResultRows = resultMap.size(); @@ -4510,6 +4682,152 @@ public class SimilarAsinTaskService { return normalized.isBlank() ? "similar-asin-result.xlsx" : normalized; } + private Map loadPersistedResultRowsForRowsWithRetry( + Long taskId, + List requestedRows) { + Map result = + loadPersistedResultRowsForRows(taskId, requestedRows); + if (requestedRows == null || requestedRows.isEmpty() || !result.isEmpty()) { + return result; + } + for (int attempt = 1; attempt <= RESULT_ROWS_READ_RETRY_LIMIT && result.isEmpty(); attempt++) { + sleepBeforeResultRowsRetry(attempt); + result = loadPersistedResultRowsForRows(taskId, requestedRows); + if (!result.isEmpty()) { + log.info("[similar-asin] source result rows recovered after retry taskId={} attempt={} rows={}", + taskId, attempt, result.size()); + } + } + return result; + } + + /** + * Select and decode one chunk payload at a time. Only rows needed by the + * current source workbook are retained in the returned map. + */ + private Map loadPersistedResultRowsForRows( + Long taskId, + List requestedRows) { + Map result = new LinkedHashMap<>(); + Set requestedKeys = resultLookupKeys(requestedRows); + if (taskId == null || requestedKeys.isEmpty()) { + return result; + } + + Long lastChunkId = null; + while (true) { + LambdaQueryWrapper query = new LambdaQueryWrapper() + .eq(TaskChunkEntity::getTaskId, taskId) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) + .orderByAsc(TaskChunkEntity::getId) + .last("LIMIT 1"); + if (lastChunkId != null) { + query.gt(TaskChunkEntity::getId, lastChunkId); + } + List page = taskChunkMapper.selectList(query); + if (page == null || page.isEmpty()) { + break; + } + TaskChunkEntity chunk = page.getFirst(); + appendRequestedResultRows(result, readChunkRows(chunk).values(), requestedKeys, false); + lastChunkId = chunk.getId(); + if (lastChunkId == null) { + break; + } + } + appendRequestedOrphanRows(taskId, requestedKeys, result); + return result; + } + + private Set resultLookupKeys(List rows) { + Set keys = new LinkedHashSet<>(); + for (SimilarAsinParsedRowVo row : rows == null ? List.of() : rows) { + String primary = rowKey(row); + if (!primary.isBlank()) { + keys.add(primary); + } + String legacy = legacyRowKey(row); + if (!legacy.isBlank()) { + keys.add(legacy); + } + } + return keys; + } + + private void appendRequestedResultRows(Map sink, + Iterable rows, + Set requestedKeys, + boolean onlyIfAbsent) { + if (rows == null || requestedKeys == null || requestedKeys.isEmpty()) { + return; + } + for (SimilarAsinResultRowDto row : rows) { + if (row == null) { + continue; + } + String primary = rowKey(row); + String legacy = legacyRowKey(row); + String matchedKey = requestedKeys.contains(primary) + ? primary + : requestedKeys.contains(legacy) ? legacy : ""; + if (matchedKey.isBlank()) { + continue; + } + if (onlyIfAbsent) { + sink.putIfAbsent(matchedKey, row); + } else { + sink.put(matchedKey, row); + } + } + } + + private void appendRequestedOrphanRows(Long taskId, + Set requestedKeys, + Map sink) { + Long lastStateId = null; + while (true) { + LambdaQueryWrapper query = new LambdaQueryWrapper() + .select(TaskScopeStateEntity::getId, + TaskScopeStateEntity::getParsedPayloadJson) + .eq(TaskScopeStateEntity::getTaskId, taskId) + .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE) + .likeRight(TaskScopeStateEntity::getScopeKey, ORPHAN_SCOPE_KEY_PREFIX) + .orderByAsc(TaskScopeStateEntity::getId) + .last("LIMIT 1"); + if (lastStateId != null) { + query.gt(TaskScopeStateEntity::getId, lastStateId); + } + List page = taskScopeStateMapper.selectList(query); + if (page == null || page.isEmpty()) { + return; + } + TaskScopeStateEntity state = page.getFirst(); + lastStateId = state.getId(); + if (lastStateId == null) { + return; + } + if (state.getParsedPayloadJson() == null || state.getParsedPayloadJson().isBlank()) { + continue; + } + try { + String payloadJson = transientPayloadStorageService.resolvePayload( + state.getParsedPayloadJson(), "read similar ASIN orphan rows failed"); + JsonNode array = objectMapper.readTree(payloadJson); + if (!array.isArray()) { + continue; + } + List matchingRows = new ArrayList<>(); + for (JsonNode node : array) { + matchingRows.add(objectMapper.treeToValue(node, SimilarAsinResultRowDto.class)); + } + appendRequestedResultRows(sink, matchingRows, requestedKeys, true); + } catch (Exception ex) { + log.warn("[similar-asin] read requested orphan rows failed taskId={} stateId={} err={}", + taskId, state.getId(), ex.getMessage()); + } + } + } + private Map loadPersistedResultRows(Long taskId) { Map result = new LinkedHashMap<>(); List chunks = taskChunkMapper.selectList(new LambdaQueryWrapper() @@ -6145,7 +6463,14 @@ public class SimilarAsinTaskService { private record SourceResultWorkbook(File file, String filename, - int rowCount) { + int rowCount, + int persistedRows, + int resultRows, + long resolvedRows, + int conformPropagated) { + private SourceResultWorkbook(File file, String filename, int rowCount) { + this(file, filename, rowCount, 0, 0, 0L, 0); + } } public record ResultDownloadInfo(String url, diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java index 131bffc3..ace2c5f1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java @@ -15,6 +15,7 @@ import org.springframework.stereotype.Component; import javax.imageio.IIOImage; import javax.imageio.ImageIO; +import javax.imageio.ImageReadParam; import javax.imageio.ImageReader; import javax.imageio.ImageWriteParam; import javax.imageio.ImageWriter; @@ -50,13 +51,16 @@ import java.util.Locale; import java.util.Map; import java.util.Objects; import java.util.Set; +import java.util.concurrent.CompletableFuture; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; import java.util.concurrent.CompletionService; import java.util.concurrent.ExecutorCompletionService; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; +import java.util.concurrent.Semaphore; import java.util.concurrent.ThreadFactory; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; @@ -77,8 +81,8 @@ public class SimilarAsinImageEmbedder { static final int DEFAULT_DOWNLOAD_TIMEOUT_SECONDS = 5; /** P2-10：retry 由 1 升到 2，配合 5s timeout 单图最坏耗时 ≈ 15s。 */ static final int DOWNLOAD_MAX_RETRY = 2; - /** 图片下载、解码、缩放都在该池执行；默认限制为 8，避免结果生成占满整机 CPU。 */ - static final int DEFAULT_DOWNLOAD_POOL_SIZE = 8; + /** 4 核生产机默认 2；构造时还会按 JVM 可见 CPU 数的一半做硬上限。 */ + static final int DEFAULT_DOWNLOAD_POOL_SIZE = 2; static final int DEFAULT_PREFETCH_TIMEOUT_SECONDS = 1800; // 单元格固定尺寸，图片在其中等比缩放（不拉伸）；resize 仍按长边 1280 px 控制堆体积。 public static final float IMAGE_ROW_HEIGHT_POINTS = 409f; @@ -115,6 +119,8 @@ public class SimilarAsinImageEmbedder { private final int prefetchTimeoutSeconds; private final OkHttpClient httpClient; private final ExecutorService downloadPool; + private final Semaphore imageProcessingSlots; + private final ConcurrentMap> inFlightResizes = new ConcurrentHashMap<>(); private final OssStorageService ossStorageService; private final Path localImageCacheDir; @@ -125,7 +131,16 @@ public class SimilarAsinImageEmbedder { ? DEFAULT_PREFETCH_TIMEOUT_SECONDS : properties.getImagePrefetchTimeoutSeconds(); this.downloadTimeoutSeconds = rawTimeout > 0 ? rawTimeout : DEFAULT_DOWNLOAD_TIMEOUT_SECONDS; - this.downloadPoolSize = rawPool > 0 ? rawPool : DEFAULT_DOWNLOAD_POOL_SIZE; + int requestedPoolSize = rawPool > 0 ? rawPool : DEFAULT_DOWNLOAD_POOL_SIZE; + int cpuBoundPoolSize = cpuBoundPoolLimit(); + this.downloadPoolSize = Math.min(requestedPoolSize, cpuBoundPoolSize); + if (requestedPoolSize > this.downloadPoolSize) { + log.warn("[similar-asin][image] image pool clamped requested={} effective={} visibleProcessors={}", + requestedPoolSize, this.downloadPoolSize, Runtime.getRuntime().availableProcessors()); + } + this.imageProcessingSlots = new Semaphore(this.downloadPoolSize, true); + log.info("[similar-asin][image] image pool initialized requested={} effective={} cpuBoundLimit={}", + requestedPoolSize, this.downloadPoolSize, cpuBoundPoolSize); this.prefetchTimeoutSeconds = rawPrefetchTimeout > 0 ? rawPrefetchTimeout : DEFAULT_PREFETCH_TIMEOUT_SECONDS; @@ -152,6 +167,11 @@ public class SimilarAsinImageEmbedder { downloadPool.shutdownNow(); } + static int cpuBoundPoolLimit() { + int visibleProcessors = Math.max(1, Runtime.getRuntime().availableProcessors()); + return Math.max(1, (visibleProcessors + 1) / 2); + } + int downloadTimeoutSeconds() { return downloadTimeoutSeconds; } @@ -251,9 +271,7 @@ public class SimilarAsinImageEmbedder { if (taskImageCache.containsKey(url)) { return; } - String downloadUrl = normalizeAndValidateDownloadUrl(url); - byte[] raw = doFetch(downloadUrl); - ResizedImage thumb = resizeImage(downloadUrl, raw); + ResizedImage thumb = fetchAndResizeDirect(url); taskImageCache.putIfAbsent(url, thumb); } catch (Exception ex) { // 预下载失败不抛出：embed() 时同 url 会再次尝试并走原有兜底链路。 @@ -605,6 +623,25 @@ public class SimilarAsinImageEmbedder { private ResizedImage fetchAndResizeDirect(String url) throws IOException { String downloadUrl = normalizeAndValidateDownloadUrl(url); + CompletableFuture owned = new CompletableFuture<>(); + CompletableFuture existing = inFlightResizes.putIfAbsent(downloadUrl, owned); + if (existing != null) { + return awaitInFlightResize(downloadUrl, existing); + } + try { + ResizedImage cached = readLocalCachedThumb(downloadUrl); + ResizedImage result = cached != null ? cached : fetchAndResizeUncached(downloadUrl); + owned.complete(result); + return result; + } catch (IOException | RuntimeException | Error ex) { + owned.completeExceptionally(ex); + throw ex; + } finally { + inFlightResizes.remove(downloadUrl, owned); + } + } + + private ResizedImage fetchAndResizeUncached(String downloadUrl) throws IOException { List candidates = downloadCandidates(downloadUrl); IOException last = null; for (int attempt = 0; attempt <= DOWNLOAD_MAX_RETRY; attempt++) { @@ -634,6 +671,30 @@ public class SimilarAsinImageEmbedder { throw last == null ? new IOException("image download failed: " + downloadUrl) : last; } + private static ResizedImage awaitInFlightResize(String url, + CompletableFuture future) throws IOException { + try { + return future.get(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + InterruptedIOException interrupted = new InterruptedIOException("image resize wait interrupted url=" + url); + interrupted.initCause(ex); + throw interrupted; + } catch (ExecutionException ex) { + Throwable cause = ex.getCause(); + if (cause instanceof IOException ioException) { + throw ioException; + } + if (cause instanceof RuntimeException runtimeException) { + throw runtimeException; + } + if (cause instanceof Error error) { + throw error; + } + throw new IOException("image resize failed url=" + url, cause); + } + } + String normalizeAndValidateDownloadUrl(String url) { String downloadUrl = ossStorageService.normalizeManagedPublicUrl(url); validateHttpsUrl(downloadUrl); @@ -825,43 +886,105 @@ public class SimilarAsinImageEmbedder { * 仍然超限才抛 ResizeOversizeException 触发文本兜底。 */ ResizedImage resizeImage(String sourceUrl, byte[] raw) throws IOException { - ensureImageWorkNotInterrupted(); - guardImageDimensions(sourceUrl, raw); - BufferedImage src = ImageIO.read(new ByteArrayInputStream(raw)); - if (src == null) { - throw new IOException("unsupported image format url=" + sourceUrl); - } - int srcW = src.getWidth(); - int srcH = src.getHeight(); - // B 方案降级顺序：固定 MAX_THUMB_SIZE 上限 → 优先调整长边像素 → 再调质量。 - // 同一 src BufferedImage 解码一次，下面 9 种组合复用，避免重复 ImageIO.read。 - ResizedImage candidate = null; - ResizedImage smallest = null; - for (int longEdge : FALLBACK_LONG_EDGES) { + acquireImageProcessingSlot(); + try { ensureImageWorkNotInterrupted(); - BufferedImage scaled = scaleAt(src, srcW, srcH, longEdge); - for (float quality : FALLBACK_QUALITIES) { - ensureImageWorkNotInterrupted(); - ResizedImage tried = encodeJpeg(scaled, quality); - if (smallest == null || tried.bytes().length < smallest.bytes().length) { - smallest = tried; + BufferedImage src = decodeForResize(sourceUrl, raw); + try { + int srcW = src.getWidth(); + int srcH = src.getHeight(); + ResizedImage candidate = null; + ResizedImage smallest = null; + for (int longEdge : FALLBACK_LONG_EDGES) { + ensureImageWorkNotInterrupted(); + BufferedImage scaled = scaleAt(src, srcW, srcH, longEdge); + try { + for (float quality : FALLBACK_QUALITIES) { + ensureImageWorkNotInterrupted(); + ResizedImage tried = encodeJpeg(scaled, quality); + if (smallest == null || tried.bytes().length < smallest.bytes().length) { + smallest = tried; + } + if (tried.bytes().length <= MAX_THUMB_SIZE_BYTES) { + candidate = tried; + break; + } + } + } finally { + scaled.flush(); + } + if (candidate != null) { + break; + } } - if (tried.bytes().length <= MAX_THUMB_SIZE_BYTES) { - candidate = tried; - break; + if (candidate != null) { + return candidate; } + int reportedSize = smallest != null ? smallest.bytes().length : -1; + throw new ResizeOversizeException(sourceUrl, reportedSize); + } finally { + src.flush(); } - if (candidate != null) { - break; + } finally { + imageProcessingSlots.release(); + } + } + + private void acquireImageProcessingSlot() throws InterruptedIOException { + try { + imageProcessingSlots.acquire(); + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + InterruptedIOException interrupted = new InterruptedIOException("image processing interrupted"); + interrupted.initCause(ex); + throw interrupted; + } + } + + private static BufferedImage decodeForResize(String sourceUrl, byte[] raw) throws IOException { + try (ImageInputStream iis = ImageIO.createImageInputStream(new ByteArrayInputStream(raw))) { + if (iis == null) { + throw new IOException("unable to create ImageInputStream url=" + sourceUrl); + } + Iterator readers = ImageIO.getImageReaders(iis); + if (!readers.hasNext()) { + throw new IOException("unsupported image format url=" + sourceUrl); + } + ImageReader reader = readers.next(); + try { + reader.setInput(iis, true, true); + int sourceWidth = reader.getWidth(0); + int sourceHeight = reader.getHeight(0); + long pixels = (long) sourceWidth * (long) sourceHeight; + if (sourceWidth <= 0 || sourceHeight <= 0 || pixels > MAX_DECODE_PIXELS) { + throw new ResizeException("image too large url=" + sourceUrl + " pixels=" + pixels); + } + ImageReadParam readParam = reader.getDefaultReadParam(); + if (isJpegReader(reader)) { + int subsampling = jpegSourceSubsampling(sourceWidth, sourceHeight); + if (subsampling > 1) { + readParam.setSourceSubsampling(subsampling, subsampling, 0, 0); + } + } + BufferedImage decoded = reader.read(0, readParam); + if (decoded == null) { + throw new IOException("unsupported image format url=" + sourceUrl); + } + return decoded; + } finally { + reader.dispose(); } } - if (candidate != null) { - return candidate; - } - // 9 组合都没压到上限：抛 ResizeOversizeException 走文本兜底， - // 同时报告 smallest 字节让运维直观知道当前压缩极限。 - int reportedSize = smallest != null ? smallest.bytes().length : -1; - throw new ResizeOversizeException(sourceUrl, reportedSize); + } + + private static boolean isJpegReader(ImageReader reader) throws IOException { + String format = reader.getFormatName(); + return "JPEG".equalsIgnoreCase(format) || "JPG".equalsIgnoreCase(format); + } + + static int jpegSourceSubsampling(int sourceWidth, int sourceHeight) { + int ratio = Math.max(sourceWidth, sourceHeight) / TARGET_LONG_EDGE_PX; + return ratio > 1 ? Integer.highestOneBit(ratio) : 1; } private static void ensureImageWorkNotInterrupted() throws InterruptedIOException { @@ -995,29 +1118,6 @@ public class SimilarAsinImageEmbedder { } } - /** 在解码整张位图前用 ImageReader 仅读取头部尺寸，避免“图像炸弹”导致 heap OOM。 */ - private static void guardImageDimensions(String sourceUrl, byte[] raw) throws IOException { - try (ImageInputStream iis = ImageIO.createImageInputStream(new ByteArrayInputStream(raw))) { - if (iis == null) { - throw new IOException("unable to create ImageInputStream url=" + sourceUrl); - } - Iterator readers = ImageIO.getImageReaders(iis); - if (!readers.hasNext()) { - throw new IOException("unsupported image format url=" + sourceUrl); - } - ImageReader reader = readers.next(); - try { - reader.setInput(iis, true, true); - long pixels = (long) reader.getWidth(0) * (long) reader.getHeight(0); - if (pixels > MAX_DECODE_PIXELS) { - throw new ResizeException("image too large url=" + sourceUrl + " pixels=" + pixels); - } - } finally { - reader.dispose(); - } - } - } - private static ThreadFactory namedFactory(String prefix) { AtomicInteger counter = new AtomicInteger(); return r -> { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java index 622ff5bb..f8f0d34a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java @@ -9,6 +9,9 @@ import lombok.Data; @Schema(description = "任务心跳请求") public class TaskHeartbeatRequest { + @Schema(description = "可选的任务模块类型；传入后可避免跨任务表探测", example = "SHOP_DATA_CRAWL") + private String moduleType; + @Schema(description = "可选的任务阶段标识", example = "crawling") private String phase; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java index 8e9f2a40..dfb2973f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java @@ -78,15 +78,15 @@ public class TaskHeartbeatService { return TaskHeartbeatVo.notAlive(null, null, "invalid taskId"); } - LambdaQueryWrapper fileQuery = new LambdaQueryWrapper() - .eq(FileTaskEntity::getId, taskId) - .last("limit 1"); - FileTaskEntity fileTask = fileTaskMapper.selectOne(fileQuery); - - LambdaQueryWrapper brandQuery = new LambdaQueryWrapper() - .eq(BrandCrawlTaskEntity::getId, taskId) - .last("limit 1"); - BrandCrawlTaskEntity brandTask = brandCrawlTaskMapper.selectOne(brandQuery); + String requestedModuleType = request == null || request.getModuleType() == null + ? null + : request.getModuleType().trim(); + FileTaskEntity fileTask = MODULE_BRAND.equalsIgnoreCase(requestedModuleType) + ? null + : selectFileTask(taskId); + BrandCrawlTaskEntity brandTask = MODULE_SHOP_DATA_CRAWL.equalsIgnoreCase(requestedModuleType) + ? null + : selectBrandTask(taskId); TaskHeartbeatVo fileResult = touchFileTaskIfRunning(fileTask, request); TaskHeartbeatVo brandResult = touchBrandTaskIfRunning(brandTask, request); @@ -110,6 +110,20 @@ public class TaskHeartbeatService { return TaskHeartbeatVo.notAlive(null, null, "task not found"); } + private FileTaskEntity selectFileTask(Long taskId) { + LambdaQueryWrapper fileQuery = new LambdaQueryWrapper() + .eq(FileTaskEntity::getId, taskId) + .last("limit 1"); + return fileTaskMapper.selectOne(fileQuery); + } + + private BrandCrawlTaskEntity selectBrandTask(Long taskId) { + LambdaQueryWrapper brandQuery = new LambdaQueryWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .last("limit 1"); + return brandCrawlTaskMapper.selectOne(brandQuery); + } + private TaskHeartbeatVo touchFileTaskIfRunning(FileTaskEntity task, TaskHeartbeatRequest request) { if (task == null) { return null; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoTransientCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoTransientCacheService.java index 75df21f9..06867aa6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoTransientCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/memory/service/ZiniaoTransientCacheService.java @@ -5,11 +5,17 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.scheduling.annotation.Scheduled; import org.springframework.stereotype.Service; +import java.nio.charset.StandardCharsets; import java.time.Duration; import java.time.LocalDateTime; +import java.util.ArrayList; +import java.util.Comparator; import java.util.List; +import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.concurrent.ConcurrentHashMap; @@ -29,6 +35,12 @@ public class ZiniaoTransientCacheService { private final ConcurrentHashMap map = new ConcurrentHashMap<>(); + @Value("${aiimage.ziniao.transient-cache-max-entries:10000}") + private int maxEntries; + + @Value("${aiimage.ziniao.transient-cache-max-payload-bytes:5242880}") + private int maxPayloadBytes; + public Optional get(String cacheType, String cacheKey, Class valueType) { Holder holder = getHolder(cacheType, cacheKey); if (holder == null) { @@ -68,8 +80,18 @@ public class ZiniaoTransientCacheService { } try { String json = objectMapper.writeValueAsString(payload); + int payloadBytes = json.getBytes(StandardCharsets.UTF_8).length; + int payloadLimit = maxPayloadBytes > 0 ? maxPayloadBytes : 5 * 1024 * 1024; + if (payloadBytes > payloadLimit) { + log.warn("[ziniao-transient] skip oversized cache type={} keyLen={} bytes={} maxBytes={}", + normalizedType, normalizedKey.length(), + payloadBytes, payloadLimit); + return; + } LocalDateTime expiresAt = LocalDateTime.now().plusSeconds(ttl.getSeconds()); + cleanupExpiredEntries(LocalDateTime.now()); map.put(compoundKey(normalizedType, normalizedKey), new Holder(json, expiresAt)); + evictToCapacity(); log.trace("[ziniao-transient] put type={} keyLen={} ttlSec={}", normalizedType, normalizedKey.length(), ttl.getSeconds()); } catch (Exception ex) { throw new BusinessException("写入紫鸟进程缓存失败"); @@ -83,6 +105,36 @@ public class ZiniaoTransientCacheService { log.trace("[ziniao-transient] delete type={}", normalizedType); } + @Scheduled(fixedDelayString = "${aiimage.ziniao.transient-cache-cleanup-delay-ms:60000}") + void cleanupExpiredEntriesScheduled() { + cleanupExpiredEntries(LocalDateTime.now()); + evictToCapacity(); + } + + private void cleanupExpiredEntries(LocalDateTime now) { + map.entrySet().removeIf(entry -> { + Holder holder = entry.getValue(); + return holder == null || holder.expiresAt == null || !holder.expiresAt.isAfter(now); + }); + } + + private void evictToCapacity() { + int limit = maxEntries > 0 ? maxEntries : 10000; + int overflow = map.size() - limit; + if (overflow <= 0) { + return; + } + List> candidates = new ArrayList<>(map.entrySet()); + candidates.sort(Comparator.comparing(entry -> entry.getValue().expiresAt, + Comparator.nullsFirst(Comparator.naturalOrder()))); + candidates.stream().limit(overflow).forEach(entry -> + map.remove(entry.getKey(), entry.getValue())); + } + + int size() { + return map.size(); + } + private Holder getHolder(String cacheType, String cacheKey) { String normalizedType = normalizeRequired(cacheType, "cacheType 不能为空"); String normalizedKey = normalizeRequired(cacheKey, "cacheKey 不能为空"); diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 654b1537..b458df86 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -227,7 +227,7 @@ aiimage: coze-submit-min-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MIN_INTERVAL_MILLIS:5000} coze-flush-pending-minutes: ${AIIMAGE_SIMILAR_ASIN_COZE_FLUSH_PENDING_MINUTES:1} coze-submit-max-retry-count: ${AIIMAGE_SIMILAR_ASIN_COZE_SUBMIT_MAX_RETRY_COUNT:5} - image-download-pool-size: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE:8} + image-download-pool-size: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_POOL_SIZE:2} image-download-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_DOWNLOAD_TIMEOUT_SECONDS:5} image-prefetch-timeout-seconds: ${AIIMAGE_SIMILAR_ASIN_IMAGE_PREFETCH_TIMEOUT_SECONDS:1800} result-file-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_RESULT_FILE_TIMEOUT_MINUTES:90} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java index 59d4b2a1..f1856d39 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataServiceTest.java @@ -43,6 +43,7 @@ import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -172,7 +173,8 @@ class DedupeTotalDataServiceTest { void duplicateExcelValueKeepsOriginalUploader() throws Exception { when(adminUserMapper.selectById(23L)).thenReturn(user(23L, "normal", "member-a")); stubWritableGroup(23L, 7L); - when(dedupeTotalDataMapper.selectOne(any())).thenReturn(data(91L, 99L)); + when(dedupeTotalDataMapper.selectExistingDataValues(List.of("B012345678"))) + .thenReturn(List.of("B012345678")); MockMultipartFile file = asinWorkbook("B012345678"); var result = service.importFromExcel(file, 7L, 23L); @@ -187,7 +189,10 @@ class DedupeTotalDataServiceTest { when(adminUserMapper.selectById(23L)).thenReturn(user(23L, "normal", "member-a")); stubWritableGroup(23L, 7L); when(dedupeTotalDataMapper.selectOne(any())).thenReturn(null); + when(dedupeTotalDataMapper.selectExistingDataValues(List.of("B012345678"))).thenReturn(List.of()); when(transactionManager.getTransaction(any())).thenReturn(transactionStatus); + when(dedupeTotalDataMapper.insertBatchIgnore(any())) + .thenThrow(new DuplicateKeyException("duplicate batch")); when(dedupeTotalDataMapper.insert(any(DedupeTotalDataEntity.class))) .thenThrow(new DuplicateKeyException("duplicate")); @@ -195,7 +200,7 @@ class DedupeTotalDataServiceTest { assertEquals(0, result.getInsertedCount()); assertEquals(1, result.getSkippedCount()); - verify(transactionManager).rollback(transactionStatus); + verify(transactionManager, times(2)).rollback(transactionStatus); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java index de2614d3..4835b7d0 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceChunkTest.java @@ -47,6 +47,7 @@ import java.util.Objects; import static org.junit.jupiter.api.Assertions.assertEquals; import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.junit.jupiter.api.Assertions.assertTrue; import static org.mockito.ArgumentMatchers.any; @@ -124,7 +125,8 @@ class ShopDataCrawlTaskServiceChunkTest { taskScopeStateMapper, transientPayloadStorageService, instanceMetadata, - dailyFileService); + dailyFileService, + null); storedChunks.clear(); storedScopes.clear(); @@ -183,6 +185,10 @@ class ShopDataCrawlTaskServiceChunkTest { assertTrue(task.getResultJson().indexOf("B001") < task.getResultJson().indexOf("B002")); verify(taskFileJobService).enqueueAssembleResult(task.getId(), MODULE_TYPE, result.getId(), "task:" + task.getId() + ":owner:instance-a"); + assertNull(result.getResultFileUrl()); + verify(excelAssemblyService, never()).writeWorkbook(any(), any()); + verify(excelAssemblyService, never()).appendWorkbook(any(), any(), any()); + verify(ossStorageService, never()).uploadResultFile(any(), eq(MODULE_TYPE)); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java index 2d8662ce..412d7103 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/shopdatacrawl/service/ShopDataCrawlTaskServiceRetentionTest.java @@ -36,15 +36,18 @@ import org.mockito.junit.jupiter.MockitoSettings; import org.mockito.quality.Strictness; import org.mockito.Spy; import org.mockito.junit.jupiter.MockitoExtension; -import org.springframework.transaction.support.TransactionSynchronization; -import org.springframework.transaction.support.TransactionSynchronizationManager; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionStatus; import java.time.LocalDate; import java.time.LocalDateTime; import java.util.List; import java.util.Set; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertThrows; import static org.mockito.ArgumentMatchers.any; @@ -94,6 +97,8 @@ class ShopDataCrawlTaskServiceRetentionTest { @Mock private TransientPayloadStorageService transientPayloadStorageService; @Mock private InstanceMetadata instanceMetadata; @Mock private ShopDataCrawlDailyFileService dailyFileService; + @Mock private PlatformTransactionManager transactionManager; + @Mock private TransactionStatus transactionStatus; @Spy private final ObjectMapper objectMapper = new ObjectMapper().findAndRegisterModules(); @InjectMocks private ShopDataCrawlTaskService service; @@ -128,6 +133,7 @@ class ShopDataCrawlTaskServiceRetentionTest { when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of()); when(dailyFileService.countObjectReferences(anyString())).thenReturn(0L); when(dailyFileService.addMember(anyLong(), anyLong(), anyLong())).thenReturn(true); + when(transactionManager.getTransaction(any())).thenReturn(transactionStatus); doAnswer(invocation -> { ShopDataCrawlDailyFileEntity entity = invocation.getArgument(0); entity.setId(301L); @@ -153,6 +159,7 @@ class ShopDataCrawlTaskServiceRetentionTest { verify(dailyFileService).insert(captor.capture()); assertEquals(BUSINESS_DATE, captor.getValue().getBusinessDate()); assertEquals(1L, captor.getValue().getVersion()); + assertEquals(currentRow.getResultFileUrl(), captor.getValue().getResultFileUrl()); } @Test @@ -169,25 +176,60 @@ class ShopDataCrawlTaskServiceRetentionTest { when(ossStorageService.readObjectBytes("result/old.xlsx")).thenReturn(new byte[]{1, 2, 3}); when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); - TransactionSynchronizationManager.initSynchronization(); - try { - service.processResultFileJob(job); + service.processResultFileJob(job); - verify(excelAssemblyService).appendWorkbook(any(), any(), eq(List.of(snapshot))); - verify(dailyFileService).update(daily); - verify(ossStorageService, never()).deleteObject("result/old.xlsx"); - assertNull(previous.getResultFileUrl()); - assertEquals("result/new.xlsx", currentRow.getResultFileUrl()); - assertEquals(3, currentRow.getRowCount()); + verify(excelAssemblyService).appendWorkbook(any(), any(), eq(List.of(snapshot))); + verify(dailyFileService).update(daily); + verify(ossStorageService).deleteObject("result/old.xlsx"); + assertNull(previous.getResultFileUrl()); + assertEquals("result/new.xlsx", currentRow.getResultFileUrl()); + assertEquals(3, currentRow.getRowCount()); + } - List synchronizations = TransactionSynchronizationManager.getSynchronizations(); - synchronizations.forEach(TransactionSynchronization::afterCommit); - verify(ossStorageService).deleteObject("result/old.xlsx"); - synchronizations.forEach(synchronization -> - synchronization.afterCompletion(TransactionSynchronization.STATUS_COMMITTED)); - } finally { - TransactionSynchronizationManager.clearSynchronization(); - } + @Test + void workbookAndSnapshotStorageRunOutsideShortTransactions() { + AtomicBoolean transactionActive = new AtomicBoolean(); + doAnswer(invocation -> { + assertFalse(transactionActive.getAndSet(true), "short transactions must not overlap"); + return transactionStatus; + }).when(transactionManager).getTransaction(any()); + doAnswer(invocation -> { + transactionActive.set(false); + return null; + }).when(transactionManager).commit(any()); + doAnswer(invocation -> { + transactionActive.set(false); + return null; + }).when(transactionManager).rollback(any()); + + ShopDataCrawlDailyFileEntity daily = daily("result/old.xlsx", 2); + when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); + when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); + doAnswer(invocation -> { + assertFalse(transactionActive.get(), "row counting must run outside the database transaction"); + return 1; + }).when(excelAssemblyService).countRows(any()); + doAnswer(invocation -> { + assertFalse(transactionActive.get(), "workbook download must run outside the database transaction"); + return new byte[]{1, 2, 3}; + }).when(ossStorageService).readObjectBytes("result/old.xlsx"); + doAnswer(invocation -> { + assertFalse(transactionActive.get(), "workbook assembly must run outside the database transaction"); + return null; + }).when(excelAssemblyService).appendWorkbook(any(), any(), any()); + doAnswer(invocation -> { + assertFalse(transactionActive.get(), "workbook upload must run outside the database transaction"); + return "result/new.xlsx"; + }).when(ossStorageService).uploadResultFile(any(), eq(MODULE_TYPE)); + doAnswer(invocation -> { + assertFalse(transactionActive.get(), "snapshot payload storage must run outside the database transaction"); + return null; + }).when(taskResultItemService).replaceTaskSnapshots(anyLong(), eq(MODULE_TYPE), any(), any()); + + service.processResultFileJob(job); + + assertFalse(transactionActive.get()); } @Test @@ -207,6 +249,26 @@ class ShopDataCrawlTaskServiceRetentionTest { assertEquals(3, currentRow.getRowCount()); } + @Test + void zeroNewRowsReuseDailyObjectWithoutWorkbookIo() { + ShopDataCrawlDailyFileEntity daily = daily("result/current.xlsx", 3); + when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); + when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(daily); + when(dailyFileService.containsResult(301L, RESULT_ID)).thenReturn(false); + when(excelAssemblyService.countRows(any())).thenReturn(0); + + service.processResultFileJob(job); + + verify(ossStorageService, never()).readObjectBytes(anyString()); + verify(excelAssemblyService, never()).writeWorkbook(any(), any()); + verify(excelAssemblyService, never()).appendWorkbook(any(), any(), any()); + verify(ossStorageService, never()).uploadResultFile(any(), anyString()); + verify(dailyFileService).addMember(301L, TASK_ID, RESULT_ID); + assertEquals("result/current.xlsx", currentRow.getResultFileUrl()); + assertEquals(10L, currentRow.getResultFileSize()); + assertEquals(3, currentRow.getRowCount()); + } + @Test void sameDaySuccessKeepsLegacyPointerOwnedByAnotherUser() { FileResultEntity previous = result(200L, 100L, "result/old.xlsx"); @@ -287,6 +349,29 @@ class ShopDataCrawlTaskServiceRetentionTest { verify(ossStorageService, never()).deleteObject("result/yesterday.xlsx"); } + @Test + void failedDatabaseCommitDeletesNewUploadButKeepsPreviousObject() { + ShopDataCrawlDailyFileEntity yesterday = daily("result/yesterday.xlsx", 4); + yesterday.setId(300L); + yesterday.setBusinessDate(BUSINESS_DATE.minusDays(1)); + when(fileResultMapper.selectList(any())).thenReturn(List.of(currentRow)); + when(dailyFileService.findForUpdate(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(null); + when(dailyFileService.findOlder(USER_ID, "hash-1", BUSINESS_DATE)).thenReturn(List.of(yesterday)); + when(ossStorageService.uploadResultFile(any(), eq(MODULE_TYPE))).thenReturn("result/new.xlsx"); + AtomicInteger commitCount = new AtomicInteger(); + doAnswer(invocation -> { + if (commitCount.incrementAndGet() == 2) { + throw new IllegalStateException("commit failed"); + } + return null; + }).when(transactionManager).commit(any()); + + assertThrows(IllegalStateException.class, () -> service.processResultFileJob(job)); + + verify(ossStorageService).deleteObject("result/new.xlsx"); + verify(ossStorageService, never()).deleteObject("result/yesterday.xlsx"); + } + @Test void deletingOneMemberRebuildsDailyWorkbookFromRemainingResults() { task.setStatus("SUCCESS"); diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java index ec465fa7..7d4b9da5 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java @@ -28,6 +28,9 @@ import java.util.Collections; import java.util.HashMap; import java.util.List; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; @@ -64,9 +67,77 @@ class SimilarAsinImageEmbedderTest { } @Test - void defaultsImageDownloadPoolToEight() { - assertEquals(8, new SimilarAsinProperties().getImageDownloadPoolSize()); - assertEquals(8, embedder.downloadPoolSize()); + void defaultsImageDownloadPoolToTwoAndCapsItByVisibleCpu() { + assertEquals(2, new SimilarAsinProperties().getImageDownloadPoolSize()); + assertEquals(Math.min(2, SimilarAsinImageEmbedder.cpuBoundPoolLimit()), embedder.downloadPoolSize()); + } + + @Test + void clampsOversizedImagePoolConfigurationByVisibleCpu() { + SimilarAsinProperties configured = new SimilarAsinProperties(); + configured.setImageDownloadPoolSize(Integer.MAX_VALUE); + SimilarAsinImageEmbedder limited = new SimilarAsinImageEmbedder(configured, createOssStorageService()); + try { + assertEquals(SimilarAsinImageEmbedder.cpuBoundPoolLimit(), limited.downloadPoolSize()); + } finally { + limited.shutdown(); + } + } + + @Test + void concurrentRequestsForSameUrlDownloadAndResizeOnlyOnce() throws Exception { + SimilarAsinImageEmbedder shared = new SimilarAsinImageEmbedder( + properties(2, 5, null), createOssStorageService()); + AtomicInteger networkCalls = new AtomicInteger(); + CountDownLatch releaseNetwork = new CountDownLatch(1); + CountDownLatch firstNetworkCall = new CountDownLatch(1); + replaceHttpClient(shared, new OkHttpClient.Builder() + .addInterceptor(chain -> { + networkCalls.incrementAndGet(); + firstNetworkCall.countDown(); + try { + if (!releaseNetwork.await(2, TimeUnit.SECONDS)) { + throw new IOException("test network release timed out"); + } + } catch (InterruptedException ex) { + Thread.currentThread().interrupt(); + throw new IOException("test network interrupted", ex); + } + return response(chain.request(), 200, "OK", createJpegBytes()); + }) + .build()); + ExecutorService callers = Executors.newFixedThreadPool(2); + CountDownLatch start = new CountDownLatch(1); + try { + Future first = callers.submit(() -> { + start.await(); + return shared.fetchAndResizeForCache("https://images.example.com/shared.jpg"); + }); + Future second = callers.submit(() -> { + start.await(); + return shared.fetchAndResizeForCache("https://images.example.com/shared.jpg"); + }); + + start.countDown(); + assertTrue(firstNetworkCall.await(1, TimeUnit.SECONDS)); + Thread.sleep(100L); + releaseNetwork.countDown(); + + assertNotNull(first.get(2, TimeUnit.SECONDS)); + assertNotNull(second.get(2, TimeUnit.SECONDS)); + assertEquals(1, networkCalls.get()); + } finally { + releaseNetwork.countDown(); + callers.shutdownNow(); + shared.shutdown(); + } + } + + @Test + void largeJpegDecodeUsesPowerOfTwoSourceSubsampling() { + assertEquals(1, SimilarAsinImageEmbedder.jpegSourceSubsampling(2559, 1200)); + assertEquals(2, SimilarAsinImageEmbedder.jpegSourceSubsampling(3200, 2400)); + assertEquals(4, SimilarAsinImageEmbedder.jpegSourceSubsampling(6000, 4000)); } @Test diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java index 99bf6c8c..6255266e 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatServiceTest.java @@ -112,11 +112,12 @@ class TaskHeartbeatServiceTest { task.setId(taskId); task.setModuleType("SHOP_DATA_CRAWL"); task.setStatus("RUNNING"); + TaskHeartbeatRequest request = new TaskHeartbeatRequest(); + request.setModuleType("SHOP_DATA_CRAWL"); when(fileTaskMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(task); - when(brandCrawlTaskMapper.selectOne(any(LambdaQueryWrapper.class))).thenReturn(null); when(fileTaskMapper.update(isNull(), any(LambdaUpdateWrapper.class))).thenReturn(1); - TaskHeartbeatVo result = service.heartbeat(taskId, new TaskHeartbeatRequest()); + TaskHeartbeatVo result = service.heartbeat(taskId, request); assertTrue(result.isAlive()); InOrder order = inOrder(shopDataCrawlTaskService, fileTaskMapper); @@ -125,6 +126,7 @@ class TaskHeartbeatServiceTest { order.verify(fileTaskMapper).update(isNull(), any(LambdaUpdateWrapper.class)); verify(shopDataCrawlTaskCacheService).touchTaskHeartbeat(taskId); verify(shopDataCrawlTaskCacheService).saveTaskCache(task); + verify(brandCrawlTaskMapper, never()).selectOne(any(LambdaQueryWrapper.class)); } @Test diff --git a/backend/ali_oss.py b/backend/ali_oss.py index b489edfb..8f657c65 100644 --- a/backend/ali_oss.py +++ b/backend/ali_oss.py @@ -7,7 +7,9 @@ import base64 import io import mimetypes +import os import re +import tempfile import threading import time @@ -27,6 +29,8 @@ from config import ( _client = None _client_lock = threading.Lock() +_REMOTE_IMAGE_MAX_BYTES = int(os.getenv("OSS_UPLOAD_IMAGE_MAX_BYTES", str(10 * 1024 * 1024))) +_REMOTE_IMAGE_TIMEOUT = (5, 30) def get_client(): @@ -65,19 +69,72 @@ def get_presigned_url(key: str, expires: int = 7 * 24 * 3600) -> str: -def upload_file(file_content: bytes, key: str): - """上传字节内容到 MinIO，返回可访问链接""" +class _LimitedReader: + """Keep a streamed upload from accepting an unexpectedly huge object.""" + + def __init__(self, source, max_bytes: int): + self._source = source + self._max_bytes = max_bytes + self._read_bytes = 0 + + def read(self, size=-1): + remaining = self._max_bytes - self._read_bytes + if remaining < 0: + raise ValueError("upload exceeds configured size limit") + read_size = remaining + 1 if size is None or size < 0 else min(size, remaining + 1) + data = self._source.read(read_size) + if not data: + return data + self._read_bytes += len(data) + if self._read_bytes > self._max_bytes: + raise ValueError("upload exceeds configured size limit") + return data + + def seek(self, offset, whence=0): + position = self._source.seek(offset, whence) + self._read_bytes = max(0, position) + return position + + def tell(self): + return self._source.tell() + + +def upload_fileobj(file_obj, key: str, max_bytes: int = 0, content_type: str = ""): + """Stream a file-like object to S3/MinIO without materializing it.""" key = key.lstrip("/") + if isinstance(file_obj, (bytes, bytearray)): + file_obj = io.BytesIO(file_obj) + if not hasattr(file_obj, "read"): + raise TypeError("file_obj must be bytes or a readable file-like object") + stream_size = None + try: + file_obj.seek(0) + file_obj.seek(0, os.SEEK_END) + stream_size = file_obj.tell() + file_obj.seek(0) + except (AttributeError, OSError, TypeError, ValueError): + try: + file_obj.seek(0) + except (AttributeError, OSError, TypeError, ValueError): + pass + if max_bytes and max_bytes > 0 and stream_size is not None and stream_size > max_bytes: + raise ValueError("upload exceeds configured size limit") client = get_client() - client.put_object( - Bucket=bucket, # 存储桶名称 - Key=key, # 对象名称 - Body=io.BytesIO(file_content) if isinstance(file_content, (bytes, bytearray)) else file_content, - ContentType=_guess_content_type(key), + body = _LimitedReader(file_obj, max_bytes) if max_bytes and max_bytes > 0 else file_obj + client.upload_fileobj( + body, + bucket, + key, + ExtraArgs={"ContentType": content_type or _guess_content_type(key)}, ) return file_url_pre + key +def upload_file(file_content: bytes, key: str): + """上传字节内容到 MinIO，返回可访问链接""" + return upload_fileobj(file_content, key) + + def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") -> str: """ 将 base64 data URL 上传到对象存储，返回图片链接 @@ -89,14 +146,19 @@ def upload_data_url(data_url: str, prefix: str = "history", key_hint: str = "") if not match: raise ValueError('无效的 data URL 格式') ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' - file_content = base64.b64decode(match.group(2)) + encoded_payload = match.group(2) + if len(encoded_payload) > ((max(_REMOTE_IMAGE_MAX_BYTES, 0) + 2) // 3) * 4: + raise ValueError("image exceeds configured upload size limit") + file_content = base64.b64decode(encoded_payload, validate=True) + if len(file_content) > _REMOTE_IMAGE_MAX_BYTES: + raise ValueError("image exceeds configured upload size limit") ts = int(time.time() * 1000) key = f"{bucket_path}{prefix}/{ts}{key_hint}.{ext}" - return upload_file(file_content, key) + return upload_fileobj(io.BytesIO(file_content), key, _REMOTE_IMAGE_MAX_BYTES) def upload_data_urls(data_urls: list, prefix: str = "history") -> list: - """批量上传 base64 图片到对象存储，返回图片链接列表""" + """批量上传图片；每张图���完成上传后立即释放其缓冲区。""" urls = [] ts = int(time.time() * 1000) for i, data_url in enumerate(data_urls or []): @@ -107,12 +169,37 @@ def upload_data_urls(data_urls: list, prefix: str = "history") -> list: if not match: continue ext = 'png' if match.group(1).lower() in ('png', 'webp') else 'jpg' - file_content = base64.b64decode(match.group(2)) - else: - file_content = requests.get(data_url).content - ext = "png" - key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}" - urls.append(upload_file(file_content, key)) + encoded_payload = match.group(2) + if len(encoded_payload) > ((max(_REMOTE_IMAGE_MAX_BYTES, 0) + 2) // 3) * 4: + raise ValueError("image exceeds configured upload size limit") + file_content = base64.b64decode(encoded_payload, validate=True) + if len(file_content) > _REMOTE_IMAGE_MAX_BYTES: + raise ValueError("image exceeds configured upload size limit") + key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}" + urls.append(upload_fileobj(io.BytesIO(file_content), key, _REMOTE_IMAGE_MAX_BYTES)) + del file_content + continue + + with requests.get(data_url, stream=True, timeout=_REMOTE_IMAGE_TIMEOUT) as response: + response.raise_for_status() + content_length = response.headers.get("Content-Length") + if content_length and int(content_length) > _REMOTE_IMAGE_MAX_BYTES: + raise ValueError("image exceeds configured upload size limit") + content_type = response.headers.get("Content-Type", "").split(";", 1)[0].strip() + extension = mimetypes.guess_extension(content_type) or ".png" + ext = extension.lstrip(".") or "png" + key = f"{bucket_path}{prefix}/{ts}_{i}.{ext}" + with tempfile.SpooledTemporaryFile(max_size=2 * 1024 * 1024, mode="w+b") as image_file: + total = 0 + for chunk in response.iter_content(chunk_size=64 * 1024): + if not chunk: + continue + total += len(chunk) + if total > _REMOTE_IMAGE_MAX_BYTES: + raise ValueError("image exceeds configured upload size limit") + image_file.write(chunk) + image_file.seek(0) + urls.append(upload_fileobj(image_file, key, _REMOTE_IMAGE_MAX_BYTES, content_type)) return urls diff --git a/backend/blueprints/admin_api.py b/backend/blueprints/admin_api.py index e7192719..6ee436c3 100644 --- a/backend/blueprints/admin_api.py +++ b/backend/blueprints/admin_api.py @@ -34,7 +34,7 @@ from werkzeug.security import generate_password_hash from utils.db import get_db from utils.auth import admin_required, login_required, get_current_admin_role -from ali_oss import upload_file as oss_upload_file +from ali_oss import upload_fileobj as oss_upload_fileobj try: from config import bucket_path, backend_java_base_url @@ -47,6 +47,10 @@ _backend_java_session_local = threading.local() _internal_token_lock = threading.Lock() IMAGE_VIDEO_DATA_PERMISSION_KEY = 'admin_image_video_task_data' SHOP_DATA_CRAWL_DATA_PERMISSION_KEY = 'admin_shop_data_crawl_task_data' +try: + VERSION_UPLOAD_MAX_BYTES = int(os.environ.get('VERSION_UPLOAD_MAX_BYTES', str(512 * 1024 * 1024))) +except ValueError: + VERSION_UPLOAD_MAX_BYTES = 512 * 1024 * 1024 ADMIN_MENU_ACCESS_CONFIG = { 'dedupe-total-data': { @@ -2723,13 +2727,28 @@ def upload_version(): return jsonify({'success': False, 'error': '请选择要上传的 zip 压缩包'}) if not (file_storage.filename or '').lower().endswith('.zip'): return jsonify({'success': False, 'error': '仅支持 .zip 格式'}) + conn = None try: - file_content = file_storage.read() - if not file_content: + if file_storage.content_length and file_storage.content_length > VERSION_UPLOAD_MAX_BYTES: + return jsonify({'success': False, 'error': '文件超过允许的大小限制'}) + file_stream = file_storage.stream + stream_size = file_storage.content_length + try: + file_stream.seek(0) + file_stream.seek(0, os.SEEK_END) + stream_size = file_stream.tell() + file_stream.seek(0) + except (AttributeError, OSError, TypeError, ValueError): + try: + file_stream.seek(0) + except (AttributeError, OSError, TypeError, ValueError): + pass + if stream_size == 0: return jsonify({'success': False, 'error': '文件为空'}) safe_key = _safe_version_key(version) key = f"{bucket_path}versions/{safe_key}.zip" - file_url = oss_upload_file(file_content, key) + file_url = oss_upload_fileobj(file_stream, key, VERSION_UPLOAD_MAX_BYTES, + 'application/zip') conn = get_db() with conn.cursor() as cur: cur.execute( @@ -2737,7 +2756,6 @@ def upload_version(): (version, file_url) ) conn.commit() - conn.close() return jsonify({ 'success': True, 'version': version, @@ -2747,6 +2765,12 @@ def upload_version(): except Exception as e: traceback.print_exc() return jsonify({'success': False, 'error': str(e)}) + finally: + if conn is not None: + try: + conn.close() + except Exception: + pass # ---------- 店铺密钥管理 ---------- @@ -2972,6 +2996,7 @@ def export_dedupe_total_data(): url, params=params, headers={'X-Internal-Token': _resolve_internal_token()}, + stream=True, timeout=60, ) except requests.RequestException: @@ -2982,14 +3007,23 @@ def export_dedupe_total_data(): error = data.get('message') or data.get('error') or '导出失败' except ValueError: error = '导出失败' + resp.close() return jsonify({'success': False, 'error': error}), resp.status_code headers = {} disposition = resp.headers.get('Content-Disposition') if disposition: headers['Content-Disposition'] = disposition + def generate(): + try: + for chunk in resp.iter_content(chunk_size=1024 * 1024): + if chunk: + yield chunk + finally: + resp.close() + return Response( - resp.content, + stream_with_context(generate()), status=resp.status_code, headers=headers, content_type=resp.headers.get( diff --git a/backend/tests/test_admin_dedupe_total_data.py b/backend/tests/test_admin_dedupe_total_data.py index 7a1f0059..a0d34adf 100644 --- a/backend/tests/test_admin_dedupe_total_data.py +++ b/backend/tests/test_admin_dedupe_total_data.py @@ -18,6 +18,12 @@ class _FakeExportResponse: 'Content-Type': 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet', } + def iter_content(self, chunk_size=None): + yield self.content + + def close(self): + return None + class AdminDedupeTotalDataTest(unittest.TestCase): def setUp(self): @@ -138,12 +144,14 @@ class AdminDedupeTotalDataTest(unittest.TestCase): response = admin_api.export_dedupe_total_data.__wrapped__() self.assertEqual(response.status_code, 200) + self.assertEqual(response.get_data(), b'xlsx') self.assertEqual(session.kwargs['params'], { 'operatorId': 7, 'username': 'operator', 'groupId': 3, }) self.assertEqual(session.kwargs['headers'], {'X-Internal-Token': 'token'}) + self.assertTrue(session.kwargs['stream']) def test_import_requires_group(self): with self.app.test_request_context(