From b4b80cb5727cf35ee016ea9879c9dff08d327bb1 Mon Sep 17 00:00:00 2001 From: super <2903208875@qq.com> Date: Fri, 22 May 2026 09:41:31 +0800 Subject: [PATCH] =?UTF-8?q?=E6=9B=B4=E6=96=B0=E8=BF=99=E4=B8=AA=E8=B4=A7?= =?UTF-8?q?=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 1 + backend-java/pom.xml | 1 + .../common/exception/BusinessException.java | 10 + .../util/CozeGroupResultPropagator.java | 216 ++++++++ .../aiimage/config/SimilarAsinProperties.java | 4 +- .../config/TransientStorageProperties.java | 1 + .../service/AppearancePatentTaskService.java | 21 + .../brand/service/BrandTaskService.java | 2 - .../service/CozeCredentialPoolService.java | 13 +- .../service/DedupeTotalDataService.java | 35 +- .../service/DeleteBrandRunService.java | 39 +- .../object/RustfsObjectStorageService.java | 24 +- .../service/PatrolDeleteResolveService.java | 25 +- .../service/QueryAsinResolveService.java | 25 +- .../client/SimilarAsinCozeClient.java | 25 +- .../model/dto/SimilarAsinResultRowDto.java | 12 + .../service/SimilarAsinTaskService.java | 71 ++- .../util/SimilarAsinImageEmbedder.java | 464 ++++++++++++++++++ .../task/service/TaskFileJobService.java | 18 + .../task/service/TaskResultFileJobWorker.java | 22 + .../src/main/resources/application.yml | 6 +- .../util/SimilarAsinImageEmbedderTest.java | 153 ++++++ 22 files changed, 1136 insertions(+), 52 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java diff --git a/.gitignore b/.gitignore index 80e257c..953c124 100644 --- a/.gitignore +++ b/.gitignore @@ -108,3 +108,4 @@ xlsx/ OPS_REDIS_MYSQL_OPTIMIZATION_NOTES.md 架构.md *ts.% +.omc \ No newline at end of file diff --git a/backend-java/pom.xml b/backend-java/pom.xml index 0c4fa85..2420199 100644 --- a/backend-java/pom.xml +++ b/backend-java/pom.xml @@ -138,6 +138,7 @@ ${java.version} ${java.version} + true org.projectlombok diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessException.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessException.java index a4d039a..589afca 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessException.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/BusinessException.java @@ -8,11 +8,21 @@ public class BusinessException extends RuntimeException { this(null, message); } + public BusinessException(String message, Throwable cause) { + super(message, cause); + this.code = null; + } + public BusinessException(Integer code, String message) { super(message); this.code = code; } + public BusinessException(Integer code, String message, Throwable cause) { + super(message, cause); + this.code = code; + } + public Integer getCode() { return code; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java b/backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java new file mode 100644 index 0000000..11a2cf3 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/common/util/CozeGroupResultPropagator.java @@ -0,0 +1,216 @@ +package com.nanri.aiimage.common.util; + +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.List; +import java.util.function.BiConsumer; +import java.util.function.Function; + +/** + * Coze 回流数据按 ID 分组传播工具。 + * + *

业务背景： + * 解析行按 Excel 行顺序排列，ID 形如 "1"、"1_1"、"1_2"、"2"、"2_1"。 + * 同一 baseId（第一个下划线 '_' 之前的部分）的连续行视为同一组——即所谓 + * "分组只看下一行数据"：按行顺序、相邻同 baseId 行才是同组，遇到不同 baseId 立即开新组。 + * + *

传播规则： + *

    + *
  1. 单行组（孤立 baseId）跳过，不做任何处理。
    2. + *
  2. 组内任一行的目标字段值"包含" {@code hitValues} 中某个候选值， + * 则把该组所有行的目标字段统一覆盖为该候选值（标准值）。
    3. + *
  3. 多个候选值同时命中时，按 {@code hitValues} 列表顺序优先匹配 + * （列表越靠前优先级越高，比如专利场景把"已侵权"放在"侵权"前面， + * 避免短串先命中导致丢失"已"字）。
    4. + *
  4. 无法解析到对应 result 行（resolver 返回 null）的 parsed 行不参与 + * 组内匹配，但仍占据分组位置——不会打断同 baseId 的连续性， + * 也不会被强行写入。
    5. + *
+ * + *

使用方式： + * 

+ *  // 专利结论列：组内任一行结论命中 "已侵权" 或 "侵权"，组内都改为该标准值
+ *  CozeGroupResultPropagator.propagateByGroup(
+ *      receivedRows,
+ *      AppearancePatentParsedRowVo::getDisplayId,
+ *      row -> findResultRow(row, resultMap),
+ *      AppearancePatentResultRowDto::getConclusion,
+ *      AppearancePatentResultRowDto::setConclusion,
+ *      List.of("已侵权", "侵权"),
+ *      "[appearance-patent] taskId=" + task.getId()
+ *  );
+ *
+ */ +public final class CozeGroupResultPropagator { + + private static final Logger log = LoggerFactory.getLogger(CozeGroupResultPropagator.class); + + /** + * 否定前缀关键字。若当前值同时包含 standard 和这些关键字之一，则不视为命中。 + * 用于规避"没有侵权""不侵权""未发现明显侵权风险"等被 contains 误判为命中"侵权"的情况。 + */ + private static final List NEGATIVE_KEYWORDS = Arrays.asList("没有", "不", "未"); + + private CozeGroupResultPropagator() { + } + + /** + * 按 baseId 连续分组、命中即组内传播。无日志上下文标签的便捷重载。 + */ + public static int propagateByGroup(List

parsedRows, + Function displayIdGetter, + Function resultRowResolver, + Function fieldGetter, + BiConsumer fieldSetter, + List hitValues) { + return propagateByGroup(parsedRows, displayIdGetter, resultRowResolver, + fieldGetter, fieldSetter, hitValues, null); + } + + /** + * 按 baseId 连续分组、命中即组内传播。 + * + * @param parsedRows 解析行（按 Excel 行顺序） + * @param displayIdGetter parsed 行 → ID 字符串（用于计算 baseId） + * @param resultRowResolver parsed 行 → 对应的 result DTO（找不到返回 null） + * @param fieldGetter result DTO → 当前目标字段值 + * @param fieldSetter result DTO ← 新目标字段值 + * @param hitValues 需要传播的标准值列表，靠前优先匹配（包含语义） + * @param logTag 日志前缀（如 "[appearance-patent] taskId=9975"），null 则只在汇总层打日志 + * @param

parsed 行类型 + * @param result DTO 类型 + * @return 实际被改写的 result 行数（用于上层日志埋点） + */ + public static int propagateByGroup(List

parsedRows, + Function displayIdGetter, + Function resultRowResolver, + Function fieldGetter, + BiConsumer fieldSetter, + List hitValues, + String logTag) { + if (parsedRows == null || parsedRows.isEmpty() + || displayIdGetter == null || resultRowResolver == null + || fieldGetter == null || fieldSetter == null + || hitValues == null || hitValues.isEmpty()) { + return 0; + } + + String tagPrefix = (logTag == null || logTag.isEmpty()) ? "" : logTag + " "; + int totalUpdated = 0; + int n = parsedRows.size(); + int i = 0; + while (i < n) { + String currentBaseId = baseId(displayIdGetter.apply(parsedRows.get(i))); + int groupStart = i; + int j = i + 1; + // 连续同 baseId 视为一组 + while (j < n && currentBaseId.equals(baseId(displayIdGetter.apply(parsedRows.get(j))))) { + j++; + } + int groupEnd = j; // 半开区间 [groupStart, groupEnd) + i = j; + + // 单行组跳过 + if (groupEnd - groupStart < 2) { + continue; + } + + // 收集组内所有能解析到 result DTO 的行，同时记录显示 ID 方便日志 + List groupResults = new ArrayList<>(groupEnd - groupStart); + List groupDisplayIds = new ArrayList<>(groupEnd - groupStart); + for (int k = groupStart; k < groupEnd; k++) { + P parsedRow = parsedRows.get(k); + R resultRow = resultRowResolver.apply(parsedRow); + if (resultRow != null) { + groupResults.add(resultRow); + groupDisplayIds.add(displayIdGetter.apply(parsedRow)); + } + } + if (groupResults.isEmpty()) { + continue; + } + + // 按 hitValues 顺序检测组内是否有命中（包含语义），命中即定下"标准值"。 + // 命中需同时满足：当前值包含 standard 且不含任何否定关键字（"没有"/"不"/"未"）， + // 避免"没有侵权""不侵权""未发现明显侵权风险"这类被 contains 误判为命中。 + String matchedStandardValue = null; + outer: + for (String standard : hitValues) { + if (standard == null || standard.isEmpty()) { + continue; + } + for (int k = 0; k < groupResults.size(); k++) { + R row = groupResults.get(k); + String currentValue = fieldGetter.apply(row); + if (currentValue == null || !currentValue.contains(standard)) { + continue; + } + if (containsNegative(currentValue)) { + log.debug("{}group-propagate skip negative baseId={} displayId={} currentValue={} candidate={}", + tagPrefix, currentBaseId, groupDisplayIds.get(k), currentValue, standard); + continue; + } + matchedStandardValue = standard; + break outer; + } + } + if (matchedStandardValue == null) { + continue; + } + + // 把组内所有 result DTO 的目标字段统一覆盖为标准值，并逐行打日志 + int groupUpdated = 0; + for (int idx = 0; idx < groupResults.size(); idx++) { + R row = groupResults.get(idx); + String currentValue = fieldGetter.apply(row); + if (matchedStandardValue.equals(currentValue)) { + continue; + } + fieldSetter.accept(row, matchedStandardValue); + groupUpdated++; + totalUpdated++; + log.info("{}group-propagate row baseId={} displayId={} oldValue={} newValue={}", + tagPrefix, currentBaseId, groupDisplayIds.get(idx), + currentValue, matchedStandardValue); + } + if (groupUpdated > 0) { + log.info("{}group-propagate group hit baseId={} groupSize={} standardValue={} updatedRows={}", + tagPrefix, currentBaseId, groupResults.size(), + matchedStandardValue, groupUpdated); + } + } + return totalUpdated; + } + + /** + * 判断当前值是否含任意否定关键字。命中关键字即视为"未发生"语义， + * 上层应跳过该值，避免被 {@link String#contains(CharSequence)} 误判命中标准值。 + */ + private static boolean containsNegative(String value) { + if (value == null || value.isEmpty()) { + return false; + } + for (String neg : NEGATIVE_KEYWORDS) { + if (value.contains(neg)) { + return true; + } + } + return false; + } + + /** + * 取 ID 中第一个 '_' 之前的部分。例如 "1" → "1"，"1_2" → "1"，null → ""。 + * 与两个 TaskService 内的 {@code baseId} 实现保持一致。 + */ + private static String baseId(String id) { + if (id == null) { + return ""; + } + String s = id.trim(); + int idx = s.indexOf('_'); + return idx > 0 ? s.substring(0, idx) : s; + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java index 783813f..854f7fe 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/SimilarAsinProperties.java @@ -15,12 +15,12 @@ public class SimilarAsinProperties { private String cozeWorkflowId = "7635328462404583478"; private String cozeToken = ""; private List cozeCredentials = new ArrayList<>(); - private int cozeCredentialStripeSize = 1; + private int cozeCredentialStripeSize = 0; private int cozeBatchSize = 10; private int cozeConnectTimeoutMillis = 10000; private int cozeReadTimeoutMillis = 60000; private int cozePollIntervalMillis = 30000; - private int cozePollTimeoutMillis = 600000; + private int cozePollTimeoutMillis = 1800000; private int staleTimeoutMinutes = 30; private String staleFinalizeCron = "0 */2 * * * *"; diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java index e819631..f1a77d6 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TransientStorageProperties.java @@ -16,4 +16,5 @@ public class TransientStorageProperties { private int readTimeoutSeconds = 60; private int writeTimeoutSeconds = 60; private int uploadMaxRetries = 3; + private int readMaxRetries = 3; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java index af01b8b..2dbcce1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java @@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.common.util.CozeGroupResultPropagator; import com.nanri.aiimage.common.util.FailedStatusRowFilter; import com.nanri.aiimage.config.AppearancePatentProperties; import com.nanri.aiimage.config.InstanceMetadata; @@ -533,6 +534,8 @@ public class AppearancePatentTaskService { taskScopeStateMapper.delete(new LambdaQueryWrapper().eq(TaskScopeStateEntity::getTaskId, taskId).eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)); taskChunkMapper.delete(new LambdaQueryWrapper().eq(TaskChunkEntity::getTaskId, taskId).eq(TaskChunkEntity::getModuleType, MODULE_TYPE)); taskCacheService.deleteTaskCache(taskId); + // 同步清理 task_file_job，避免被删除任务遗留的 PENDING/FAILED 行被 TaskResultFileJobWorker 反复扫描出 task not found。 + taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE); fileTaskMapper.deleteById(taskId); } @@ -541,6 +544,8 @@ public class AppearancePatentTaskService { if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) { throw new BusinessException("记录不存在"); } + // 同步清理 task_file_job，避免被删除历史记录残留的 ASSEMBLE_RESULT job 被反复扫描出 result record not found。 + taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, resultId); fileResultMapper.deleteById(resultId); } @@ -2916,6 +2921,22 @@ public class AppearancePatentTaskService { throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件"); } validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap); + // 公共分组处理：同 baseId 连续行视为一组（如 1、1_1、1_2），组内任一行结论命中 + // "已侵权"/"侵权"（包含匹配，"已侵权"优先），则组内所有行结论统一为该标准值。 + // 单行组跳过。仅修改 conclusion 列，不影响其他列。 + int conclusionPropagated = CozeGroupResultPropagator.propagateByGroup( + receivedRows, + AppearancePatentParsedRowVo::getDisplayId, + row -> findResultRow(row, resultMap), + AppearancePatentResultRowDto::getConclusion, + AppearancePatentResultRowDto::setConclusion, + List.of("已侵权", "侵权"), + "[appearance-patent] taskId=" + task.getId() + ); + if (conclusionPropagated > 0) { + log.info("[appearance-patent] group propagate conclusion taskId={} updatedRows={}", + task.getId(), conclusionPropagated); + } File outputDir = new File(storageProperties.getLocalTempDir(), "appearance-patent-result"); if (!outputDir.exists() && !outputDir.mkdirs()) { throw new BusinessException("创建结果目录失败"); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java index 8bdcc44..16672a2 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java @@ -45,7 +45,6 @@ import org.apache.poi.ss.usermodel.Workbook; import org.apache.poi.ss.usermodel.WorkbookFactory; import org.apache.poi.xssf.streaming.SXSSFWorkbook; import org.springframework.stereotype.Service; -import org.springframework.transaction.annotation.Transactional; import java.io.File; import java.io.FileInputStream; @@ -1208,7 +1207,6 @@ public class BrandTaskService { } } - @Transactional public void failStaleRunningTasks() { DistributedJobLockService.LockHandle lockHandle = distributedJobLockService.tryLock("brand:stale-check", STALE_CHECK_LOCK_TTL); if (lockHandle == null) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java index f79a63c..a9482a1 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/coze/service/CozeCredentialPoolService.java @@ -69,14 +69,15 @@ public class CozeCredentialPoolService { if (credentials == null || credentials.isEmpty()) { return null; } - int safeStripeSize = Math.max(1, stripeSize); - // 自检：当 stripe < credentials.size() 时凭据轮换会偏向首张凭据， - // 通常说明 stripeSize 配置错（推荐 stripe == credentials.size()）。 - // 仅 WARN 一次，避免日志噪音；不阻断启动，留运维自行调整。 - if (safeStripeSize < credentials.size() && moduleType != null && !moduleType.isBlank()) { + // stripeSize ≤ 0 时自动按 credentials.size() 适配；正数则按配置走。 + // 注意：stripe=1 与 stripe=credentials.size() 在轮换正确性上等价，差异只在"每个凭据连续选中次数"。 + int safeStripeSize = stripeSize <= 0 ? credentials.size() : stripeSize; + // 仅当 stripe>1 且小于凭据数时 WARN：此时凭据被切到下一张前会连续选 stripe 次，但未覆盖所有凭据就回到首张，存在偏向。 + if (safeStripeSize > 1 && safeStripeSize < credentials.size() && moduleType != null && !moduleType.isBlank()) { if (stripeWarnedModules.putIfAbsent(moduleType, Boolean.TRUE) == null) { log.warn("[coze-credential] stripeSize({}) < credentials.size({}) for moduleType={}, " - + "round-robin will be biased; consider setting stripeSize == credentials.size()", + + "round-robin may be biased; consider setting stripeSize == credentials.size() " + + "or leave it 0/negative to auto-adapt", safeStripeSize, credentials.size(), moduleType); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java index abcf23b..8687a16 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/dedupe/service/DedupeTotalDataService.java @@ -19,7 +19,10 @@ import org.apache.poi.ss.usermodel.Row; import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.ss.usermodel.Workbook; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import org.springframework.web.multipart.MultipartFile; import java.io.ByteArrayInputStream; @@ -41,9 +44,16 @@ public class DedupeTotalDataService { private static final int COMPARE_BATCH_SIZE = 5000; private final DedupeTotalDataMapper dedupeTotalDataMapper; + private final PlatformTransactionManager transactionManager; private final Map importProgressMap = new ConcurrentHashMap<>(); private final Map deleteImportProgressMap = new ConcurrentHashMap<>(); + private TransactionTemplate newRequiresNewTemplate() { + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return template; + } + public DedupeTotalDataPageVo page(long page, long pageSize, String keyword) { long safePage = Math.max(page, 1); long safePageSize = Math.min(Math.max(pageSize, 1), 100); @@ -281,7 +291,6 @@ public class DedupeTotalDataService { } } - @Transactional public DedupeTotalDataImportVo importFromExcel(MultipartFile file) { if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 文件"); @@ -295,7 +304,6 @@ public class DedupeTotalDataService { } } - @Transactional public DedupeTotalDataImportVo deleteFromExcel(MultipartFile file) { if (file == null || file.isEmpty()) { throw new BusinessException("请上传 xlsx 文件"); @@ -309,7 +317,6 @@ public class DedupeTotalDataService { } } - @Transactional private DedupeTotalDataImportVo importFromExcelInternal(InputStream inputStream, String filename, DedupeTotalDataImportProgressVo progress) { String lowerFilename = filename == null ? "" : filename.toLowerCase(); if (!(lowerFilename.endsWith(".xlsx") || lowerFilename.endsWith(".xls"))) { @@ -397,10 +404,21 @@ public class DedupeTotalDataService { } continue; } - DedupeTotalDataEntity entity = new DedupeTotalDataEntity(); - entity.setDataValue(dataValue); - dedupeTotalDataMapper.insert(entity); - insertedCount++; + final String pendingDataValue = dataValue; + Boolean inserted = newRequiresNewTemplate().execute(status -> { + if (existsDataValue(pendingDataValue)) { + return false; + } + DedupeTotalDataEntity entity = new DedupeTotalDataEntity(); + entity.setDataValue(pendingDataValue); + dedupeTotalDataMapper.insert(entity); + return true; + }); + if (Boolean.TRUE.equals(inserted)) { + insertedCount++; + } else { + skippedCount++; + } if (progress != null) { progress.setProcessedRows(rowNum); progress.setAsinCount(asinCount); @@ -422,7 +440,6 @@ public class DedupeTotalDataService { } } - @Transactional private DedupeTotalDataImportVo deleteFromExcelInternal(InputStream inputStream, String filename, DedupeTotalDataImportProgressVo progress) { String lowerFilename = filename == null ? "" : filename.toLowerCase(); if (!(lowerFilename.endsWith(".xlsx") || lowerFilename.endsWith(".xls"))) { @@ -501,7 +518,7 @@ public class DedupeTotalDataService { continue; } - int deletedThisRow = deleteByDataValue(dataValue); + int deletedThisRow = newRequiresNewTemplate().execute(status -> deleteByDataValue(dataValue)); if (deletedThisRow > 0) { deletedCount += deletedThisRow; } else { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java index f829eee..171af3d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandRunService.java @@ -49,7 +49,10 @@ import org.apache.poi.ss.usermodel.Sheet; import org.apache.poi.ss.usermodel.Workbook; import org.apache.poi.xssf.streaming.SXSSFWorkbook; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.io.File; @@ -88,6 +91,24 @@ public class DeleteBrandRunService { private final TaskPressureProperties taskPressureProperties; private final TaskDistributedLockService taskDistributedLockService; private final TaskFileJobService taskFileJobService; + private final PlatformTransactionManager transactionManager; + + private TransactionTemplate newRequiresNewTemplate() { + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return template; + } + + private T inNewTransaction(java.util.function.Supplier action) { + return newRequiresNewTemplate().execute(status -> action.get()); + } + + private void inNewTransactionVoid(Runnable action) { + newRequiresNewTemplate().execute(status -> { + action.run(); + return null; + }); + } private boolean isUsableIndexedMatch(ZiniaoShopMatchResultVo matchResult) { if (matchResult == null) { @@ -1254,7 +1275,7 @@ public class DeleteBrandRunService { return Math.max(0L, (endExclusive - startInclusive) / 1_000_000L); } - @Transactional + @Transactional(propagation = org.springframework.transaction.annotation.Propagation.NEVER) public void submitResult(Long taskId, DeleteBrandSubmitResultRequest request) { if (taskId == null || taskId <= 0) { throw new BusinessException("taskId invalid"); @@ -1290,11 +1311,13 @@ public class DeleteBrandRunService { } if (!"RUNNING".equals(task.getStatus())) { - task.setStatus("RUNNING"); - task.setErrorMessage(null); - task.setUpdatedAt(LocalDateTime.now()); - task.setFinishedAt(null); - fileTaskMapper.updateById(task); + inNewTransactionVoid(() -> { + task.setStatus("RUNNING"); + task.setErrorMessage(null); + task.setUpdatedAt(LocalDateTime.now()); + task.setFinishedAt(null); + fileTaskMapper.updateById(task); + }); deleteBrandTaskCacheService.saveTaskCache(task); } @@ -1338,7 +1361,9 @@ public class DeleteBrandRunService { if (isFileCompleted(fileDto)) { log.info("[DeleteBrand] File chunk transmission fully reached via chunkTotal for taskId: {}, file: {}", taskId, fileDto.getSourceFilename()); - enqueueCompletedFileAssembleJob(task, fileIdentity, parsedFile); + final String identityForTx = fileIdentity; + final DeleteBrandParsedFileCacheDto parsedForTx = parsedFile; + inNewTransactionVoid(() -> enqueueCompletedFileAssembleJob(task, identityForTx, parsedForTx)); shouldTryFinalize = true; } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java index faf75c9..a2e6b55 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/file/service/object/RustfsObjectStorageService.java @@ -69,14 +69,24 @@ public class RustfsObjectStorageService { if (!isConfigured()) { throw new IllegalStateException("transient storage is not configured"); } - try (var stream = buildClient().getObject(GetObjectArgs.builder() - .bucket(properties.getBucket()) - .object(objectKey) - .build())) { - return new String(stream.readAllBytes(), StandardCharsets.UTF_8); - } catch (Exception ex) { - throw new IllegalStateException("failed to read payload from transient storage", ex); + Exception last = null; + int maxRetries = Math.max(1, properties.getReadMaxRetries()); + for (int attempt = 1; attempt <= maxRetries; attempt++) { + try (var stream = buildClient().getObject(GetObjectArgs.builder() + .bucket(properties.getBucket()) + .object(objectKey) + .build())) { + return new String(stream.readAllBytes(), StandardCharsets.UTF_8); + } catch (Exception ex) { + last = ex; + if (attempt < maxRetries) { + // 与 uploadText 对齐的线性退避，避免 rustfs 短暂抖动直接打成 read coze batch failed。 + log.warn("[rustfs] read failed, retrying objectKey={} attempt={}/{} err={}", objectKey, attempt, maxRetries, ex.getMessage()); + sleepQuietly(500L * attempt); + } + } } + throw new IllegalStateException("failed to read payload from transient storage", last); } public void deleteObject(String objectKey) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteResolveService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteResolveService.java index a23ab0a..1a6eb10 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteResolveService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteResolveService.java @@ -18,7 +18,10 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService; import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; import java.util.ArrayList; @@ -33,6 +36,13 @@ public class PatrolDeleteResolveService { private final PatrolDeleteShopCandidateMapper candidateMapper; private final PatrolDeleteConditionMapper conditionMapper; private final ZiniaoShopSwitchService ziniaoShopSwitchService; + private final PlatformTransactionManager transactionManager; + + private TransactionTemplate newRequiresNewTemplate() { + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return template; + } public List listCandidates(Long userId) { validateUserId(userId); @@ -51,14 +61,13 @@ public class PatrolDeleteResolveService { return list; } - @Transactional public ProductRiskCandidateVo addCandidate(ProductRiskCandidateAddRequest request) { validateUserId(request.getUserId()); String normalized = ziniaoShopSwitchService.normalizeShopName(request.getShopName()); if (normalized.isBlank()) { throw new BusinessException("店铺名不能为空"); } - ZiniaoShopMatchResultVo indexHit = ziniaoShopSwitchService.findIndexedStoreByName(normalized, false); + ZiniaoShopMatchResultVo indexHit = resolveIndexedShop(normalized); if (indexHit == null || !indexHit.isMatched()) { String hint = indexHit != null && indexHit.getMatchMessage() != null && !indexHit.getMatchMessage().isBlank() ? indexHit.getMatchMessage() @@ -68,9 +77,17 @@ public class PatrolDeleteResolveService { if (ZiniaoShopIndexService.MATCH_STATUS_CONFLICT.equals(indexHit.getMatchStatus())) { throw new BusinessException(indexHit.getMatchMessage() != null ? indexHit.getMatchMessage() : "存在多个同名店铺，请人工确认"); } + return newRequiresNewTemplate().execute(status -> persistCandidate(request.getUserId(), normalized)); + } + + private ZiniaoShopMatchResultVo resolveIndexedShop(String normalizedShopName) { + return ziniaoShopSwitchService.findIndexedStoreByName(normalizedShopName, false); + } + + private ProductRiskCandidateVo persistCandidate(Long userId, String normalized) { PatrolDeleteShopCandidateEntity existing = candidateMapper.selectOne( new LambdaQueryWrapper() - .eq(PatrolDeleteShopCandidateEntity::getUserId, request.getUserId()) + .eq(PatrolDeleteShopCandidateEntity::getUserId, userId) .eq(PatrolDeleteShopCandidateEntity::getShopName, normalized) .last("limit 1")); if (existing != null) { @@ -81,7 +98,7 @@ public class PatrolDeleteResolveService { return vo; } PatrolDeleteShopCandidateEntity entity = new PatrolDeleteShopCandidateEntity(); - entity.setUserId(request.getUserId()); + entity.setUserId(userId); entity.setShopName(normalized); entity.setCreatedAt(LocalDateTime.now()); candidateMapper.insert(entity); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java index d57234c..5c04c4a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinResolveService.java @@ -17,7 +17,10 @@ import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopIndexService; import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService; import lombok.RequiredArgsConstructor; import org.springframework.stereotype.Service; +import org.springframework.transaction.PlatformTransactionManager; +import org.springframework.transaction.TransactionDefinition; import org.springframework.transaction.annotation.Transactional; +import org.springframework.transaction.support.TransactionTemplate; import java.time.LocalDateTime; import java.util.ArrayList; @@ -34,6 +37,13 @@ public class QueryAsinResolveService { private final QueryAsinShopCandidateMapper candidateMapper; private final ZiniaoShopSwitchService ziniaoShopSwitchService; private final QueryAsinMapper queryAsinMapper; + private final PlatformTransactionManager transactionManager; + + private TransactionTemplate newRequiresNewTemplate() { + TransactionTemplate template = new TransactionTemplate(transactionManager); + template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW); + return template; + } public List listCandidates(Long userId) { validateUserId(userId); @@ -52,14 +62,13 @@ public class QueryAsinResolveService { return list; } - @Transactional public ProductRiskCandidateVo addCandidate(ProductRiskCandidateAddRequest request) { validateUserId(request.getUserId()); String normalized = ziniaoShopSwitchService.normalizeShopName(request.getShopName()); if (normalized.isBlank()) { throw new BusinessException("店铺名不能为空"); } - ZiniaoShopMatchResultVo indexHit = ziniaoShopSwitchService.findIndexedStoreByName(normalized, false); + ZiniaoShopMatchResultVo indexHit = resolveIndexedShop(normalized); if (indexHit == null || !indexHit.isMatched()) { String hint = indexHit != null && indexHit.getMatchMessage() != null && !indexHit.getMatchMessage().isBlank() ? indexHit.getMatchMessage() @@ -69,9 +78,17 @@ public class QueryAsinResolveService { if (ZiniaoShopIndexService.MATCH_STATUS_CONFLICT.equals(indexHit.getMatchStatus())) { throw new BusinessException(indexHit.getMatchMessage() != null ? indexHit.getMatchMessage() : "存在多个同名店铺，请人工确认"); } + return newRequiresNewTemplate().execute(status -> persistCandidate(request.getUserId(), normalized)); + } + + private ZiniaoShopMatchResultVo resolveIndexedShop(String normalizedShopName) { + return ziniaoShopSwitchService.findIndexedStoreByName(normalizedShopName, false); + } + + private ProductRiskCandidateVo persistCandidate(Long userId, String normalized) { QueryAsinShopCandidateEntity existing = candidateMapper.selectOne( new LambdaQueryWrapper() - .eq(QueryAsinShopCandidateEntity::getUserId, request.getUserId()) + .eq(QueryAsinShopCandidateEntity::getUserId, userId) .eq(QueryAsinShopCandidateEntity::getShopName, normalized) .last("limit 1")); if (existing != null) { @@ -82,7 +99,7 @@ public class QueryAsinResolveService { return vo; } QueryAsinShopCandidateEntity entity = new QueryAsinShopCandidateEntity(); - entity.setUserId(request.getUserId()); + entity.setUserId(userId); entity.setShopName(normalized); entity.setCreatedAt(LocalDateTime.now()); candidateMapper.insert(entity); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java index 3133d8d..cbc73ec 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java @@ -286,7 +286,8 @@ public class SimilarAsinCozeClient { pooledCredential.maxConcurrent()); } List credentials = configuredCredentials(); - int stripeSize = Math.max(1, properties.getCozeCredentialStripeSize()); + int configuredStripe = properties.getCozeCredentialStripeSize(); + int stripeSize = configuredStripe <= 0 ? Math.max(1, credentials.size()) : configuredStripe; long cursor = Math.max(0L, credentialCursor.getAndIncrement()); int index = (int) ((cursor / stripeSize) % credentials.size()); return credentials.get(index); @@ -497,7 +498,16 @@ public class SimilarAsinCozeClient { text(firstNonNull(node.get("patent_reason"), firstNonNull(firstNonNull(node.get("patentReason"), node.get("patent reason")), firstNonNull(itemNode.get("patent_reason"), - firstNonNull(itemNode.get("patentReason"), itemNode.get("patent reason")))))) + firstNonNull(itemNode.get("patentReason"), itemNode.get("patent reason")))))), + text(firstNonNull( + firstNonNull(node.get("main_url"), firstNonNull(node.get("mainUrl"), firstNonNull(node.get("main_image_url"), firstNonNull(node.get("mainImageUrl"), firstNonNull(node.get("main_img"), node.get("mainImg")))))), + firstNonNull(itemNode.get("main_url"), firstNonNull(itemNode.get("mainUrl"), firstNonNull(itemNode.get("main_image_url"), firstNonNull(itemNode.get("mainImageUrl"), firstNonNull(itemNode.get("main_img"), itemNode.get("mainImg")))))))), + text(firstNonNull( + firstNonNull(node.get("puzzle_img1"), firstNonNull(node.get("puzzleImg1"), firstNonNull(node.get("puzzle_img_1"), node.get("puzzleImg_1")))), + firstNonNull(itemNode.get("puzzle_img1"), firstNonNull(itemNode.get("puzzleImg1"), firstNonNull(itemNode.get("puzzle_img_1"), itemNode.get("puzzleImg_1")))))), + text(firstNonNull( + firstNonNull(node.get("puzzle_img2"), firstNonNull(node.get("puzzleImg2"), firstNonNull(node.get("puzzle_img_2"), node.get("puzzleImg_2")))), + firstNonNull(itemNode.get("puzzle_img2"), firstNonNull(itemNode.get("puzzleImg2"), firstNonNull(itemNode.get("puzzle_img_2"), itemNode.get("puzzleImg_2")))))) )); } } @@ -616,6 +626,9 @@ public class SimilarAsinCozeClient { row.setTitleReason(result.titleReason()); row.setAppearanceReason(result.appearanceReason()); row.setPatentReason(result.patentReason()); + row.setMainUrl(result.mainUrl()); + row.setPuzzleImg1(result.puzzleImg1()); + row.setPuzzleImg2(result.puzzleImg2()); normalizeRequiredBusinessFields(row); } @@ -695,6 +708,9 @@ public class SimilarAsinCozeClient { row.setTitleReason(source.getTitleReason()); row.setAppearanceReason(source.getAppearanceReason()); row.setPatentReason(source.getPatentReason()); + row.setMainUrl(source.getMainUrl()); + row.setPuzzleImg1(source.getPuzzleImg1()); + row.setPuzzleImg2(source.getPuzzleImg2()); return row; } @@ -1240,7 +1256,10 @@ public class SimilarAsinCozeClient { String status, String titleReason, String appearanceReason, - String patentReason + String patentReason, + String mainUrl, + String puzzleImg1, + String puzzleImg2 ) { } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java index 01bd1dd..b0fd34e 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/model/dto/SimilarAsinResultRowDto.java @@ -112,6 +112,18 @@ public class SimilarAsinResultRowDto { @Schema(description = "Coze patent reason", accessMode = Schema.AccessMode.READ_ONLY) private String patentReason; + @JsonAlias({"main_url", "mainUrl", "main_image_url", "mainImageUrl", "main_img", "mainImg", "主图URL", "主图链接"}) + @Schema(description = "Coze 回包中的亚马逊主图 URL，最终 xlsx 中以嵌入图片形式呈现", example = "https://m.media-amazon.com/images/I/xxx.jpg", accessMode = Schema.AccessMode.READ_ONLY) + private String mainUrl; + + @JsonAlias({"puzzle_img1", "puzzleImg1", "puzzle_img_1", "puzzleImg_1", "拼图1"}) + @Schema(description = "Coze 回包中阿里巴巴搜图候选 1 的 URL，最终 xlsx 中以嵌入图片形式呈现", example = "https://cbu01.alicdn.com/img/ibank/xxx.jpg", accessMode = Schema.AccessMode.READ_ONLY) + private String puzzleImg1; + + @JsonAlias({"puzzle_img2", "puzzleImg2", "puzzle_img_2", "puzzleImg_2", "拼图2"}) + @Schema(description = "Coze 回包中阿里巴巴搜图候选 2 的 URL，最终 xlsx 中以嵌入图片形式呈现", example = "https://cbu01.alicdn.com/img/ibank/yyy.jpg", accessMode = Schema.AccessMode.READ_ONLY) + private String puzzleImg2; + public String getUrl() { if (url != null && !url.isBlank()) { return url; diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java index 83ab112..7633e5f 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java @@ -8,6 +8,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.common.util.CozeGroupResultPropagator; import com.nanri.aiimage.common.util.FailedStatusRowFilter; import com.nanri.aiimage.config.InstanceMetadata; import com.nanri.aiimage.config.SimilarAsinProperties; @@ -33,6 +34,7 @@ import com.nanri.aiimage.modules.similarasin.model.vo.SimilarAsinParseVo; import com.nanri.aiimage.modules.similarasin.model.vo.SimilarAsinTaskBatchVo; import com.nanri.aiimage.modules.similarasin.model.vo.SimilarAsinTaskDetailVo; import com.nanri.aiimage.modules.similarasin.model.vo.SimilarAsinTaskItemVo; +import com.nanri.aiimage.modules.similarasin.util.SimilarAsinImageEmbedder; import com.nanri.aiimage.modules.file.service.LocalFileStorageService; import com.nanri.aiimage.modules.file.service.oss.OssStorageService; import com.nanri.aiimage.modules.task.mapper.FileResultMapper; @@ -54,6 +56,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.poi.ss.usermodel.Cell; import org.apache.poi.ss.usermodel.CellStyle; import org.apache.poi.ss.usermodel.DataFormatter; +import org.apache.poi.ss.usermodel.Drawing; import org.apache.poi.ss.usermodel.Font; import org.apache.poi.ss.usermodel.Row; import org.apache.poi.ss.usermodel.Sheet; @@ -88,6 +91,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; import java.util.function.Supplier; import java.util.regex.Pattern; import java.util.zip.ZipEntry; @@ -143,9 +147,16 @@ public class SimilarAsinTaskService { "是否符合类目", "不符合理由", "产品类目", - "状态" + "状态", + "主图", + "阿里巴巴图片1", + "阿里巴巴图片2" ); + private static final int IMG_COL_MAIN = 9; + private static final int IMG_COL_PUZZLE1 = 10; + private static final int IMG_COL_PUZZLE2 = 11; + private final LocalFileStorageService localFileStorageService; private final OssStorageService ossStorageService; private final StorageProperties storageProperties; @@ -166,6 +177,7 @@ public class SimilarAsinTaskService { private final DistributedJobLockService distributedJobLockService; private final InstanceMetadata instanceMetadata; private final CozeCredentialPoolService cozeCredentialPoolService; + private final SimilarAsinImageEmbedder imageEmbedder; @Autowired @Qualifier("cozeTaskExecutor") private TaskExecutor cozeTaskExecutor; @@ -613,6 +625,8 @@ public class SimilarAsinTaskService { taskScopeStateMapper.delete(new LambdaQueryWrapper().eq(TaskScopeStateEntity::getTaskId, taskId).eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)); taskChunkMapper.delete(new LambdaQueryWrapper().eq(TaskChunkEntity::getTaskId, taskId).eq(TaskChunkEntity::getModuleType, MODULE_TYPE)); taskCacheService.deleteTaskCache(taskId); + // 同步清理 task_file_job，避免被删除任务遗留的 PENDING/FAILED 行被 TaskResultFileJobWorker 反复扫描出 task not found。 + taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE); fileTaskMapper.deleteById(taskId); } } @@ -622,6 +636,8 @@ public class SimilarAsinTaskService { if (row == null || !MODULE_TYPE.equals(row.getModuleType()) || !Objects.equals(userId, row.getUserId())) { throw new BusinessException("记录不存在"); } + // 同步清理 task_file_job，避免被删除历史记录残留的 ASSEMBLE_RESULT job 被反复扫描出 result record not found。 + taskFileJobService.deleteResultJobs(row.getTaskId(), MODULE_TYPE, resultId); fileResultMapper.deleteById(resultId); } @@ -2353,13 +2369,29 @@ public class SimilarAsinTaskService { log.warn("[similar-asin] coze pending submit retry failed taskId={} stateId={} jobId={} rows={} err={}", state.getTaskId(), state.getId(), context.jobId(), batchRows.size(), message); if (isCozeThrottleLockTimeout(message)) { + if (isCozeStateTimedOut(state)) { + Map> allRowsByBaseId = loadAllRowsByBaseId(task); + String finalMessage = "Coze pending submit timeout after waiting credentials: " + message; + mergeCozeRowsIntoChunk(task, + null, + null, + cozeClient.markRowsFailed(batchRows, finalMessage), + allRowsByBaseId); + markCozeStateTerminal(state, COZE_STATUS_FAILED, finalMessage); + maybeFinalizeCozeJobLocked(state.getTaskId(), context); + log.warn("[similar-asin] coze pending submit force-failed by timeout taskId={} stateId={} jobId={} rows={}", + state.getTaskId(), state.getId(), context.jobId(), batchRows.size()); + return; + } keepPendingCozeSubmitState(state, message); return; } int nextAttemptCount = cozeAttemptCount(state) + 1; - if (nextAttemptCount >= MAX_COZE_SUBMIT_RETRY_COUNT) { + if (nextAttemptCount >= MAX_COZE_SUBMIT_RETRY_COUNT || isCozeStateTimedOut(state)) { Map> allRowsByBaseId = loadAllRowsByBaseId(task); - String finalMessage = message + " after " + nextAttemptCount + " submit attempts"; + String finalMessage = isCozeStateTimedOut(state) + ? message + " (timeout after " + properties.getCozePollTimeoutMillis() + "ms)" + : message + " after " + nextAttemptCount + " submit attempts"; mergeCozeRowsIntoChunk(task, null, null, @@ -2986,6 +3018,22 @@ public class SimilarAsinTaskService { if (!parsed.getAllItems().isEmpty() && resultMap.isEmpty()) { throw new BusinessException("相似ASIN检测结果为空，请稍后重试生成结果文件"); } + // 公共分组处理：同 baseId 连续行视为一组（如 1、1_1、1_2），组内任一行的"是否符合类目" + // 命中"不符合"（包含匹配），则组内所有行的"是否符合类目"统一为标准值"不符合"。 + // 单行组跳过。仅修改 isConform 列，不影响其他列。 + int conformPropagated = CozeGroupResultPropagator.propagateByGroup( + parsed.getAllItems(), + SimilarAsinParsedRowVo::getDisplayId, + row -> findResultRow(row, resultMap), + SimilarAsinResultRowDto::getIsConform, + SimilarAsinResultRowDto::setIsConform, + List.of("不符合"), + "[similar-asin] taskId=" + task.getId() + ); + if (conformPropagated > 0) { + log.info("[similar-asin] group propagate isConform taskId={} updatedRows={}", + task.getId(), conformPropagated); + } File outputDir = new File(storageProperties.getLocalTempDir(), "similar-asin-result"); if (!outputDir.exists() && !outputDir.mkdirs()) { throw new BusinessException("创建结果目录失败"); @@ -3253,6 +3301,12 @@ public class SimilarAsinTaskService { font.setBold(true); headerStyle.setFont(font); + sheet.setColumnWidth(IMG_COL_MAIN, SimilarAsinImageEmbedder.IMAGE_COL_WIDTH_CHARS * 256); + sheet.setColumnWidth(IMG_COL_PUZZLE1, SimilarAsinImageEmbedder.IMAGE_COL_WIDTH_CHARS * 256); + sheet.setColumnWidth(IMG_COL_PUZZLE2, SimilarAsinImageEmbedder.IMAGE_COL_WIDTH_CHARS * 256); + Drawing patriarch = sheet.createDrawingPatriarch(); + Map taskImageCache = new ConcurrentHashMap<>(); + List resultHeaders = new ArrayList<>(RESULT_HEADERS); Row header = sheet.createRow(0); for (int i = 0; i < resultHeaders.size(); i++) { @@ -3264,7 +3318,8 @@ public class SimilarAsinTaskService { int rowIndex = 1; for (SimilarAsinParsedRowVo parsedRow : rowsToWrite == null ? List.of() : rowsToWrite) { SimilarAsinResultRowDto resultRow = findResultRow(parsedRow, resultMap); - Row row = sheet.createRow(rowIndex++); + Row row = sheet.createRow(rowIndex); + row.setHeightInPoints(SimilarAsinImageEmbedder.IMAGE_ROW_HEIGHT_POINTS); int col = 0; row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId())); row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), "")); @@ -3275,11 +3330,17 @@ public class SimilarAsinTaskService { row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getReason())); row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getCategory())); row.createCell(col).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getStatus(), "")); + if (resultRow != null) { + imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_MAIN, resultRow.getMainUrl(), row, taskImageCache); + imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE1, resultRow.getPuzzleImg1(), row, taskImageCache); + imageEmbedder.embed(sheet, patriarch, workbook, rowIndex, IMG_COL_PUZZLE2, resultRow.getPuzzleImg2(), row, taskImageCache); + } + rowIndex++; } workbook.write(fos); workbook.dispose(); } catch (Exception ex) { - throw new BusinessException("生成相似ASIN检测结果失败"); + throw new BusinessException("生成相似ASIN检测结果失败", ex); } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java new file mode 100644 index 0000000..5ced792 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java @@ -0,0 +1,464 @@ +package com.nanri.aiimage.modules.similarasin.util; + +import jakarta.annotation.PreDestroy; +import lombok.extern.slf4j.Slf4j; +import okhttp3.Dns; +import okhttp3.OkHttpClient; +import okhttp3.Request; +import okhttp3.Response; +import okhttp3.ResponseBody; +import org.apache.poi.ss.usermodel.ClientAnchor; +import org.apache.poi.ss.usermodel.Drawing; +import org.apache.poi.ss.usermodel.Row; +import org.apache.poi.ss.usermodel.Sheet; +import org.apache.poi.ss.usermodel.Workbook; +import org.apache.poi.xssf.streaming.SXSSFWorkbook; +import org.springframework.stereotype.Component; + +import javax.imageio.IIOImage; +import javax.imageio.ImageIO; +import javax.imageio.ImageReader; +import javax.imageio.ImageWriteParam; +import javax.imageio.ImageWriter; +import javax.imageio.stream.ImageInputStream; +import javax.imageio.stream.ImageOutputStream; +import java.awt.Graphics2D; +import java.awt.RenderingHints; +import java.awt.image.BufferedImage; +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.IOException; +import java.io.InputStream; +import java.net.InetAddress; +import java.net.URI; +import java.net.URISyntaxException; +import java.net.UnknownHostException; +import java.time.Duration; +import java.util.ArrayList; +import java.util.Iterator; +import java.util.List; +import java.util.Locale; +import java.util.Map; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicInteger; + +/** + * similar-asin 结果 xlsx 的图片嵌入器：下载、缩略图压缩、任务内缓存、POI 嵌入与单图失败兜底集中在一处。 + * 按 ralplan 共识 .omc/plans/similar-asin-coze-image-embed.md T4 / T4.5 / T5 实现。 + */ +@Component +@Slf4j +public class SimilarAsinImageEmbedder { + + static final int DOWNLOAD_TIMEOUT_SECONDS = 5; + static final int DOWNLOAD_MAX_RETRY = 1; + static final int DOWNLOAD_POOL_SIZE = 8; + public static final float IMAGE_ROW_HEIGHT_POINTS = 150f; + public static final int IMAGE_COL_WIDTH_CHARS = 30; + static final int TARGET_LONG_EDGE_PX = 300; + static final float JPEG_QUALITY = 0.85f; + static final int MAX_THUMB_SIZE_BYTES = 40 * 1024; + static final int MAX_DOWNLOAD_BYTES = 5 * 1024 * 1024; + static final int MAX_DECODE_PIXELS = 6000 * 6000; + + private static final String UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " + + "(KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36"; + + private final OkHttpClient httpClient = new OkHttpClient.Builder() + .connectTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS)) + .readTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS)) + .writeTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS)) + .callTimeout(Duration.ofSeconds(DOWNLOAD_TIMEOUT_SECONDS * 2L)) + .retryOnConnectionFailure(false) + .followRedirects(false) + .followSslRedirects(false) + .dns(new SafeDns(Dns.SYSTEM)) + .build(); + + private final ExecutorService downloadPool = Executors.newFixedThreadPool( + DOWNLOAD_POOL_SIZE, namedFactory("similar-asin-image-dl")); + + @PreDestroy + public void shutdown() { + downloadPool.shutdownNow(); + } + + /** + * 嵌入单元格：下载 → resize → POI addPicture → createPicture，全链路任一失败即 URL 文本兜底。 + * taskImageCache 由调用方在 writeResultWorkbook 入口创建（任务级 ConcurrentHashMap），方法返回随栈帧释放。 + */ + public void embed(Sheet sheet, + Drawing patriarch, + SXSSFWorkbook wb, + int rowIdx, + int colIdx, + String url, + Row row, + Map taskImageCache) { + if (url == null || url.isBlank()) { + return; + } + String trimmedUrl = url.trim(); + byte[] thumb; + try { + thumb = taskImageCache.get(trimmedUrl); + boolean cacheHit = thumb != null; + if (!cacheHit) { + byte[] raw = downloadWithRetry(trimmedUrl); + thumb = resizeImage(trimmedUrl, raw); + taskImageCache.put(trimmedUrl, thumb); + log.info("[similar-asin][image] cache-miss url={} bytes={}", trimmedUrl, thumb.length); + } else { + log.debug("[similar-asin][image] cache-hit url={} bytes={}", trimmedUrl, thumb.length); + } + } catch (UnsupportedUrlException ex) { + log.warn("[similar-asin][image] unsupported-url url={} reason={}", trimmedUrl, ex.getMessage()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } catch (DownloadOversizeException ex) { + log.warn("[similar-asin][image] download-oversize url={} bytes={}", trimmedUrl, ex.size()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } catch (ResizeOversizeException ex) { + log.warn("[similar-asin][image] resize-oversize url={} bytes={}", ex.url(), ex.size()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } catch (ResizeException ex) { + log.warn("[similar-asin][image] resize-fail url={} err={}", trimmedUrl, ex.getMessage()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } catch (IOException | TimeoutException ex) { + log.warn("[similar-asin][image] download-fail url={} err={}", trimmedUrl, ex.getMessage()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } catch (RuntimeException ex) { + log.warn("[similar-asin][image] embed-fail url={} err={}", trimmedUrl, ex.getMessage()); + row.createCell(colIdx).setCellValue(trimmedUrl); + return; + } + try { + int picIdx = wb.addPicture(thumb, Workbook.PICTURE_TYPE_JPEG); + ClientAnchor anchor = wb.getCreationHelper().createClientAnchor(); + anchor.setCol1(colIdx); + anchor.setRow1(rowIdx); + anchor.setCol2(colIdx + 1); + anchor.setRow2(rowIdx + 1); + anchor.setAnchorType(ClientAnchor.AnchorType.MOVE_AND_RESIZE); + patriarch.createPicture(anchor, picIdx); + } catch (RuntimeException ex) { + log.warn("[similar-asin][image] embed-fail url={} err={}", trimmedUrl, ex.getMessage()); + row.createCell(colIdx).setCellValue(trimmedUrl); + } + } + + private byte[] downloadWithRetry(String url) throws IOException, TimeoutException { + validateHttpsUrl(url); + IOException last = null; + for (int attempt = 0; attempt <= DOWNLOAD_MAX_RETRY; attempt++) { + Future future = downloadPool.submit(() -> doFetch(url)); + try { + return future.get(DOWNLOAD_TIMEOUT_SECONDS * 2L, TimeUnit.SECONDS); + } catch (java.util.concurrent.ExecutionException ee) { + Throwable cause = ee.getCause(); + if (cause instanceof DownloadOversizeException doe) { + throw doe; + } + if (cause instanceof IOException io) { + last = io; + continue; + } + if (cause instanceof RuntimeException re) { + throw re; + } + throw new IOException("image download failed: " + cause.getMessage(), cause); + } catch (java.util.concurrent.TimeoutException te) { + future.cancel(true); + throw te; + } catch (InterruptedException ie) { + Thread.currentThread().interrupt(); + throw new IOException("image download interrupted", ie); + } + } + throw last != null ? last : new IOException("image download failed without cause"); + } + + static void validateHttpsUrl(String url) { + URI uri; + try { + uri = new URI(url); + } catch (URISyntaxException ex) { + throw new UnsupportedUrlException("invalid url syntax: " + url); + } + String scheme = uri.getScheme(); + if (scheme == null || !scheme.toLowerCase(Locale.ROOT).equals("https")) { + throw new UnsupportedUrlException("only https is allowed, got=" + scheme + " url=" + url); + } + String host = uri.getHost(); + if (host == null || host.isBlank()) { + throw new UnsupportedUrlException("missing host: " + url); + } + String lowerHost = host.toLowerCase(Locale.ROOT); + if (lowerHost.equals("localhost") || lowerHost.endsWith(".localhost")) { + throw new UnsupportedUrlException("localhost not allowed url=" + url); + } + if (lowerHost.startsWith("127.") || lowerHost.startsWith("10.") + || lowerHost.startsWith("192.168.") + || lowerHost.startsWith("169.254.") + || lowerHost.equals("0.0.0.0") + || lowerHost.equals("::1") + || lowerHost.startsWith("[::1") + || lowerHost.startsWith("[fc") + || lowerHost.startsWith("[fd")) { + throw new UnsupportedUrlException("private/loopback host not allowed url=" + url); + } + if (lowerHost.startsWith("172.")) { + String[] parts = lowerHost.split("\\."); + if (parts.length >= 2) { + try { + int second = Integer.parseInt(parts[1]); + if (second >= 16 && second <= 31) { + throw new UnsupportedUrlException("private host not allowed url=" + url); + } + } catch (NumberFormatException ignored) { + // 非纯数字，跳过 172.16/12 私网判断 + } + } + } + } + + private byte[] doFetch(String url) throws IOException { + Request req = new Request.Builder() + .url(url) + .header("User-Agent", UA) + .get() + .build(); + try (Response resp = httpClient.newCall(req).execute()) { + if (!resp.isSuccessful()) { + throw new IOException("http " + resp.code() + " for " + url); + } + ResponseBody body = resp.body(); + if (body == null) { + throw new IOException("empty body for " + url); + } + long advertised = body.contentLength(); + if (advertised > MAX_DOWNLOAD_BYTES) { + throw new DownloadOversizeException(url, advertised); + } + try (InputStream in = body.byteStream()) { + ByteArrayOutputStream out = new ByteArrayOutputStream(); + byte[] buf = new byte[8192]; + int total = 0; + int n; + while ((n = in.read(buf)) != -1) { + total += n; + if (total > MAX_DOWNLOAD_BYTES) { + throw new DownloadOversizeException(url, total); + } + out.write(buf, 0, n); + } + return out.toByteArray(); + } + } + } + + /** + * 长边 300px 等比缩放 + JPEG q=0.85 输出；ImageWriter / ImageOutputStream 双层 try-finally 释放 native 资源。 + * 出口校验字节数 ≤ MAX_THUMB_SIZE_BYTES，超限抛 ResizeOversizeException 触发文本兜底。 + */ + byte[] resizeImage(String sourceUrl, byte[] raw) throws IOException { + guardImageDimensions(sourceUrl, raw); + BufferedImage src = ImageIO.read(new ByteArrayInputStream(raw)); + if (src == null) { + throw new IOException("unsupported image format url=" + sourceUrl); + } + int srcW = src.getWidth(); + int srcH = src.getHeight(); + double ratio = (double) Math.max(srcW, srcH) / TARGET_LONG_EDGE_PX; + int dstW = ratio > 1 ? Math.max(1, (int) Math.round(srcW / ratio)) : srcW; + int dstH = ratio > 1 ? Math.max(1, (int) Math.round(srcH / ratio)) : srcH; + BufferedImage dst = new BufferedImage(dstW, dstH, BufferedImage.TYPE_INT_RGB); + Graphics2D g = dst.createGraphics(); + try { + g.setRenderingHint(RenderingHints.KEY_INTERPOLATION, RenderingHints.VALUE_INTERPOLATION_BILINEAR); + g.setRenderingHint(RenderingHints.KEY_RENDERING, RenderingHints.VALUE_RENDER_QUALITY); + g.drawImage(src, 0, 0, dstW, dstH, null); + } finally { + g.dispose(); + } + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + Iterator writers = ImageIO.getImageWritersByFormatName("jpg"); + if (!writers.hasNext()) { + throw new IOException("no JPEG writer available"); + } + ImageWriter writer = writers.next(); + try { + ImageWriteParam param = writer.getDefaultWriteParam(); + param.setCompressionMode(ImageWriteParam.MODE_EXPLICIT); + param.setCompressionQuality(JPEG_QUALITY); + ImageOutputStream ios = ImageIO.createImageOutputStream(baos); + try { + writer.setOutput(ios); + writer.write(null, new IIOImage(dst, null, null), param); + } finally { + ios.close(); + } + } finally { + writer.dispose(); + } + if (baos.size() > MAX_THUMB_SIZE_BYTES) { + throw new ResizeOversizeException(sourceUrl, baos.size()); + } + return baos.toByteArray(); + } + + /** 在解码整张位图前用 ImageReader 仅读取头部尺寸，避免“图像炸弹”导致 heap OOM。 */ + private static void guardImageDimensions(String sourceUrl, byte[] raw) throws IOException { + try (ImageInputStream iis = ImageIO.createImageInputStream(new ByteArrayInputStream(raw))) { + if (iis == null) { + throw new IOException("unable to create ImageInputStream url=" + sourceUrl); + } + Iterator readers = ImageIO.getImageReaders(iis); + if (!readers.hasNext()) { + throw new IOException("unsupported image format url=" + sourceUrl); + } + ImageReader reader = readers.next(); + try { + reader.setInput(iis, true, true); + long pixels = (long) reader.getWidth(0) * (long) reader.getHeight(0); + if (pixels > MAX_DECODE_PIXELS) { + throw new ResizeException("image too large url=" + sourceUrl + " pixels=" + pixels); + } + } finally { + reader.dispose(); + } + } + } + + private static ThreadFactory namedFactory(String prefix) { + AtomicInteger counter = new AtomicInteger(); + return r -> { + Thread t = new Thread(r, prefix + "-" + counter.incrementAndGet()); + t.setDaemon(true); + return t; + }; + } + + /** + * 在 OkHttp 解析 host 后立刻按 IP 复检，关闭 DNS rebinding 通道。 + * 校验项：loopback / linkLocal / siteLocal / anyLocal / multicast / IPv4-mapped IPv6 私网 / CGNAT 100.64/10。 + */ + static class SafeDns implements Dns { + private final Dns delegate; + + SafeDns(Dns delegate) { + this.delegate = delegate; + } + + @Override + public List lookup(String hostname) throws UnknownHostException { + List resolved = delegate.lookup(hostname); + List safe = new ArrayList<>(resolved.size()); + for (InetAddress addr : resolved) { + if (isPrivateIp(addr)) { + throw new UnknownHostException( + "blocked private/loopback ip host=" + hostname + " ip=" + addr.getHostAddress()); + } + safe.add(addr); + } + if (safe.isEmpty()) { + throw new UnknownHostException("no public ip for host=" + hostname); + } + return safe; + } + } + + static boolean isPrivateIp(InetAddress addr) { + if (addr.isLoopbackAddress() || addr.isLinkLocalAddress() || addr.isSiteLocalAddress() + || addr.isAnyLocalAddress() || addr.isMulticastAddress()) { + return true; + } + byte[] raw = addr.getAddress(); + if (raw.length == 4) { + int b0 = raw[0] & 0xff; + int b1 = raw[1] & 0xff; + int b2 = raw[2] & 0xff; + int b3 = raw[3] & 0xff; + // CGNAT 100.64.0.0/10 + if (b0 == 100 && b1 >= 64 && b1 <= 127) { + return true; + } + // 受限广播 255.255.255.255 + if (b0 == 255 && b1 == 255 && b2 == 255 && b3 == 255) { + return true; + } + } else if (raw.length == 16) { + // RFC 4193 IPv6 ULA fc00::/7（高 7 位为 1111110），与 validateHttpsUrl 的 `[fc`/`[fd` 字面量拦截对齐 + if ((raw[0] & 0xfe) == 0xfc) { + return true; + } + } + return false; + } + + /** URL 协议/host 不在白名单时抛出，触发文本兜底。 */ + public static class UnsupportedUrlException extends RuntimeException { + public UnsupportedUrlException(String message) { + super(message); + } + } + + /** 下载链路体积超限保护：上限 MAX_DOWNLOAD_BYTES（5MB），覆盖 Content-Length 与流式累加两条路径。 */ + public static class DownloadOversizeException extends IOException { + private final String url; + private final long size; + + public DownloadOversizeException(String url, long size) { + super("download oversize url=" + url + " size=" + size); + this.url = url; + this.size = size; + } + + public String url() { + return url; + } + + public long size() { + return size; + } + } + + /** resize 链路非 oversize 的失败（解码异常、像素炸弹等）。 */ + public static class ResizeException extends RuntimeException { + public ResizeException(String message) { + super(message); + } + } + + /** + * resize 出口字节硬上限保护：worst case 6000 行 × 3 列 × 40KB × 2 副本 = 1440MB； + * 因此线上限制建议结合 `taskImageCache` 命中率与单任务行数评估，超出预算需缩小 MAX_THUMB_SIZE_BYTES 或限制行数。 + */ + public static class ResizeOversizeException extends ResizeException { + private final String url; + private final int size; + + public ResizeOversizeException(String url, int size) { + super("resize oversize url=" + url + " size=" + size); + this.url = url; + this.size = size; + } + + public String url() { + return url; + } + + public int size() { + return size; + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java index 2c2c073..312823d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java @@ -223,9 +223,12 @@ public class TaskFileJobService { if (jobId == null || jobId <= 0) { return false; } + // 仅在状态从非 PENDING 切换到 PENDING 时更新并发布事件， + // 避免对已在排队中的 job 重复触发 dispatch（同一 jobId 在多个 dispatch 线程并发处理）。 int updated = taskFileJobMapper.update(null, new LambdaUpdateWrapper() .eq(TaskFileJobEntity::getId, jobId) .ne(TaskFileJobEntity::getStatus, "SUCCESS") + .ne(TaskFileJobEntity::getStatus, "PENDING") .set(TaskFileJobEntity::getStatus, "PENDING") .set(TaskFileJobEntity::getErrorMessage, message) .set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()) @@ -267,6 +270,21 @@ public class TaskFileJobService { .set(TaskFileJobEntity::getFinishedAt, retryCount >= 5 ? LocalDateTime.now() : null)); } + /** + * 终态失败：把 retryCount 直接拉满到上限，让 listRunnableJobs 不再扫到该 job。 + * 适用于 task/result 已经不存在等无法通过重试恢复的场景。 + */ + public void markFailedPermanent(TaskFileJobEntity job, String message) { + taskFileJobMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskFileJobEntity::getId, job.getId()) + .ne(TaskFileJobEntity::getStatus, "SUCCESS") + .set(TaskFileJobEntity::getStatus, "FAILED") + .set(TaskFileJobEntity::getRetryCount, 5) + .set(TaskFileJobEntity::getErrorMessage, message == null ? "结果文件生成失败" : message) + .set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()) + .set(TaskFileJobEntity::getFinishedAt, LocalDateTime.now())); + } + public TaskFileJobEntity findAssembleJob(Long taskId, String moduleType, Long resultId) { return findJob(taskId, moduleType, resultId, JOB_TYPE_ASSEMBLE_RESULT); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java index 5b110df..d0af6aa 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java @@ -1,5 +1,6 @@ package com.nanri.aiimage.modules.task.service; +import com.nanri.aiimage.common.exception.BusinessException; import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskService; import com.nanri.aiimage.config.InstanceMetadata; import com.nanri.aiimage.modules.brand.service.BrandTaskService; @@ -162,12 +163,33 @@ public class TaskResultFileJobWorker { return; } String message = ex.getMessage() == null ? "结果文件生成失败" : ex.getMessage(); + // 孤儿 job：fileTask/fileResult 已被删除却残留 task_file_job 行，重试 5 次也救不回来， + // 直接拉满 retryCount 让 worker 跳过，避免每 15 秒刷一次 task not found 警告日志。 + if (isOrphanJobFailure(ex, message)) { + log.warn("[task-file-job] process aborted because owning task/result no longer exists jobId={} taskId={} moduleType={} resultId={} msg={}", + job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message); + taskFileJobService.markFailedPermanent(job, message); + return; + } log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={}", job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(), message); taskFileJobService.markFailed(job, message); } } + private boolean isOrphanJobFailure(Exception ex, String message) { + if (!(ex instanceof BusinessException)) { + return false; + } + if (message == null) { + return false; + } + return message.contains("task not found") + || message.contains("result record not found") + || message.contains("任务不存在") + || message.contains("记录不存在"); + } + private String resolveResultFileUrl(TaskFileJobEntity job) { if ("BRAND".equals(job.getModuleType())) { return brandTaskService.resolveResultObjectKey(job.getTaskId()); diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index ccbf2a6..2fa1a2e 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -23,7 +23,7 @@ spring: idle-timeout: ${AIIMAGE_DB_POOL_IDLE_TIMEOUT_MS:600000} max-lifetime: ${AIIMAGE_DB_POOL_MAX_LIFETIME_MS:1500000} keepalive-time: ${AIIMAGE_DB_POOL_KEEPALIVE_TIME_MS:120000} - leak-detection-threshold: ${AIIMAGE_DB_POOL_LEAK_DETECT_MS:0} + leak-detection-threshold: ${AIIMAGE_DB_POOL_LEAK_DETECT_MS:15000} data: redis: username: ${AIIMAGE_REDIS_USERNAME:} @@ -167,11 +167,11 @@ aiimage: coze-workflow-id: ${AIIMAGE_SIMILAR_ASIN_COZE_WORKFLOW_ID:7639708860686024756} coze-token: ${AIIMAGE_SIMILAR_ASIN_COZE_TOKEN:} coze-batch-size: ${AIIMAGE_SIMILAR_ASIN_COZE_BATCH_SIZE:10} - coze-credential-stripe-size: ${AIIMAGE_SIMILAR_ASIN_COZE_CREDENTIAL_STRIPE_SIZE:1} + coze-credential-stripe-size: ${AIIMAGE_SIMILAR_ASIN_COZE_CREDENTIAL_STRIPE_SIZE:0} coze-connect-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_CONNECT_TIMEOUT_MILLIS:10000} coze-read-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_READ_TIMEOUT_MILLIS:60000} coze-poll-interval-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_INTERVAL_MILLIS:30000} - coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:600000} + coze-poll-timeout-millis: ${AIIMAGE_SIMILAR_ASIN_COZE_POLL_TIMEOUT_MILLIS:1800000} stale-timeout-minutes: ${AIIMAGE_SIMILAR_ASIN_STALE_TIMEOUT_MINUTES:30} stale-finalize-cron: ${AIIMAGE_SIMILAR_ASIN_STALE_FINALIZE_CRON:0 */2 * * * *} coze-include-legacy-api-key: ${AIIMAGE_SIMILAR_ASIN_COZE_INCLUDE_LEGACY_API_KEY:true} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java new file mode 100644 index 0000000..86b4b0f --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java @@ -0,0 +1,153 @@ +package com.nanri.aiimage.modules.similarasin.util; + +import org.junit.jupiter.api.Test; + +import javax.imageio.ImageIO; +import java.awt.Color; +import java.awt.Graphics2D; +import java.awt.image.BufferedImage; +import java.io.ByteArrayOutputStream; +import java.net.InetAddress; +import java.net.UnknownHostException; +import java.util.Collections; +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertThrows; +import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assertions.assertFalse; + +class SimilarAsinImageEmbedderTest { + + private final SimilarAsinImageEmbedder embedder = new SimilarAsinImageEmbedder(); + + @Test + void resizeImageProducesThumbnailUnderHardCap() throws Exception { + BufferedImage src = new BufferedImage(1200, 1600, BufferedImage.TYPE_INT_RGB); + Graphics2D g = src.createGraphics(); + try { + g.setColor(new Color(0x33, 0x66, 0x99)); + g.fillRect(0, 0, 1200, 1600); + } finally { + g.dispose(); + } + ByteArrayOutputStream baos = new ByteArrayOutputStream(); + ImageIO.write(src, "jpg", baos); + + byte[] thumb = embedder.resizeImage("https://example.com/big.jpg", baos.toByteArray()); + + assertNotNull(thumb); + assertTrue(thumb.length > 0, "thumb 不可为空"); + assertTrue(thumb.length <= SimilarAsinImageEmbedder.MAX_THUMB_SIZE_BYTES, + "thumb 字节应 <= 40KB 上限，实际=" + thumb.length); + + BufferedImage decoded = ImageIO.read(new java.io.ByteArrayInputStream(thumb)); + assertNotNull(decoded, "thumb 应可被 ImageIO 重新解码"); + int longEdge = Math.max(decoded.getWidth(), decoded.getHeight()); + assertEquals(SimilarAsinImageEmbedder.TARGET_LONG_EDGE_PX, longEdge, + "长边应等于 TARGET_LONG_EDGE_PX=300"); + } + + @Test + void resizeOversizeExceptionCarriesUrlAndSize() { + SimilarAsinImageEmbedder.ResizeOversizeException ex = + new SimilarAsinImageEmbedder.ResizeOversizeException("https://example.com/huge.jpg", 99999); + assertEquals("https://example.com/huge.jpg", ex.url()); + assertEquals(99999, ex.size()); + assertTrue(ex.getMessage().contains("oversize")); + assertTrue(ex.getMessage().contains("99999")); + } + + @Test + void resizeUnsupportedFormatThrowsIOException() { + byte[] notAnImage = "not an image at all, just some bytes".getBytes(); + Exception ex = assertThrows(Exception.class, + () -> embedder.resizeImage("https://example.com/broken.bin", notAnImage)); + assertTrue(ex instanceof java.io.IOException || ex instanceof RuntimeException, + "应为 IOException 或 RuntimeException 兜底，实际=" + ex.getClass().getSimpleName()); + assertTrue(ex.getMessage() == null + || ex.getMessage().toLowerCase().contains("unsupported") + || ex.getMessage().toLowerCase().contains("unable"), + "异常消息应反映不支持/无法解析"); + } + + @Test + void safeDnsRejectsPrivateIpsOnLookup() { + SimilarAsinImageEmbedder.SafeDns safeDns = new SimilarAsinImageEmbedder.SafeDns( + hostname -> Collections.singletonList(InetAddress.getByName("127.0.0.1"))); + UnknownHostException ex = assertThrows(UnknownHostException.class, + () -> safeDns.lookup("rebound.example.com")); + assertTrue(ex.getMessage().contains("blocked private/loopback ip"), + "异常消息应说明被阻断，实际=" + ex.getMessage()); + } + + @Test + void safeDnsAllowsPublicIpResolution() throws Exception { + SimilarAsinImageEmbedder.SafeDns safeDns = new SimilarAsinImageEmbedder.SafeDns( + hostname -> Collections.singletonList(InetAddress.getByName("8.8.8.8"))); + List result = safeDns.lookup("dns.google"); + assertEquals(1, result.size()); + assertEquals("8.8.8.8", result.get(0).getHostAddress()); + } + + @Test + void safeDnsRejectsMixedResultsContainingPrivateIp() { + SimilarAsinImageEmbedder.SafeDns safeDns = new SimilarAsinImageEmbedder.SafeDns( + hostname -> List.of( + InetAddress.getByName("8.8.8.8"), + InetAddress.getByName("10.0.0.5"))); + // 任一结果落入私网即整体阻断，关闭 DNS rebinding 通道 + assertThrows(UnknownHostException.class, + () -> safeDns.lookup("partial-rebound.example.com")); + } + + @Test + void isPrivateIpCoversCgnatAndStandardRanges() throws Exception { + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("127.0.0.1")), "loopback"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("10.0.0.1")), "10/8"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("192.168.1.1")), "192.168/16"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("169.254.0.1")), "link-local"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("172.16.0.1")), "site-local 172.16"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("100.64.0.1")), "CGNAT lower"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("100.127.255.255")), "CGNAT upper"); + assertFalse(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("8.8.8.8")), "public DNS"); + assertFalse(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("100.63.0.1")), "below CGNAT 段"); + assertFalse(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("100.128.0.1")), "above CGNAT 段"); + // 受限广播 + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("255.255.255.255")), "受限广播"); + // IPv6 ULA fc00::/7 + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("fc00::1")), "ULA fc00::"); + assertTrue(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("fd12:3456:789a::1")), "ULA fd::"); + // IPv6 公网放行 + assertFalse(SimilarAsinImageEmbedder.isPrivateIp(InetAddress.getByName("2001:4860:4860::8888")), "Google IPv6 DNS"); + } + + @Test + void validateHttpsUrlRejectsHttpAndPrivateHosts() { + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("http://example.com/a.jpg")); + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("https://localhost/a.jpg")); + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("https://127.0.0.1/a.jpg")); + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("https://10.0.0.5/a.jpg")); + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("https://192.168.1.1/a.jpg")); + assertThrows(SimilarAsinImageEmbedder.UnsupportedUrlException.class, + () -> SimilarAsinImageEmbedder.validateHttpsUrl("https://172.16.0.1/a.jpg")); + // 公网 https 应通过 + SimilarAsinImageEmbedder.validateHttpsUrl("https://m.media-amazon.com/images/I/abc.jpg"); + SimilarAsinImageEmbedder.validateHttpsUrl("https://cbu01.alicdn.com/img/ibank/xxx.jpg"); + } + + @Test + void downloadOversizeExceptionCarriesUrlAndSize() { + SimilarAsinImageEmbedder.DownloadOversizeException ex = + new SimilarAsinImageEmbedder.DownloadOversizeException("https://example.com/big.jpg", 12345678L); + assertEquals("https://example.com/big.jpg", ex.url()); + assertEquals(12345678L, ex.size()); + assertTrue(ex.getMessage().contains("oversize")); + } +}