From e98a1a12079d06457614ec5042477300b6fd00f2 Mon Sep 17 00:00:00 2001 From: super <2903208875@qq.com> Date: Wed, 3 Jun 2026 17:14:50 +0800 Subject: [PATCH] feat: update task routing and coze handling --- app/web_source/brand.html | 4 +- backend-java/docs/patrol-delete-python-api.md | 2 +- .../exception/GlobalExceptionHandler.java | 28 +++ .../exception/TaskOwnerMismatchException.java | 36 +++ .../service/TaskOwnerForwardService.java | 156 +++++++++++++ .../util/CozeGroupResultPropagator.java | 8 +- .../config/AppearancePatentProperties.java | 5 + .../aiimage/config/BrandCheckProperties.java | 15 ++ .../config/DeleteBrandProgressProperties.java | 18 +- .../config/InstanceRoutingProperties.java | 17 ++ .../aiimage/config/PropertiesConfig.java | 2 +- .../aiimage/config/RequestTraceFilter.java | 48 ++-- .../TaskImageCacheCleanupProperties.java | 15 ++ .../config/TaskOperationLockConfig.java | 21 +- .../client/AppearancePatentCozeClient.java | 91 +++++++- .../service/AppearancePatentTaskService.java | 193 +++++++++++----- .../brand/client/BrandCheckClient.java | 200 +++++++++++++++++ .../BrandTaskProgressCacheService.java | 23 ++ .../service/DeleteBrandStaleTaskService.java | 67 +----- .../client/SimilarAsinCozeClient.java | 80 ++++++- .../model/dto/SimilarAsinResultRowDto.java | 70 +++++- .../service/SimilarAsinTaskService.java | 194 +++++++++++----- .../util/SimilarAsinImageEmbedder.java | 109 +++++++-- .../controller/TaskHeartbeatController.java | 36 +++ .../task/mapper/TaskImageCacheMapper.java | 8 +- .../task/model/dto/TaskHeartbeatRequest.java | 23 ++ .../model/entity/TaskImageCacheEntity.java | 2 +- .../task/model/vo/TaskHeartbeatVo.java | 39 ++++ .../task/service/TaskHeartbeatService.java | 211 ++++++++++++++++++ .../service/TaskImageCacheCleanupService.java | 69 ++++++ .../src/main/resources/application.yml | 40 +++- .../util/CozeGroupResultPropagatorTest.java | 80 +++++++ .../AppearancePatentCozeClientTest.java | 64 ++++++ .../brand/client/BrandCheckClientTest.java | 29 +++ .../client/SimilarAsinCozeClientTest.java | 86 +++++++ .../util/SimilarAsinImageEmbedderTest.java | 46 ++++ backend/blueprints/admin_api.py | 2 +- 37 files changed, 1883 insertions(+), 254 deletions(-) create mode 100644 backend-java/src/main/java/com/nanri/aiimage/common/exception/TaskOwnerMismatchException.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/config/InstanceRoutingProperties.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/config/TaskImageCacheCleanupProperties.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskHeartbeatVo.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java create mode 100644 backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskImageCacheCleanupService.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/common/util/CozeGroupResultPropagatorTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClientTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/brand/client/BrandCheckClientTest.java create mode 100644 backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClientTest.java diff --git a/app/web_source/brand.html b/app/web_source/brand.html index 4d0766e..1d37aaa 100644 --- a/app/web_source/brand.html +++ b/app/web_source/brand.html @@ -647,7 +647,7 @@
前端工具
- 采集数据 + 采集数据 变体分析 外观专利检测 @@ -1447,4 +1447,4 @@ - \ No newline at end of file + diff --git a/backend-java/docs/patrol-delete-python-api.md b/backend-java/docs/patrol-delete-python-api.md index f57a3ed..fcd712a 100644 --- a/backend-java/docs/patrol-delete-python-api.md +++ b/backend-java/docs/patrol-delete-python-api.md @@ -107,7 +107,7 @@ Content-Type: application/json 当前超时配置来自后端配置项： - `aiimage.delete-brand-progress.patrol-delete-stale-timeout-minutes` -- `aiimage.delete-brand-progress.patrol-delete-initial-timeout-minutes` +- Python should call `POST /api/tasks/{taskId}/heartbeat` every minute; timeout now uses only the normal stale-timeout setting. ## 6. 历史清理 diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java index 8e172a8..8abbc59 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/GlobalExceptionHandler.java @@ -1,8 +1,13 @@ package com.nanri.aiimage.common.exception; import com.nanri.aiimage.common.api.ApiResponse; +import com.nanri.aiimage.common.service.TaskOwnerForwardService; +import com.nanri.aiimage.config.TaskOperationLockConfig; +import jakarta.servlet.http.HttpServletRequest; import jakarta.validation.ConstraintViolationException; +import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; +import org.springframework.http.ResponseEntity; import org.springframework.web.bind.MethodArgumentNotValidException; import org.springframework.web.bind.annotation.ExceptionHandler; import org.springframework.web.bind.annotation.RestControllerAdvice; @@ -11,9 +16,32 @@ import org.springframework.web.context.request.async.AsyncRequestNotUsableExcept import java.io.IOException; @Slf4j +@RequiredArgsConstructor @RestControllerAdvice public class GlobalExceptionHandler { + private final TaskOwnerForwardService taskOwnerForwardService; + + @ExceptionHandler(TaskOwnerMismatchException.class) + public Object handleTaskOwnerMismatchException(TaskOwnerMismatchException ex, HttpServletRequest request) { + TaskOperationLockConfig.releaseRequestLock(request); + try { + ResponseEntity response = taskOwnerForwardService.forwardCurrentRequest(ex, request); + return ResponseEntity.status(response.getStatusCode()) + .headers(response.getHeaders()) + .body(response.getBody()); + } catch (BusinessException forwardEx) { + return forwardEx.getCode() == null + ? ApiResponse.fail(forwardEx.getMessage()) + : ApiResponse.fail(forwardEx.getCode(), forwardEx.getMessage()); + } catch (Exception forwardEx) { + log.warn("[instance-routing] forward failed taskId={} operation={} owner={} current={} msg={}", + ex.getTaskId(), ex.getOperation(), ex.getOwnerInstanceId(), ex.getCurrentInstanceId(), + forwardEx.getMessage(), forwardEx); + return ApiResponse.fail(40903, "任务归属实例转发失败: " + forwardEx.getMessage()); + } + } + @ExceptionHandler(BusinessException.class) public ApiResponse handleBusinessException(BusinessException ex) { if (Integer.valueOf(40901).equals(ex.getCode())) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/exception/TaskOwnerMismatchException.java b/backend-java/src/main/java/com/nanri/aiimage/common/exception/TaskOwnerMismatchException.java new file mode 100644 index 0000000..d568034 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/common/exception/TaskOwnerMismatchException.java @@ -0,0 +1,36 @@ +package com.nanri.aiimage.common.exception; + +public class TaskOwnerMismatchException extends BusinessException { + + private final Long taskId; + private final String operation; + private final String ownerInstanceId; + private final String currentInstanceId; + + public TaskOwnerMismatchException(Long taskId, + String operation, + String ownerInstanceId, + String currentInstanceId) { + super(40903, "该任务已绑定到另一台服务实例处理，请通过原实例继续处理"); + this.taskId = taskId; + this.operation = operation; + this.ownerInstanceId = ownerInstanceId; + this.currentInstanceId = currentInstanceId; + } + + public Long getTaskId() { + return taskId; + } + + public String getOperation() { + return operation; + } + + public String getOwnerInstanceId() { + return ownerInstanceId; + } + + public String getCurrentInstanceId() { + return currentInstanceId; + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java new file mode 100644 index 0000000..9d7bd57 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/common/service/TaskOwnerForwardService.java @@ -0,0 +1,156 @@ +package com.nanri.aiimage.common.service; + +import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.exception.TaskOwnerMismatchException; +import com.nanri.aiimage.config.InstanceRoutingProperties; +import jakarta.servlet.http.HttpServletRequest; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpHeaders; +import org.springframework.http.HttpMethod; +import org.springframework.http.MediaType; +import org.springframework.http.ResponseEntity; +import org.springframework.http.client.SimpleClientHttpRequestFactory; +import org.springframework.stereotype.Service; +import org.springframework.util.StreamUtils; +import org.springframework.web.client.RestClient; +import org.springframework.web.util.ContentCachingRequestWrapper; + +import java.io.IOException; +import java.util.Enumeration; +import java.util.List; +import java.util.Set; + +@Slf4j +@Service +@RequiredArgsConstructor +public class TaskOwnerForwardService { + + public static final String FORWARDED_HEADER = "X-AIIMAGE-Owner-Forwarded"; + + private static final Set HOP_BY_HOP_HEADERS = Set.of( + "connection", + "keep-alive", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", + "transfer-encoding", + "upgrade", + "host", + "content-length" + ); + + private final InstanceRoutingProperties properties; + + private volatile RestClient sharedRestClient; + + public ResponseEntity forwardCurrentRequest(TaskOwnerMismatchException ex, HttpServletRequest request) { + if (request == null) { + throw new BusinessException(40903, "无法获取当前请求，不能转发到任务归属实例"); + } + if (hasAlreadyForwarded(request)) { + throw new BusinessException(40903, "任务归属实例转发检测到循环，请检查实例路由配置"); + } + HttpMethod method = HttpMethod.valueOf(request.getMethod()); + String url = resolveUrl(ex, currentPathAndQuery(request)); + byte[] body = requestBody(request); + HttpHeaders headers = copyForwardHeaders(request, ex.getCurrentInstanceId()); + log.info("[instance-routing] forward {} taskId={} operation={} owner={} current={} url={}", + method, ex.getTaskId(), ex.getOperation(), ex.getOwnerInstanceId(), ex.getCurrentInstanceId(), url); + return restClient().method(method) + .uri(url) + .headers(target -> target.addAll(headers)) + .body(body) + .retrieve() + .toEntity(byte[].class); + } + + private String resolveUrl(TaskOwnerMismatchException ex, String path) { + String owner = ex.getOwnerInstanceId(); + String baseUrl = properties.getRoutes().get(owner); + if (baseUrl == null || baseUrl.isBlank()) { + throw new BusinessException(40903, "任务归属实例未配置服务路由：" + owner); + } + return stripTrailingSlash(baseUrl) + ensureLeadingSlash(path); + } + + private RestClient restClient() { + RestClient client = sharedRestClient; + if (client != null) { + return client; + } + synchronized (this) { + if (sharedRestClient == null) { + SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory(); + requestFactory.setConnectTimeout(properties.getConnectTimeoutMillis()); + requestFactory.setReadTimeout(properties.getReadTimeoutMillis()); + sharedRestClient = RestClient.builder().requestFactory(requestFactory).build(); + } + return sharedRestClient; + } + } + + private static boolean hasAlreadyForwarded(HttpServletRequest request) { + String value = request.getHeader(FORWARDED_HEADER); + return value != null && !value.isBlank(); + } + + private static String currentPathAndQuery(HttpServletRequest request) { + String path = request.getRequestURI(); + String query = request.getQueryString(); + return query == null || query.isBlank() ? path : path + "?" + query; + } + + private static byte[] requestBody(HttpServletRequest request) { + if (request instanceof ContentCachingRequestWrapper wrapper) { + byte[] body = wrapper.getContentAsByteArray(); + return body == null ? new byte[0] : body; + } + try { + return StreamUtils.copyToByteArray(request.getInputStream()); + } catch (IOException ex) { + throw new BusinessException("读取转发请求体失败：" + ex.getMessage()); + } + } + + private static HttpHeaders copyForwardHeaders(HttpServletRequest request, String currentInstanceId) { + HttpHeaders headers = new HttpHeaders(); + Enumeration headerNames = request.getHeaderNames(); + while (headerNames != null && headerNames.hasMoreElements()) { + String headerName = headerNames.nextElement(); + if (headerName == null || HOP_BY_HOP_HEADERS.contains(headerName.toLowerCase())) { + continue; + } + Enumeration values = request.getHeaders(headerName); + while (values != null && values.hasMoreElements()) { + headers.add(headerName, values.nextElement()); + } + } + headers.set(FORWARDED_HEADER, currentInstanceId == null ? "unknown-instance" : currentInstanceId); + if (!headers.containsKey(HttpHeaders.ACCEPT)) { + headers.setAccept(List.of(MediaType.APPLICATION_JSON)); + } + if (!headers.containsKey(HttpHeaders.CONTENT_TYPE) + && request.getContentType() != null + && !request.getContentType().isBlank()) { + headers.set(HttpHeaders.CONTENT_TYPE, request.getContentType()); + } + return headers; + } + + private static String stripTrailingSlash(String value) { + String result = value == null ? "" : value.trim(); + while (result.endsWith("/")) { + result = result.substring(0, result.length() - 1); + } + return result; + } + + private static String ensureLeadingSlash(String value) { + if (value == null || value.isBlank()) { + return "/"; + } + return value.startsWith("/") ? value : "/" + value; + } +} 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 index 11a2cf3..0c76629 100644 --- 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 @@ -50,9 +50,9 @@ public final class CozeGroupResultPropagator { /** * 否定前缀关键字。若当前值同时包含 standard 和这些关键字之一，则不视为命中。 - * 用于规避"没有侵权""不侵权""未发现明显侵权风险"等被 contains 误判为命中"侵权"的情况。 + * 用于规避"没有侵权""无侵权""不侵权""未发现明显侵权风险"等被 contains 误判为命中"侵权"的情况。 */ - private static final List NEGATIVE_KEYWORDS = Arrays.asList("没有", "不", "未"); + private static final List NEGATIVE_KEYWORDS = Arrays.asList("没有", "无", "不", "未"); private CozeGroupResultPropagator() { } @@ -134,8 +134,8 @@ public final class CozeGroupResultPropagator { } // 按 hitValues 顺序检测组内是否有命中（包含语义），命中即定下"标准值"。 - // 命中需同时满足：当前值包含 standard 且不含任何否定关键字（"没有"/"不"/"未"）， - // 避免"没有侵权""不侵权""未发现明显侵权风险"这类被 contains 误判为命中。 + // 命中需同时满足：当前值包含 standard 且不含任何否定关键字（"没有"/"无"/"不"/"未"）， + // 避免"没有侵权""无侵权""不侵权""未发现明显侵权风险"这类被 contains 误判为命中。 String matchedStandardValue = null; outer: for (String standard : hitValues) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java index a8a676f..6f7e61d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java @@ -23,6 +23,11 @@ public class AppearancePatentProperties { private int cozePollTimeoutMillis = 600000; private int staleTimeoutMinutes = 30; private String staleFinalizeCron = "0 */2 * * * *"; + /** + * 末尾不足一批的数据等待该时长后强制提交 Coze。 + * Python 回传超时后会直接封口；这个配置处理 Python 慢回传但仍未超时的零头批次。 + */ + private int cozeFlushPendingMinutes = 1; @Data public static class CozeCredential { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java new file mode 100644 index 0000000..7273002 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/BrandCheckProperties.java @@ -0,0 +1,15 @@ +package com.nanri.aiimage.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@Data +@ConfigurationProperties(prefix = "aiimage.brand-check") +public class BrandCheckProperties { + private String baseUrl = "http://47.110.241.161:16890"; + private String path = "/brand_check"; + private String token = "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9"; + private String defaultStrategy = "Terms"; + private int connectTimeoutMillis = 10000; + private int readTimeoutMillis = 60000; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java index 7c79101..4081166 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java @@ -7,7 +7,6 @@ import org.springframework.boot.context.properties.ConfigurationProperties; @ConfigurationProperties(prefix = "aiimage.delete-brand-progress") public class DeleteBrandProgressProperties { private long heartbeatTimeoutMinutes = 15; - private long deleteBrandInitialTimeoutMinutes = 20; private String staleCheckCron = "*/30 * * * * *"; /** @@ -19,24 +18,25 @@ public class DeleteBrandProgressProperties { * 商品风险（PRODUCT_RISK_RESOLVE）RUNNING 超时自动收尾/失败，按分钟计算； * 与删除品牌共用同一定时调度。 */ - private long productRiskStaleTimeoutMinutes = 20; - private long productRiskInitialTimeoutMinutes = 20; + private long productRiskStaleTimeoutMinutes = 30; /** * Price track RUNNING timeout auto-finalize/fail threshold in minutes. */ - private long priceTrackStaleTimeoutMinutes = 20; - private long priceTrackInitialTimeoutMinutes = 20; + private long priceTrackStaleTimeoutMinutes = 30; /** * Shop match RUNNING timeout auto-finalize/fail threshold in minutes. */ - private long shopMatchStaleTimeoutMinutes = 20; - private long shopMatchInitialTimeoutMinutes = 20; + private long shopMatchStaleTimeoutMinutes = 30; /** * Patrol delete RUNNING timeout auto-finalize/fail threshold in minutes. */ - private long patrolDeleteStaleTimeoutMinutes = 20; - private long patrolDeleteInitialTimeoutMinutes = 20; + private long patrolDeleteStaleTimeoutMinutes = 30; + + /** + * Query ASIN RUNNING timeout auto-finalize/fail threshold in minutes. + */ + private long queryAsinStaleTimeoutMinutes = 30; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/InstanceRoutingProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/InstanceRoutingProperties.java new file mode 100644 index 0000000..7b92714 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/InstanceRoutingProperties.java @@ -0,0 +1,17 @@ +package com.nanri.aiimage.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +import java.util.LinkedHashMap; +import java.util.Map; + +@Data +@ConfigurationProperties(prefix = "aiimage.instance-routing") +public class InstanceRoutingProperties { + + private Map routes = new LinkedHashMap<>(); + private int connectTimeoutMillis = 3000; + private int readTimeoutMillis = 300000; + private int requestBodyCacheLimitBytes = 104857600; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java index 52f1837..b3b2f0a 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/PropertiesConfig.java @@ -4,6 +4,6 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.context.annotation.Configuration; @Configuration -@EnableConfigurationProperties({OssProperties.class, TransientStorageProperties.class, StorageProperties.class, BrandProgressProperties.class, DeleteBrandProgressProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class, TaskPressureProperties.class, AppearancePatentProperties.class, SimilarAsinProperties.class}) +@EnableConfigurationProperties({OssProperties.class, TransientStorageProperties.class, StorageProperties.class, BrandProgressProperties.class, DeleteBrandProgressProperties.class, BrandCheckProperties.class, ZiniaoProperties.class, ModuleCleanupProperties.class, TaskPressureProperties.class, TaskImageCacheCleanupProperties.class, AppearancePatentProperties.class, SimilarAsinProperties.class, InstanceRoutingProperties.class}) public class PropertiesConfig { } diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java index ae607cb..ab022c9 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/RequestTraceFilter.java @@ -6,8 +6,10 @@ import jakarta.servlet.http.HttpServletRequest; import jakarta.servlet.http.HttpServletResponse; import org.slf4j.Logger; import org.slf4j.LoggerFactory; +import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Component; import org.springframework.web.filter.OncePerRequestFilter; +import org.springframework.web.util.ContentCachingRequestWrapper; import java.io.IOException; import java.util.Optional; @@ -18,27 +20,31 @@ public class RequestTraceFilter extends OncePerRequestFilter { private static final Logger log = LoggerFactory.getLogger(RequestTraceFilter.class); private final InstanceMetadata instanceMetadata; + private final int requestBodyCacheLimitBytes; - public RequestTraceFilter(InstanceMetadata instanceMetadata) { + public RequestTraceFilter(InstanceMetadata instanceMetadata, + @Value("${aiimage.instance-routing.request-body-cache-limit-bytes:104857600}") int requestBodyCacheLimitBytes) { this.instanceMetadata = instanceMetadata; + this.requestBodyCacheLimitBytes = Math.max(1024 * 1024, requestBodyCacheLimitBytes); } @Override protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain) throws ServletException, IOException { long start = System.currentTimeMillis(); + HttpServletRequest requestToUse = wrapRequestIfNeeded(request, requestBodyCacheLimitBytes); String remoteAddr = firstNonBlank( - request.getHeader("X-Forwarded-For"), - request.getHeader("X-Real-IP"), - request.getRemoteAddr() + requestToUse.getHeader("X-Forwarded-For"), + requestToUse.getHeader("X-Real-IP"), + requestToUse.getRemoteAddr() ); - String forwardedProto = request.getHeader("X-Forwarded-Proto"); - String forwardedHost = firstNonBlank(request.getHeader("X-Forwarded-Host"), request.getHeader("Host")); - String forwardedPort = request.getHeader("X-Forwarded-Port"); + String forwardedProto = requestToUse.getHeader("X-Forwarded-Proto"); + String forwardedHost = firstNonBlank(requestToUse.getHeader("X-Forwarded-Host"), requestToUse.getHeader("Host")); + String forwardedPort = requestToUse.getHeader("X-Forwarded-Port"); String requestId = firstNonBlank( - request.getHeader("X-Request-Id"), - request.getHeader("X-Amzn-Trace-Id"), - request.getHeader("Traceparent") + requestToUse.getHeader("X-Request-Id"), + requestToUse.getHeader("X-Amzn-Trace-Id"), + requestToUse.getHeader("Traceparent") ); response.setHeader("X-AIIMAGE-Instance", instanceMetadata.getInstanceId()); @@ -47,7 +53,7 @@ public class RequestTraceFilter extends OncePerRequestFilter { response.setHeader("X-AIIMAGE-Instance-Stable", String.valueOf(instanceMetadata.isStable())); try { - filterChain.doFilter(request, response); + filterChain.doFilter(requestToUse, response); } finally { long costMs = System.currentTimeMillis() - start; log.info( @@ -56,20 +62,34 @@ public class RequestTraceFilter extends OncePerRequestFilter { instanceMetadata.getSource(), instanceMetadata.isStable(), instanceMetadata.getHostname(), - request.getMethod(), - request.getRequestURI(), + requestToUse.getMethod(), + requestToUse.getRequestURI(), response.getStatus(), remoteAddr, blankToDash(forwardedHost), blankToDash(forwardedProto), blankToDash(forwardedPort), blankToDash(requestId), - blankToDash(request.getHeader("User-Agent")), + blankToDash(requestToUse.getHeader("User-Agent")), costMs ); } } + private static HttpServletRequest wrapRequestIfNeeded(HttpServletRequest request, int requestBodyCacheLimitBytes) { + if (request instanceof ContentCachingRequestWrapper) { + return request; + } + String method = request.getMethod(); + if ("POST".equalsIgnoreCase(method) + || "PUT".equalsIgnoreCase(method) + || "PATCH".equalsIgnoreCase(method) + || "DELETE".equalsIgnoreCase(method)) { + return new ContentCachingRequestWrapper(request, requestBodyCacheLimitBytes); + } + return request; + } + private static String firstNonBlank(String... values) { for (String value : values) { if (value != null && !value.isBlank()) { diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskImageCacheCleanupProperties.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskImageCacheCleanupProperties.java new file mode 100644 index 0000000..d45bbb1 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskImageCacheCleanupProperties.java @@ -0,0 +1,15 @@ +package com.nanri.aiimage.config; + +import lombok.Data; +import org.springframework.boot.context.properties.ConfigurationProperties; + +@Data +@ConfigurationProperties(prefix = "aiimage.task-image-cache-cleanup") +public class TaskImageCacheCleanupProperties { + + private boolean enabled = true; + private String cron = "0 30 2 * * *"; + private int retentionDays = 3; + private int batchSize = 5000; + private int maxBatchesPerRun = 200; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java b/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java index 9f39e47..66fd751 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java +++ b/backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java @@ -20,6 +20,8 @@ import java.util.regex.Pattern; @RequiredArgsConstructor public class TaskOperationLockConfig implements WebMvcConfigurer { + private static final String LOCK_ATTRIBUTE = TaskOperationLockConfig.class.getName() + ".LOCK"; + private final TaskDistributedLockService taskDistributedLockService; @Override @@ -28,6 +30,19 @@ public class TaskOperationLockConfig implements WebMvcConfigurer { .addPathPatterns("/api/**"); } + public static boolean releaseRequestLock(HttpServletRequest request) { + if (request == null) { + return false; + } + Object lockHandle = request.getAttribute(LOCK_ATTRIBUTE); + if (lockHandle instanceof TaskDistributedLockService.LockHandle handle) { + request.removeAttribute(LOCK_ATTRIBUTE); + handle.close(); + return true; + } + return false; + } + @Slf4j private static final class TaskOperationLockInterceptor implements HandlerInterceptor { @@ -36,7 +51,6 @@ public class TaskOperationLockConfig implements WebMvcConfigurer { private static final Set MUTATING_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE"); private static final long RESULT_SUBMIT_WAIT_MILLIS = 30 * 1000L; private static final long DELETE_WAIT_MILLIS = 60 * 1000L; - private static final String LOCK_ATTRIBUTE = TaskOperationLockInterceptor.class.getName() + ".LOCK"; private final TaskDistributedLockService taskDistributedLockService; @@ -83,10 +97,7 @@ public class TaskOperationLockConfig implements WebMvcConfigurer { HttpServletResponse response, Object handler, Exception ex) { - Object lockHandle = request.getAttribute(LOCK_ATTRIBUTE); - if (lockHandle instanceof TaskDistributedLockService.LockHandle handle) { - handle.close(); - } + releaseRequestLock(request); } } } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java index 0d05abc..5248765 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java @@ -4,6 +4,7 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.config.AppearancePatentProperties; import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto; +import com.nanri.aiimage.modules.brand.client.BrandCheckClient; import com.nanri.aiimage.modules.coze.service.CozeCredentialPoolService; import lombok.RequiredArgsConstructor; import lombok.extern.slf4j.Slf4j; @@ -29,10 +30,14 @@ public class AppearancePatentCozeClient { private static final String MODULE_TYPE = "APPEARANCE_PATENT"; private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + private static final String INFRINGEMENT = "侵权"; + private static final String NO_INFRINGEMENT = "无侵权"; + private static final String BRAND_QUERY_FAILED = "商标查询失败"; private final AppearancePatentProperties properties; private final ObjectMapper objectMapper; private final CozeCredentialPoolService cozeCredentialPoolService; + private final BrandCheckClient brandCheckClient; private final AtomicLong credentialCursor = new AtomicLong(); public List inspect(List rows, String prompt, String apiKey) { @@ -361,7 +366,7 @@ public class AppearancePatentCozeClient { List rowIds = rows.stream().map(row -> nonBlank(row.getId(), "")).toList(); List asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList(); List countries = rows.stream().map(row -> nonBlank(row.getCountry(), "")).toList(); - List skus = rows.stream().map(row -> nonBlank(row.getSku(), "")).toList(); + List skus = rows.stream().map(row -> sanitizeSku(row.getSku())).toList(); List titles = rows.stream().map(row -> nonBlank(row.getTitle(), row.getAsin())).toList(); List urls = rows.stream().map(row -> nonBlank(row.getUrl(), "")).toList(); @@ -580,10 +585,15 @@ public class AppearancePatentCozeClient { if (row == null || result == null) { return; } - row.setTitleRisk(result.title()); + if (isNoTitleRisk(result)) { + row.setTitleRisk(NO_INFRINGEMENT); + } else { + BrandCheckClient.BrandCheckBatchResult brandCheck = brandCheckClient.checkTitleText(result.title(), "Terms"); + row.setTitleRisk(buildTitleRisk(result.title(), brandCheck)); + } row.setAppearanceRisk(result.appearance()); row.setPatentRisk(result.patent()); - row.setConclusion(result.result()); + row.setConclusion(buildConclusion(row.getTitleRisk(), row.getAppearanceRisk(), result.result())); row.setStatus(result.status()); row.setTitleReason(result.titleReason()); row.setAppearanceReason(result.appearanceReason()); @@ -591,6 +601,55 @@ public class AppearancePatentCozeClient { row.setScore(result.score()); } + private boolean isNoTitleRisk(CozeResult result) { + return result != null + && "无".equals(normalize(result.title())) + && "无".equals(normalize(result.titleReason())); + } + + String buildTitleRisk(String cozeTitle, BrandCheckClient.BrandCheckBatchResult brandCheck) { + List brands = brandCheck == null ? List.of() : brandCheck.brands(); + if (brands == null || brands.isEmpty()) { + return firstNonBlank(cozeTitle, ""); + } + if (brandCheck.hasFailedData()) { + return INFRINGEMENT; + } + if (brandCheck.hasQueryFailedData()) { + return BRAND_QUERY_FAILED; + } + return NO_INFRINGEMENT; + } + + String buildConclusion(String titleRisk, String appearanceRisk, String fallbackConclusion) { + if (isInfringement(titleRisk) || isInfringement(appearanceRisk)) { + return INFRINGEMENT; + } + if (isNoInfringement(titleRisk) && isNoInfringement(appearanceRisk)) { + return NO_INFRINGEMENT; + } + if (isBrandQueryFailed(titleRisk)) { + return BRAND_QUERY_FAILED; + } + return firstNonBlank(fallbackConclusion, ""); + } + + private boolean isInfringement(String value) { + String normalized = normalize(value); + return normalized.contains("侵权") + && !normalized.contains("无侵权") + && !normalized.contains("未侵权"); + } + + private boolean isNoInfringement(String value) { + String normalized = normalize(value); + return normalized.contains("无侵权") || normalized.contains("未侵权"); + } + + private boolean isBrandQueryFailed(String value) { + return normalize(value).contains(BRAND_QUERY_FAILED); + } + private RestClient restClient() { SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory(); requestFactory.setConnectTimeout(properties.getCozeConnectTimeoutMillis()); @@ -1012,6 +1071,32 @@ public class AppearancePatentCozeClient { return value == null || value.isBlank() ? fallback : value; } + private String sanitizeSku(String value) { + String normalized = normalize(value); + if (normalized.isBlank() || isProbablyDescriptionSku(normalized)) { + return ""; + } + return value; + } + + private boolean isProbablyDescriptionSku(String value) { + String normalized = normalize(value); + if (normalized.isBlank()) { + return false; + } + String lower = normalized.toLowerCase(Locale.ROOT); + return normalized.length() > 120 + || normalized.contains("\n") + || lower.contains(" minimum tensile ") + || lower.contains(" multipurpose:") + || lower.contains(" strong, durable") + || lower.contains(" product includes ") + || lower.contains(" perfect gift") + || lower.contains(" high quality ") + || lower.contains(" easy to ") + || lower.contains(" capacity:"); + } + private String firstNonBlank(String preferred, String fallback) { return preferred == null || preferred.isBlank() ? fallback : preferred.trim(); } 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 8d5ff8b..ef0e97b 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 @@ -7,6 +7,7 @@ import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.ObjectMapper; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.exception.TaskOwnerMismatchException; import com.nanri.aiimage.common.service.DistributedJobLockService; import com.nanri.aiimage.common.util.CozeGroupResultPropagator; import com.nanri.aiimage.common.util.FailedStatusRowFilter; @@ -108,6 +109,7 @@ public class AppearancePatentTaskService { private static final String COZE_STATUS_FAILED = "FAILED"; private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet"; private static final String CONTENT_TYPE_ZIP = "application/zip"; + private static final int CHUNK_PAYLOAD_MERGE_RETRY_LIMIT = 3; private static final int RESULT_ROWS_READ_RETRY_LIMIT = 3; private static final long RESULT_ROWS_READ_RETRY_DELAY_MS = 500L; private static final Duration TASK_LOCK_TTL = Duration.ofMinutes(5); @@ -572,7 +574,8 @@ public class AppearancePatentTaskService { public void finalizeStaleTasks() { if (transactionManager != null) { - LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); + LocalDateTime now = LocalDateTime.now(); + LocalDateTime threshold = now.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); List tasks = listStaleFinalizeCandidates(threshold); for (FileTaskEntity task : tasks) { @@ -596,7 +599,8 @@ public class AppearancePatentTaskService { } return; } - LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); + LocalDateTime now = LocalDateTime.now(); + LocalDateTime threshold = now.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); List tasks = listStaleFinalizeCandidates(threshold); for (FileTaskEntity task : tasks) { @@ -647,8 +651,8 @@ public class AppearancePatentTaskService { return fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE) .eq(FileTaskEntity::getStatus, STATUS_RUNNING) - .lt(FileTaskEntity::getCreatedAt, threshold) - .orderByAsc(FileTaskEntity::getCreatedAt) + .lt(FileTaskEntity::getUpdatedAt, threshold) + .orderByAsc(FileTaskEntity::getUpdatedAt) .last("limit 200")); } @@ -811,22 +815,6 @@ public class AppearancePatentTaskService { long activeAssembleJobs = taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE); log.info("[appearance-patent] stale recovery probe taskId={} uploadComplete={} pendingCozeStates={} activeAssembleJobs={} persistedRows={}", taskId, uploadComplete, pendingCozeStates, activeAssembleJobs, hasPersistedResultRows(taskId)); - if (isJavaSideProcessing(taskId)) { - // 防止 Coze 永远 pending 时 stale-recovery 永久 defer： - // 超过 stale-timeout-minutes × 4 仍未推进的 RUNNING 任务，强制走 finalize 链路。 - long deferCeilingMinutes = Math.max(1, properties.getStaleTimeoutMinutes()) * 4L; - LocalDateTime updatedAt = task.getUpdatedAt(); - if (updatedAt != null - && Duration.between(updatedAt, LocalDateTime.now()).toMinutes() >= deferCeilingMinutes) { - log.warn("[appearance-patent] stale recovery defer ceiling exceeded, forcing finalize taskId={} updatedAt={} ceilingMinutes={} pendingCozeStates={} activeAssembleJobs={}", - taskId, updatedAt, deferCeilingMinutes, pendingCozeStates, activeAssembleJobs); - return false; - } - touchJavaSideTaskActivity(taskId); - log.info("[appearance-patent] stale recovery deferred because Java-side processing is still active taskId={} uploadComplete={} pendingCozeStates={} activeAssembleJobs={}", - taskId, uploadComplete, pendingCozeStates, activeAssembleJobs); - return true; - } if (!hasPersistedResultRows(taskId)) { log.info("[appearance-patent] stale recovery aborted because no persisted rows taskId={}", taskId); return false; @@ -839,10 +827,10 @@ public class AppearancePatentTaskService { return false; } int pendingRows = collectPendingCozeCandidates(task, loadSubmittedChunks(taskId)).size(); - log.warn("[appearance-patent] python heartbeat timed out, forcing coze flush taskId={} pendingRows={} pendingCozeStates={} activeAssembleJobs={} forcedScopes={}", + log.warn("[appearance-patent] Python 回传心跳超时，已封口上传并继续 Coze/文件收尾 taskId={} pendingRows={} pendingCozeStates={} activeAssembleJobs={} forcedScopes={}", taskId, pendingRows, pendingCozeStates, activeAssembleJobs, forcedScopes); } else { - log.warn("[appearance-patent] stale running task resuming coze/file assembly after python timeout taskId={} pendingCozeStates={} activeAssembleJobs={}", + log.warn("[appearance-patent] Python 超时恢复继续推进 Coze/文件收尾 taskId={} pendingCozeStates={} activeAssembleJobs={}", taskId, pendingCozeStates, activeAssembleJobs); } submitCozeForSubmittedChunk(new SubmitContext(task, null, null, 0, true, null)); @@ -1029,32 +1017,52 @@ public class AppearancePatentTaskService { } private void mergeChunkPayload(Long taskId, String scopeHash, Integer chunkIndex, List rows) { - TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() - .eq(TaskChunkEntity::getTaskId, taskId) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) - .eq(TaskChunkEntity::getScopeHash, scopeHash) - .eq(TaskChunkEntity::getChunkIndex, chunkIndex) - .last("limit 1")); - if (chunk == null || rows == null || rows.isEmpty()) { + if (rows == null || rows.isEmpty()) { return; } - Map persistedRows = readChunkRows(chunk); - for (AppearancePatentResultRowDto row : rows) { - persistedRows.put(rowKey(row), row); - } - String payloadJson = writeJson(rows == null ? List.of() : rows, "结果序列化失败"); - payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "appearance patent chunk payload merge failed"); - String oldPayload = chunk.getPayloadJson(); - String storedPayload = storeSharedChunkPayloadVersioned(taskId, scopeHash, chunkIndex, payloadJson); - chunk.setPayloadJson(storedPayload); - chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); - chunk.setUpdatedAt(LocalDateTime.now()); - int updated = taskChunkMapper.updateById(chunk); - if (updated <= 0) { + for (int attempt = 1; attempt <= CHUNK_PAYLOAD_MERGE_RETRY_LIMIT; attempt++) { + TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() + .eq(TaskChunkEntity::getTaskId, taskId) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) + .eq(TaskChunkEntity::getScopeHash, scopeHash) + .eq(TaskChunkEntity::getChunkIndex, chunkIndex) + .last("limit 1")); + if (chunk == null) { + return; + } + Map persistedRows = readChunkRows(chunk); + for (AppearancePatentResultRowDto row : rows) { + persistedRows.put(rowKey(row), row); + } + String payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "appearance patent chunk payload merge failed"); + String oldPayload = chunk.getPayloadJson(); + String oldPayloadHash = chunk.getPayloadHash(); + String newPayloadHash = DigestUtil.sha256Hex(payloadJson); + String storedPayload = storeSharedChunkPayloadVersioned(taskId, scopeHash, chunkIndex, payloadJson); + LambdaUpdateWrapper updateWrapper = new LambdaUpdateWrapper() + .eq(TaskChunkEntity::getId, chunk.getId()) + .set(TaskChunkEntity::getPayloadJson, storedPayload) + .set(TaskChunkEntity::getPayloadHash, newPayloadHash) + .set(TaskChunkEntity::getUpdatedAt, LocalDateTime.now()); + if (oldPayloadHash == null) { + updateWrapper.isNull(TaskChunkEntity::getPayloadHash); + } else { + updateWrapper.eq(TaskChunkEntity::getPayloadHash, oldPayloadHash); + } + int updated = taskChunkMapper.update(null, updateWrapper); + if (updated > 0) { + log.debug("[appearance-patent] chunk payload replaced taskId={} scopeHash={} chunk={} oldPayload={} newPayload={} attempt={}", + taskId, scopeHash, chunkIndex, oldPayload, storedPayload, attempt); + transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); + return; + } transientPayloadStorageService.deletePayloadIfPresent(storedPayload); - throw new IllegalStateException("appearance patent chunk payload update failed"); + if (attempt < CHUNK_PAYLOAD_MERGE_RETRY_LIMIT) { + log.warn("[appearance-patent] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}", + taskId, scopeHash, chunkIndex, attempt, CHUNK_PAYLOAD_MERGE_RETRY_LIMIT); + } } - transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); + throw new IllegalStateException("appearance patent chunk payload update conflict"); } private List expandRows(List representatives, @@ -1438,6 +1446,9 @@ public class AppearancePatentTaskService { int cozeWorkUnits = countCozeWorkUnits(chunks, Math.max(1, properties.getCozeBatchSize())); int plannedCozeUnits = Math.max(cozeWorkUnits, countAllCozeStates(task.getId())); int totalProgressUnits = Math.max(3, plannedCozeUnits + 3); + if (countPendingCozeStates(task.getId()) > 0) { + finalizeTimedOutCozeStatesForTask(task.getId()); + } if (countPendingCozeStates(task.getId()) > 0) { taskFileJobService.touchRunning(job.getId()); touchJavaSideTaskActivity(task.getId()); @@ -1605,6 +1616,16 @@ public class AppearancePatentTaskService { int batchSize = Math.max(1, properties.getCozeBatchSize()); List candidates = collectPendingCozeCandidates(task, chunks); boolean flushRemainder = isResultSubmissionComplete(task.getId()); + if (!flushRemainder && !candidates.isEmpty()) { + LocalDateTime jobUpdatedAt = job.getUpdatedAt(); + long pendingFlushMillis = cozeFlushPendingMillis(); + if (jobUpdatedAt != null + && Duration.between(jobUpdatedAt, LocalDateTime.now()).toMillis() >= pendingFlushMillis) { + flushRemainder = true; + log.warn("[appearance-patent] Coze 零头批次等待超时，强制提交 taskId={} jobId={} pendingRows={} batchSize={} jobUpdatedAt={} flushAfterMillis={}", + task.getId(), job.getId(), candidates.size(), batchSize, jobUpdatedAt, pendingFlushMillis); + } + } int submitLimit = (candidates.size() / batchSize) * batchSize; if (flushRemainder && submitLimit < candidates.size()) { submitLimit = candidates.size(); @@ -1869,7 +1890,7 @@ public class AppearancePatentTaskService { : ""; if (!poll.hasPayload() && failureMessage.isBlank()) { failureMessage = isCozeStateTimedOut(state) - ? "Coze async workflow poll timeout" + ? "Coze 异步工作流轮询超时" : "Coze async workflow completed without output"; } List batchRows = readCozeBatchRows(state); @@ -1940,6 +1961,43 @@ public class AppearancePatentTaskService { } } + private void finalizeTimedOutCozeStatesForTask(Long taskId) { + if (taskId == null || taskId <= 0) { + return; + } + List states = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .eq(TaskScopeStateEntity::getTaskId, taskId) + .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE) + .in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING)) + .orderByAsc(TaskScopeStateEntity::getCozeSubmittedAt) + .last("limit 50")); + if (states == null || states.isEmpty()) { + return; + } + for (TaskScopeStateEntity state : states) { + if (state == null || state.getId() == null || !isCozeStateTimedOut(state)) { + continue; + } + CozeBatchContext context = readCozeBatchContext(state); + if (context == null || context.resultId() == null) { + markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze 批次上下文缺失"); + continue; + } + List batchRows = readCozeBatchRows(state); + FileTaskEntity task = fileTaskMapper.selectById(taskId); + if (task != null) { + Map> allRowsByBaseId = loadAllRowsByBaseId(task); + mergeCozeRowsIntoSubmittedChunks(task, + cozeClient.markRowsFailed(batchRows, "Coze 异步工作流轮询超时"), + allRowsByBaseId); + } + markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze 异步工作流轮询超时"); + maybeFinalizeCozeJobLocked(taskId, context); + log.warn("[appearance-patent] 文件任务超时兜底已将 Coze pending 批次置为失败 taskId={} stateId={} jobId={}", + taskId, state.getId(), context.jobId()); + } + } + private boolean retryFailedCozeBatchState(TaskScopeStateEntity state, CozeBatchContext context, List batchRows, @@ -2499,6 +2557,10 @@ public class AppearancePatentTaskService { return Duration.between(state.getCozeSubmittedAt(), LocalDateTime.now()).toMillis() >= timeoutMillis; } + private long cozeFlushPendingMillis() { + return Math.max(1, properties.getCozeFlushPendingMinutes()) * 60_000L; + } + private int cozeAttemptCount(TaskScopeStateEntity state) { return state == null || state.getCozeAttemptCount() == null ? 0 : state.getCozeAttemptCount(); } @@ -2549,11 +2611,6 @@ public class AppearancePatentTaskService { return count != null && count > 0; } - private boolean isJavaSideProcessing(Long taskId) { - return countPendingCozeStates(taskId) > 0 - || taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE) > 0; - } - private void touchJavaSideTaskActivity(Long taskId) { if (taskId == null || taskId <= 0) { return; @@ -2664,13 +2721,27 @@ public class AppearancePatentTaskService { log.warn("[appearance-patent] 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 提交凭证超时：" + message; + mergeCozeRowsIntoSubmittedChunks(task, + cozeClient.markRowsFailed(batchRows, finalMessage), + allRowsByBaseId); + markCozeStateTerminal(state, COZE_STATUS_FAILED, finalMessage); + maybeFinalizeCozeJobLocked(state.getTaskId(), context); + log.warn("[appearance-patent] Coze pending 提交等待超时，已将批次置为失败 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 + "（超过 " + properties.getCozePollTimeoutMillis() + "ms）" + : message + "，已重试提交 " + nextAttemptCount + " 次"; mergeCozeRowsIntoSubmittedChunks(task, cozeClient.markRowsFailed(batchRows, finalMessage), allRowsByBaseId); @@ -2711,7 +2782,11 @@ public class AppearancePatentTaskService { } log.warn("[appearance-patent] reject task operation because owner is another instance taskId={} operation={} owner={} current={}", task == null ? null : task.getId(), operation, owner, currentInstanceId()); - throw new BusinessException(40903, "该任务已绑定到另一台服务实例处理，请通过原实例继续处理"); + throw new TaskOwnerMismatchException( + task == null ? null : task.getId(), + operation, + owner, + currentInstanceId()); } private boolean isCozeStateOwnedByCurrentInstance(TaskScopeStateEntity state) { @@ -3909,6 +3984,7 @@ public class AppearancePatentTaskService { if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) { throw new BusinessException("任务不存在"); } + ensureTaskOwnedByCurrentInstance(task, "parsed payload"); return readParsedPayload(task); } @@ -3917,6 +3993,7 @@ public class AppearancePatentTaskService { if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) { throw new BusinessException("任务不存在"); } + ensureTaskOwnedByCurrentInstance(task, "queue payload"); AppearancePatentParsedPayloadDto payload = readParsedPayload(task); AppearancePatentParsedPayloadDto queuePayload = new AppearancePatentParsedPayloadDto(); queuePayload.setAiPrompt(payload.getAiPrompt()); @@ -4004,6 +4081,8 @@ public class AppearancePatentTaskService { } catch (Exception ex) { log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}", chunk.getTaskId(), chunk.getChunkIndex(), ex.getMessage()); + throw new BusinessException("appearance patent chunk payload read failed chunk=" + + chunk.getChunkIndex() + ": " + ex.getMessage(), ex); } return rows; } @@ -4097,13 +4176,6 @@ public class AppearancePatentTaskService { return firstNonBlank(row.getConclusion(), ""); } - /** - * 导出最终 xlsx 时的状态展示： - * - 如果 status 是 markFailed 合成的（failureSyntheticStatus=true）， - * 且 row 实际并没有任何风险维度结果，导出层展示错误信息（有则放入，无则留空）， - * 避免用户原样重新上传时该行被反复识别为失败行重新触发 Coze。 - * - 真正业务侧失败 / Python 真实回传 FAILED 仍按原值显示。 - */ private String userFacingStatus(AppearancePatentResultRowDto row) { if (row == null) { return ""; @@ -4112,7 +4184,6 @@ public class AppearancePatentTaskService { if (!row.isFailureSyntheticStatus()) { return original; } - // 合成状态：有错误信息则放入错误信息，没有则留空 return firstNonBlank(row.getError(), ""); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java new file mode 100644 index 0000000..683d4e6 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/client/BrandCheckClient.java @@ -0,0 +1,200 @@ +package com.nanri.aiimage.modules.brand.client; + +import com.fasterxml.jackson.annotation.JsonAlias; +import com.nanri.aiimage.config.BrandCheckProperties; +import lombok.Data; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.http.HttpHeaders; +import org.springframework.http.MediaType; +import org.springframework.http.client.SimpleClientHttpRequestFactory; +import org.springframework.stereotype.Component; +import org.springframework.web.client.RestClient; + +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; + +@Component +@RequiredArgsConstructor +@Slf4j +public class BrandCheckClient { + + private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8); + + private final BrandCheckProperties properties; + private volatile RestClient sharedRestClient; + + public BrandCheckResponse check(String brand) { + return check(brand, properties.getDefaultStrategy()); + } + + public BrandCheckResponse check(String brand, String strategy) { + String normalizedBrand = normalize(brand); + if (normalizedBrand.isBlank()) { + return BrandCheckResponse.empty(); + } + Map body = new LinkedHashMap<>(); + body.put("brand", normalizedBrand); + body.put("strategy", normalize(strategy).isBlank() ? "Terms" : normalize(strategy)); + return restClient().post() + .uri(joinUrl(properties.getBaseUrl(), properties.getPath())) + .headers(headers -> { + headers.set(HttpHeaders.ACCEPT, MediaType.APPLICATION_JSON_VALUE); + headers.setContentType(APPLICATION_JSON_UTF8); + if (properties.getToken() != null && !properties.getToken().isBlank()) { + headers.set("X-Token", properties.getToken().trim()); + } + }) + .body(body) + .retrieve() + .body(BrandCheckResponse.class); + } + + public BrandCheckBatchResult checkAll(List brands, String strategy) { + List distinctBrands = distinctNonBlank(brands); + List failedData = new ArrayList<>(); + List queryFailedData = new ArrayList<>(); + for (String brand : distinctBrands) { + try { + BrandCheckResponse response = check(brand, strategy); + if (response == null) { + queryFailedData.add(brand); + continue; + } + failedData.addAll(nullToEmpty(response.getFaildData())); + queryFailedData.addAll(nullToEmpty(response.getQueryFaildData())); + } catch (Exception ex) { + log.warn("[brand-check] request failed brand={} strategy={} err={}", brand, strategy, ex.getMessage()); + queryFailedData.add(brand); + } + } + return new BrandCheckBatchResult(distinctBrands, failedData, queryFailedData); + } + + public BrandCheckBatchResult checkTitleText(String titleText) { + return checkTitleText(titleText, properties.getDefaultStrategy()); + } + + public BrandCheckBatchResult checkTitleText(String titleText, String strategy) { + return checkAll(splitTitleText(titleText), strategy); + } + + public List splitTitleText(String titleText) { + String normalized = normalize(titleText) + .replace('，', ',') + .replace('、', ',') + .replace('；', ',') + .replace(';', ',') + .replace('\n', ',') + .replace('\r', ',') + .replace('\t', ','); + if (normalized.isBlank()) { + return List.of(); + } + List result = new ArrayList<>(); + for (String item : normalized.split(",")) { + String brand = cleanTitleToken(item); + if (!brand.isBlank() && !result.contains(brand)) { + result.add(brand); + } + } + return result; + } + + private RestClient restClient() { + RestClient client = sharedRestClient; + if (client != null) { + return client; + } + synchronized (this) { + if (sharedRestClient == null) { + SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory(); + requestFactory.setConnectTimeout(properties.getConnectTimeoutMillis()); + requestFactory.setReadTimeout(properties.getReadTimeoutMillis()); + sharedRestClient = RestClient.builder().requestFactory(requestFactory).build(); + } + return sharedRestClient; + } + } + + private List nullToEmpty(List values) { + return values == null ? List.of() : values; + } + + private String cleanTitleToken(String value) { + String normalized = normalize(value); + while (normalized.startsWith("'") + || normalized.startsWith("\"") + || normalized.startsWith("“") + || normalized.startsWith("‘")) { + normalized = normalized.substring(1).trim(); + } + while (normalized.endsWith("'") + || normalized.endsWith("\"") + || normalized.endsWith("”") + || normalized.endsWith("’")) { + normalized = normalized.substring(0, normalized.length() - 1).trim(); + } + return normalized; + } + + private List distinctNonBlank(List values) { + List result = new ArrayList<>(); + if (values == null) { + return result; + } + for (String value : values) { + String normalized = normalize(value); + if (!normalized.isBlank() && !result.contains(normalized)) { + result.add(normalized); + } + } + return result; + } + + private String joinUrl(String baseUrl, String path) { + String base = baseUrl == null ? "" : baseUrl.trim(); + String suffix = path == null ? "" : path.trim(); + if (base.endsWith("/") && suffix.startsWith("/")) { + return base + suffix.substring(1); + } + if (!base.endsWith("/") && !suffix.startsWith("/")) { + return base + "/" + suffix; + } + return base + suffix; + } + + private String normalize(String value) { + return value == null ? "" : value.trim(); + } + + @Data + public static class BrandCheckResponse { + @JsonAlias({"faild_data", "failed_data"}) + private List faildData = new ArrayList<>(); + + @JsonAlias({"query_faild_data", "query_failed_data"}) + private List queryFaildData = new ArrayList<>(); + + private static BrandCheckResponse empty() { + return new BrandCheckResponse(); + } + } + + public record BrandCheckBatchResult( + List brands, + List faildData, + List queryFaildData + ) { + public boolean hasFailedData() { + return faildData != null && !faildData.isEmpty(); + } + + public boolean hasQueryFailedData() { + return queryFaildData != null && !queryFaildData.isEmpty(); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java index 26f35bd..03a652d 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskProgressCacheService.java @@ -80,6 +80,29 @@ public class BrandTaskProgressCacheService { } } + public void touchHeartbeat(Long taskId, String phase, Integer current, Integer total) { + String key = buildKey(taskId); + String now = String.valueOf(Instant.now().toEpochMilli()); + Map values = new LinkedHashMap<>(); + values.put("updated_at", now); + values.put("last_heartbeat_at", now); + if (phase != null && !phase.isBlank()) { + values.put("phase", normalizePhase(phase)); + } + if (current != null) { + values.put("current_line", String.valueOf(Math.max(current, 0))); + } + if (total != null) { + values.put("total_lines", String.valueOf(Math.max(total, 0))); + } + try { + stringRedisTemplate.opsForHash().putAll(key, values); + stringRedisTemplate.expire(key, ttl()); + } catch (Exception ex) { + log.warn("[brand-progress-cache] touch heartbeat degraded taskId={} msg={}", taskId, ex.getMessage()); + } + } + public void markFailed(Long taskId, String message) { String key = buildKey(taskId); String now = String.valueOf(Instant.now().toEpochMilli()); diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java index b1b49b5..84146e3 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java @@ -154,12 +154,10 @@ public class DeleteBrandStaleTaskService { private void failStaleDeleteBrandTasks() { long minutes = Math.max(1L, deleteBrandProgressProperties.getHeartbeatTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getDeleteBrandInitialTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND) @@ -187,9 +185,6 @@ public class DeleteBrandStaleTaskService { if (lastHeartbeatAt > 0L && nowMillis - lastHeartbeatAt < staleTimeoutMillis) { continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_DELETE_BRAND, task.getId()); if (taskLockHandle == null) { @@ -209,16 +204,14 @@ public class DeleteBrandStaleTaskService { log.warn("[stale-check] delete-brand finalize threw taskId={} msg={}", task.getId(), ex.getMessage()); } - log.warn("[stale-check] delete-brand failing stale task -> taskId={} updatedAt={} createdAt={} lastHeartbeatAt={} timeoutMinutes={} initialTimeoutMinutes={} completedScopes={} hasStartedProgress={} initialThreshold={}", + log.warn("[stale-check] delete-brand failing stale task -> taskId={} updatedAt={} createdAt={} lastHeartbeatAt={} timeoutMinutes={} completedScopes={} hasStartedProgress={}", task.getId(), task.getUpdatedAt(), task.getCreatedAt(), lastHeartbeatAt, minutes, - initialMinutes, completedScopeCount, - hasStartedProgress, - initialThreshold); + hasStartedProgress); int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() .eq(FileTaskEntity::getId, task.getId()) .eq(FileTaskEntity::getModuleType, MODULE_TYPE_DELETE_BRAND) @@ -239,12 +232,10 @@ public class DeleteBrandStaleTaskService { private ProductRiskStaleCheckStats failStaleProductRiskResolveTasks() { ProductRiskStaleCheckStats stats = new ProductRiskStaleCheckStats(); long minutes = Math.max(1L, deleteBrandProgressProperties.getProductRiskStaleTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getProductRiskInitialTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRODUCT_RISK) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -256,8 +247,8 @@ public class DeleteBrandStaleTaskService { } Map heartbeatByTaskId = productRiskTaskCacheService.getTaskHeartbeatMillisBatch( runningTasks.stream().map(FileTaskEntity::getId).toList()); - log.info("[stale-check] product-risk candidates={} threshold={} timeoutMinutes={} initialThreshold={} initialTimeoutMinutes={}", - runningTasks.size(), threshold, minutes, initialThreshold, initialMinutes); + log.info("[stale-check] product-risk candidates={} threshold={} timeoutMinutes={}", + runningTasks.size(), threshold, minutes); for (FileTaskEntity task : runningTasks) { long lastPayloadHeartbeatMillis = heartbeatByTaskId.getOrDefault(task.getId(), 0L); boolean hasFinishedRows = (task.getSuccessFileCount() != null && task.getSuccessFileCount() > 0) @@ -280,12 +271,6 @@ public class DeleteBrandStaleTaskService { taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_PRODUCT_RISK)); continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - stats.skippedTaskCount++; - log.info("[stale-check] product-risk skip initial-grace taskId={} createdAt={} initialThreshold={}", - task.getId(), task.getCreatedAt(), initialThreshold); - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRODUCT_RISK, task.getId()); if (taskLockHandle == null) { stats.skippedTaskCount++; @@ -323,12 +308,10 @@ public class DeleteBrandStaleTaskService { private PriceTrackStaleCheckStats failStalePriceTrackTasks() { PriceTrackStaleCheckStats stats = new PriceTrackStaleCheckStats(); long minutes = Math.max(1L, deleteBrandProgressProperties.getPriceTrackStaleTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getPriceTrackInitialTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRICE_TRACK) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -340,8 +323,8 @@ public class DeleteBrandStaleTaskService { } Map heartbeatByTaskId = priceTrackTaskCacheService.getTaskHeartbeatMillisBatch( runningTasks.stream().map(FileTaskEntity::getId).toList()); - log.info("[stale-check] price-track candidates={} threshold={} timeoutMinutes={} initialThreshold={} initialTimeoutMinutes={}", - runningTasks.size(), threshold, minutes, initialThreshold, initialMinutes); + log.info("[stale-check] price-track candidates={} threshold={} timeoutMinutes={}", + runningTasks.size(), threshold, minutes); for (FileTaskEntity task : runningTasks) { long lastHeartbeatMillis = heartbeatByTaskId.getOrDefault(task.getId(), 0L); boolean hasFinishedRows = (task.getSuccessFileCount() != null && task.getSuccessFileCount() > 0) @@ -362,10 +345,6 @@ public class DeleteBrandStaleTaskService { taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_PRICE_TRACK)); continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - stats.skippedTaskCount++; - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRICE_TRACK, task.getId()); if (taskLockHandle == null) { stats.skippedTaskCount++; @@ -404,12 +383,10 @@ public class DeleteBrandStaleTaskService { private ShopMatchStaleCheckStats failStaleShopMatchTasks() { ShopMatchStaleCheckStats stats = new ShopMatchStaleCheckStats(); long minutes = Math.max(1L, deleteBrandProgressProperties.getShopMatchStaleTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getShopMatchInitialTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_SHOP_MATCH) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -421,8 +398,8 @@ public class DeleteBrandStaleTaskService { } Map heartbeatByTaskId = shopMatchTaskCacheService.getTaskHeartbeatMillisBatch( runningTasks.stream().map(FileTaskEntity::getId).toList()); - log.info("[stale-check] shop-match candidates={} threshold={} timeoutMinutes={} initialThreshold={} initialTimeoutMinutes={}", - runningTasks.size(), threshold, minutes, initialThreshold, initialMinutes); + log.info("[stale-check] shop-match candidates={} threshold={} timeoutMinutes={}", + runningTasks.size(), threshold, minutes); for (FileTaskEntity task : runningTasks) { long lastHeartbeatMillis = heartbeatByTaskId.getOrDefault(task.getId(), 0L); boolean hasFinishedRows = (task.getSuccessFileCount() != null && task.getSuccessFileCount() > 0) @@ -445,12 +422,6 @@ public class DeleteBrandStaleTaskService { taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_SHOP_MATCH)); continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - stats.skippedTaskCount++; - log.info("[stale-check] shop-match skip initial-grace taskId={} createdAt={} initialThreshold={}", - task.getId(), task.getCreatedAt(), initialThreshold); - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_SHOP_MATCH, task.getId()); if (taskLockHandle == null) { stats.skippedTaskCount++; @@ -489,12 +460,10 @@ public class DeleteBrandStaleTaskService { private ShopMatchStaleCheckStats failStalePatrolDeleteTasks() { ShopMatchStaleCheckStats stats = new ShopMatchStaleCheckStats(); long minutes = Math.max(1L, deleteBrandProgressProperties.getPatrolDeleteStaleTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getPatrolDeleteInitialTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_PATROL_DELETE) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -526,10 +495,6 @@ public class DeleteBrandStaleTaskService { taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_PATROL_DELETE)); continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - stats.skippedTaskCount++; - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PATROL_DELETE, task.getId()); if (taskLockHandle == null) { stats.skippedTaskCount++; @@ -566,13 +531,11 @@ public class DeleteBrandStaleTaskService { private ShopMatchStaleCheckStats failStaleQueryAsinTasks() { ShopMatchStaleCheckStats stats = new ShopMatchStaleCheckStats(); - long minutes = Math.max(1L, deleteBrandProgressProperties.getPatrolDeleteStaleTimeoutMinutes()); - long initialMinutes = Math.max(minutes, deleteBrandProgressProperties.getPatrolDeleteInitialTimeoutMinutes()); + long minutes = Math.max(1L, deleteBrandProgressProperties.getQueryAsinStaleTimeoutMinutes()); long staleTimeoutMillis = Duration.ofMinutes(minutes).toMillis(); long nowMillis = System.currentTimeMillis(); LocalDateTime now = LocalDateTime.now(); LocalDateTime threshold = now.minusMinutes(minutes); - LocalDateTime initialThreshold = now.minusMinutes(initialMinutes); java.util.List runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE_QUERY_ASIN) .eq(FileTaskEntity::getStatus, "RUNNING") @@ -604,10 +567,6 @@ public class DeleteBrandStaleTaskService { taskFileJobService.countActiveAssembleJobs(task.getId(), MODULE_TYPE_QUERY_ASIN)); continue; } - if (!hasStartedProgress && isWithinInitialGrace(task, initialThreshold)) { - stats.skippedTaskCount++; - continue; - } TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_QUERY_ASIN, task.getId()); if (taskLockHandle == null) { stats.skippedTaskCount++; @@ -643,14 +602,6 @@ public class DeleteBrandStaleTaskService { return stats; } - private boolean isWithinInitialGrace(FileTaskEntity task, LocalDateTime initialThreshold) { - if (task == null || initialThreshold == null) { - return false; - } - LocalDateTime baseline = task.getUpdatedAt() != null ? task.getUpdatedAt() : task.getCreatedAt(); - return baseline != null && baseline.isAfter(initialThreshold); - } - private TaskDistributedLockService.LockHandle acquireTaskLock(String moduleType, Long taskId) { TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(moduleType, taskId, 0L); if (lockHandle == null) { 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 4cec407..23181af 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 @@ -14,6 +14,7 @@ import org.springframework.stereotype.Component; import org.springframework.util.StreamUtils; import org.springframework.web.client.RestClient; +import java.math.BigDecimal; import java.nio.charset.StandardCharsets; import java.util.ArrayList; import java.util.LinkedHashMap; @@ -370,8 +371,9 @@ public class SimilarAsinCozeClient { List skus = rows.stream().map(row -> safeText(row.getSku())).toList(); List urls = rows.stream().map(this::primaryImageUrl).toList(); List> urlLists = rows.stream().map(this::imageUrls).toList(); + List>> alibabaLists = rows.stream().map(this::alibabaItems).toList(); - List> items = buildItemObjects(asins, titles, skus, urls, urlLists); + List> items = buildItemObjects(asins, titles, skus, urls, urlLists, alibabaLists); logCozeItemsDiff(rows, items); Map parameters = new LinkedHashMap<>(); parameters.put("items", items); @@ -408,22 +410,26 @@ public class SimilarAsinCozeClient { List rowUrls = safeUrls(row.getUrls()); String rowUrl = safeText(row.getUrl()); Object outUrl = item.get("url"); + Object outAlibaba = item.get("alibaba"); Object outTargetUrls = item.get("target_urls"); int outTargetSize = (outTargetUrls instanceof List list) ? list.size() : 0; + int outAlibabaSize = (outAlibaba instanceof List list) ? list.size() : 0; boolean urlMatch = String.valueOf(outUrl == null ? "" : outUrl).equals(rowUrl); boolean targetMatch = (outTargetUrls instanceof List outList) && outList.size() == rowUrls.size() && outList.equals(rowUrls); - log.info("[similar-asin] coze item idx={} asin={} title={} sku={} rowUrl={} rowUrlsSize={} rowUrlsHead={} rowUrlsTail={} outUrl={} outTargetSize={} outTargetHead={} outTargetTail={} urlMatch={} targetUrlsMatch={}", + log.info("[similar-asin] coze item idx={} asin={} title={} sku={} price={} rowUrl={} rowUrlsSize={} rowUrlsHead={} rowUrlsTail={} outUrl={} outAlibabaSize={} outTargetSize={} outTargetHead={} outTargetTail={} urlMatch={} targetUrlsMatch={}", i, safeText(row.getAsin()), abbreviate(safeText(row.getTitle()), 80), safeText(row.getSku()), + safeText(row.getPrice()), abbreviate(rowUrl, 200), rowUrls.size(), rowUrls.isEmpty() ? "" : abbreviate(rowUrls.get(0), 200), rowUrls.size() <= 1 ? "" : abbreviate(rowUrls.get(rowUrls.size() - 1), 200), abbreviate(String.valueOf(outUrl == null ? "" : outUrl), 200), + outAlibabaSize, outTargetSize, (outTargetUrls instanceof List headList && !headList.isEmpty()) ? abbreviate(String.valueOf(headList.get(0)), 200) : "", @@ -468,7 +474,8 @@ public class SimilarAsinCozeClient { List titles, List skus, List urls, - List> urlLists) { + List> urlLists, + List>> alibabaLists) { // legacy flag：保留切回旧字段顺序 {asin, sku, url, target_urls, title} 的开关， // 默认 false 使用当前顺序 {asin, url, target_urls, title, sku}。 boolean useLegacyOrder = properties.isCozeUseLegacyItemFieldOrder(); @@ -476,16 +483,19 @@ public class SimilarAsinCozeClient { for (int i = 0; i < asins.size(); i++) { String url = urls.get(i); List imageUrls = urlLists.get(i); + List> alibabaItems = alibabaLists.get(i); Map item = new LinkedHashMap<>(); if (useLegacyOrder) { item.put("asin", asins.get(i)); item.put("sku", skus.get(i)); item.put("url", url); + item.put("alibaba", alibabaItems); item.put("target_urls", imageUrls); item.put("title", titles.get(i)); } else { item.put("asin", asins.get(i)); item.put("url", url); + item.put("alibaba", alibabaItems); item.put("target_urls", imageUrls); item.put("title", titles.get(i)); item.put("sku", skus.get(i)); @@ -726,6 +736,7 @@ public class SimilarAsinCozeClient { row.setSku(source.getSku()); row.setPrice(source.getPrice()); row.setUrls(source.getUrls()); + row.setAlibaba(source.getAlibaba()); row.setTitle(source.getTitle()); row.setError(source.getError()); row.setDone(source.getDone()); @@ -1213,6 +1224,69 @@ public class SimilarAsinCozeClient { return primaryUrl.isBlank() ? List.of() : List.of(primaryUrl); } + private List> alibabaItems(SimilarAsinResultRowDto row) { + if (row == null) { + return List.of(); + } + List explicitItems = row.getAlibaba(); + if (explicitItems != null && !explicitItems.isEmpty()) { + List> result = new ArrayList<>(explicitItems.size()); + for (SimilarAsinResultRowDto.AlibabaItem source : explicitItems) { + if (source == null) { + continue; + } + String sourceUrl = safeText(source.getUrl()); + Object sourcePrice = cozePrice(source.getRawPrice()); + if (sourceUrl.isBlank() && isBlankPrice(sourcePrice)) { + continue; + } + Map item = new LinkedHashMap<>(); + item.put("url", sourceUrl); + item.put("price", sourcePrice); + result.add(item); + } + return result; + } + List imageUrls = imageUrls(row); + if (imageUrls.isEmpty()) { + return List.of(); + } + Object rowPrice = cozePrice(row.getPrice()); + List> result = new ArrayList<>(imageUrls.size()); + for (String imageUrl : imageUrls) { + Map item = new LinkedHashMap<>(); + item.put("url", imageUrl); + item.put("price", rowPrice); + result.add(item); + } + return result; + } + + private Object cozePrice(Object value) { + if (value == null) { + return ""; + } + if (value instanceof BigDecimal decimal) { + return decimal.stripTrailingZeros(); + } + if (value instanceof Number number) { + return new BigDecimal(number.toString()).stripTrailingZeros(); + } + String text = safeText(String.valueOf(value)); + if (text.isBlank()) { + return ""; + } + try { + return new BigDecimal(text).stripTrailingZeros(); + } catch (NumberFormatException ex) { + return text; + } + } + + private boolean isBlankPrice(Object value) { + return value == null || value instanceof String text && text.isBlank(); + } + private String cozeGroupKey(SimilarAsinResultRowDto row) { if (row == null) { return ""; 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 f6cc6e1..cb2c9c3 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 @@ -44,6 +44,10 @@ public class SimilarAsinResultRowDto { @Schema(description = "商品价格。来自 Excel 解析或 Python 回传。", example = "12.29") private String price; + @JsonAlias({"alibaba", "alibaba_items", "alibabaItems", "alibaba_products", "alibabaProducts"}) + @Schema(description = "Alibaba candidates returned by Python, each item contains url and price") + private List alibaba = new ArrayList<>(); + @Schema(description = "商品主图 URL。Python 端解析的代表图。允许传字符串或数组（数组取首个非空），与 urls 互不影响。", example = "https://webstatic.aiproxy.vip/output/20260425/103322/demo.jpg") private String url; @@ -150,7 +154,11 @@ public class SimilarAsinResultRowDto { public List getUrls() { // 直接返回 Python 回传的 urls，不再把 url 强推到首位、不去重、不截断。 - return cleanUrls(urls); + List cleaned = cleanUrls(urls); + if (!cleaned.isEmpty()) { + return cleaned; + } + return alibabaUrls(); } @JsonSetter("urls") @@ -169,12 +177,30 @@ public class SimilarAsinResultRowDto { this.urls = normalized; } + public List getAlibaba() { + if (alibaba == null || alibaba.isEmpty()) { + return new ArrayList<>(); + } + List result = new ArrayList<>(alibaba.size()); + for (AlibabaItem item : alibaba) { + if (item == null || item.getUrl().isBlank() && item.getPrice().isBlank()) { + continue; + } + result.add(item); + } + return result; + } + + public void setAlibaba(List alibaba) { + this.alibaba = alibaba == null ? new ArrayList<>() : new ArrayList<>(alibaba); + } + public boolean hasImageUrl() { // url（主图）和 urls（同类商品图）任一存在即可作为可送 Coze 的素材。 if (url != null && !url.isBlank()) { return true; } - return !cleanUrls(urls).isEmpty(); + return !getUrls().isEmpty(); } private static void appendUrls(List result, Object value) { @@ -227,4 +253,44 @@ public class SimilarAsinResultRowDto { } return result; } + + private List alibabaUrls() { + List items = getAlibaba(); + if (items.isEmpty()) { + return new ArrayList<>(); + } + List result = new ArrayList<>(items.size()); + for (AlibabaItem item : items) { + String itemUrl = item.getUrl(); + if (!itemUrl.isBlank()) { + result.add(itemUrl); + } + } + return result; + } + + @Data + @Schema(description = "Alibaba candidate") + public static class AlibabaItem { + + @JsonAlias({"url", "image_url", "imageUrl", "img_url", "imgUrl", "link"}) + @Schema(description = "Alibaba image or product URL") + private String url; + + @JsonAlias({"price", "浠锋牸"}) + @Schema(description = "Alibaba candidate price") + private Object price; + + public String getUrl() { + return url == null ? "" : url.trim(); + } + + public String getPrice() { + return price == null ? "" : String.valueOf(price).trim(); + } + + public Object getRawPrice() { + return price; + } + } } 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 0901886..e2b2928 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 @@ -9,6 +9,7 @@ import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.node.ArrayNode; import com.fasterxml.jackson.databind.node.ObjectNode; import com.nanri.aiimage.common.exception.BusinessException; +import com.nanri.aiimage.common.exception.TaskOwnerMismatchException; import com.nanri.aiimage.common.service.DistributedJobLockService; import com.nanri.aiimage.common.util.CozeGroupResultPropagator; import com.nanri.aiimage.common.util.FailedStatusRowFilter; @@ -205,6 +206,8 @@ public class SimilarAsinTaskService { "id", "asin", "国家", + "卖家名称", + "品牌", "是否有货", "相似度", "是否符合类目", @@ -216,9 +219,9 @@ public class SimilarAsinTaskService { "阿里巴巴图片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 static final int IMG_COL_MAIN = 11; + private static final int IMG_COL_PUZZLE1 = 12; + private static final int IMG_COL_PUZZLE2 = 13; private final LocalFileStorageService localFileStorageService; private final OssStorageService ossStorageService; @@ -471,6 +474,7 @@ public class SimilarAsinTaskService { if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) { throw new BusinessException("任务不存在"); } + ensureTaskOwnedByCurrentInstance(task, "activate"); if (STATUS_SUCCESS.equals(task.getStatus()) || STATUS_FAILED.equals(task.getStatus())) { throw new BusinessException("任务已结束"); } @@ -805,7 +809,8 @@ public class SimilarAsinTaskService { public void finalizeStaleTasks() { if (transactionManager != null) { - LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); + LocalDateTime now = LocalDateTime.now(); + LocalDateTime threshold = now.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); List tasks = listStaleFinalizeCandidates(threshold); for (FileTaskEntity task : tasks) { @@ -829,7 +834,8 @@ public class SimilarAsinTaskService { } return; } - LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); + LocalDateTime now = LocalDateTime.now(); + LocalDateTime threshold = now.minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes())); long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); List tasks = listStaleFinalizeCandidates(threshold); for (FileTaskEntity task : tasks) { @@ -880,8 +886,8 @@ public class SimilarAsinTaskService { return fileTaskMapper.selectList(new LambdaQueryWrapper() .eq(FileTaskEntity::getModuleType, MODULE_TYPE) .eq(FileTaskEntity::getStatus, STATUS_RUNNING) - .lt(FileTaskEntity::getCreatedAt, threshold) - .orderByAsc(FileTaskEntity::getCreatedAt) + .lt(FileTaskEntity::getUpdatedAt, threshold) + .orderByAsc(FileTaskEntity::getUpdatedAt) .last("limit 200")); } @@ -1071,22 +1077,6 @@ public class SimilarAsinTaskService { boolean uploadComplete = isResultSubmissionComplete(taskId); long pendingCozeStates = countPendingCozeStates(taskId); long activeAssembleJobs = taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE); - if (isJavaSideProcessing(taskId)) { - // 防止 Coze 永远 pending 时 stale-recovery 永久 defer： - // 超过 stale-timeout-minutes × 4 仍未推进的 RUNNING 任务，强制走 finalize 链路。 - long deferCeilingMinutes = Math.max(1, properties.getStaleTimeoutMinutes()) * 4L; - LocalDateTime updatedAt = task.getUpdatedAt(); - if (updatedAt != null - && Duration.between(updatedAt, LocalDateTime.now()).toMinutes() >= deferCeilingMinutes) { - log.warn("[similar-asin] stale recovery defer ceiling exceeded, forcing finalize taskId={} updatedAt={} ceilingMinutes={} pendingCozeStates={} activeAssembleJobs={}", - taskId, updatedAt, deferCeilingMinutes, pendingCozeStates, activeAssembleJobs); - return false; - } - touchJavaSideTaskActivity(taskId); - log.info("[similar-asin] stale recovery deferred because Java-side processing is still active taskId={} uploadComplete={} pendingCozeStates={} activeAssembleJobs={}", - taskId, uploadComplete, pendingCozeStates, activeAssembleJobs); - return true; - } if (!hasPersistedResultRows(taskId)) { return false; } @@ -1095,10 +1085,10 @@ public class SimilarAsinTaskService { if (forcedScopes <= 0) { return false; } - log.warn("[similar-asin] python heartbeat timed out, forcing finalize pipeline taskId={} pendingCozeStates={} activeAssembleJobs={} forcedScopes={}", + log.warn("[similar-asin] Python 回传心跳超时，已封口上传并继续 Coze/文件收尾 taskId={} pendingCozeStates={} activeAssembleJobs={} forcedScopes={}", taskId, pendingCozeStates, activeAssembleJobs, forcedScopes); } else { - log.warn("[similar-asin] stale running task resuming coze/file assembly after python timeout taskId={} pendingCozeStates={} activeAssembleJobs={}", + log.warn("[similar-asin] Python 超时恢复继续推进 Coze/文件收尾 taskId={} pendingCozeStates={} activeAssembleJobs={}", taskId, pendingCozeStates, activeAssembleJobs); } submitCozeForSubmittedChunk(new SubmitContext(task, null, null, null, true, null)); @@ -1272,32 +1262,48 @@ public class SimilarAsinTaskService { } private void mergeChunkPayload(Long taskId, String scopeHash, Integer chunkIndex, List rows) { - TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() - .eq(TaskChunkEntity::getTaskId, taskId) - .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) - .eq(TaskChunkEntity::getScopeHash, scopeHash) - .eq(TaskChunkEntity::getChunkIndex, chunkIndex) - .last("limit 1")); - if (chunk == null || rows == null || rows.isEmpty()) { + if (rows == null || rows.isEmpty()) { return; } - Map persistedRows = readChunkRows(chunk); - for (SimilarAsinResultRowDto row : rows) { - persistedRows.put(rowKey(row), row); - } - String payloadJson = writeJson(rows == null ? List.of() : rows, "结果序列化失败"); - payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败"); - String oldPayload = chunk.getPayloadJson(); - String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); - chunk.setPayloadJson(storedPayload); - chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson)); - chunk.setUpdatedAt(LocalDateTime.now()); - int updated = taskChunkMapper.updateById(chunk); - if (updated <= 0) { + int maxAttempts = 3; + for (int attempt = 1; attempt <= maxAttempts; attempt++) { + TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper() + .eq(TaskChunkEntity::getTaskId, taskId) + .eq(TaskChunkEntity::getModuleType, MODULE_TYPE) + .eq(TaskChunkEntity::getScopeHash, scopeHash) + .eq(TaskChunkEntity::getChunkIndex, chunkIndex) + .last("limit 1")); + if (chunk == null) { + return; + } + Map persistedRows = readChunkRows(chunk); + for (SimilarAsinResultRowDto row : rows) { + persistedRows.put(rowKey(row), row); + } + String payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "相似ASIN分片载荷合并失败"); + String oldPayload = chunk.getPayloadJson(); + String oldPayloadHash = chunk.getPayloadHash(); + String newPayloadHash = DigestUtil.sha256Hex(payloadJson); + String storedPayload = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson); + int updated = taskChunkMapper.update(null, new LambdaUpdateWrapper() + .eq(TaskChunkEntity::getId, chunk.getId()) + .eq(TaskChunkEntity::getPayloadHash, oldPayloadHash) + .set(TaskChunkEntity::getPayloadJson, storedPayload) + .set(TaskChunkEntity::getPayloadHash, newPayloadHash) + .set(TaskChunkEntity::getUpdatedAt, LocalDateTime.now())); + if (updated > 0) { + log.debug("[similar-asin] chunk payload replaced taskId={} scopeHash={} chunk={} oldPayload={} newPayload={} attempt={}", + taskId, scopeHash, chunkIndex, oldPayload, storedPayload, attempt); + transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); + return; + } transientPayloadStorageService.deletePayloadIfPresent(storedPayload); - throw new IllegalStateException("相似ASIN分片载荷更新失败"); + if (attempt < maxAttempts) { + log.warn("[similar-asin] chunk payload CAS conflict, retry merge taskId={} scopeHash={} chunk={} attempt={}/{}", + taskId, scopeHash, chunkIndex, attempt, maxAttempts); + } } - transientPayloadStorageService.deleteReplacedPayloadIfNeeded(oldPayload, storedPayload); + throw new IllegalStateException("相似ASIN分片载荷更新失败"); } private List expandRows(List rows, @@ -1482,7 +1488,8 @@ public class SimilarAsinTaskService { } String url = row.getUrl(); List urls = row.getUrls(); - log.info("[similar-asin] python inbound idx={} groupKey={} rowToken={} id={} asin={} country={} title={} sku={} url={} urlsSize={} urlsHead={} urlsTail={}", + List alibaba = row.getAlibaba(); + log.info("[similar-asin] python inbound idx={} groupKey={} rowToken={} id={} asin={} country={} title={} sku={} price={} url={} urlsSize={} alibabaSize={} urlsHead={} urlsTail={}", i, normalize(row.getGroupKey()), normalize(row.getRowToken()), @@ -1491,8 +1498,10 @@ public class SimilarAsinTaskService { normalize(row.getCountry()), abbreviateForLog(row.getTitle(), 80), normalize(row.getSku()), + normalize(row.getPrice()), abbreviateForLog(url, 200), urls == null ? 0 : urls.size(), + alibaba == null ? 0 : alibaba.size(), urls == null || urls.isEmpty() ? "" : abbreviateForLog(urls.get(0), 200), urls == null || urls.size() <= 1 ? "" : abbreviateForLog(urls.get(urls.size() - 1), 200)); } @@ -1698,6 +1707,9 @@ public class SimilarAsinTaskService { int batchSize = resolveCozeBatchSize(readImgSwitch(task)); int cozeWorkUnits = countCozeWorkUnits(chunks, batchSize); int totalProgressUnits = Math.max(3, cozeWorkUnits + 3); + if (countPendingCozeStates(task.getId()) > 0) { + finalizeTimedOutCozeStatesForTask(task.getId()); + } if (countPendingCozeStates(task.getId()) > 0) { taskFileJobService.touchRunning(job.getId()); touchJavaSideTaskActivity(task.getId()); @@ -2397,7 +2409,16 @@ public class SimilarAsinTaskService { FileTaskEntity task = taskForPoll(state.getTaskId()); if (task != null) { Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); - mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); + try { + mergeCozeRowsIntoChunk(task, context.chunkScopeHash(), context.chunkIndex(), cozeRows, allRowsByBaseId); + } catch (Exception mergeEx) { + if (failureMessage.isBlank()) { + throw mergeEx; + } + log.warn("[similar-asin] coze failed result merge failed, mark state terminal anyway taskId={} stateId={} executeId={} err={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), + firstNonBlank(mergeEx.getMessage(), mergeEx.getClass().getSimpleName())); + } } } markCozeStateTerminal(state, @@ -2411,11 +2432,17 @@ public class SimilarAsinTaskService { FileTaskEntity task = taskForPoll(state.getTaskId()); if (task != null) { Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); - mergeCozeRowsIntoChunk(task, - context.chunkScopeHash(), - context.chunkIndex(), - cozeClient.markRowsFailed(batchRows, message), - allRowsByBaseId); + try { + mergeCozeRowsIntoChunk(task, + context.chunkScopeHash(), + context.chunkIndex(), + cozeClient.markRowsFailed(batchRows, message), + allRowsByBaseId); + } catch (Exception mergeEx) { + log.warn("[similar-asin] coze timeout fallback merge failed, mark state terminal anyway taskId={} stateId={} executeId={} err={}", + state.getTaskId(), state.getId(), state.getCozeExecuteId(), + firstNonBlank(mergeEx.getMessage(), mergeEx.getClass().getSimpleName())); + } } markCozeStateTerminal(state, COZE_STATUS_FAILED, message); maybeFinalizeCozeJob(state.getTaskId(), context); @@ -2427,6 +2454,45 @@ public class SimilarAsinTaskService { } } + private void finalizeTimedOutCozeStatesForTask(Long taskId) { + if (taskId == null || taskId <= 0) { + return; + } + List states = taskScopeStateMapper.selectList(new LambdaQueryWrapper() + .eq(TaskScopeStateEntity::getTaskId, taskId) + .eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE) + .in(TaskScopeStateEntity::getCozeStatus, List.of(COZE_STATUS_SUBMITTED, COZE_STATUS_RUNNING)) + .orderByAsc(TaskScopeStateEntity::getCozeSubmittedAt) + .last("limit 50")); + if (states == null || states.isEmpty()) { + return; + } + for (TaskScopeStateEntity state : states) { + if (state == null || state.getId() == null || !isCozeStateTimedOut(state)) { + continue; + } + CozeBatchContext context = readCozeBatchContext(state); + if (context == null || context.resultId() == null) { + markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze 批次上下文缺失"); + continue; + } + List batchRows = readCozeBatchRows(state); + FileTaskEntity task = taskForPoll(taskId); + if (task != null) { + Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); + mergeCozeRowsIntoChunk(task, + context.chunkScopeHash(), + context.chunkIndex(), + cozeClient.markRowsFailed(batchRows, "Coze 异步工作流轮询超时"), + allRowsByBaseId); + } + markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze 异步工作流轮询超时"); + maybeFinalizeCozeJob(taskId, context); + log.warn("[similar-asin] 文件任务超时兜底已将 Coze pending 批次置为失败 taskId={} stateId={} jobId={}", + taskId, state.getId(), context.jobId()); + } + } + private boolean retryFailedCozeBatchState(TaskScopeStateEntity state, CozeBatchContext context, List batchRows, @@ -2913,7 +2979,7 @@ public class SimilarAsinTaskService { if (CozeFailureClassifier.isThrottleLockTimeout(message)) { if (isCozeStateTimedOut(state)) { Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); - String finalMessage = "Coze pending submit timeout after waiting credentials: " + message; + String finalMessage = "等待 Coze 提交凭证超时：" + message; mergeCozeRowsIntoChunk(task, null, null, @@ -2933,7 +2999,7 @@ public class SimilarAsinTaskService { Map> allRowsByBaseId = allRowsByBaseIdForPoll(task); String finalMessage = isCozeStateTimedOut(state) ? message + " (timeout after " + properties.getCozePollTimeoutMillis() + "ms)" - : message + " after " + nextAttemptCount + " submit attempts"; + : message + "，已重试提交 " + nextAttemptCount + " 次"; mergeCozeRowsIntoChunk(task, null, null, @@ -3734,11 +3800,6 @@ public class SimilarAsinTaskService { return hasCompletedScope; } - private boolean isJavaSideProcessing(Long taskId) { - return countPendingCozeStates(taskId) > 0 - || taskFileJobService.countActiveAssembleJobs(taskId, MODULE_TYPE) > 0; - } - private void touchJavaSideTaskActivity(Long taskId) { if (taskId == null || taskId <= 0) { return; @@ -3795,7 +3856,11 @@ public class SimilarAsinTaskService { } log.warn("[similar-asin] reject task operation because owner is another instance taskId={} operation={} owner={} current={}", task == null ? null : task.getId(), operation, owner, currentInstanceId()); - throw new BusinessException(40903, "This task is bound to another service instance"); + throw new TaskOwnerMismatchException( + task == null ? null : task.getId(), + operation, + owner, + currentInstanceId()); } private boolean isCozeStateOwnedByCurrentInstance(TaskScopeStateEntity state) { @@ -4461,6 +4526,8 @@ public class SimilarAsinTaskService { row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getDisplayId(), parsedRow.getSourceId())); row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getAsin(), "")); row.createCell(col++).setCellValue(firstNonBlank(parsedRow.getCountry(), "")); + row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "卖家名称", "卖家名", "卖家", "店铺名称", "店铺名", "seller name", "seller_name", "seller-name", "sellername", "store name", "shop name")); + row.createCell(col++).setCellValue(readValueByHeader(parsedRow, "品牌", "brand")); row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getIsStock())); row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getSimilarity())); row.createCell(col++).setCellValue(resultRow == null ? "" : userFacingCozeCellValue(resultRow, resultRow.getIsConform())); @@ -5197,6 +5264,7 @@ public class SimilarAsinTaskService { if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || userId != null && !Objects.equals(userId, task.getUserId())) { throw new BusinessException("任务不存在"); } + ensureTaskOwnedByCurrentInstance(task, "parsed payload"); return readParsedPayload(task); } @@ -5305,6 +5373,8 @@ public class SimilarAsinTaskService { log.warn("[similar-asin] read chunk payload failed taskId={} chunk={} crossInstance={} err={}", chunk.getTaskId(), chunk.getChunkIndex(), crossInstance, msg); recordChunkReadFailure(chunk, crossInstance, msg); + throw new BusinessException("similar ASIN chunk payload read failed chunk=" + + chunk.getChunkIndex() + ": " + msg, ex); } return rows; } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java index 5f986ef..fa51f5c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedder.java @@ -92,6 +92,10 @@ public class SimilarAsinImageEmbedder { 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 static final String IMAGE_ACCEPT = "image/avif,image/webp,image/apng,image/svg+xml,image/*,*/*;q=0.8"; + private static final String DOWNLOAD_ACCEPT = "*/*"; + private static final String ACCEPT_LANGUAGE = "en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7"; + private static final String AMAZON_REFERER = "https://www.amazon.com/"; private final int downloadTimeoutSeconds; private final int downloadPoolSize; @@ -108,9 +112,9 @@ public class SimilarAsinImageEmbedder { .readTimeout(Duration.ofSeconds(downloadTimeoutSeconds)) .writeTimeout(Duration.ofSeconds(downloadTimeoutSeconds)) .callTimeout(Duration.ofSeconds(downloadTimeoutSeconds * 2L)) - .retryOnConnectionFailure(false) - .followRedirects(false) - .followSslRedirects(false) + .retryOnConnectionFailure(true) + .followRedirects(true) + .followSslRedirects(true) .dns(new SafeDns(Dns.SYSTEM)) .build(); this.downloadPool = Executors.newFixedThreadPool(downloadPoolSize, namedFactory("similar-asin-image-dl")); @@ -129,6 +133,10 @@ public class SimilarAsinImageEmbedder { return downloadPoolSize; } + OkHttpClient httpClient() { + return httpClient; + } + /** * P2-8 prefetch：assemble 阶段第一遍扫所有 url，并行预下载到 taskImageCache。 * embed() 第二遍写入时若 cache 命中直接复用，否则回退到现在的串行下载逻辑（保持兜底）。 @@ -174,7 +182,7 @@ public class SimilarAsinImageEmbedder { taskImageCache.putIfAbsent(url, thumb); } catch (Exception ex) { // 预下载失败不抛出：embed() 时同 url 会再次尝试并走原有兜底链路。 - log.debug("[similar-asin][image] prefetch-fail url={} err={}", url, ex.getMessage()); + log.debug("[similar-asin][image] prefetch-fail url={} err={}", url, errorSummary(ex)); } }; futures.add(completion.submit(task, null)); @@ -229,7 +237,7 @@ public class SimilarAsinImageEmbedder { byte[] raw = downloadWithRetry(trimmed); return resizeImage(trimmed, raw); } catch (Exception ex) { - log.debug("[similar-asin][image] prefetch-cache-fail url={} err={}", trimmed, ex.getMessage()); + log.debug("[similar-asin][image] prefetch-cache-fail url={} err={}", trimmed, errorSummary(ex)); return null; } } @@ -273,6 +281,7 @@ public class SimilarAsinImageEmbedder { int colIdx, Row row, Map taskImageCache) { + long startedNanos = System.nanoTime(); try { ResizedImage thumb = taskImageCache.get(trimmedUrl); boolean cacheHit = thumb != null; @@ -294,11 +303,14 @@ public class SimilarAsinImageEmbedder { } catch (ResizeOversizeException ex) { log.warn("[similar-asin][image] resize-oversize url={} bytes={}", ex.url(), ex.size()); } catch (ResizeException ex) { - log.warn("[similar-asin][image] resize-fail url={} err={}", trimmedUrl, ex.getMessage()); + log.warn("[similar-asin][image] resize-fail url={} elapsedMs={} err={}", + trimmedUrl, elapsedMs(startedNanos), errorSummary(ex)); } catch (IOException | TimeoutException ex) { - log.warn("[similar-asin][image] download-fail url={} err={}", trimmedUrl, ex.getMessage()); + log.warn("[similar-asin][image] download-fail url={} elapsedMs={} err={}", + trimmedUrl, elapsedMs(startedNanos), errorSummary(ex)); } catch (RuntimeException ex) { - log.warn("[similar-asin][image] embed-fail url={} err={}", trimmedUrl, ex.getMessage()); + log.warn("[similar-asin][image] embed-fail url={} elapsedMs={} err={}", + trimmedUrl, elapsedMs(startedNanos), errorSummary(ex)); } row.createCell(colIdx).setCellValue(trimmedUrl); return null; @@ -334,10 +346,12 @@ public class SimilarAsinImageEmbedder { private byte[] downloadWithRetry(String url) throws IOException, TimeoutException { validateHttpsUrl(url); IOException last = null; + TimeoutException lastTimeout = null; + long waitSeconds = downloadTimeoutSeconds * 2L; for (int attempt = 0; attempt <= DOWNLOAD_MAX_RETRY; attempt++) { Future future = downloadPool.submit(() -> doFetch(url)); try { - return future.get(downloadTimeoutSeconds * 2L, TimeUnit.SECONDS); + return future.get(waitSeconds, TimeUnit.SECONDS); } catch (java.util.concurrent.ExecutionException ee) { Throwable cause = ee.getCause(); if (cause instanceof DownloadOversizeException doe) { @@ -345,6 +359,8 @@ public class SimilarAsinImageEmbedder { } if (cause instanceof IOException io) { last = io; + lastTimeout = null; + logRetry(url, attempt, io); continue; } if (cause instanceof RuntimeException re) { @@ -353,15 +369,31 @@ public class SimilarAsinImageEmbedder { throw new IOException("image download failed: " + cause.getMessage(), cause); } catch (java.util.concurrent.TimeoutException te) { future.cancel(true); - throw te; + TimeoutException wrapped = new TimeoutException("image download timeout after attempt " + + (attempt + 1) + "/" + (DOWNLOAD_MAX_RETRY + 1) + + ", waitSeconds=" + waitSeconds); + wrapped.initCause(te); + lastTimeout = wrapped; + last = null; + logRetry(url, attempt, wrapped); } catch (InterruptedException ie) { Thread.currentThread().interrupt(); throw new IOException("image download interrupted", ie); } } + if (lastTimeout != null) { + throw lastTimeout; + } throw last != null ? last : new IOException("image download failed without cause"); } + private void logRetry(String url, int attempt, Exception ex) { + if (attempt < DOWNLOAD_MAX_RETRY) { + log.debug("[similar-asin][image] download-retry url={} nextAttempt={}/{} err={}", + url, attempt + 2, DOWNLOAD_MAX_RETRY + 1, errorSummary(ex)); + } + } + static void validateHttpsUrl(String url) { URI uri; try { @@ -407,11 +439,7 @@ public class SimilarAsinImageEmbedder { } private byte[] doFetch(String url) throws IOException { - Request req = new Request.Builder() - .url(url) - .header("User-Agent", UA) - .get() - .build(); + Request req = buildImageRequest(url); try (Response resp = httpClient.newCall(req).execute()) { if (!resp.isSuccessful()) { throw new IOException("http " + resp.code() + " for " + url); @@ -441,6 +469,57 @@ public class SimilarAsinImageEmbedder { } } + static Request buildImageRequest(String url) { + Request.Builder builder = new Request.Builder() + .url(url) + .header("User-Agent", UA) + .header("Accept-Language", ACCEPT_LANGUAGE) + .header("Cache-Control", "no-cache") + .get(); + if (isCozeSignedImageUrl(url)) { + // Coze/TOS signed image links behave like direct file downloads. + // A foreign Referer can be rejected, so keep this close to a browser address-bar download. + builder.header("Accept", DOWNLOAD_ACCEPT); + } else { + builder.header("Accept", IMAGE_ACCEPT) + .header("Referer", AMAZON_REFERER); + } + return builder.build(); + } + + static boolean isCozeSignedImageUrl(String url) { + try { + URI uri = new URI(url); + String host = uri.getHost(); + return host != null && host.toLowerCase(Locale.ROOT).endsWith("-bot-platform-tos-sign.coze.cn"); + } catch (URISyntaxException ex) { + return false; + } + } + + static long elapsedMs(long startedNanos) { + return TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startedNanos); + } + + static String errorSummary(Throwable ex) { + if (ex == null) { + return "unknown"; + } + StringBuilder sb = new StringBuilder(ex.getClass().getSimpleName()); + String message = ex.getMessage(); + sb.append(": "); + sb.append(message == null || message.isBlank() ? "" : message); + Throwable cause = ex.getCause(); + if (cause != null) { + sb.append("; cause=").append(cause.getClass().getSimpleName()); + String causeMessage = cause.getMessage(); + if (causeMessage != null && !causeMessage.isBlank()) { + sb.append(": ").append(causeMessage); + } + } + return sb.toString(); + } + /** * 等比缩放到长边 TARGET_LONG_EDGE_PX，JPEG q=0.75 输出；返回字节 + 实际像素，供调用方按图自适应单元格尺寸。 * 出口校验字节数 ≤ MAX_THUMB_SIZE_BYTES，超限按 MAX_THUMB_SIZE → 长边 → 质量的顺序迭代降级， diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java new file mode 100644 index 0000000..ede6019 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/controller/TaskHeartbeatController.java @@ -0,0 +1,36 @@ +package com.nanri.aiimage.modules.task.controller; + +import com.nanri.aiimage.common.api.ApiResponse; +import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; +import com.nanri.aiimage.modules.task.model.vo.TaskHeartbeatVo; +import com.nanri.aiimage.modules.task.service.TaskHeartbeatService; +import io.swagger.v3.oas.annotations.Operation; +import io.swagger.v3.oas.annotations.Parameter; +import io.swagger.v3.oas.annotations.tags.Tag; +import jakarta.validation.Valid; +import lombok.RequiredArgsConstructor; +import org.springframework.web.bind.annotation.PathVariable; +import org.springframework.web.bind.annotation.PostMapping; +import org.springframework.web.bind.annotation.RequestBody; +import org.springframework.web.bind.annotation.RequestMapping; +import org.springframework.web.bind.annotation.RestController; + +@RestController +@RequiredArgsConstructor +@RequestMapping("/api/tasks") +@Tag(name = "Task heartbeat", description = "Unified heartbeat API for Python task workers") +public class TaskHeartbeatController { + + private final TaskHeartbeatService taskHeartbeatService; + + @PostMapping("/{taskId}/heartbeat") + @Operation( + summary = "Task heartbeat", + description = "Python calls this during execution. The backend resolves biz_file_task or brand task by taskId. user_id is optional.") + public ApiResponse heartbeat( + @Parameter(description = "Task ID", required = true, example = "200") + @PathVariable Long taskId, + @Valid @RequestBody(required = false) TaskHeartbeatRequest request) { + return ApiResponse.success(taskHeartbeatService.heartbeat(taskId, request)); + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java index 7f13150..47fa1dd 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/mapper/TaskImageCacheMapper.java @@ -2,11 +2,14 @@ package com.nanri.aiimage.modules.task.mapper; import com.baomidou.mybatisplus.core.mapper.BaseMapper; import com.nanri.aiimage.modules.task.model.entity.TaskImageCacheEntity; +import org.apache.ibatis.annotations.Delete; import org.apache.ibatis.annotations.Mapper; import org.apache.ibatis.annotations.Param; import org.apache.ibatis.annotations.Select; import org.apache.ibatis.annotations.Update; +import java.time.LocalDateTime; + /** * P2-12：图片缩略图缓存 mapper。提供 LRU 命中刷新、按 url_hash 直读字节两个轻量入口， * 其余 CRUD 走 {@link BaseMapper} 默认实现。 @@ -15,7 +18,7 @@ import org.apache.ibatis.annotations.Update; public interface TaskImageCacheMapper extends BaseMapper { /** - * 命中时同步 bumping last_used_at，便于后续按 LRU 清理 7 天前未用记录。 + * 命中时同步 bumping last_used_at，便于后续按 LRU 清理过期未用记录。 */ @Update("UPDATE biz_task_image_cache SET last_used_at = NOW(3) WHERE url_hash = #{urlHash}") int touchLastUsed(@Param("urlHash") String urlHash); @@ -26,4 +29,7 @@ public interface TaskImageCacheMapper extends BaseMapper { */ @Select("SELECT image_bytes FROM biz_task_image_cache WHERE url_hash = #{urlHash} LIMIT 1") byte[] selectBytesByUrlHash(@Param("urlHash") String urlHash); + + @Delete("DELETE FROM biz_task_image_cache WHERE last_used_at < #{cutoff} ORDER BY last_used_at LIMIT #{limit}") + int deleteExpiredBatch(@Param("cutoff") LocalDateTime cutoff, @Param("limit") int limit); } diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java new file mode 100644 index 0000000..2adc8cb --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/dto/TaskHeartbeatRequest.java @@ -0,0 +1,23 @@ +package com.nanri.aiimage.modules.task.model.dto; + +import com.fasterxml.jackson.annotation.JsonProperty; +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +@Data +@Schema(description = "Task heartbeat request") +public class TaskHeartbeatRequest { + + @JsonProperty("user_id") + @Schema(description = "Optional task owner user ID. Python workers can omit it.", example = "453") + private Long userId; + + @Schema(description = "Optional phase label", example = "crawling") + private String phase; + + @Schema(description = "Optional current progress", example = "42") + private Integer current; + + @Schema(description = "Optional total progress", example = "1100") + private Integer total; +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/TaskImageCacheEntity.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/TaskImageCacheEntity.java index bffff0f..4848b3c 100644 --- a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/TaskImageCacheEntity.java +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/entity/TaskImageCacheEntity.java @@ -16,7 +16,7 @@ import java.time.LocalDateTime; *
  • {@link #urlHash}：sha256 lowercase hex，作为唯一键避免 1024 长 URL 命中索引长度限制；
    • *
  • {@link #imageBytes}：已 resize 后的 JPEG 缩略图字节，最大约 300KB（保持与 * {@code SimilarAsinImageEmbedder.MAX_THUMB_SIZE_BYTES} 对齐）；
    • - *
  • {@link #lastUsedAt}：用于后续 LRU 清理（>7 天未用）。
    • + *
  • {@link #lastUsedAt}：用于后续 LRU 清理（默认 >3 天未用）。
    • * */ @Data diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskHeartbeatVo.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskHeartbeatVo.java new file mode 100644 index 0000000..214fb03 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/model/vo/TaskHeartbeatVo.java @@ -0,0 +1,39 @@ +package com.nanri.aiimage.modules.task.model.vo; + +import io.swagger.v3.oas.annotations.media.Schema; +import lombok.Data; + +@Data +@Schema(description = "Task heartbeat response") +public class TaskHeartbeatVo { + + @Schema(description = "Whether the task can continue", example = "true") + private boolean alive; + + @Schema(description = "Task status", example = "RUNNING") + private String status; + + @Schema(description = "Resolved task module", example = "PRODUCT_RISK_RESOLVE") + private String moduleType; + + @Schema(description = "Message", example = "ok") + private String message; + + public static TaskHeartbeatVo alive(String moduleType, String status) { + TaskHeartbeatVo vo = new TaskHeartbeatVo(); + vo.setAlive(true); + vo.setModuleType(moduleType); + vo.setStatus(status); + vo.setMessage("ok"); + return vo; + } + + public static TaskHeartbeatVo notAlive(String moduleType, String status, String message) { + TaskHeartbeatVo vo = new TaskHeartbeatVo(); + vo.setAlive(false); + vo.setModuleType(moduleType); + vo.setStatus(status); + vo.setMessage(message); + return vo; + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java new file mode 100644 index 0000000..b722ad3 --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskHeartbeatService.java @@ -0,0 +1,211 @@ +package com.nanri.aiimage.modules.task.service; + +import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper; +import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper; +import com.nanri.aiimage.modules.appearancepatent.service.AppearancePatentTaskCacheService; +import com.nanri.aiimage.modules.brand.mapper.BrandCrawlTaskMapper; +import com.nanri.aiimage.modules.brand.model.entity.BrandCrawlTaskEntity; +import com.nanri.aiimage.modules.brand.service.BrandTaskProgressCacheService; +import com.nanri.aiimage.modules.deletebrand.service.DeleteBrandTaskCacheService; +import com.nanri.aiimage.modules.patroldelete.service.PatrolDeleteTaskCacheService; +import com.nanri.aiimage.modules.pricetrack.service.PriceTrackTaskCacheService; +import com.nanri.aiimage.modules.productrisk.service.ProductRiskTaskCacheService; +import com.nanri.aiimage.modules.queryasin.service.QueryAsinTaskCacheService; +import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskCacheService; +import com.nanri.aiimage.modules.similarasin.service.SimilarAsinTaskCacheService; +import com.nanri.aiimage.modules.task.mapper.FileTaskMapper; +import com.nanri.aiimage.modules.task.model.dto.TaskHeartbeatRequest; +import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity; +import com.nanri.aiimage.modules.task.model.vo.TaskHeartbeatVo; +import lombok.RequiredArgsConstructor; +import org.springframework.stereotype.Service; + +import java.time.LocalDateTime; +import java.util.LinkedHashMap; +import java.util.Map; + +@Service +@RequiredArgsConstructor +public class TaskHeartbeatService { + + private static final String STATUS_RUNNING = "RUNNING"; + private static final String BRAND_STATUS_RUNNING = "running"; + + private static final String MODULE_DELETE_BRAND = "DELETE_BRAND"; + private static final String MODULE_PRODUCT_RISK = "PRODUCT_RISK_RESOLVE"; + private static final String MODULE_PRICE_TRACK = "PRICE_TRACK"; + private static final String MODULE_SHOP_MATCH = "SHOP_MATCH"; + private static final String MODULE_PATROL_DELETE = "PATROL_DELETE"; + private static final String MODULE_QUERY_ASIN = "QUERY_ASIN"; + private static final String MODULE_APPEARANCE_PATENT = "APPEARANCE_PATENT"; + private static final String MODULE_SIMILAR_ASIN = "SIMILAR_ASIN"; + private static final String MODULE_BRAND = "BRAND"; + + private final FileTaskMapper fileTaskMapper; + private final BrandCrawlTaskMapper brandCrawlTaskMapper; + private final ProductRiskTaskCacheService productRiskTaskCacheService; + private final PriceTrackTaskCacheService priceTrackTaskCacheService; + private final ShopMatchTaskCacheService shopMatchTaskCacheService; + private final PatrolDeleteTaskCacheService patrolDeleteTaskCacheService; + private final QueryAsinTaskCacheService queryAsinTaskCacheService; + private final AppearancePatentTaskCacheService appearancePatentTaskCacheService; + private final SimilarAsinTaskCacheService similarAsinTaskCacheService; + private final DeleteBrandTaskCacheService deleteBrandTaskCacheService; + private final BrandTaskProgressCacheService brandTaskProgressCacheService; + + public TaskHeartbeatVo heartbeat(Long taskId, TaskHeartbeatRequest request) { + if (taskId == null || taskId <= 0) { + return TaskHeartbeatVo.notAlive(null, null, "invalid taskId"); + } + Long userId = request == null ? null : request.getUserId(); + + LambdaQueryWrapper fileQuery = new LambdaQueryWrapper() + .eq(FileTaskEntity::getId, taskId) + .last("limit 1"); + if (userId != null && userId > 0) { + fileQuery.eq(FileTaskEntity::getUserId, userId); + } + FileTaskEntity fileTask = fileTaskMapper.selectOne(fileQuery); + + LambdaQueryWrapper brandQuery = new LambdaQueryWrapper() + .eq(BrandCrawlTaskEntity::getId, taskId) + .last("limit 1"); + if (userId != null && userId > 0) { + brandQuery.eq(BrandCrawlTaskEntity::getUserId, userId); + } + BrandCrawlTaskEntity brandTask = brandCrawlTaskMapper.selectOne(brandQuery); + + TaskHeartbeatVo fileResult = touchFileTaskIfRunning(fileTask, request); + TaskHeartbeatVo brandResult = touchBrandTaskIfRunning(brandTask, request); + + if (fileResult != null && fileResult.isAlive() && brandResult != null && brandResult.isAlive()) { + return TaskHeartbeatVo.alive(fileResult.getModuleType() + "," + MODULE_BRAND, STATUS_RUNNING); + } + if (fileResult != null && fileResult.isAlive()) { + return fileResult; + } + if (brandResult != null && brandResult.isAlive()) { + return brandResult; + } + if (fileResult != null) { + return fileResult; + } + if (brandResult != null) { + return brandResult; + } + return TaskHeartbeatVo.notAlive(null, null, "task not found"); + } + + private TaskHeartbeatVo touchFileTaskIfRunning(FileTaskEntity task, TaskHeartbeatRequest request) { + if (task == null) { + return null; + } + String moduleType = task.getModuleType() == null ? "" : task.getModuleType(); + String status = task.getStatus(); + if (!STATUS_RUNNING.equals(status)) { + return TaskHeartbeatVo.notAlive(moduleType, status, "task is not running"); + } + LocalDateTime now = LocalDateTime.now(); + int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(FileTaskEntity::getId, task.getId()) + .eq(FileTaskEntity::getUserId, task.getUserId()) + .eq(FileTaskEntity::getStatus, STATUS_RUNNING) + .set(FileTaskEntity::getUpdatedAt, now)); + if (updated <= 0) { + FileTaskEntity latest = fileTaskMapper.selectById(task.getId()); + return TaskHeartbeatVo.notAlive(moduleType, latest == null ? status : latest.getStatus(), "task is not running"); + } + touchModuleHeartbeat(moduleType, task.getId(), request); + task.setUpdatedAt(now); + saveFileTaskCache(moduleType, task); + return TaskHeartbeatVo.alive(moduleType, STATUS_RUNNING); + } + + private TaskHeartbeatVo touchBrandTaskIfRunning(BrandCrawlTaskEntity task, TaskHeartbeatRequest request) { + if (task == null) { + return null; + } + String status = task.getStatus(); + if (!BRAND_STATUS_RUNNING.equals(status)) { + return TaskHeartbeatVo.notAlive(MODULE_BRAND, status, "task is not running"); + } + int updated = brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper() + .eq(BrandCrawlTaskEntity::getId, task.getId()) + .eq(BrandCrawlTaskEntity::getUserId, task.getUserId()) + .eq(BrandCrawlTaskEntity::getStatus, BRAND_STATUS_RUNNING) + .set(BrandCrawlTaskEntity::getUpdatedAt, LocalDateTime.now())); + if (updated <= 0) { + BrandCrawlTaskEntity latest = brandCrawlTaskMapper.selectById(task.getId()); + return TaskHeartbeatVo.notAlive(MODULE_BRAND, latest == null ? status : latest.getStatus(), "task is not running"); + } + brandTaskProgressCacheService.touchHeartbeat( + task.getId(), + request == null ? null : request.getPhase(), + request == null ? null : request.getCurrent(), + request == null ? null : request.getTotal()); + return TaskHeartbeatVo.alive(MODULE_BRAND, BRAND_STATUS_RUNNING); + } + + private void touchModuleHeartbeat(String moduleType, Long taskId, TaskHeartbeatRequest request) { + switch (moduleType) { + case MODULE_PRODUCT_RISK -> { + productRiskTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_PRICE_TRACK -> { + priceTrackTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_SHOP_MATCH -> { + shopMatchTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_PATROL_DELETE -> { + patrolDeleteTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_QUERY_ASIN -> { + queryAsinTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_APPEARANCE_PATENT -> { + appearancePatentTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_SIMILAR_ASIN -> { + similarAsinTaskCacheService.touchTaskHeartbeat(taskId); + } + case MODULE_DELETE_BRAND -> { + deleteBrandTaskCacheService.saveProgress(taskId, buildDeleteBrandHeartbeatProgress(request), true); + } + default -> { + } + } + } + + private Map buildDeleteBrandHeartbeatProgress(TaskHeartbeatRequest request) { + String now = String.valueOf(System.currentTimeMillis()); + Map values = new LinkedHashMap<>(); + values.put("last_heartbeat_at", now); + values.put("updated_at", now); + if (request != null) { + putIfPresent(values, "phase", request.getPhase()); + putIfPresent(values, "current", request.getCurrent()); + putIfPresent(values, "total", request.getTotal()); + } + return values; + } + + private void saveFileTaskCache(String moduleType, FileTaskEntity task) { + switch (moduleType) { + case MODULE_PRODUCT_RISK -> productRiskTaskCacheService.saveTaskCache(task); + case MODULE_PRICE_TRACK -> priceTrackTaskCacheService.saveTaskCache(task); + case MODULE_SHOP_MATCH -> shopMatchTaskCacheService.saveTaskCache(task); + case MODULE_PATROL_DELETE -> patrolDeleteTaskCacheService.saveTaskCache(task); + case MODULE_QUERY_ASIN -> queryAsinTaskCacheService.saveTaskCache(task); + case MODULE_DELETE_BRAND -> deleteBrandTaskCacheService.saveTaskCache(task); + default -> { + } + } + } + + private void putIfPresent(Map values, String key, Object value) { + if (value != null) { + values.put(key, String.valueOf(value)); + } + } +} diff --git a/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskImageCacheCleanupService.java b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskImageCacheCleanupService.java new file mode 100644 index 0000000..3483fba --- /dev/null +++ b/backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskImageCacheCleanupService.java @@ -0,0 +1,69 @@ +package com.nanri.aiimage.modules.task.service; + +import com.nanri.aiimage.common.service.DistributedJobLockService; +import com.nanri.aiimage.config.TaskImageCacheCleanupProperties; +import com.nanri.aiimage.modules.task.mapper.TaskImageCacheMapper; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; +import org.springframework.scheduling.annotation.Scheduled; +import org.springframework.stereotype.Service; + +import java.time.Duration; +import java.time.LocalDateTime; + +@Service +@RequiredArgsConstructor +@Slf4j +public class TaskImageCacheCleanupService { + + private static final Duration CLEANUP_LOCK_TTL = Duration.ofMinutes(30); + + private final TaskImageCacheMapper taskImageCacheMapper; + private final TaskImageCacheCleanupProperties cleanupProperties; + private final DistributedJobLockService distributedJobLockService; + + @Scheduled(cron = "${aiimage.task-image-cache-cleanup.cron:0 30 2 * * *}") + public void cleanupExpiredImageCache() { + if (!cleanupProperties.isEnabled()) { + return; + } + DistributedJobLockService.LockHandle lockHandle = + distributedJobLockService.tryLock("task-image-cache:cleanup", CLEANUP_LOCK_TTL); + if (lockHandle == null) { + log.info("[task-image-cache-cleanup] skip because another instance holds the distributed lock"); + return; + } + try (lockHandle) { + int retentionDays = Math.max(1, cleanupProperties.getRetentionDays()); + int batchSize = Math.max(1, cleanupProperties.getBatchSize()); + int maxBatches = Math.max(1, cleanupProperties.getMaxBatchesPerRun()); + LocalDateTime cutoff = LocalDateTime.now().minusDays(retentionDays); + + int deletedTotal = 0; + int batches = 0; + while (batches < maxBatches) { + int deleted = taskImageCacheMapper.deleteExpiredBatch(cutoff, batchSize); + if (deleted <= 0) { + break; + } + deletedTotal += deleted; + batches++; + if (!lockHandle.renew(CLEANUP_LOCK_TTL)) { + log.warn("[task-image-cache-cleanup] stop because lock renew failed deletedTotal={} batches={}", + deletedTotal, batches); + break; + } + if (deleted < batchSize) { + break; + } + } + + if (deletedTotal > 0 || batches >= maxBatches) { + log.info("[task-image-cache-cleanup] finished cutoff={} retentionDays={} deletedTotal={} batches={} batchSize={} maxBatches={}", + cutoff, retentionDays, deletedTotal, batches, batchSize, maxBatches); + } + } catch (Exception ex) { + log.warn("[task-image-cache-cleanup] failed msg={}", ex.getMessage(), ex); + } + } +} diff --git a/backend-java/src/main/resources/application.yml b/backend-java/src/main/resources/application.yml index 8a4cded..a620098 100644 --- a/backend-java/src/main/resources/application.yml +++ b/backend-java/src/main/resources/application.yml @@ -75,6 +75,14 @@ aiimage: # For IDEA/local startup, set -Daiimage.instance-id= or configure the AIIMAGE_INSTANCE_ID env var. # Avoid relying on container hostname/container ID, otherwise task owner continuity may break after restart/redeploy. instance-id: ${AIIMAGE_INSTANCE_ID:local-dev} + instance-routing: + routes: + server-110: ${AIIMAGE_INSTANCE_ROUTE_SERVER_110:http://192.168.0.171:18080} + server-111: ${AIIMAGE_INSTANCE_ROUTE_SERVER_111:http://47.111.163.154:18080} + server-121: ${AIIMAGE_INSTANCE_ROUTE_SERVER_121:http://192.168.0.170:18080} + connect-timeout-millis: ${AIIMAGE_INSTANCE_ROUTE_CONNECT_TIMEOUT_MILLIS:3000} + read-timeout-millis: ${AIIMAGE_INSTANCE_ROUTE_READ_TIMEOUT_MILLIS:300000} + request-body-cache-limit-bytes: ${AIIMAGE_INSTANCE_ROUTE_REQUEST_BODY_CACHE_LIMIT_BYTES:104857600} oss: region: ${AIIMAGE_OSS_REGION:cn-hangzhou} endpoint: ${AIIMAGE_OSS_ENDPOINT:oss-cn-hangzhou.aliyuncs.com} @@ -104,21 +112,17 @@ aiimage: brand-progress: ttl-hours: ${AIIMAGE_BRAND_PROGRESS_TTL_HOURS:24} failed-ttl-hours: ${AIIMAGE_BRAND_PROGRESS_FAILED_TTL_HOURS:2} - heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:15} + heartbeat-timeout-minutes: ${AIIMAGE_BRAND_PROGRESS_HEARTBEAT_TIMEOUT_MINUTES:30} stale-check-cron: ${AIIMAGE_BRAND_PROGRESS_STALE_CHECK_CRON:0 */2 * * * *} delete-brand-progress: - heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:15} - delete-brand-initial-timeout-minutes: ${AIIMAGE_DELETE_BRAND_INITIAL_TIMEOUT_MINUTES:20} + heartbeat-timeout-minutes: ${AIIMAGE_DELETE_BRAND_HEARTBEAT_TIMEOUT_MINUTES:30} stale-check-cron: ${AIIMAGE_DELETE_BRAND_STALE_CHECK_CRON:*/30 * * * * *} finalize-check-cron: ${AIIMAGE_DELETE_BRAND_FINALIZE_CHECK_CRON:30 */2 * * * *} - product-risk-stale-timeout-minutes: ${AIIMAGE_PRODUCT_RISK_STALE_TIMEOUT_MINUTES:20} - product-risk-initial-timeout-minutes: ${AIIMAGE_PRODUCT_RISK_INITIAL_TIMEOUT_MINUTES:20} - price-track-stale-timeout-minutes: ${AIIMAGE_PRICE_TRACK_STALE_TIMEOUT_MINUTES:20} - price-track-initial-timeout-minutes: ${AIIMAGE_PRICE_TRACK_INITIAL_TIMEOUT_MINUTES:20} - shop-match-stale-timeout-minutes: ${AIIMAGE_SHOP_MATCH_STALE_TIMEOUT_MINUTES:20} - shop-match-initial-timeout-minutes: ${AIIMAGE_SHOP_MATCH_INITIAL_TIMEOUT_MINUTES:20} - patrol-delete-stale-timeout-minutes: ${AIIMAGE_PATROL_DELETE_STALE_TIMEOUT_MINUTES:20} - patrol-delete-initial-timeout-minutes: ${AIIMAGE_PATROL_DELETE_INITIAL_TIMEOUT_MINUTES:20} + product-risk-stale-timeout-minutes: ${AIIMAGE_PRODUCT_RISK_STALE_TIMEOUT_MINUTES:30} + price-track-stale-timeout-minutes: ${AIIMAGE_PRICE_TRACK_STALE_TIMEOUT_MINUTES:30} + shop-match-stale-timeout-minutes: ${AIIMAGE_SHOP_MATCH_STALE_TIMEOUT_MINUTES:30} + patrol-delete-stale-timeout-minutes: ${AIIMAGE_PATROL_DELETE_STALE_TIMEOUT_MINUTES:30} + query-asin-stale-timeout-minutes: ${AIIMAGE_QUERY_ASIN_STALE_TIMEOUT_MINUTES:30} module-cleanup: enabled: ${AIIMAGE_MODULE_CLEANUP_ENABLED:true} cron: ${AIIMAGE_MODULE_CLEANUP_CRON:0 0 0 * * *} @@ -132,6 +136,12 @@ aiimage: scope-payload-buffer-retention-hours: ${AIIMAGE_TASK_SCOPE_PAYLOAD_BUFFER_RETENTION_HOURS:24} scope-payload-recovery-max-files: ${AIIMAGE_TASK_SCOPE_PAYLOAD_RECOVERY_MAX_FILES:200} scope-payload-cleanup-cron: ${AIIMAGE_TASK_SCOPE_PAYLOAD_CLEANUP_CRON:15 */30 * * * *} + task-image-cache-cleanup: + enabled: ${AIIMAGE_TASK_IMAGE_CACHE_CLEANUP_ENABLED:true} + cron: ${AIIMAGE_TASK_IMAGE_CACHE_CLEANUP_CRON:0 30 2 * * *} + retention-days: ${AIIMAGE_TASK_IMAGE_CACHE_RETENTION_DAYS:3} + batch-size: ${AIIMAGE_TASK_IMAGE_CACHE_CLEANUP_BATCH_SIZE:5000} + max-batches-per-run: ${AIIMAGE_TASK_IMAGE_CACHE_CLEANUP_MAX_BATCHES_PER_RUN:200} result-file-job: mq-enabled: ${AIIMAGE_RESULT_FILE_JOB_MQ_ENABLED:true} topic: ${AIIMAGE_RESULT_FILE_JOB_TOPIC:aiimage-result-file-job} @@ -149,6 +159,13 @@ aiimage: batch-size: ${AIIMAGE_RESULT_FILE_JOB_BATCH_SIZE:20} coze-task: max-concurrent: ${AIIMAGE_COZE_TASK_MAX_CONCURRENT:12} + brand-check: + base-url: ${AIIMAGE_BRAND_CHECK_BASE_URL:http://47.110.241.161:16890} + path: ${AIIMAGE_BRAND_CHECK_PATH:/brand_check} + token: ${AIIMAGE_BRAND_CHECK_TOKEN:eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9} + default-strategy: ${AIIMAGE_BRAND_CHECK_DEFAULT_STRATEGY:Terms} + connect-timeout-millis: ${AIIMAGE_BRAND_CHECK_CONNECT_TIMEOUT_MILLIS:10000} + read-timeout-millis: ${AIIMAGE_BRAND_CHECK_READ_TIMEOUT_MILLIS:60000} appearance-patent: coze-base-url: ${AIIMAGE_APPEARANCE_PATENT_COZE_BASE_URL:https://api.coze.cn} coze-workflow-path: ${AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH:/v1/workflow/run} @@ -159,6 +176,7 @@ aiimage: coze-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS:60000} coze-poll-interval-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_INTERVAL_MILLIS:30000} coze-poll-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_POLL_TIMEOUT_MILLIS:600000} + coze-flush-pending-minutes: ${AIIMAGE_APPEARANCE_PATENT_COZE_FLUSH_PENDING_MINUTES:1} stale-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES:30} stale-finalize-cron: ${AIIMAGE_APPEARANCE_PATENT_STALE_FINALIZE_CRON:0 */2 * * * *} similar-asin: diff --git a/backend-java/src/test/java/com/nanri/aiimage/common/util/CozeGroupResultPropagatorTest.java b/backend-java/src/test/java/com/nanri/aiimage/common/util/CozeGroupResultPropagatorTest.java new file mode 100644 index 0000000..299577f --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/common/util/CozeGroupResultPropagatorTest.java @@ -0,0 +1,80 @@ +package com.nanri.aiimage.common.util; + +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.junit.jupiter.api.Assertions.assertEquals; + +class CozeGroupResultPropagatorTest { + + @Test + void doesNotTreatNoInfringementAsInfringementHit() { + List parsedRows = List.of( + new ParsedRow("20_1"), + new ParsedRow("20_2") + ); + List resultRows = List.of( + new ResultRow("无侵权"), + new ResultRow("无侵权") + ); + + int updatedRows = CozeGroupResultPropagator.propagateByGroup( + parsedRows, + ParsedRow::displayId, + row -> resultRows.get(parsedRows.indexOf(row)), + ResultRow::conclusion, + ResultRow::setConclusion, + List.of("已侵权", "侵权") + ); + + assertEquals(0, updatedRows); + assertEquals("无侵权", resultRows.get(0).conclusion()); + assertEquals("无侵权", resultRows.get(1).conclusion()); + } + + @Test + void propagatesWhenGroupHasRealInfringementHit() { + List parsedRows = List.of( + new ParsedRow("20_1"), + new ParsedRow("20_2") + ); + List resultRows = List.of( + new ResultRow("无侵权"), + new ResultRow("侵权") + ); + + int updatedRows = CozeGroupResultPropagator.propagateByGroup( + parsedRows, + ParsedRow::displayId, + row -> resultRows.get(parsedRows.indexOf(row)), + ResultRow::conclusion, + ResultRow::setConclusion, + List.of("已侵权", "侵权") + ); + + assertEquals(1, updatedRows); + assertEquals("侵权", resultRows.get(0).conclusion()); + assertEquals("侵权", resultRows.get(1).conclusion()); + } + + private record ParsedRow(String displayId) { + } + + private static final class ResultRow { + + private String conclusion; + + private ResultRow(String conclusion) { + this.conclusion = conclusion; + } + + private String conclusion() { + return conclusion; + } + + private void setConclusion(String conclusion) { + this.conclusion = conclusion; + } + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClientTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClientTest.java new file mode 100644 index 0000000..2ec59aa --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClientTest.java @@ -0,0 +1,64 @@ +package com.nanri.aiimage.modules.appearancepatent.client; + +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.AppearancePatentProperties; +import com.nanri.aiimage.config.BrandCheckProperties; +import com.nanri.aiimage.modules.brand.client.BrandCheckClient; +import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.ArgumentMatchers.anyString; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; + +class AppearancePatentCozeClientTest { + + private final AppearancePatentCozeClient client = new AppearancePatentCozeClient( + new AppearancePatentProperties(), + new ObjectMapper(), + null, + new BrandCheckClient(new BrandCheckProperties()) + ); + + @Test + void titleRiskIsInfringementWhenBrandCheckHasFailedData() { + BrandCheckClient.BrandCheckBatchResult result = + new BrandCheckClient.BrandCheckBatchResult(List.of("阿凡达"), List.of("阿凡达"), List.of()); + + assertThat(client.buildTitleRisk("阿凡达", result)).isEqualTo("侵权"); + } + + @Test + void conclusionIsInfringementWhenTitleOrAppearanceIsInfringement() { + assertThat(client.buildConclusion("侵权", "无侵权", "")).isEqualTo("侵权"); + assertThat(client.buildConclusion("无侵权", "侵权", "")).isEqualTo("侵权"); + assertThat(client.buildConclusion("无侵权", "无侵权", "")).isEqualTo("无侵权"); + } + + @Test + void skipBrandCheckWhenCozeTitleAndReasonAreNone() throws Exception { + BrandCheckClient brandCheckClient = mock(BrandCheckClient.class); + AppearancePatentCozeClient cozeClient = new AppearancePatentCozeClient( + new AppearancePatentProperties(), + new ObjectMapper(), + null, + brandCheckClient + ); + AppearancePatentResultRowDto row = new AppearancePatentResultRowDto(); + row.setId("1"); + String dataText = """ + {"data":[{"row_id":"1","title":"无","title_reason":"无","appearance":"无侵权","result":"无侵权"}]} + """; + + List merged = cozeClient.mergeRowsFromDataText(List.of(row), dataText); + + assertThat(merged).hasSize(1); + assertThat(merged.get(0).getTitleRisk()).isEqualTo("无侵权"); + assertThat(merged.get(0).getConclusion()).isEqualTo("无侵权"); + verify(brandCheckClient, never()).checkTitleText(anyString(), anyString()); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/brand/client/BrandCheckClientTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/client/BrandCheckClientTest.java new file mode 100644 index 0000000..3a4f1e4 --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/brand/client/BrandCheckClientTest.java @@ -0,0 +1,29 @@ +package com.nanri.aiimage.modules.brand.client; + +import com.nanri.aiimage.config.BrandCheckProperties; +import org.junit.jupiter.api.Test; + +import java.util.List; + +import static org.assertj.core.api.Assertions.assertThat; + +class BrandCheckClientTest { + + @Test + void splitTitleTextSupportsCommonSeparatorsAndQuotes() { + BrandCheckClient client = new BrandCheckClient(new BrandCheckProperties()); + + List brands = client.splitTitleText("'阿凡达,任天堂'； Disney、LEGO\nSony"); + + assertThat(brands).containsExactly("阿凡达", "任天堂", "Disney", "LEGO", "Sony"); + } + + @Test + void splitTitleTextDeduplicatesBlankValues() { + BrandCheckClient client = new BrandCheckClient(new BrandCheckProperties()); + + List brands = client.splitTitleText(" 任天堂, ,任天堂，Sony "); + + assertThat(brands).containsExactly("任天堂", "Sony"); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClientTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClientTest.java new file mode 100644 index 0000000..0e0a80a --- /dev/null +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClientTest.java @@ -0,0 +1,86 @@ +package com.nanri.aiimage.modules.similarasin.client; + +import com.fasterxml.jackson.core.type.TypeReference; +import com.fasterxml.jackson.databind.ObjectMapper; +import com.nanri.aiimage.config.SimilarAsinProperties; +import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto; +import org.junit.jupiter.api.Test; + +import java.math.BigDecimal; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +class SimilarAsinCozeClientTest { + + private final ObjectMapper objectMapper = new ObjectMapper(); + + @Test + @SuppressWarnings("unchecked") + void buildParametersIncludesAlibabaPriceFromPythonPayload() throws Exception { + List rows = objectMapper.readValue(""" + [ + { + "asin": "B0TEST123", + "url": "https://m.media-amazon.com/images/I/main.jpg", + "alibaba": [ + {"url": "https://cbu01.alicdn.com/img/ibank/a.jpg", "price": 12.80}, + {"url": "https://cbu01.alicdn.com/img/ibank/b.jpg", "price": 19.99} + ], + "title": "Test title", + "sku": "SKU-1" + } + ] + """, new TypeReference<>() { + }); + SimilarAsinCozeClient client = new SimilarAsinCozeClient(new SimilarAsinProperties(), objectMapper, null); + + Method method = SimilarAsinCozeClient.class.getDeclaredMethod( + "buildParameters", List.class, String.class, String.class, boolean.class); + method.setAccessible(true); + Map parameters = (Map) method.invoke(client, rows, "", "", true); + + List> items = (List>) parameters.get("items"); + Map item = items.getFirst(); + List> alibaba = (List>) item.get("alibaba"); + + assertEquals("B0TEST123", item.get("asin")); + assertEquals("https://m.media-amazon.com/images/I/main.jpg", item.get("url")); + assertEquals("Test title", item.get("title")); + assertEquals("SKU-1", item.get("sku")); + assertEquals(2, alibaba.size()); + assertEquals("https://cbu01.alicdn.com/img/ibank/a.jpg", alibaba.get(0).get("url")); + assertEquals(0, new BigDecimal("12.8").compareTo((BigDecimal) alibaba.get(0).get("price"))); + assertEquals("https://cbu01.alicdn.com/img/ibank/b.jpg", alibaba.get(1).get("url")); + assertEquals(0, new BigDecimal("19.99").compareTo((BigDecimal) alibaba.get(1).get("price"))); + assertTrue(rows.getFirst().hasImageUrl()); + } + + @Test + @SuppressWarnings("unchecked") + void buildParametersFallsBackAlibabaFromLegacyUrlsAndTopLevelPrice() throws Exception { + SimilarAsinResultRowDto row = new SimilarAsinResultRowDto(); + row.setAsin("B0LEGACY1"); + row.setUrl("https://m.media-amazon.com/images/I/main.jpg"); + row.setUrls(List.of("https://cbu01.alicdn.com/img/ibank/legacy-a.jpg")); + row.setPrice("8.50"); + row.setTitle("Legacy title"); + row.setSku("SKU-LEGACY"); + SimilarAsinCozeClient client = new SimilarAsinCozeClient(new SimilarAsinProperties(), objectMapper, null); + + Method method = SimilarAsinCozeClient.class.getDeclaredMethod( + "buildParameters", List.class, String.class, String.class, boolean.class); + method.setAccessible(true); + Map parameters = (Map) method.invoke(client, List.of(row), "", "", false); + + List> items = (List>) parameters.get("items"); + List> alibaba = (List>) items.getFirst().get("alibaba"); + + assertEquals(1, alibaba.size()); + assertEquals("https://cbu01.alicdn.com/img/ibank/legacy-a.jpg", alibaba.getFirst().get("url")); + assertEquals(0, new BigDecimal("8.5").compareTo((BigDecimal) alibaba.getFirst().get("price"))); + } +} diff --git a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java index b3ddc41..4af7244 100644 --- a/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java +++ b/backend-java/src/test/java/com/nanri/aiimage/modules/similarasin/util/SimilarAsinImageEmbedderTest.java @@ -7,13 +7,16 @@ import java.awt.Color; import java.awt.Graphics2D; import java.awt.image.BufferedImage; import java.io.ByteArrayOutputStream; +import java.io.IOException; import java.net.InetAddress; import java.net.UnknownHostException; import java.util.Collections; import java.util.List; +import java.util.concurrent.TimeoutException; import static org.junit.jupiter.api.Assertions.assertEquals; 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.junit.jupiter.api.Assertions.assertFalse; @@ -154,4 +157,47 @@ class SimilarAsinImageEmbedderTest { assertEquals(12345678L, ex.size()); assertTrue(ex.getMessage().contains("oversize")); } + + @Test + void httpClientKeepsImageDownloadResilienceEnabled() { + assertTrue(embedder.httpClient().retryOnConnectionFailure(), "OkHttp should retry transient connection failures"); + assertTrue(embedder.httpClient().followRedirects(), "image CDN redirects should be followed"); + assertTrue(embedder.httpClient().followSslRedirects(), "signed CDN downloads should follow browser-like SSL redirects"); + } + + @Test + void buildImageRequestUsesBrowserLikeHeaders() { + var req = SimilarAsinImageEmbedder.buildImageRequest("https://m.media-amazon.com/images/I/abc.jpg"); + + assertEquals("GET", req.method()); + assertTrue(req.header("User-Agent").contains("Chrome"), "User-Agent should look browser-like"); + assertTrue(req.header("Accept").contains("image/"), "Accept should prefer images"); + assertEquals("https://www.amazon.com/", req.header("Referer")); + assertEquals("no-cache", req.header("Cache-Control")); + assertNotNull(req.header("Accept-Language")); + } + + @Test + void buildImageRequestUsesDownloadHeadersForCozeSignedImages() { + var req = SimilarAsinImageEmbedder.buildImageRequest( + "https://lf9-bot-platform-tos-sign.coze.cn/bot-studio-bot-platform/bot_files/1/image/jpeg/2/merged_image.jpg?x-expires=1&x-signature=a"); + + assertEquals("GET", req.method()); + assertTrue(req.header("User-Agent").contains("Chrome"), "User-Agent should look browser-like"); + assertEquals("*/*", req.header("Accept")); + assertNull(req.header("Referer"), "Coze signed download links should not carry an Amazon referer"); + assertEquals("no-cache", req.header("Cache-Control")); + assertTrue(SimilarAsinImageEmbedder.isCozeSignedImageUrl(req.url().toString())); + } + + @Test + void errorSummaryDoesNotHideEmptyTimeoutMessage() { + TimeoutException timeout = new TimeoutException(); + assertEquals("TimeoutException: ", SimilarAsinImageEmbedder.errorSummary(timeout)); + + IOException io = new IOException("outer", new TimeoutException("inner")); + String summary = SimilarAsinImageEmbedder.errorSummary(io); + assertTrue(summary.contains("IOException: outer")); + assertTrue(summary.contains("cause=TimeoutException: inner")); + } } diff --git a/backend/blueprints/admin_api.py b/backend/blueprints/admin_api.py index 9cf37bd..5d0a380 100644 --- a/backend/blueprints/admin_api.py +++ b/backend/blueprints/admin_api.py @@ -196,7 +196,7 @@ def _ensure_column_sort_schema(): try: with conn.cursor() as cur: try: - cur.execute("ALTER TABLE columns ADD COLUMN sort_order INT NOT NULL DEFAULT 0 COMMENT '鑿滃崟鎺掑簭' AFTER route_path") + cur.execute("ALTER TABLE columns ADD COLUMN sort_order INT NOT NULL DEFAULT 0 COMMENT '菜单排序' AFTER route_path") except Exception: pass try: