后台更新服务优化

2026-08-11 14:50:21 +08:00
parent 38216a1885
commit 5b1ccad40e
24 changed files with 1623 additions and 502 deletions
@@ -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.
@@ -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;
@@ -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 堆，
@@ -2678,9 +2678,10 @@ public class AppearancePatentTaskService {
            markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze pending submit task missing");
            return;
        }
-        Map<String, AppearancePatentResultRowDto> currentRows = loadPersistedResultRows(task.getId());
+        Map<String, AppearancePatentResultRowDto> currentRows = loadPersistedResultRowsForResultRows(
+                task.getId(), batchRows);
        List<AppearancePatentResultRowDto> currentBatchRows = batchRows.stream()
-                .map(row -> currentRows.get(rowKey(row)))
+                .map(row -> findPersistedResultRow(row, currentRows))
                .filter(Objects::nonNull)
                .toList();
        if (currentBatchRows.size() == batchRows.size()
@@ -3006,24 +3007,40 @@ public class AppearancePatentTaskService {

    private void assembleResultWorkbook(FileTaskEntity task, FileResultEntity result) {
        AppearancePatentParsedPayloadDto parsed = readParsedPayload(task);
-        Map<String, AppearancePatentResultRowDto> resultMap = loadPersistedResultRowsWithRetry(task.getId(), parsed.getAllItems().size());
-        List<AppearancePatentParsedRowVo> 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<AppearancePatentParsedRowVo> parsedRows = parsed.getAllItems() == null
+                ? List.of() : parsed.getAllItems();
+        List<SourceRows> sourceRows = splitRowsBySourceFile(parsed, parsedRows, result.getSourceFilename());
+        if (sourceRows.isEmpty()) {
            throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件");
        }
+        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<SourceResultWorkbook> workbooks = new ArrayList<>();
+        File zip = null;
+        try {
+            for (SourceRows item : sourceRows) {
+                Map<String, AppearancePatentResultRowDto> resultMap =
+                        loadPersistedResultRowsForRowsWithRetry(task.getId(), item.rows());
+                List<AppearancePatentParsedRowVo> 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);
-        // 公共分组处理：同 baseId 连续行视为一组（如 1、1_1、1_2），组内任一行结论命中
-        // "已侵权"/"侵权"（包含匹配，"已侵权"优先），则组内所有行结论统一为该标准值。
-        // 单行组跳过。仅修改 conclusion 列，不影响其他列。
-        int conclusionPropagated = CozeGroupResultPropagator.propagateByGroup(
+                conclusionPropagated += CozeGroupResultPropagator.propagateByGroup(
                        receivedRows,
                        AppearancePatentParsedRowVo::getDisplayId,
                        row -> findResultRow(row, resultMap),
@@ -3032,19 +3049,6 @@ public class AppearancePatentTaskService {
                        List.of("已侵权", "侵权"),
                        "[appearance-patent] taskId=" + task.getId()
                );
-        if (conclusionPropagated > 0) {
-            log.info("[appearance-patent] group propagate conclusion taskId={} updatedRows={}",
-                    task.getId(), conclusionPropagated);
-        }
-        File outputDir = new File(storageProperties.getLocalTempDir(), "appearance-patent-result");
-        if (!outputDir.exists() && !outputDir.mkdirs()) {
-            throw new BusinessException("创建结果目录失败");
-        }
-        List<SourceRows> sourceRows = splitRowsBySourceFile(parsed, receivedRows, result.getSourceFilename());
-        List<SourceResultWorkbook> workbooks = new ArrayList<>();
-        File zip = null;
-        try {
-            for (SourceRows item : sourceRows) {
                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<String, AppearancePatentResultRowDto> loadPersistedResultRows(Long taskId) {
-        Map<String, AppearancePatentResultRowDto> result = new LinkedHashMap<>();
-        List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
-                .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<String, AppearancePatentResultRowDto> loadPersistedResultRowsWithRetry(Long taskId, int expectedParsedRows) {
-        Map<String, AppearancePatentResultRowDto> result = loadPersistedResultRows(taskId);
-        if (expectedParsedRows <= 0 || !result.isEmpty()) {
+    private Map<String, AppearancePatentResultRowDto> loadPersistedResultRowsForRowsWithRetry(
+            Long taskId, List<AppearancePatentParsedRowVo> requestedRows) {
+        Set<String> requestedKeys = resultLookupKeysForParsedRows(requestedRows);
+        Map<String, AppearancePatentResultRowDto> 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<String, AppearancePatentResultRowDto> loadPersistedResultRowsForResultRows(
+            Long taskId, List<AppearancePatentResultRowDto> requestedRows) {
+        Set<String> requestedKeys = resultLookupKeysForResultRows(requestedRows);
+        return loadPersistedResultRowsForKeys(taskId, requestedKeys);
+    }
+
+    private Map<String, AppearancePatentResultRowDto> loadPersistedResultRowsForKeys(
+            Long taskId, Set<String> requestedKeys) {
+        Map<String, AppearancePatentResultRowDto> result = new LinkedHashMap<>();
+        if (taskId == null || requestedKeys == null || requestedKeys.isEmpty()) {
+            return result;
+        }
+        Long lastChunkId = null;
+        while (true) {
+            LambdaQueryWrapper<TaskChunkEntity> query = new LambdaQueryWrapper<TaskChunkEntity>()
+                    .eq(TaskChunkEntity::getTaskId, taskId)
+                    .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
+                    .orderByAsc(TaskChunkEntity::getId)
+                    .last("LIMIT 1");
+            if (lastChunkId != null) {
+                query.gt(TaskChunkEntity::getId, lastChunkId);
+            }
+            List<TaskChunkEntity> 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<String> resultLookupKeysForParsedRows(List<AppearancePatentParsedRowVo> rows) {
+        Set<String> keys = new LinkedHashSet<>();
+        for (AppearancePatentParsedRowVo row : rows == null ? List.<AppearancePatentParsedRowVo>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<String> resultLookupKeysForResultRows(List<AppearancePatentResultRowDto> rows) {
+        Set<String> keys = new LinkedHashSet<>();
+        for (AppearancePatentResultRowDto row : rows == null ? List.<AppearancePatentResultRowDto>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<String, AppearancePatentResultRowDto> sink,
+                                           Iterable<AppearancePatentResultRowDto> rows,
+                                           Set<String> 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<String, AppearancePatentResultRowDto> 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);
@@ -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<byte[]> export(
+    public ResponseEntity<StreamingResponseBody> 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
@@ -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<DedupeTotalDataEntity>
    @SelectProvider(type = SqlProvider.class, method = "selectExistingDataValues")
    List<String> selectExistingDataValues(@Param("values") List<String> 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("""
+            <script>
+            INSERT IGNORE INTO biz_dedupe_total_data
+                (data_value, group_id, uploader_user_id, uploader_username)
+            VALUES
+            <foreach collection="rows" item="row" separator=",">
+                (#{row.dataValue}, #{row.groupId}, #{row.uploaderUserId}, #{row.uploaderUsername})
+            </foreach>
+            </script>
+            """)
+    int insertBatchIgnore(@Param("rows") List<DedupeTotalDataEntity> rows);
+
    class SqlProvider {
        public String selectExistingDataValues(@Param("values") List<String> values) {
            StringJoiner placeholders = new StringJoiner(", ");
@@ -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<DedupeTotalDataEntity> query = new LambdaQueryWrapper<DedupeTotalDataEntity>()
-                .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<DedupeTotalDataEntity> 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<DedupeTotalDataEntity> rows, Map<Long, String> 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,9 +171,22 @@ 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);
+            Long lastId = null;
+            int rowIndex = 1;
+            while (true) {
+                LambdaQueryWrapper<DedupeTotalDataEntity> pageQuery = buildExportQuery(
+                        username, startDate, endDate, groupId, scope);
+                if (lastId != null) {
+                    pageQuery.lt(DedupeTotalDataEntity::getId, lastId);
+                }
+                List<DedupeTotalDataEntity> rows = dedupeTotalDataMapper.selectList(
+                        pageQuery.last("LIMIT " + EXPORT_PAGE_SIZE));
+                if (rows == null || rows.isEmpty()) {
+                    break;
+                }
+                Map<Long, String> 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());
@@ -163,6 +195,15 @@ public class DedupeTotalDataService {
                            : 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);
            sheet.setColumnWidth(2, 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<DedupeTotalDataEntity> buildExportQuery(String username,
+                                                                        LocalDate startDate,
+                                                                        LocalDate endDate,
+                                                                        Long groupId,
+                                                                        AccessScope scope) {
+        String safeUsername = username == null ? "" : username.trim();
+        LambdaQueryWrapper<DedupeTotalDataEntity> query = new LambdaQueryWrapper<DedupeTotalDataEntity>()
+                .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<String> seenInFile = new HashSet<>();
+            List<String> 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);
+                pendingValues.add(asin);
+                if (pendingValues.size() >= IMPORT_BATCH_SIZE) {
+                    ImportBatchResult batch = insertImportBatch(
+                            pendingValues, groupId, uploaderUserId, uploaderUsername);
+                    insertedCount += batch.insertedCount();
+                    skippedCount += batch.skippedCount();
+                    pendingValues.clear();
                }
-                    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);
+                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<String> values,
+                                                Long groupId,
+                                                Long uploaderUserId,
+                                                String uploaderUsername) {
+        if (values == null || values.isEmpty()) {
+            return new ImportBatchResult(0, 0);
+        }
+        Set<String> existingValues = findExistingComparableValues(values);
+        List<DedupeTotalDataEntity> 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> T executeInNewTransaction(Supplier<T> 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) {
@@ -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<String> 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<String> response = httpClient().send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
-            String responseBody = response.body() == null ? "" : response.body();
+            HttpResponse<InputStream> 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<String> response = httpClient().send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
-            String responseBody = response.body() == null ? "" : response.body();
+            HttpResponse<InputStream> 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) {
@@ -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,9 +1600,6 @@ public class ShopDataCrawlTaskService {
        }
        Map<Long, ShopDataCrawlResultItemVo> snapshotsByResultId = indexSnapshotByResultId(successItems);
        LocalDate businessDate = dailyFileService.currentBusinessDate();
-        List<String> uploadedObjectKeys = new ArrayList<>();
-        List<String> obsoleteObjectKeys = new ArrayList<>();
-        try {
        for (FileResultEntity row : rows) {
            if (!Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) {
                continue;
@@ -1610,25 +1611,24 @@ public class ShopDataCrawlTaskService {
            if (snapshot == null) {
                throw new BusinessException("店铺结果快照不存在，无法生成累计文件");
            }
-                DailyAggregationResult result = aggregateDailyResult(
-                        task, row, snapshot, businessDate, uploadedObjectKeys);
-                obsoleteObjectKeys.addAll(result.obsoleteObjectKeys());
+            aggregateDailyResult(task, row, snapshot, businessDate);
        }
+
+        // Keep task status persistence short-lived. Workbook/OSS work and the
+        // snapshot payload upload are completed outside this transaction.
        updateTaskStatusFromRows(task, rows);
-            persistSnapshotJson(task, buildSnapshotFromDb(task, rows));
+        List<ShopDataCrawlResultItemVo> finalSnapshots = buildSnapshotFromDb(task, rows);
+        persistSnapshotJson(task, finalSnapshots);
+        executeShortTransaction(() -> {
            fileTaskMapper.updateById(task);
-        } catch (RuntimeException ex) {
-            registerRollbackObjectCleanup(uploadedObjectKeys);
-            throw ex;
-        }
-        registerDailyObjectLifecycle(uploadedObjectKeys, obsoleteObjectKeys);
+            return null;
+        });
    }

    private DailyAggregationResult aggregateDailyResult(FileTaskEntity task,
                                                         FileResultEntity row,
                                                         ShopDataCrawlResultItemVo snapshot,
-                                                         LocalDate businessDate,
-                                                         List<String> uploadedObjectKeys) {
+                                                         LocalDate businessDate) {
        Long userId = row.getUserId() != null ? row.getUserId() : task.getUserId();
        String shopKey = dailyFileService.shopKey(row);
        String shopKeyHash = dailyFileService.shopKeyHash(shopKey);
@@ -1639,46 +1639,73 @@ 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<ShopDataCrawlDailyFileEntity> olderFiles = dailyFileService.findOlder(userId, shopKeyHash, businessDate);
-            Set<String> 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));
+            DailyWorkbookArtifact artifact = assembleDailyWorkbook(
+                    task, snapshot, preparation.dailyFile(), addedRowCount);
+            try {
+                persistedResult = executeShortTransaction(() -> persistDailyAggregation(
+                        task, row, userId, shopKey, shopKeyHash, businessDate,
+                        preparation, artifact, addedRowCount));
+                if (persistedResult.discardUploadedObject() && artifact.uploaded()) {
+                    deleteObjectQuietly(artifact.objectKey());
+                }
+                return persistedResult;
+            } catch (RuntimeException ex) {
+                if (artifact.uploaded()) {
+                    deleteObjectQuietly(artifact.objectKey());
+                }
+                throw ex;
+            }
+        } finally {
+            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",
@@ -1686,38 +1713,62 @@ public class ShopDataCrawlTaskService {
                "daily-" + UUID.randomUUID()));
        File baseXlsx = FileUtil.file(workRoot, "base.xlsx");
        File outputXlsx = FileUtil.file(workRoot, filename);
-            String objectKey = null;
        try {
-                if (dailyFile != null && !blank(dailyFile.getResultFileUrl())) {
+            if (dailyFile != null && !blank(existingObjectKey)) {
                try {
-                        Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(dailyFile.getResultFileUrl()));
+                    Files.write(baseXlsx.toPath(), ossStorageService.readObjectBytes(existingObjectKey));
                } 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));
            }
-                if (dailyFile != null && addedRowCount == 0) {
-                    objectKey = dailyFile.getResultFileUrl();
-                } else {
-                    objectKey = ossStorageService.uploadResultFile(outputXlsx, MODULE_TYPE);
-                    uploadedObjectKeys.add(objectKey);
+            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<ShopDataCrawlDailyFileEntity> olderFiles = dailyFileService.findOlder(
+                userId, shopKeyHash, businessDate);
+        Set<String> 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<FileResultEntity> shopRows = findShopResultRows(userId, row);
        clearShopResultPointers(shopRows, row.getId());
        row.setResultFilename(filename);
        row.setResultFileUrl(objectKey);
-                row.setResultFileSize(outputXlsx.length());
+        row.setResultFileSize(artifact.fileSize());
        row.setResultContentType(CONTENT_TYPE_XLSX);
        row.setRowCount(dailyFile == null
                ? addedRowCount
@@ -1758,15 +1809,48 @@ public class ShopDataCrawlTaskService {
            dailyFileService.deleteDailyFile(older.getId());
        }
        obsoleteObjectKeys.remove(objectKey);
-                return new DailyAggregationResult(new ArrayList<>(obsoleteObjectKeys));
-            } finally {
-                FileUtil.del(baseXlsx);
-                FileUtil.del(outputXlsx);
-                FileUtil.del(workRoot);
+        return new DailyAggregationResult(new ArrayList<>(obsoleteObjectKeys), false);
    }
-        } finally {
-            releaseDailyLockAfterTransaction(lock);
+
+    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) {
@@ -1795,6 +1879,17 @@ public class ShopDataCrawlTaskService {
        fileResultMapper.updateById(row);
    }

+    private <T> T executeShortTransaction(Supplier<T> 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<FileResultEntity> 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<String> uploadedObjectKeys, List<String> obsoleteObjectKeys) {
-        Set<String> uploaded = uploadedObjectKeys == null ? Set.of() : new HashSet<>(uploadedObjectKeys);
-        Set<String> 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<String> 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<String> obsoleteObjectKeys) {
+    private record DailyAggregationPreparation(ShopDataCrawlDailyFileEntity dailyFile,
+                                               boolean alreadyArchived) {
+    }
+
+    private record DailyWorkbookArtifact(String objectKey,
+                                         long fileSize,
+                                         boolean uploaded,
+                                         String filename) {
+    }
+
+    private record DailyAggregationResult(List<String> obsoleteObjectKeys,
+                                          boolean discardUploadedObject) {
    }

    public void cleanupResultFileJob(TaskFileJobEntity job) {
@@ -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;
@@ -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<TaskScopeStateEntity> candidates = new ArrayList<>(stateIds.size());
-        for (Long stateId : stateIds) {
-            if (stateId == null) {
-                continue;
-            }
-            TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
+        List<TaskScopeStateEntity> refreshedStates = taskScopeStateMapper.selectList(
+                new LambdaQueryWrapper<TaskScopeStateEntity>()
+                        .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.<TaskScopeStateEntity>of()
+                : refreshedStates) {
            if (state == null
                    || state.getCozeExecuteId() == null
                    || state.getCozeExecuteId().isBlank()) {
@@ -1953,13 +1980,8 @@ 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<Long, Future<SimilarAsinCozeClient.CozePollResponse>> futures = new LinkedHashMap<>();
-        try {
        long prefetchStart = System.currentTimeMillis();
        for (TaskScopeStateEntity state : candidates) {
            final Long stateId = state.getId();
@@ -1969,14 +1991,8 @@ public class SimilarAsinTaskService {
                continue;
            }
            final String credentialName = context.credentialName();
-                Future<SimilarAsinCozeClient.CozePollResponse> future = prefetchPool.submit(() -> {
-                    concurrency.acquire();
-                    try {
-                        return cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName));
-                    } finally {
-                        concurrency.release();
-                    }
-                });
+            Future<SimilarAsinCozeClient.CozePollResponse> future = cozePollPrefetchExecutor.submit(
+                        () -> cozeClient.pollWorkflow(executeId, cozeClient.credentialByName(credentialName)));
            futures.put(stateId, future);
        }
        for (Map.Entry<Long, Future<SimilarAsinCozeClient.CozePollResponse>> entry : futures.entrySet()) {
@@ -2000,9 +2016,6 @@ public class SimilarAsinTaskService {
        log.info("[similar-asin] coze poll prefetch finished candidates={} cached={} costMs={}",
                candidates.size(), cache.cozePollResponseCache.size(),
                System.currentTimeMillis() - prefetchStart);
-        } finally {
-            prefetchPool.shutdownNow();
-        }
    }

    /**
@@ -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> 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<SourceResultWorkbook> workbooks = new ArrayList<>();
+        List<File> 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<Future<SourceResultWorkbook>> 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<String, SimilarAsinResultRowDto> 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<SourceResultWorkbook> 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<String, SimilarAsinResultRowDto> 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<String, SimilarAsinResultRowDto> loadPersistedResultRowsForRowsWithRetry(
+            Long taskId,
+            List<SimilarAsinParsedRowVo> requestedRows) {
+        Map<String, SimilarAsinResultRowDto> 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<String, SimilarAsinResultRowDto> loadPersistedResultRowsForRows(
+            Long taskId,
+            List<SimilarAsinParsedRowVo> requestedRows) {
+        Map<String, SimilarAsinResultRowDto> result = new LinkedHashMap<>();
+        Set<String> requestedKeys = resultLookupKeys(requestedRows);
+        if (taskId == null || requestedKeys.isEmpty()) {
+            return result;
+        }
+
+        Long lastChunkId = null;
+        while (true) {
+            LambdaQueryWrapper<TaskChunkEntity> query = new LambdaQueryWrapper<TaskChunkEntity>()
+                    .eq(TaskChunkEntity::getTaskId, taskId)
+                    .eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
+                    .orderByAsc(TaskChunkEntity::getId)
+                    .last("LIMIT 1");
+            if (lastChunkId != null) {
+                query.gt(TaskChunkEntity::getId, lastChunkId);
+            }
+            List<TaskChunkEntity> 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<String> resultLookupKeys(List<SimilarAsinParsedRowVo> rows) {
+        Set<String> keys = new LinkedHashSet<>();
+        for (SimilarAsinParsedRowVo row : rows == null ? List.<SimilarAsinParsedRowVo>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<String, SimilarAsinResultRowDto> sink,
+                                           Iterable<SimilarAsinResultRowDto> rows,
+                                           Set<String> 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<String> requestedKeys,
+                                           Map<String, SimilarAsinResultRowDto> sink) {
+        Long lastStateId = null;
+        while (true) {
+            LambdaQueryWrapper<TaskScopeStateEntity> query = new LambdaQueryWrapper<TaskScopeStateEntity>()
+                    .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<TaskScopeStateEntity> 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<SimilarAsinResultRowDto> 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<String, SimilarAsinResultRowDto> loadPersistedResultRows(Long taskId) {
        Map<String, SimilarAsinResultRowDto> result = new LinkedHashMap<>();
        List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
@@ -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,
@@ -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<String, CompletableFuture<ResizedImage>> 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<ResizedImage> owned = new CompletableFuture<>();
+        CompletableFuture<ResizedImage> 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<String> 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<ResizedImage> 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,21 +886,19 @@ public class SimilarAsinImageEmbedder {
     * 仍然超限才抛 ResizeOversizeException 触发文本兜底。
     */
    ResizedImage resizeImage(String sourceUrl, byte[] raw) throws IOException {
+        acquireImageProcessingSlot();
+        try {
            ensureImageWorkNotInterrupted();
-        guardImageDimensions(sourceUrl, raw);
-        BufferedImage src = ImageIO.read(new ByteArrayInputStream(raw));
-        if (src == null) {
-            throw new IOException("unsupported image format url=" + sourceUrl);
-        }
+            BufferedImage src = decodeForResize(sourceUrl, raw);
+            try {
                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) {
                    ensureImageWorkNotInterrupted();
                    BufferedImage scaled = scaleAt(src, srcW, srcH, longEdge);
+                    try {
                        for (float quality : FALLBACK_QUALITIES) {
                            ensureImageWorkNotInterrupted();
                            ResizedImage tried = encodeJpeg(scaled, quality);
@@ -851,6 +910,9 @@ public class SimilarAsinImageEmbedder {
                                break;
                            }
                        }
+                    } finally {
+                        scaled.flush();
+                    }
                    if (candidate != null) {
                        break;
                    }
@@ -858,10 +920,71 @@ public class SimilarAsinImageEmbedder {
                if (candidate != null) {
                    return candidate;
                }
-        // 9 组合都没压到上限：抛 ResizeOversizeException 走文本兜底，
-        // 同时报告 smallest 字节让运维直观知道当前压缩极限。
                int reportedSize = smallest != null ? smallest.bytes().length : -1;
                throw new ResizeOversizeException(sourceUrl, reportedSize);
+            } finally {
+                src.flush();
+            }
+        } 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<ImageReader> 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();
+            }
+        }
+    }
+
+    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<ImageReader> 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 -> {
@@ -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;

@@ -78,15 +78,15 @@ public class TaskHeartbeatService {
            return TaskHeartbeatVo.notAlive(null, null, "invalid taskId");
        }

-        LambdaQueryWrapper<FileTaskEntity> fileQuery = new LambdaQueryWrapper<FileTaskEntity>()
-                .eq(FileTaskEntity::getId, taskId)
-                .last("limit 1");
-        FileTaskEntity fileTask = fileTaskMapper.selectOne(fileQuery);
-
-        LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
-                .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<FileTaskEntity> fileQuery = new LambdaQueryWrapper<FileTaskEntity>()
+                .eq(FileTaskEntity::getId, taskId)
+                .last("limit 1");
+        return fileTaskMapper.selectOne(fileQuery);
+    }
+
+    private BrandCrawlTaskEntity selectBrandTask(Long taskId) {
+        LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
+                .eq(BrandCrawlTaskEntity::getId, taskId)
+                .last("limit 1");
+        return brandCrawlTaskMapper.selectOne(brandQuery);
+    }
+
    private TaskHeartbeatVo touchFileTaskIfRunning(FileTaskEntity task, TaskHeartbeatRequest request) {
        if (task == null) {
            return null;
@@ -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<String, Holder> 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 <T> Optional<T> get(String cacheType, String cacheKey, Class<T> 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<Map.Entry<String, Holder>> 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 不能为空");
@@ -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}
@@ -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
@@ -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
@@ -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);

        verify(excelAssemblyService).appendWorkbook(any(), any(), eq(List.of(snapshot)));
        verify(dailyFileService).update(daily);
-            verify(ossStorageService, never()).deleteObject("result/old.xlsx");
+        verify(ossStorageService).deleteObject("result/old.xlsx");
        assertNull(previous.getResultFileUrl());
        assertEquals("result/new.xlsx", currentRow.getResultFileUrl());
        assertEquals(3, currentRow.getRowCount());
-
-            List<TransactionSynchronization> 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");
@@ -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<SimilarAsinImageEmbedder.ResizedImage> first = callers.submit(() -> {
+                start.await();
+                return shared.fetchAndResizeForCache("https://images.example.com/shared.jpg");
+            });
+            Future<SimilarAsinImageEmbedder.ResizedImage> 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
@@ -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
@@ -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"
+            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_file(file_content, key))
+            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


@@ -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(
@@ -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(