feat: update task routing and coze handling

This commit is contained in:
super
2026-06-03 17:14:50 +08:00
parent ea37d82d73
commit e98a1a1207
37 changed files with 1883 additions and 254 deletions

View File

@@ -647,7 +647,7 @@
<div class="nav-section" data-column-key="brand_front_tools">
<div class="nav-section-title">前端工具</div>
<div class="nav-section-items">
<span class="nav-item disabled">采集数据</span>
<a class="nav-item" href="/new_web_source/collect-data.html">采集数据</a>
<span class="nav-item disabled">变体分析</span>
<button type="button" class="nav-item active" data-panel="brandCheck">品牌检测</button>
<a class="nav-item" href="/new_web_source/appearance-patent.html">外观专利检测</a>
@@ -1447,4 +1447,4 @@
</script>
</body>
</html>
</html>

View File

@@ -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. 历史清理

View File

@@ -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<byte[]> 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<Void> handleBusinessException(BusinessException ex) {
if (Integer.valueOf(40901).equals(ex.getCode())) {

View File

@@ -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;
}
}

View File

@@ -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<String> 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<byte[]> 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<String> headerNames = request.getHeaderNames();
while (headerNames != null && headerNames.hasMoreElements()) {
String headerName = headerNames.nextElement();
if (headerName == null || HOP_BY_HOP_HEADERS.contains(headerName.toLowerCase())) {
continue;
}
Enumeration<String> 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;
}
}

View File

@@ -50,9 +50,9 @@ public final class CozeGroupResultPropagator {
/**
* 否定前缀关键字。若当前值同时包含 standard 和这些关键字之一,则不视为命中。
* 用于规避"没有侵权""不侵权""未发现明显侵权风险"等被 contains 误判为命中"侵权"的情况。
* 用于规避"没有侵权""无侵权""不侵权""未发现明显侵权风险"等被 contains 误判为命中"侵权"的情况。
*/
private static final List<String> NEGATIVE_KEYWORDS = Arrays.asList("没有", "", "");
private static final List<String> 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) {

View File

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

View File

@@ -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;
}

View File

@@ -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_RESOLVERUNNING 超时自动收尾/失败,按分钟计算;
* 与删除品牌共用同一定时调度。
*/
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;
}

View File

@@ -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<String, String> routes = new LinkedHashMap<>();
private int connectTimeoutMillis = 3000;
private int readTimeoutMillis = 300000;
private int requestBodyCacheLimitBytes = 104857600;
}

View File

@@ -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 {
}

View File

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

View File

@@ -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;
}

View File

@@ -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<String> 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);
}
}
}

View File

@@ -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<AppearancePatentResultRowDto> inspect(List<AppearancePatentResultRowDto> rows, String prompt, String apiKey) {
@@ -361,7 +366,7 @@ public class AppearancePatentCozeClient {
List<String> rowIds = rows.stream().map(row -> nonBlank(row.getId(), "")).toList();
List<String> asins = rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList();
List<String> countries = rows.stream().map(row -> nonBlank(row.getCountry(), "")).toList();
List<String> skus = rows.stream().map(row -> nonBlank(row.getSku(), "")).toList();
List<String> skus = rows.stream().map(row -> sanitizeSku(row.getSku())).toList();
List<String> titles = rows.stream().map(row -> nonBlank(row.getTitle(), row.getAsin())).toList();
List<String> 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<String> 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();
}

View File

@@ -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<FileTaskEntity> 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<FileTaskEntity> tasks = listStaleFinalizeCandidates(threshold);
for (FileTaskEntity task : tasks) {
@@ -647,8 +651,8 @@ public class AppearancePatentTaskService {
return fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<AppearancePatentResultRowDto> rows) {
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
.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<String, AppearancePatentResultRowDto> 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<TaskChunkEntity>()
.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<String, AppearancePatentResultRowDto> 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<TaskChunkEntity> updateWrapper = new LambdaUpdateWrapper<TaskChunkEntity>()
.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<AppearancePatentResultRowDto> expandRows(List<AppearancePatentResultRowDto> 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<CozeCandidate> 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<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
@@ -1940,6 +1961,43 @@ public class AppearancePatentTaskService {
}
}
private void finalizeTimedOutCozeStatesForTask(Long taskId) {
if (taskId == null || taskId <= 0) {
return;
}
List<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
.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<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task != null) {
Map<String, List<AppearancePatentParsedRowVo>> 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<AppearancePatentResultRowDto> 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<String, List<AppearancePatentParsedRowVo>> 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<String, List<AppearancePatentParsedRowVo>> 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(), "");
}

View File

@@ -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<String, Object> 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<String> brands, String strategy) {
List<String> distinctBrands = distinctNonBlank(brands);
List<Object> failedData = new ArrayList<>();
List<Object> 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<String> splitTitleText(String titleText) {
String normalized = normalize(titleText)
.replace('', ',')
.replace('、', ',')
.replace('', ',')
.replace(';', ',')
.replace('\n', ',')
.replace('\r', ',')
.replace('\t', ',');
if (normalized.isBlank()) {
return List.of();
}
List<String> 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<Object> nullToEmpty(List<Object> 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<String> distinctNonBlank(List<String> values) {
List<String> 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<Object> faildData = new ArrayList<>();
@JsonAlias({"query_faild_data", "query_failed_data"})
private List<Object> queryFaildData = new ArrayList<>();
private static BrandCheckResponse empty() {
return new BrandCheckResponse();
}
}
public record BrandCheckBatchResult(
List<String> brands,
List<Object> faildData,
List<Object> queryFaildData
) {
public boolean hasFailedData() {
return faildData != null && !faildData.isEmpty();
}
public boolean hasQueryFailedData() {
return queryFaildData != null && !queryFaildData.isEmpty();
}
}
}

View File

@@ -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<String, String> 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());

View File

@@ -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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<FileTaskEntity>()
.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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRODUCT_RISK)
.eq(FileTaskEntity::getStatus, "RUNNING")
@@ -256,8 +247,8 @@ public class DeleteBrandStaleTaskService {
}
Map<Long, Long> 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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_PRICE_TRACK)
.eq(FileTaskEntity::getStatus, "RUNNING")
@@ -340,8 +323,8 @@ public class DeleteBrandStaleTaskService {
}
Map<Long, Long> 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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE_SHOP_MATCH)
.eq(FileTaskEntity::getStatus, "RUNNING")
@@ -421,8 +398,8 @@ public class DeleteBrandStaleTaskService {
}
Map<Long, Long> 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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<FileTaskEntity> runningTasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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) {

View File

@@ -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<String> skus = rows.stream().map(row -> safeText(row.getSku())).toList();
List<String> urls = rows.stream().map(this::primaryImageUrl).toList();
List<List<String>> urlLists = rows.stream().map(this::imageUrls).toList();
List<List<Map<String, Object>>> alibabaLists = rows.stream().map(this::alibabaItems).toList();
List<Map<String, Object>> items = buildItemObjects(asins, titles, skus, urls, urlLists);
List<Map<String, Object>> items = buildItemObjects(asins, titles, skus, urls, urlLists, alibabaLists);
logCozeItemsDiff(rows, items);
Map<String, Object> parameters = new LinkedHashMap<>();
parameters.put("items", items);
@@ -408,22 +410,26 @@ public class SimilarAsinCozeClient {
List<String> 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<String> titles,
List<String> skus,
List<String> urls,
List<List<String>> urlLists) {
List<List<String>> urlLists,
List<List<Map<String, Object>>> 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<String> imageUrls = urlLists.get(i);
List<Map<String, Object>> alibabaItems = alibabaLists.get(i);
Map<String, Object> 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<Map<String, Object>> alibabaItems(SimilarAsinResultRowDto row) {
if (row == null) {
return List.of();
}
List<SimilarAsinResultRowDto.AlibabaItem> explicitItems = row.getAlibaba();
if (explicitItems != null && !explicitItems.isEmpty()) {
List<Map<String, Object>> 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<String, Object> item = new LinkedHashMap<>();
item.put("url", sourceUrl);
item.put("price", sourcePrice);
result.add(item);
}
return result;
}
List<String> imageUrls = imageUrls(row);
if (imageUrls.isEmpty()) {
return List.of();
}
Object rowPrice = cozePrice(row.getPrice());
List<Map<String, Object>> result = new ArrayList<>(imageUrls.size());
for (String imageUrl : imageUrls) {
Map<String, Object> 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 "";

View File

@@ -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<AlibabaItem> 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<String> getUrls() {
// 直接返回 Python 回传的 urls不再把 url 强推到首位、不去重、不截断。
return cleanUrls(urls);
List<String> cleaned = cleanUrls(urls);
if (!cleaned.isEmpty()) {
return cleaned;
}
return alibabaUrls();
}
@JsonSetter("urls")
@@ -169,12 +177,30 @@ public class SimilarAsinResultRowDto {
this.urls = normalized;
}
public List<AlibabaItem> getAlibaba() {
if (alibaba == null || alibaba.isEmpty()) {
return new ArrayList<>();
}
List<AlibabaItem> 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<AlibabaItem> 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<String> result, Object value) {
@@ -227,4 +253,44 @@ public class SimilarAsinResultRowDto {
}
return result;
}
private List<String> alibabaUrls() {
List<AlibabaItem> items = getAlibaba();
if (items.isEmpty()) {
return new ArrayList<>();
}
List<String> 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;
}
}
}

View File

@@ -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<FileTaskEntity> 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<FileTaskEntity> tasks = listStaleFinalizeCandidates(threshold);
for (FileTaskEntity task : tasks) {
@@ -880,8 +886,8 @@ public class SimilarAsinTaskService {
return fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.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<SimilarAsinResultRowDto> rows) {
TaskChunkEntity chunk = taskChunkMapper.selectOne(new LambdaQueryWrapper<TaskChunkEntity>()
.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<String, SimilarAsinResultRowDto> 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<TaskChunkEntity>()
.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<String, SimilarAsinResultRowDto> 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<TaskChunkEntity>()
.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<SimilarAsinResultRowDto> expandRows(List<SimilarAsinResultRowDto> rows,
@@ -1482,7 +1488,8 @@ public class SimilarAsinTaskService {
}
String url = row.getUrl();
List<String> urls = row.getUrls();
log.info("[similar-asin] python inbound idx={} groupKey={} rowToken={} id={} asin={} country={} title={} sku={} url={} urlsSize={} urlsHead={} urlsTail={}",
List<SimilarAsinResultRowDto.AlibabaItem> 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<String, List<SimilarAsinParsedRowVo>> 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<String, List<SimilarAsinParsedRowVo>> 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<TaskScopeStateEntity> states = taskScopeStateMapper.selectList(new LambdaQueryWrapper<TaskScopeStateEntity>()
.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<SimilarAsinResultRowDto> batchRows = readCozeBatchRows(state);
FileTaskEntity task = taskForPoll(taskId);
if (task != null) {
Map<String, List<SimilarAsinParsedRowVo>> 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<SimilarAsinResultRowDto> batchRows,
@@ -2913,7 +2979,7 @@ public class SimilarAsinTaskService {
if (CozeFailureClassifier.isThrottleLockTimeout(message)) {
if (isCozeStateTimedOut(state)) {
Map<String, List<SimilarAsinParsedRowVo>> 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<String, List<SimilarAsinParsedRowVo>> 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;
}

View File

@@ -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 prefetchassemble 阶段第一遍扫所有 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<String, ResizedImage> 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<byte[]> 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() ? "<empty>" : 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_PXJPEG q=0.75 输出;返回字节 + 实际像素,供调用方按图自适应单元格尺寸。
* 出口校验字节数 ≤ MAX_THUMB_SIZE_BYTES超限按 MAX_THUMB_SIZE → 长边 → 质量的顺序迭代降级,

View File

@@ -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<TaskHeartbeatVo> 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));
}
}

View File

@@ -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<TaskImageCacheEntity> {
/**
* 命中时同步 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<TaskImageCacheEntity> {
*/
@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);
}

View File

@@ -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;
}

View File

@@ -16,7 +16,7 @@ import java.time.LocalDateTime;
* <li>{@link #urlHash}sha256 lowercase hex作为唯一键避免 1024 长 URL 命中索引长度限制;</li>
* <li>{@link #imageBytes}:已 resize 后的 JPEG 缩略图字节,最大约 300KB保持与
* {@code SimilarAsinImageEmbedder.MAX_THUMB_SIZE_BYTES} 对齐);</li>
* <li>{@link #lastUsedAt}:用于后续 LRU 清理(>7 天未用)。</li>
* <li>{@link #lastUsedAt}:用于后续 LRU 清理(默认 >3 天未用)。</li>
* </ul>
*/
@Data

View File

@@ -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;
}
}

View File

@@ -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<FileTaskEntity> fileQuery = new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getId, taskId)
.last("limit 1");
if (userId != null && userId > 0) {
fileQuery.eq(FileTaskEntity::getUserId, userId);
}
FileTaskEntity fileTask = fileTaskMapper.selectOne(fileQuery);
LambdaQueryWrapper<BrandCrawlTaskEntity> brandQuery = new LambdaQueryWrapper<BrandCrawlTaskEntity>()
.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<FileTaskEntity>()
.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<BrandCrawlTaskEntity>()
.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<String, String> buildDeleteBrandHeartbeatProgress(TaskHeartbeatRequest request) {
String now = String.valueOf(System.currentTimeMillis());
Map<String, String> 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<String, String> values, String key, Object value) {
if (value != null) {
values.put(key, String.valueOf(value));
}
}
}

View File

@@ -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);
}
}
}

View File

@@ -75,6 +75,14 @@ aiimage:
# For IDEA/local startup, set -Daiimage.instance-id=<stable-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:

View File

@@ -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<ParsedRow> parsedRows = List.of(
new ParsedRow("20_1"),
new ParsedRow("20_2")
);
List<ResultRow> 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<ParsedRow> parsedRows = List.of(
new ParsedRow("20_1"),
new ParsedRow("20_2")
);
List<ResultRow> 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;
}
}
}

View File

@@ -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<AppearancePatentResultRowDto> 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());
}
}

View File

@@ -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<String> brands = client.splitTitleText("'阿凡达,任天堂' Disney、LEGO\nSony");
assertThat(brands).containsExactly("阿凡达", "任天堂", "Disney", "LEGO", "Sony");
}
@Test
void splitTitleTextDeduplicatesBlankValues() {
BrandCheckClient client = new BrandCheckClient(new BrandCheckProperties());
List<String> brands = client.splitTitleText(" 任天堂, ,任天堂Sony ");
assertThat(brands).containsExactly("任天堂", "Sony");
}
}

View File

@@ -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<SimilarAsinResultRowDto> 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<String, Object> parameters = (Map<String, Object>) method.invoke(client, rows, "", "", true);
List<Map<String, Object>> items = (List<Map<String, Object>>) parameters.get("items");
Map<String, Object> item = items.getFirst();
List<Map<String, Object>> alibaba = (List<Map<String, Object>>) 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<String, Object> parameters = (Map<String, Object>) method.invoke(client, List.of(row), "", "", false);
List<Map<String, Object>> items = (List<Map<String, Object>>) parameters.get("items");
List<Map<String, Object>> alibaba = (List<Map<String, Object>>) 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")));
}
}

View File

@@ -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: <empty>", 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"));
}
}

View File

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