super/crawler-plugin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

增加rufts

Browse Source
This commit is contained in:
super
2026-04-27 18:45:33 +08:00
parent 052bb31aea
commit 5105bf7049
35 changed files with 2098 additions and 248 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1056
backend-java/jstack-6488.txt Normal file
View File

File diff suppressed because it is too large Load Diff

2
backend-java/src/main/java/com/nanri/aiimage/config/AppearancePatentProperties.java
View File

@@ -11,6 +11,8 @@ public class AppearancePatentProperties {
private String cozeWorkflowId = "7632683471312355338";
private String cozeToken = "";
private int cozeBatchSize = 10;
private int cozeConnectTimeoutMillis = 10000;
private int cozeReadTimeoutMillis = 60000;
private int staleTimeoutMinutes = 20;
private String staleFinalizeCron = "0 */2 * * * *";
}

1
backend-java/src/main/java/com/nanri/aiimage/config/DeleteBrandProgressProperties.java
View File

@@ -7,6 +7,7 @@ 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 * * * * *";
/**

271
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/client/AppearancePatentCozeClient.java
View File

@@ -9,11 +9,14 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.http.MediaType;
import org.springframework.http.client.SimpleClientHttpRequestFactory;
import org.springframework.stereotype.Component;
import org.springframework.util.StreamUtils;
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.Locale;
import java.util.Map;
@Component
@@ -33,29 +36,72 @@ public class AppearancePatentCozeClient {
return rows.stream().map(this::copy).toList();
}
try {
String raw = postWorkflow(rows, prompt);
List<CozeResult> results = parseResults(raw);
List<AppearancePatentResultRowDto> merged = new ArrayList<>();
for (int i = 0; i < rows.size(); i++) {
AppearancePatentResultRowDto row = copy(rows.get(i));
if (i < results.size()) {
CozeResult result = results.get(i);
row.setTitleRisk(result.title());
row.setAppearanceRisk(result.appearance());
row.setPatentRisk(result.patent());
row.setConclusion(result.result());
}
merged.add(row);
}
return merged;
return inspectWithFallback(rows, prompt);
} catch (Exception ex) {
log.warn("[appearance-patent] coze batch failed size={} err={}", rows.size(), ex.getMessage());
return rows.stream().map(this::copy).toList();
String failureMessage = failureMessage(ex);
log.warn("[appearance-patent] coze batch failed size={} err={}", rows.size(), failureMessage);
return rows.stream().map(this::copy).map(row -> markFailed(row, failureMessage)).toList();
}
}
private List<AppearancePatentResultRowDto> inspectWithFallback(List<AppearancePatentResultRowDto> rows, String prompt) {
try {
if (rows.size() == 1) {
return inspectSingleRowWithRetry(rows, prompt);
}
InspectAttempt attempt = inspectOnce(rows, prompt);
if (attempt.resolvedCount() < rows.size()) {
throw new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
}
return attempt.mergedRows();
} catch (Exception ex) {
if (shouldSplitBatch(rows, ex)) {
int middle = rows.size() / 2;
log.warn("[appearance-patent] coze batch fallback split size={} left={} right={} err={}",
rows.size(), middle, rows.size() - middle, failureMessage(ex));
List<AppearancePatentResultRowDto> merged = new ArrayList<>(rows.size());
merged.addAll(inspectWithFallback(rows.subList(0, middle), prompt));
merged.addAll(inspectWithFallback(rows.subList(middle, rows.size()), prompt));
return merged;
}
throw propagate(ex);
}
}
private List<AppearancePatentResultRowDto> inspectSingleRowWithRetry(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
AppearancePatentResultRowDto row = rows.getFirst();
PartialCozeResultException lastFailure = null;
for (int attemptIndex = 1; attemptIndex <= 3; attemptIndex++) {
InspectAttempt attempt = inspectOnce(rows, prompt);
if (attempt.resolvedCount() == rows.size()) {
return attempt.mergedRows();
}
log.warn("[appearance-patent] coze single unresolved attempt={} rowId={} asin={} country={} title={} url={} raw={}",
attemptIndex,
row.getId(),
row.getAsin(),
row.getCountry(),
abbreviate(row.getTitle(), 120),
abbreviate(row.getUrl(), 120),
abbreviate(attempt.raw(), 500));
lastFailure = new PartialCozeResultException(attempt.resolvedCount(), rows.size(), attempt.rawResultCount());
}
throw lastFailure == null ? new PartialCozeResultException(0, rows.size(), 0) : lastFailure;
}
private InspectAttempt inspectOnce(List<AppearancePatentResultRowDto> rows, String prompt) throws Exception {
String raw = postWorkflow(rows, prompt);
List<CozeResult> results = parseResults(raw);
List<AppearancePatentResultRowDto> merged = mergeRows(rows, results);
return new InspectAttempt(raw, merged, resolvedCount(merged), results.size());
}
private String postWorkflow(List<AppearancePatentResultRowDto> rows, String prompt) {
Map<String, Object> parameters = new LinkedHashMap<>();
parameters.put("row_id_list", rows.stream().map(row -> nonBlank(row.getId(), "")).toList());
parameters.put("asin_list", rows.stream().map(row -> nonBlank(row.getAsin(), "")).toList());
parameters.put("country_list", rows.stream().map(row -> nonBlank(row.getCountry(), "")).toList());
parameters.put("row_key_list", rows.stream().map(this::rowKey).toList());
parameters.put("title_list", rows.stream().map(row -> nonBlank(row.getTitle(), row.getAsin())).toList());
parameters.put("url_list", rows.stream().map(row -> nonBlank(row.getUrl(), "")).toList());
parameters.put("prompt", prompt == null ? "" : prompt);
@@ -71,7 +117,10 @@ public class AppearancePatentCozeClient {
headers.setContentType(MediaType.APPLICATION_JSON);
});
request.body(body);
return request.retrieve().body(String.class);
return request.exchange((clientRequest, clientResponse) -> {
byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());
return responseBytes.length == 0 ? "" : new String(responseBytes, StandardCharsets.UTF_8);
});
}
private List<CozeResult> parseResults(String raw) throws Exception {
@@ -89,6 +138,9 @@ public class AppearancePatentCozeClient {
if (array.isArray()) {
for (JsonNode node : array) {
results.add(new CozeResult(
text(firstNonNull(node.get("row_id"), firstNonNull(node.get("rowId"), node.get("id")))),
text(node.get("asin")),
text(firstNonNull(node.get("country"), node.get("site"))),
text(node.get("title")),
text(node.get("appearance")),
text(firstNonNull(node.get("patent"), node.get("patent "))),
@@ -99,10 +151,51 @@ public class AppearancePatentCozeClient {
return results;
}
private List<AppearancePatentResultRowDto> mergeRows(List<AppearancePatentResultRowDto> rows, List<CozeResult> results) {
Map<String, CozeResult> resultByCompositeKey = new LinkedHashMap<>();
Map<String, CozeResult> resultByRowId = new LinkedHashMap<>();
for (CozeResult result : results) {
String compositeKey = rowKey(result.rowId(), result.asin(), result.country());
if (!compositeKey.isBlank()) {
resultByCompositeKey.putIfAbsent(compositeKey, result);
}
String rowIdKey = normalize(result.rowId());
if (!rowIdKey.isBlank()) {
resultByRowId.putIfAbsent(rowIdKey, result);
}
}
List<AppearancePatentResultRowDto> merged = new ArrayList<>(rows.size());
boolean allowIndexFallback = results.size() == rows.size();
for (int i = 0; i < rows.size(); i++) {
AppearancePatentResultRowDto row = copy(rows.get(i));
CozeResult result = resultByCompositeKey.get(rowKey(row.getId(), row.getAsin(), row.getCountry()));
if (result == null) {
result = resultByRowId.get(normalize(row.getId()));
}
if (result == null && allowIndexFallback && i < results.size()) {
result = results.get(i);
}
applyResult(row, result);
merged.add(row);
}
return merged;
}
private void applyResult(AppearancePatentResultRowDto row, CozeResult result) {
if (row == null || result == null) {
return;
}
row.setTitleRisk(result.title());
row.setAppearanceRisk(result.appearance());
row.setPatentRisk(result.patent());
row.setConclusion(result.result());
}
private RestClient restClient() {
SimpleClientHttpRequestFactory requestFactory = new SimpleClientHttpRequestFactory();
requestFactory.setConnectTimeout(10000);
requestFactory.setReadTimeout(60000);
requestFactory.setConnectTimeout(properties.getCozeConnectTimeoutMillis());
requestFactory.setReadTimeout(properties.getCozeReadTimeoutMillis());
return RestClient.builder().requestFactory(requestFactory).build();
}
@@ -122,6 +215,67 @@ public class AppearancePatentCozeClient {
return row;
}
private AppearancePatentResultRowDto markFailed(AppearancePatentResultRowDto row, String failureMessage) {
if (row.getError() == null || row.getError().isBlank()) {
row.setError(failureMessage);
}
if (row.getConclusion() == null || row.getConclusion().isBlank()) {
row.setConclusion(failureMessage);
}
return row;
}
private boolean shouldSplitBatch(List<AppearancePatentResultRowDto> rows, Exception ex) {
return rows != null && rows.size() > 1 && isRetryableBatchFailure(ex);
}
private boolean isRetryableBatchFailure(Exception ex) {
if (ex instanceof PartialCozeResultException) {
return true;
}
String message = ex == null ? "" : nonBlank(ex.getMessage(), "");
return message.contains("Workflow node execution limit exceeded")
|| message.contains("Read timed out")
|| message.contains("Connection reset")
|| message.contains("I/O error on POST request")
|| message.toLowerCase(Locale.ROOT).contains("timeout");
}
private int resolvedCount(List<AppearancePatentResultRowDto> rows) {
int resolved = 0;
for (AppearancePatentResultRowDto row : rows) {
if (hasResolvedCozeFields(row)) {
resolved++;
}
}
return resolved;
}
private boolean hasResolvedCozeFields(AppearancePatentResultRowDto row) {
if (row == null) {
return false;
}
return !normalize(row.getTitleRisk()).isBlank()
|| !normalize(row.getAppearanceRisk()).isBlank()
|| !normalize(row.getPatentRisk()).isBlank()
|| !normalize(row.getConclusion()).isBlank();
}
private String abbreviate(String value, int maxLength) {
String normalized = value == null ? "" : value.trim();
if (normalized.length() <= maxLength) {
return normalized;
}
return normalized.substring(0, Math.max(0, maxLength - 3)) + "...";
}
private RuntimeException propagate(Exception ex) {
if (ex instanceof RuntimeException runtimeException) {
return runtimeException;
}
return new IllegalStateException(nonBlank(ex.getMessage(), "Coze call failed"), ex);
}
private JsonNode firstNonNull(JsonNode left, JsonNode right) {
return left == null || left.isNull() ? right : left;
}
@@ -134,6 +288,38 @@ public class AppearancePatentCozeClient {
return value == null || value.isBlank() ? fallback : value;
}
private String normalize(String value) {
return value == null ? "" : value.replace("\ufeff", "").replace("\u3000", " ").trim();
}
private String rowKey(AppearancePatentResultRowDto row) {
if (row == null) {
return "";
}
return rowKey(row.getId(), row.getAsin(), row.getCountry());
}
private String rowKey(String rowId, String asin, String country) {
return normalize(rowId) + "::" + normalize(asin).toUpperCase(Locale.ROOT) + "::" + normalize(country);
}
private String failureMessage(Exception ex) {
if (ex instanceof PartialCozeResultException partial) {
return "Coze返回结果不完整(" + partial.resolvedCount() + "/" + partial.expectedCount() + ")";
}
String message = ex == null ? null : ex.getMessage();
if (message == null || message.isBlank()) {
return "Coze调用失败";
}
if (message.contains("Workflow node execution limit exceeded")) {
return "Coze工作流节点执行超限，请检查工作流配置";
}
if (message.contains("Read timed out")) {
return "Coze调用超时，请稍后重试";
}
return "Coze调用失败: " + message;
}
private String stripBearer(String token) {
String normalized = token == null ? "" : token.trim();
return normalized.regionMatches(true, 0, "Bearer ", 0, 7) ? normalized.substring(7).trim() : normalized;
@@ -151,6 +337,49 @@ public class AppearancePatentCozeClient {
return base + suffix;
}
private record CozeResult(String title, String appearance, String patent, String result) {
private record CozeResult(
String rowId,
String asin,
String country,
String title,
String appearance,
String patent,
String result
) {
}
private record InspectAttempt(
String raw,
List<AppearancePatentResultRowDto> mergedRows,
int resolvedCount,
int rawResultCount
) {
}
private static final class PartialCozeResultException extends RuntimeException {
private final int resolvedCount;
private final int expectedCount;
private final int rawResultCount;
private PartialCozeResultException(int resolvedCount, int expectedCount, int rawResultCount) {
super("partial-result resolved=" + resolvedCount + "/" + expectedCount + " raw=" + rawResultCount);
this.resolvedCount = resolvedCount;
this.expectedCount = expectedCount;
this.rawResultCount = rawResultCount;
}
private int resolvedCount() {
return resolvedCount;
}
private int expectedCount() {
return expectedCount;
}
@SuppressWarnings("unused")
private int rawResultCount() {
return rawResultCount;
}
}
}

1
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/model/vo/AppearancePatentParsedRowVo.java
View File

@@ -16,6 +16,7 @@ public class AppearancePatentParsedRowVo {
@Schema(description = "前端展示和 Python 回传使用的 id。整数 id 原样保留，子数据第一条如 2_1 原样保留。", example = "2_1")
private String displayId;
@Schema(description = "亚马逊 ASIN。", example = "B0CJ8SNXXV")
private String groupKey;
private String asin;
@Schema(description = "国家或站点。", example = "英国")
private String country;

8
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskCacheService.java
View File

@@ -22,14 +22,14 @@ public class AppearancePatentTaskCacheService {
private final StringRedisTemplate stringRedisTemplate;
private final ObjectMapper objectMapper;
public void appendPendingRow(Long taskId, Integer chunkIndex, AppearancePatentResultRowDto row) {
if (taskId == null || taskId <= 0 || chunkIndex == null || row == null) {
public void appendPendingRow(Long taskId, String scopeHash, Integer chunkIndex, AppearancePatentResultRowDto row) {
if (taskId == null || taskId <= 0 || scopeHash == null || scopeHash.isBlank() || chunkIndex == null || row == null) {
return;
}
try {
stringRedisTemplate.opsForList().rightPush(
pendingRowsKey(taskId),
objectMapper.writeValueAsString(new PendingRow(chunkIndex, row))
objectMapper.writeValueAsString(new PendingRow(scopeHash, chunkIndex, row))
);
stringRedisTemplate.expire(pendingRowsKey(taskId), Duration.ofHours(TTL_HOURS));
touchTaskHeartbeat(taskId);
@@ -139,6 +139,6 @@ public class AppearancePatentTaskCacheService {
return "appearance-patent:task:heartbeat:" + taskId;
}
public record PendingRow(Integer chunkIndex, AppearancePatentResultRowDto row) {
public record PendingRow(String scopeHash, Integer chunkIndex, AppearancePatentResultRowDto row) {
}
}

415
backend-java/src/main/java/com/nanri/aiimage/modules/appearancepatent/service/AppearancePatentTaskService.java
View File

@@ -49,7 +49,10 @@ import org.apache.poi.xssf.streaming.SXSSFWorkbook;
import org.springframework.dao.DuplicateKeyException;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
import java.io.File;
import java.io.FileInputStream;
@@ -65,6 +68,7 @@ import java.util.Locale;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Supplier;
@Service
@RequiredArgsConstructor
@@ -101,6 +105,7 @@ public class AppearancePatentTaskService {
private final AppearancePatentProperties properties;
private final TaskFileJobService taskFileJobService;
private final TransientPayloadStorageService transientPayloadStorageService;
private final PlatformTransactionManager transactionManager;
@Transactional
public AppearancePatentParseVo parseAndCreateTask(AppearancePatentParseRequest request) {
@@ -259,8 +264,16 @@ public class AppearancePatentTaskService {
return vo;
}
@Transactional
public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) {
if (transactionManager != null) {
SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request));
flushPendingRowsIfNeeded(context.task(), context.forceFlush());
inNewTransaction(() -> {
completeSubmittedChunk(context);
return null;
});
return;
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("任务不存在");
@@ -302,19 +315,21 @@ public class AppearancePatentTaskService {
taskChunkMapper.insert(chunk);
} catch (DuplicateKeyException ex) {
inserted = false;
transientPayloadStorageService.deletePayloadIfPresent(storedPayload);
// storeChunkPayload uses a deterministic key like chunk-{index}. When two
// concurrent callbacks submit the same chunk, deleting the loser payload here
// can also remove the winner's shared object and break later assembly.
log.info("[appearance-patent] duplicate chunk inserted concurrently taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
}
if (inserted) {
for (AppearancePatentResultRowDto rawRow : rawRows) {
taskCacheService.appendPendingRow(taskId, chunkIndex, rawRow);
taskCacheService.appendPendingRow(taskId, scopeHash, chunkIndex, rawRow);
}
}
} else {
log.info("[appearance-patent] duplicate chunk ignored taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
}
flushPendingRowsIfNeeded(task, scopeHash, Boolean.TRUE.equals(request.getDone()));
flushPendingRowsIfNeeded(task, Boolean.TRUE.equals(request.getDone()));
TaskScopeStateEntity scope = taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getTaskId, taskId)
@@ -394,8 +409,28 @@ public class AppearancePatentTaskService {
}
@Scheduled(cron = "${aiimage.appearance-patent.stale-finalize-cron:0 */2 * * * *}")
@Transactional
public void finalizeStaleTasks() {
if (transactionManager != null) {
LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes()));
long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
.eq(FileTaskEntity::getStatus, STATUS_RUNNING)
.lt(FileTaskEntity::getUpdatedAt, threshold)
.last("limit 50"));
for (FileTaskEntity task : tasks) {
long heartbeatMillis = taskCacheService.getTaskHeartbeatMillis(task.getId());
if (heartbeatMillis > thresholdMillis) {
continue;
}
flushPendingRowsIfNeeded(task, true);
inNewTransaction(() -> {
finalizeStaleTask(task.getId(), "Python interrupted before uploading final appearance patent result");
return null;
});
}
return;
}
LocalDateTime threshold = LocalDateTime.now().minusMinutes(Math.max(1, properties.getStaleTimeoutMinutes()));
long thresholdMillis = threshold.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
@@ -408,11 +443,168 @@ public class AppearancePatentTaskService {
if (heartbeatMillis > thresholdMillis) {
continue;
}
flushPendingRowsIfNeeded(task, DigestUtil.sha256Hex("appearance-patent-" + task.getId()), true);
flushPendingRowsIfNeeded(task, true);
finalizeTask(task, "Python interrupted before uploading final appearance patent result", allRowCount(task), true);
}
}
private SubmitContext persistSubmittedChunk(Long taskId, AppearancePatentSubmitResultRequest request) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("浠诲姟涓嶅瓨鍦?");
}
if (!STATUS_RUNNING.equals(task.getStatus())) {
throw new BusinessException("浠诲姟涓嶆槸杩愯涓姸鎬?");
}
int chunkIndex = request.getChunkIndex() == null ? 0 : request.getChunkIndex();
int chunkTotal = request.getChunkTotal() == null ? 1 : request.getChunkTotal();
boolean done = Boolean.TRUE.equals(request.getDone());
String scopeKey = firstNonBlank(request.getSubmissionId(), "task:" + taskId);
String scopeHash = DigestUtil.sha256Hex(scopeKey);
taskCacheService.touchTaskHeartbeat(taskId);
TaskChunkEntity existing = 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 (existing == null) {
List<AppearancePatentResultRowDto> rawRows = request.getItems() == null ? List.of() : request.getItems();
String payloadJson = writeJson(rawRows, "缁撴灉搴忓垪鍖栧け璐?");
TaskChunkEntity chunk = new TaskChunkEntity();
chunk.setTaskId(taskId);
chunk.setModuleType(MODULE_TYPE);
chunk.setScopeKey(scopeKey);
chunk.setScopeHash(scopeHash);
chunk.setChunkIndex(chunkIndex);
chunk.setChunkTotal(chunkTotal);
String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
chunk.setPayloadJson(storedPayload);
chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson));
chunk.setCreatedAt(LocalDateTime.now());
chunk.setUpdatedAt(LocalDateTime.now());
boolean inserted = true;
try {
taskChunkMapper.insert(chunk);
} catch (DuplicateKeyException ex) {
inserted = false;
log.info("[appearance-patent] duplicate chunk inserted concurrently taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
}
if (inserted) {
for (AppearancePatentResultRowDto rawRow : rawRows) {
taskCacheService.appendPendingRow(taskId, scopeHash, chunkIndex, rawRow);
}
}
} else {
log.info("[appearance-patent] duplicate chunk ignored taskId={} scope={} chunk={}", taskId, scopeKey, chunkIndex);
}
upsertScopeState(taskId, scopeKey, scopeHash, chunkTotal, request.getError(), done, false);
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
return new SubmitContext(task, scopeKey, scopeHash, done, request.getError());
}
private void completeSubmittedChunk(SubmitContext context) {
FileTaskEntity task = fileTaskMapper.selectById(context.task().getId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("浠诲姟涓嶅瓨鍦?");
}
if (!STATUS_RUNNING.equals(task.getStatus())) {
log.info("[appearance-patent] skip completion because task already finalized taskId={} status={}",
task.getId(), task.getStatus());
return;
}
upsertScopeState(task.getId(), context.scopeKey(), context.scopeHash(), null, context.error(), context.forceFlush(), true);
if (context.forceFlush() || context.error() != null && !context.error().isBlank()) {
finalizeTask(task, context.error(), allRowCount(task), true);
return;
}
task.setUpdatedAt(LocalDateTime.now());
fileTaskMapper.updateById(task);
}
private void finalizeStaleTask(Long taskId, String error) {
FileTaskEntity task = fileTaskMapper.selectById(taskId);
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !STATUS_RUNNING.equals(task.getStatus())) {
return;
}
finalizeTask(task, error, allRowCount(task), true);
}
private void upsertScopeState(Long taskId,
String scopeKey,
String scopeHash,
Integer chunkTotal,
String error,
boolean completed,
boolean cozeDone) {
TaskScopeStateEntity scope = taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
.eq(TaskScopeStateEntity::getScopeHash, scopeHash)
.last("limit 1"));
LocalDateTime now = LocalDateTime.now();
if (scope == null) {
scope = new TaskScopeStateEntity();
scope.setTaskId(taskId);
scope.setModuleType(MODULE_TYPE);
scope.setScopeKey(scopeKey);
scope.setScopeHash(scopeHash);
scope.setCreatedAt(now);
}
if (chunkTotal != null) {
scope.setChunkTotal(chunkTotal);
}
scope.setReceivedChunkCount(countChunks(taskId, scopeHash));
scope.setLastChunkAt(now);
scope.setLastError(error);
scope.setCompleted(completed ? 1 : 0);
scope.setUpdatedAt(now);
scope.setStateJson(cozeDone
? "{\"phase\":\"RECEIVED\",\"coze\":\"DONE\"}"
: "{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
if (scope.getId() == null) {
try {
taskScopeStateMapper.insert(scope);
return;
} catch (DuplicateKeyException ex) {
log.info("[appearance-patent] duplicate scope state inserted concurrently taskId={} scope={}", taskId, scopeKey);
scope = taskScopeStateMapper.selectOne(new LambdaQueryWrapper<TaskScopeStateEntity>()
.eq(TaskScopeStateEntity::getTaskId, taskId)
.eq(TaskScopeStateEntity::getModuleType, MODULE_TYPE)
.eq(TaskScopeStateEntity::getScopeHash, scopeHash)
.last("limit 1"));
if (scope == null) {
log.info("[appearance-patent] scope state winner not committed yet, skip duplicate updater taskId={} scope={}",
taskId, scopeKey);
return;
}
if (chunkTotal != null) {
scope.setChunkTotal(chunkTotal);
}
scope.setReceivedChunkCount(countChunks(taskId, scopeHash));
scope.setLastChunkAt(now);
scope.setLastError(error);
scope.setCompleted(completed ? 1 : 0);
scope.setUpdatedAt(now);
scope.setStateJson(cozeDone
? "{\"phase\":\"RECEIVED\",\"coze\":\"DONE\"}"
: "{\"phase\":\"RECEIVED\",\"coze\":\"PENDING\"}");
}
}
taskScopeStateMapper.updateById(scope);
}
private <T> T inNewTransaction(Supplier<T> action) {
TransactionTemplate template = new TransactionTemplate(transactionManager);
template.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
return template.execute(status -> action.get());
}
private List<AppearancePatentResultRowDto> applyCozeInBatches(List<AppearancePatentResultRowDto> items, FileTaskEntity task) {
if (items == null || items.isEmpty()) {
return List.of();
@@ -426,8 +618,10 @@ public class AppearancePatentTaskService {
return result;
}
private void flushPendingRowsIfNeeded(FileTaskEntity task, String scopeHash, boolean force) {
private void flushPendingRowsIfNeeded(FileTaskEntity task, boolean force) {
int batchSize = Math.max(1, properties.getCozeBatchSize());
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
String prompt = readAiPrompt(task);
while (force ? taskCacheService.pendingRowCount(task.getId()) > 0 : taskCacheService.pendingRowCount(task.getId()) >= batchSize) {
int drainSize = force ? batchSize : Math.max(1, batchSize);
List<AppearancePatentTaskCacheService.PendingRow> pendingRows = taskCacheService.drainPendingRows(task.getId(), drainSize);
@@ -437,27 +631,47 @@ public class AppearancePatentTaskService {
List<AppearancePatentResultRowDto> rawRows = pendingRows.stream()
.map(AppearancePatentTaskCacheService.PendingRow::row)
.toList();
List<AppearancePatentResultRowDto> cozeRows = cozeClient.inspect(rawRows, readAiPrompt(task));
List<AppearancePatentResultRowDto> cozeRows = cozeClient.inspect(rawRows, prompt);
Map<String, Map<String, AppearancePatentResultRowDto>> chunkRows = new LinkedHashMap<>();
for (int i = 0; i < pendingRows.size(); i++) {
AppearancePatentTaskCacheService.PendingRow pending = pendingRows.get(i);
AppearancePatentResultRowDto resultRow = i < cozeRows.size() ? cozeRows.get(i) : pending.row();
List<AppearancePatentResultRowDto> persistedRows = expandRows(task, List.of(resultRow));
updateChunkPayload(task.getId(), scopeHash, pending.chunkIndex(), persistedRows);
List<AppearancePatentResultRowDto> expandedRows = expandRows(List.of(resultRow), allRowsByBaseId);
String chunkKey = pending.scopeHash() + "::" + pending.chunkIndex();
Map<String, AppearancePatentResultRowDto> mergedRows =
chunkRows.computeIfAbsent(chunkKey, key -> new LinkedHashMap<>());
for (AppearancePatentResultRowDto expandedRow : expandedRows) {
mergedRows.put(rowKey(expandedRow.getId(), expandedRow.getAsin(), expandedRow.getCountry()), expandedRow);
}
}
for (Map.Entry<String, Map<String, AppearancePatentResultRowDto>> entry : chunkRows.entrySet()) {
int delimiter = entry.getKey().lastIndexOf("::");
if (delimiter <= 0) {
continue;
}
String scopeHash = entry.getKey().substring(0, delimiter);
Integer chunkIndex = Integer.parseInt(entry.getKey().substring(delimiter + 2));
mergeChunkPayload(task.getId(), scopeHash, chunkIndex, new ArrayList<>(entry.getValue().values()));
}
}
}
private void updateChunkPayload(Long taskId, String scopeHash, Integer chunkIndex, List<AppearancePatentResultRowDto> rows) {
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) {
if (chunk == null || rows == null || rows.isEmpty()) {
return;
}
Map<String, AppearancePatentResultRowDto> persistedRows = readChunkRows(chunk);
for (AppearancePatentResultRowDto row : rows) {
persistedRows.put(rowKey(row.getId(), row.getAsin(), row.getCountry()), row);
}
String payloadJson = writeJson(rows == null ? List.of() : rows, "结果序列化失败");
payloadJson = writeJson(new ArrayList<>(persistedRows.values()), "appearance patent chunk payload merge failed");
String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(chunk.getPayloadJson(), storedPayload);
chunk.setPayloadJson(storedPayload);
@@ -466,16 +680,15 @@ public class AppearancePatentTaskService {
taskChunkMapper.updateById(chunk);
}
private List<AppearancePatentResultRowDto> expandRows(FileTaskEntity task, List<AppearancePatentResultRowDto> representatives) {
private List<AppearancePatentResultRowDto> expandRows(List<AppearancePatentResultRowDto> representatives,
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId) {
if (representatives == null || representatives.isEmpty()) {
return List.of();
}
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
List<AppearancePatentResultRowDto> result = new ArrayList<>();
Set<String> emittedKeys = new LinkedHashSet<>();
for (AppearancePatentResultRowDto representative : representatives) {
String baseId = baseId(representative.getId());
List<AppearancePatentParsedRowVo> siblings = allRowsByBaseId.get(baseId);
List<AppearancePatentParsedRowVo> siblings = resolveSiblingRows(representative, allRowsByBaseId);
if (siblings == null || siblings.isEmpty()) {
result.add(representative);
continue;
@@ -491,14 +704,40 @@ public class AppearancePatentTaskService {
return result;
}
private Map<String, List<AppearancePatentParsedRowVo>> loadAllRowsByBaseId(FileTaskEntity task) {
Map<String, List<AppearancePatentParsedRowVo>> result = new LinkedHashMap<>();
try {
for (AppearancePatentParsedRowVo row : readParsedPayload(task).getAllItems()) {
result.computeIfAbsent(baseId(row.getDisplayId()), key -> new ArrayList<>()).add(row);
private List<AppearancePatentParsedRowVo> resolveSiblingRows(AppearancePatentResultRowDto representative,
Map<String, List<AppearancePatentParsedRowVo>> groupedRows) {
if (representative == null || groupedRows == null || groupedRows.isEmpty()) {
return List.of();
}
String representativeKey = rowKey(representative.getId(), representative.getAsin(), representative.getCountry());
for (List<AppearancePatentParsedRowVo> rows : groupedRows.values()) {
for (AppearancePatentParsedRowVo row : rows) {
String rowKey = rowKey(row.getDisplayId(), row.getAsin(), row.getCountry());
if (Objects.equals(representativeKey, rowKey)) {
return rows;
}
}
}
return List.of();
}
private Map<String, List<AppearancePatentParsedRowVo>> loadAllRowsByBaseId(FileTaskEntity task) {
try {
return groupRowsByBaseId(readParsedPayload(task).getAllItems());
} catch (Exception ex) {
log.warn("[appearance-patent] read all rows failed taskId={} err={}", task.getId(), ex.getMessage());
return new LinkedHashMap<>();
}
}
private Map<String, List<AppearancePatentParsedRowVo>> groupRowsByBaseId(List<AppearancePatentParsedRowVo> rows) {
Map<String, List<AppearancePatentParsedRowVo>> result = new LinkedHashMap<>();
if (rows == null) {
return result;
}
for (AppearancePatentParsedRowVo row : rows) {
String key = firstNonBlank(normalize(row.getGroupKey()), baseId(row.getDisplayId()));
result.computeIfAbsent(key, ignored -> new ArrayList<>()).add(row);
}
return result;
}
@@ -517,8 +756,8 @@ public class AppearancePatentTaskService {
row.setId(sibling.getDisplayId());
row.setAsin(sibling.getAsin());
row.setCountry(sibling.getCountry());
row.setUrl(sibling.getUrl());
row.setTitle(sibling.getTitle());
row.setUrl(firstNonBlank(representative.getUrl(), sibling.getUrl()));
row.setTitle(firstNonBlank(representative.getTitle(), sibling.getTitle()));
row.setError(representative.getError());
row.setTitleRisk(representative.getTitleRisk());
row.setAppearanceRisk(representative.getAppearanceRisk());
@@ -644,22 +883,7 @@ public class AppearancePatentTaskService {
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.orderByAsc(TaskChunkEntity::getChunkIndex));
for (TaskChunkEntity chunk : chunks) {
if (chunk.getPayloadJson() == null || chunk.getPayloadJson().isBlank()) {
continue;
}
try {
String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read appearance patent chunk failed");
JsonNode array = objectMapper.readTree(payloadJson);
if (!array.isArray()) {
continue;
}
for (JsonNode node : array) {
AppearancePatentResultRowDto row = objectMapper.treeToValue(node, AppearancePatentResultRowDto.class);
result.put(rowKey(row.getId(), row.getAsin(), row.getCountry()), row);
}
} catch (Exception ex) {
log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}", taskId, chunk.getChunkIndex(), ex.getMessage());
}
result.putAll(readChunkRows(chunk));
}
return result;
}
@@ -672,26 +896,35 @@ public class AppearancePatentTaskService {
font.setBold(true);
headerStyle.setFont(font);
List<String> resultHeaders = new ArrayList<>(RESULT_HEADERS);
resultHeaders.add(4, "title");
resultHeaders.add(5, "url");
Row header = sheet.createRow(0);
for (int i = 0; i < RESULT_HEADERS.size(); i++) {
for (int i = 0; i < resultHeaders.size(); i++) {
Cell cell = header.createCell(i);
cell.setCellValue(RESULT_HEADERS.get(i));
cell.setCellValue(resultHeaders.get(i));
cell.setCellStyle(headerStyle);
}
int rowIndex = 1;
for (AppearancePatentParsedRowVo parsedRow : parsed.getAllItems()) {
AppearancePatentResultRowDto resultRow = resultMap.get(rowKey(parsedRow.getDisplayId(), parsedRow.getAsin(), parsedRow.getCountry()));
String missingReason = "";
if (resultRow == null) {
missingReason = hasPromptFields(parsedRow) ? "未匹配到检测结果" : "未送检：缺少标题或图片";
}
Row row = sheet.createRow(rowIndex++);
int col = 0;
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, "价格", "price"));
row.createCell(col++).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getTitleRisk(), ""));
row.createCell(col++).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getAppearanceRisk(), ""));
row.createCell(col++).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getPatentRisk(), ""));
row.createCell(col).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getConclusion(), ""));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getTitle(), "") : firstNonBlank(resultRow.getTitle(), parsedRow.getTitle()));
row.createCell(col++).setCellValue(resultRow == null ? firstNonBlank(parsedRow.getUrl(), "") : firstNonBlank(resultRow.getUrl(), parsedRow.getUrl()));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : firstNonBlank(resultRow.getTitleRisk(), ""));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : firstNonBlank(resultRow.getAppearanceRisk(), ""));
row.createCell(col++).setCellValue(resultRow == null ? missingReason : firstNonBlank(resultRow.getPatentRisk(), ""));
row.createCell(col).setCellValue(resultRow == null ? "未送检" : firstNonBlank(resultRow.getConclusion(), ""));
}
workbook.write(fos);
workbook.dispose();
@@ -713,13 +946,18 @@ public class AppearancePatentTaskService {
int idCol = findRequiredHeader(headerMap, "id");
int asinCol = findRequiredHeader(headerMap, "asin");
int countryCol = findRequiredHeader(headerMap, "国家", "country");
int urlCol = findOptionalHeader(headerMap, "url", "rul", "链接", "商品链接");
int titleCol = findOptionalHeader(headerMap, "标题", "title", "商品标题");
int urlCol = findOptionalHeaderExact(headerMap,
"url", "rul", "link", "image", "img", "pic", "picture",
"链接", "商品链接", "图片", "商品图片", "主图", "商品主图", "图片链接", "主图链接");
int titleCol = findOptionalHeaderExact(headerMap,
"标题", "title", "listing title", "product title", "商品标题", "商品名称", "产品名称");
List<AppearancePatentParsedRowVo> accepted = new ArrayList<>();
List<AppearancePatentParsedRowVo> allRows = new ArrayList<>();
int total = 0;
int dropped = 0;
String currentBlockBaseId = "";
String currentGroupKey = "";
for (int i = 1; i <= sheet.getLastRowNum(); i++) {
Row row = sheet.getRow(i);
if (row == null) {
@@ -740,6 +978,12 @@ public class AppearancePatentTaskService {
vo.setRowIndex(i + 1);
vo.setSourceId(id);
vo.setDisplayId(normalizeDisplayId(id));
String rowBaseId = baseId(vo.getDisplayId());
if (!Objects.equals(currentBlockBaseId, rowBaseId)) {
currentBlockBaseId = rowBaseId;
currentGroupKey = rowBaseId + "@" + vo.getRowIndex();
}
vo.setGroupKey(currentGroupKey);
vo.setAsin(asin);
vo.setCountry(country);
vo.setUrl(urlCol >= 0 ? cell(row, urlCol, formatter) : "");
@@ -752,6 +996,10 @@ public class AppearancePatentTaskService {
dropped++;
}
}
hydratePromptFields(allRows);
if (accepted.isEmpty()) {
throw new BusinessException("no valid appearance patent rows");
}
return new ParsedWorkbook(total, dropped, headers, accepted, allRows);
} catch (BusinessException ex) {
throw ex;
@@ -761,6 +1009,33 @@ public class AppearancePatentTaskService {
}
}
private void hydratePromptFields(List<AppearancePatentParsedRowVo> rows) {
if (rows == null || rows.isEmpty()) {
return;
}
Map<String, List<AppearancePatentParsedRowVo>> rowsByBaseId = groupRowsByBaseId(rows);
for (List<AppearancePatentParsedRowVo> siblings : rowsByBaseId.values()) {
String title = "";
String url = "";
for (AppearancePatentParsedRowVo sibling : siblings) {
title = firstNonBlank(title, sibling.getTitle());
url = firstNonBlank(url, sibling.getUrl());
}
for (AppearancePatentParsedRowVo sibling : siblings) {
if (normalize(sibling.getTitle()).isBlank()) {
sibling.setTitle(title);
}
if (normalize(sibling.getUrl()).isBlank()) {
sibling.setUrl(url);
}
}
}
}
private boolean hasPromptFields(AppearancePatentParsedRowVo row) {
return row != null && (!normalize(row.getTitle()).isBlank() || !normalize(row.getUrl()).isBlank());
}
private Map<String, Integer> buildHeaderMap(Row header, DataFormatter formatter) {
Map<String, Integer> map = new LinkedHashMap<>();
for (int i = 0; i < header.getLastCellNum(); i++) {
@@ -808,6 +1083,23 @@ public class AppearancePatentTaskService {
return -1;
}
private int findOptionalHeaderExact(Map<String, Integer> map, String... names) {
for (Map.Entry<String, Integer> entry : map.entrySet()) {
String normalizedHeader = normalizeHeaderAlias(entry.getKey());
for (String name : names) {
if (normalizedHeader.equals(normalizeHeaderAlias(name))) {
return entry.getValue();
}
}
}
return -1;
}
private String normalizeHeaderAlias(String value) {
String normalized = normalize(value).toLowerCase(Locale.ROOT);
return normalized.replaceAll("[\\s_\\-()（）\\[\\]{}:：/\\\\]+", "");
}
private String cell(Row row, int col, DataFormatter formatter) {
return col < 0 ? "" : normalize(formatter.formatCellValue(row.getCell(col)));
}
@@ -933,6 +1225,28 @@ public class AppearancePatentTaskService {
return new AppearancePatentParsedPayloadDto();
}
private Map<String, AppearancePatentResultRowDto> readChunkRows(TaskChunkEntity chunk) {
Map<String, AppearancePatentResultRowDto> rows = new LinkedHashMap<>();
if (chunk == null || chunk.getPayloadJson() == null || chunk.getPayloadJson().isBlank()) {
return rows;
}
try {
String payloadJson = transientPayloadStorageService.resolvePayload(chunk.getPayloadJson(), "read appearance patent chunk failed");
JsonNode array = objectMapper.readTree(payloadJson);
if (!array.isArray()) {
return rows;
}
for (JsonNode node : array) {
AppearancePatentResultRowDto row = objectMapper.treeToValue(node, AppearancePatentResultRowDto.class);
rows.put(rowKey(row.getId(), row.getAsin(), row.getCountry()), row);
}
} catch (Exception ex) {
log.warn("[appearance-patent] read chunk payload failed taskId={} chunk={} err={}",
chunk.getTaskId(), chunk.getChunkIndex(), ex.getMessage());
}
return rows;
}
private String rowKey(String id, String asin, String country) {
return normalize(id) + "::" + normalize(asin).toUpperCase(Locale.ROOT) + "::" + normalize(country);
}
@@ -996,6 +1310,13 @@ public class AppearancePatentTaskService {
}
}
private record SubmitContext(FileTaskEntity task,
String scopeKey,
String scopeHash,
boolean forceFlush,
String error) {
}
private record ParsedWorkbook(int totalRows, int droppedRows, List<String> headers, List<AppearancePatentParsedRowVo> acceptedRows, List<AppearancePatentParsedRowVo> allRows) {
}
}

51
backend-java/src/main/java/com/nanri/aiimage/modules/deletebrand/service/DeleteBrandStaleTaskService.java
View File

@@ -27,7 +27,6 @@ import java.nio.file.attribute.FileTime;
import java.time.Duration;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;
import java.util.Map;
@@ -47,6 +46,7 @@ public class DeleteBrandStaleTaskService {
private final FileTaskMapper fileTaskMapper;
private final DeleteBrandTaskCacheService deleteBrandTaskCacheService;
private final DeleteBrandTaskStorageService deleteBrandTaskStorageService;
private final DeleteBrandRunService deleteBrandRunService;
private final ProductRiskTaskService productRiskTaskService;
private final ProductRiskTaskCacheService productRiskTaskCacheService;
@@ -110,7 +110,13 @@ public class DeleteBrandStaleTaskService {
}
private void failStaleDeleteBrandTasks() {
LocalDateTime threshold = LocalDateTime.now().minusMinutes(deleteBrandProgressProperties.getHeartbeatTimeoutMinutes());
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)
@@ -120,36 +126,35 @@ public class DeleteBrandStaleTaskService {
for (FileTaskEntity task : runningTasks) {
Map<Object, Object> progress = deleteBrandTaskCacheService.getProgress(task.getId());
boolean hasProgress = progress != null && !progress.isEmpty() && progress.containsKey("last_heartbeat_at");
boolean isActive = false;
if (progress != null && progress.containsKey("has_progress")) {
Object hpObj = progress.get("has_progress");
if (hpObj instanceof Boolean) {
isActive = (Boolean) hpObj;
} else if (hpObj instanceof String) {
isActive = Boolean.parseBoolean((String) hpObj);
}
}
long lastHeartbeatAt = 0L;
if (hasProgress) {
if (progress != null && !progress.isEmpty() && progress.containsKey("last_heartbeat_at")) {
try {
lastHeartbeatAt = Long.parseLong(String.valueOf(progress.get("last_heartbeat_at")));
} catch (Exception ignored) {
}
}
boolean hasStartedProgress = lastHeartbeatAt > 0L;
if (!hasStartedProgress) {
hasStartedProgress = deleteBrandTaskStorageService.countCompletedScopes(task.getId()) > 0
|| deleteBrandTaskStorageService.countParsedPayloadScopes(task.getId()) > 0;
}
if (lastHeartbeatAt > 0 && isActive) {
LocalDateTime lastActivityTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(lastHeartbeatAt), ZoneId.systemDefault());
if (lastActivityTime.isAfter(threshold)) {
continue;
}
} else {
LocalDateTime queuedThreshold = LocalDateTime.now().minusHours(12);
if (task.getCreatedAt() != null && task.getCreatedAt().isAfter(queuedThreshold)) {
if (lastHeartbeatAt > 0L && nowMillis - lastHeartbeatAt < staleTimeoutMillis) {
continue;
}
if (!hasStartedProgress && task.getCreatedAt() != null && task.getCreatedAt().isAfter(initialThreshold)) {
continue;
}
try {
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
FileTaskEntity refreshed = fileTaskMapper.selectById(task.getId());
if (refreshed != null && !"RUNNING".equals(refreshed.getStatus())) {
deleteBrandTaskCacheService.saveTaskCache(refreshed);
continue;
}
} catch (Exception ex) {
log.warn("[stale-check] delete-brand finalize threw taskId={} msg={}", task.getId(), ex.getMessage());
}
int updated = fileTaskMapper.update(null, new LambdaUpdateWrapper<FileTaskEntity>()

49
backend-java/src/main/java/com/nanri/aiimage/modules/patroldelete/service/PatrolDeleteTaskService.java
View File

@@ -49,6 +49,9 @@ public class PatrolDeleteTaskService {
private static final String MODULE_TYPE = "PATROL_DELETE";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final int RESULT_PENDING = -1;
private static final int RESULT_FAILED = 0;
private static final int RESULT_SUCCESS = 1;
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
private final FileTaskMapper fileTaskMapper;
@@ -264,12 +267,13 @@ public class PatrolDeleteTaskService {
result.setSourceFilename(normalizedShopName);
result.setSourceFileUrl(item.getShopId());
result.setUserId(request.getUserId());
result.setSuccess(null);
result.setSuccess(RESULT_PENDING);
result.setCreatedAt(now);
fileResultMapper.insert(result);
snapshots.add(toSnapshotVo(result, item, task.getStatus(), null));
}
persistTaskJson(task, uniqueItems, snapshots);
taskCacheService.saveTaskCache(task);
PatrolDeleteCreateTaskVo vo = new PatrolDeleteCreateTaskVo();
vo.setTaskId(task.getId());
@@ -326,6 +330,7 @@ public class PatrolDeleteTaskService {
persistSnapshotJson(task, snapshots);
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
tryFinalizeTask(taskId, false);
}
@@ -402,6 +407,7 @@ public class PatrolDeleteTaskService {
persistSnapshotJson(task, snapshots);
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
return changed;
}
@@ -448,6 +454,7 @@ public class PatrolDeleteTaskService {
updateTaskStatusFromRows(task, rows);
persistSnapshotJson(task, buildSnapshotFromDb(task, rows));
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
}
private long countTasks(Long userId, List<String> statuses) {
@@ -502,9 +509,9 @@ public class PatrolDeleteTaskService {
private Map<Long, Map<Long, PatrolDeleteResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
Map<Long, Map<Long, PatrolDeleteResultItemVo>> out = new LinkedHashMap<>();
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
List<PatrolDeleteResultItemVo> snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, PatrolDeleteResultItemVo.class);
List<PatrolDeleteResultItemVo> snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
if (snapshots.isEmpty()) {
snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, PatrolDeleteResultItemVo.class);
}
out.put(entry.getKey(), indexSnapshotByResultId(snapshots));
}
@@ -518,7 +525,7 @@ public class PatrolDeleteTaskService {
item.setShopName(firstNonBlank(item.getShopName(), entity.getSourceFilename()));
item.setShopId(firstNonBlank(item.getShopId(), entity.getSourceFileUrl()));
item.setTaskStatus(task != null ? task.getStatus() : item.getTaskStatus());
item.setSuccess(entity.getSuccess() == null ? item.getSuccess() : entity.getSuccess() == 1);
item.setSuccess(toSuccessFlag(entity.getSuccess(), item.getSuccess()));
item.setError(!blank(entity.getErrorMessage()) ? entity.getErrorMessage() : item.getError());
item.setCreatedAt(entity.getCreatedAt());
item.setFinishedAt(task != null ? task.getFinishedAt() : item.getFinishedAt());
@@ -575,7 +582,7 @@ public class PatrolDeleteTaskService {
vo.setMatchStatus(item.getMatchStatus());
vo.setMatchMessage(item.getMatchMessage());
vo.setTaskStatus(taskStatus);
vo.setSuccess(result.getSuccess() == null ? null : result.getSuccess() == 1);
vo.setSuccess(toSuccessFlag(result.getSuccess(), null));
vo.setError(result.getErrorMessage());
vo.setCreatedAt(result.getCreatedAt());
vo.setFinishedAt(finishedAt);
@@ -616,7 +623,7 @@ public class PatrolDeleteTaskService {
item.setShopName(row.getSourceFilename());
item.setShopId(row.getSourceFileUrl());
item.setTaskStatus(task == null ? null : task.getStatus());
item.setSuccess(row.getSuccess() == null ? null : row.getSuccess() == 1);
item.setSuccess(toSuccessFlag(row.getSuccess(), null));
item.setError(row.getErrorMessage());
item.setCreatedAt(row.getCreatedAt());
item.setFinishedAt(task == null ? null : task.getFinishedAt());
@@ -642,8 +649,8 @@ public class PatrolDeleteTaskService {
}
private void updateTaskStatusFromRows(FileTaskEntity task, List<FileResultEntity> rows) {
long successCount = rows.stream().filter(row -> Integer.valueOf(1).equals(row.getSuccess())).count();
long failedCount = rows.stream().filter(row -> Integer.valueOf(0).equals(row.getSuccess())).count();
long successCount = rows.stream().filter(row -> Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())).count();
long failedCount = rows.stream().filter(row -> Integer.valueOf(RESULT_FAILED).equals(row.getSuccess())).count();
long pendingCount = rows.stream().filter(row -> !isResultFinished(row)).count();
task.setSuccessFileCount((int) successCount);
task.setFailedFileCount((int) failedCount);
@@ -796,11 +803,11 @@ public class PatrolDeleteTaskService {
.filter(item -> Boolean.TRUE.equals(item.getSuccess()))
.toList();
if (!successItems.isEmpty()) {
String filename = safeFileStem("patrol-delete-" + task.getId()) + ".xlsx";
String filename = buildTaskWorkbookFilename(task);
int rowCount = excelAssemblyService.countRows(successItems);
FileResultEntity firstSuccessRow = null;
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
if (Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(null);
row.setResultFileSize(0L);
@@ -843,7 +850,7 @@ public class PatrolDeleteTaskService {
throw new BusinessException("没有可生成的巡店删除结果");
}
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "patrol-delete-result", String.valueOf(task.getId())));
String filename = safeFileStem("patrol-delete-" + task.getId()) + ".xlsx";
String filename = buildTaskWorkbookFilename(task);
File xlsx = FileUtil.file(workRoot, filename);
try {
excelAssemblyService.writeWorkbook(xlsx, successItems);
@@ -851,7 +858,7 @@ public class PatrolDeleteTaskService {
long fileSize = xlsx.length();
int rowCount = excelAssemblyService.countRows(successItems);
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
if (Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(objectKey);
row.setResultFileSize(fileSize);
@@ -866,13 +873,13 @@ public class PatrolDeleteTaskService {
}
private void markResultSuccess(FileResultEntity row) {
row.setSuccess(1);
row.setSuccess(RESULT_SUCCESS);
row.setErrorMessage(null);
fileResultMapper.updateById(row);
}
private void markResultFailed(FileResultEntity row, String message) {
row.setSuccess(0);
row.setSuccess(RESULT_FAILED);
row.setErrorMessage(blankToNull(message));
row.setResultFilename(null);
row.setResultFileUrl(null);
@@ -883,7 +890,14 @@ public class PatrolDeleteTaskService {
}
private boolean isResultFinished(FileResultEntity row) {
return Integer.valueOf(1).equals(row.getSuccess()) || Integer.valueOf(0).equals(row.getSuccess());
return Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess()) || Integer.valueOf(RESULT_FAILED).equals(row.getSuccess());
}
private Boolean toSuccessFlag(Integer dbValue, Boolean fallback) {
if (dbValue == null || Integer.valueOf(RESULT_PENDING).equals(dbValue)) {
return fallback;
}
return Integer.valueOf(RESULT_SUCCESS).equals(dbValue);
}
private List<PatrolDeleteResultItemVo> parseTaskSnapshots(String json) {
@@ -972,6 +986,11 @@ public class PatrolDeleteTaskService {
return !blank(first) ? first : second;
}
private String buildTaskWorkbookFilename(FileTaskEntity task) {
Long taskId = task == null ? null : task.getId();
return safeFileStem("巡店删除-" + (taskId == null ? "result" : taskId)) + ".xlsx";
}
private String safeFileStem(String value) {
String raw = value == null ? "result" : value.trim();
String safe = raw.replaceAll("[\\\\/:*?\"<>|]+", "_");

49
backend-java/src/main/java/com/nanri/aiimage/modules/queryasin/service/QueryAsinTaskService.java
View File

@@ -49,6 +49,9 @@ public class QueryAsinTaskService {
private static final String MODULE_TYPE = "QUERY_ASIN";
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
private static final int RESULT_PENDING = -1;
private static final int RESULT_FAILED = 0;
private static final int RESULT_SUCCESS = 1;
private static final String INTERRUPTED_MESSAGE = "Python 在该店铺结果提交完成前中断";
private final FileTaskMapper fileTaskMapper;
@@ -270,12 +273,13 @@ public class QueryAsinTaskService {
result.setSourceFilename(normalizedShopName);
result.setSourceFileUrl(item.getShopId());
result.setUserId(request.getUserId());
result.setSuccess(null);
result.setSuccess(RESULT_PENDING);
result.setCreatedAt(now);
fileResultMapper.insert(result);
snapshots.add(toSnapshotVo(result, item, task.getStatus(), null));
}
persistTaskJson(task, uniqueItems, snapshots);
taskCacheService.saveTaskCache(task);
QueryAsinCreateTaskVo vo = new QueryAsinCreateTaskVo();
vo.setTaskId(task.getId());
@@ -332,6 +336,7 @@ public class QueryAsinTaskService {
persistSnapshotJson(task, snapshots);
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
tryFinalizeTask(taskId, false);
}
@@ -408,6 +413,7 @@ public class QueryAsinTaskService {
persistSnapshotJson(task, snapshots);
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
return changed;
}
@@ -454,6 +460,7 @@ public class QueryAsinTaskService {
updateTaskStatusFromRows(task, rows);
persistSnapshotJson(task, buildSnapshotFromDb(task, rows));
fileTaskMapper.updateById(task);
taskCacheService.saveTaskCache(task);
}
private long countTasks(Long userId, List<String> statuses) {
@@ -508,9 +515,9 @@ public class QueryAsinTaskService {
private Map<Long, Map<Long, QueryAsinResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
Map<Long, Map<Long, QueryAsinResultItemVo>> out = new LinkedHashMap<>();
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
List<QueryAsinResultItemVo> snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, QueryAsinResultItemVo.class);
List<QueryAsinResultItemVo> snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
if (snapshots.isEmpty()) {
snapshots = parseTaskSnapshots(entry.getValue().getResultJson());
snapshots = taskResultItemService.listResultSnapshots(entry.getKey(), MODULE_TYPE, QueryAsinResultItemVo.class);
}
out.put(entry.getKey(), indexSnapshotByResultId(snapshots));
}
@@ -524,7 +531,7 @@ public class QueryAsinTaskService {
item.setShopName(firstNonBlank(item.getShopName(), entity.getSourceFilename()));
item.setShopId(firstNonBlank(item.getShopId(), entity.getSourceFileUrl()));
item.setTaskStatus(task != null ? task.getStatus() : item.getTaskStatus());
item.setSuccess(entity.getSuccess() == null ? item.getSuccess() : entity.getSuccess() == 1);
item.setSuccess(toSuccessFlag(entity.getSuccess(), item.getSuccess()));
item.setError(!blank(entity.getErrorMessage()) ? entity.getErrorMessage() : item.getError());
item.setCreatedAt(entity.getCreatedAt());
item.setFinishedAt(task != null ? task.getFinishedAt() : item.getFinishedAt());
@@ -581,7 +588,7 @@ public class QueryAsinTaskService {
vo.setMatchStatus(item.getMatchStatus());
vo.setMatchMessage(item.getMatchMessage());
vo.setTaskStatus(taskStatus);
vo.setSuccess(result.getSuccess() == null ? null : result.getSuccess() == 1);
vo.setSuccess(toSuccessFlag(result.getSuccess(), null));
vo.setError(result.getErrorMessage());
vo.setCreatedAt(result.getCreatedAt());
vo.setFinishedAt(finishedAt);
@@ -622,7 +629,7 @@ public class QueryAsinTaskService {
item.setShopName(row.getSourceFilename());
item.setShopId(row.getSourceFileUrl());
item.setTaskStatus(task == null ? null : task.getStatus());
item.setSuccess(row.getSuccess() == null ? null : row.getSuccess() == 1);
item.setSuccess(toSuccessFlag(row.getSuccess(), null));
item.setError(row.getErrorMessage());
item.setCreatedAt(row.getCreatedAt());
item.setFinishedAt(task == null ? null : task.getFinishedAt());
@@ -648,8 +655,8 @@ public class QueryAsinTaskService {
}
private void updateTaskStatusFromRows(FileTaskEntity task, List<FileResultEntity> rows) {
long successCount = rows.stream().filter(row -> Integer.valueOf(1).equals(row.getSuccess())).count();
long failedCount = rows.stream().filter(row -> Integer.valueOf(0).equals(row.getSuccess())).count();
long successCount = rows.stream().filter(row -> Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())).count();
long failedCount = rows.stream().filter(row -> Integer.valueOf(RESULT_FAILED).equals(row.getSuccess())).count();
long pendingCount = rows.stream().filter(row -> !isResultFinished(row)).count();
task.setSuccessFileCount((int) successCount);
task.setFailedFileCount((int) failedCount);
@@ -816,11 +823,11 @@ public class QueryAsinTaskService {
.filter(item -> Boolean.TRUE.equals(item.getSuccess()))
.toList();
if (!successItems.isEmpty()) {
String filename = safeFileStem("query-asin-" + task.getId()) + ".xlsx";
String filename = buildTaskWorkbookFilename(task);
int rowCount = excelAssemblyService.countRows(successItems);
FileResultEntity firstSuccessRow = null;
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
if (Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(null);
row.setResultFileSize(0L);
@@ -863,7 +870,7 @@ public class QueryAsinTaskService {
throw new BusinessException("没有可生成的查询ASIN结果");
}
File workRoot = FileUtil.mkdir(FileUtil.file(System.getProperty("java.io.tmpdir"), "query-asin-result", String.valueOf(task.getId())));
String filename = safeFileStem("query-asin-" + task.getId()) + ".xlsx";
String filename = buildTaskWorkbookFilename(task);
File xlsx = FileUtil.file(workRoot, filename);
try {
excelAssemblyService.writeWorkbook(xlsx, successItems);
@@ -871,7 +878,7 @@ public class QueryAsinTaskService {
long fileSize = xlsx.length();
int rowCount = excelAssemblyService.countRows(successItems);
for (FileResultEntity row : rows) {
if (Integer.valueOf(1).equals(row.getSuccess())) {
if (Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess())) {
row.setResultFilename(filename);
row.setResultFileUrl(objectKey);
row.setResultFileSize(fileSize);
@@ -886,13 +893,13 @@ public class QueryAsinTaskService {
}
private void markResultSuccess(FileResultEntity row) {
row.setSuccess(1);
row.setSuccess(RESULT_SUCCESS);
row.setErrorMessage(null);
fileResultMapper.updateById(row);
}
private void markResultFailed(FileResultEntity row, String message) {
row.setSuccess(0);
row.setSuccess(RESULT_FAILED);
row.setErrorMessage(blankToNull(message));
row.setResultFilename(null);
row.setResultFileUrl(null);
@@ -903,7 +910,14 @@ public class QueryAsinTaskService {
}
private boolean isResultFinished(FileResultEntity row) {
return Integer.valueOf(1).equals(row.getSuccess()) || Integer.valueOf(0).equals(row.getSuccess());
return Integer.valueOf(RESULT_SUCCESS).equals(row.getSuccess()) || Integer.valueOf(RESULT_FAILED).equals(row.getSuccess());
}
private Boolean toSuccessFlag(Integer dbValue, Boolean fallback) {
if (dbValue == null || Integer.valueOf(RESULT_PENDING).equals(dbValue)) {
return fallback;
}
return Integer.valueOf(RESULT_SUCCESS).equals(dbValue);
}
private List<QueryAsinResultItemVo> parseTaskSnapshots(String json) {
@@ -1034,6 +1048,11 @@ public class QueryAsinTaskService {
return asin == null ? "" : asin.trim().toUpperCase();
}
private String buildTaskWorkbookFilename(FileTaskEntity task) {
Long taskId = task == null ? null : task.getId();
return safeFileStem("查询ASIN-" + (taskId == null ? "result" : taskId)) + ".xlsx";
}
private String safeFileStem(String value) {
String raw = value == null ? "result" : value.trim();
String safe = raw.replaceAll("[\\\\/:*?\"<>|]+", "_");

190
backend-java/src/main/java/com/nanri/aiimage/modules/shopkey/service/QueryAsinService.java
View File

@@ -15,6 +15,7 @@ import com.nanri.aiimage.modules.shopkey.model.vo.QueryAsinItemVo;
import com.nanri.aiimage.modules.shopkey.model.vo.QueryAsinPageVo;
import com.nanri.aiimage.modules.shopkey.model.vo.ShopManageGroupItemVo;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.web.multipart.MultipartFile;
@@ -30,9 +31,11 @@ import java.util.Locale;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Predicate;
@Service
@RequiredArgsConstructor
@Slf4j
public class QueryAsinService {
private static final List<String> SUPPORTED_COUNTRIES = List.of("DE", "UK", "FR", "IT", "ES");
@@ -364,13 +367,32 @@ public class QueryAsinService {
}
private String normalizeCountry(String country) {
String normalized = normalizeBlank(country).toUpperCase(Locale.ROOT);
String normalized = normalizeCountryAlias(country);
if (!SUPPORTED_COUNTRIES.contains(normalized)) {
throw new BusinessException("国家参数不支持: " + country);
}
return normalized;
}
private String normalizeCountryAlias(String country) {
String normalized = normalizeBlank(country);
if (normalized.isEmpty()) {
return "";
}
String upper = normalized.toUpperCase(Locale.ROOT);
if (SUPPORTED_COUNTRIES.contains(upper)) {
return upper;
}
return switch (normalizeHeaderKey(normalized)) {
case "\u5fb7\u56fd", "\u5fb7", "de", "germany", "german", "deutschland" -> "DE";
case "\u82f1\u56fd", "\u82f1", "uk", "gb", "greatbritain", "britain", "unitedkingdom" -> "UK";
case "\u6cd5\u56fd", "\u6cd5", "fr", "france", "french" -> "FR";
case "\u610f\u5927\u5229", "\u610f", "it", "italy", "italian" -> "IT";
case "\u897f\u73ed\u7259", "\u897f", "es", "spain", "spanish", "espana" -> "ES";
default -> "";
};
}
private String normalizeAsin(String asin) {
String normalized = normalizeRequired(asin, "ASIN 不能为空").toUpperCase(Locale.ROOT);
if (normalized.length() > 64) {
@@ -505,13 +527,19 @@ public class QueryAsinService {
}
}
headerLoaded = true;
if (findGroupIndex() == null && (fallbackGroupId == null || fallbackGroupId <= 0)) {
Integer resolvedGroupIndex = resolveGroupHeaderIndex();
Integer resolvedShopIndex = resolveShopHeaderIndex();
boolean hasWideCountries = hasWideCountryColumns();
boolean hasSingleCountryAsin = hasSingleCountryAsinColumns();
log.info("[query-asin-import] headerMap={} normalizedKeys={} groupIndex={} shopIndex={} hasWideCountries={} hasSingleCountryAsin={}",
headerMap, headerIndexByKey.keySet(), resolvedGroupIndex, resolvedShopIndex, hasWideCountries, hasSingleCountryAsin);
if (resolvedGroupIndex == null && (fallbackGroupId == null || fallbackGroupId <= 0)) {
throw new BusinessException("Excel 缺少分组列，请在页面选择分组或提供分组列");
}
if (findShopIndex() == null) {
if (resolvedShopIndex == null) {
throw new BusinessException("Excel 缺少店铺名列");
}
if (SUPPORTED_COUNTRIES.stream().noneMatch(country -> findCountryIndex(country) != null)) {
if (!hasWideCountries && !hasSingleCountryAsin) {
throw new BusinessException("Excel 至少需要包含一个国家 ASIN 列");
}
}
@@ -524,7 +552,7 @@ public class QueryAsinService {
progress.setProcessedRows(rowIndex);
Long groupId = resolveGroupId(rowMap);
String shopName = cell(rowMap, findShopIndex());
String shopName = cell(rowMap, resolveShopHeaderIndex());
if (groupId == null || shopName.isEmpty()) {
skippedCount++;
updateProgress();
@@ -532,27 +560,8 @@ public class QueryAsinService {
}
QueryAsinEntity row = newImportEntity(groupId, shopName);
int rowAsinCount = 0;
int rowAcceptedCount = 0;
for (String country : SUPPORTED_COUNTRIES) {
Integer index = findCountryIndex(country);
if (index == null) {
continue;
}
String asin = normalizeBlank(cell(rowMap, index)).toUpperCase(Locale.ROOT);
if (asin.isEmpty()) {
continue;
}
if (asin.length() > 64) {
skippedCount++;
continue;
}
asinCount++;
rowAsinCount++;
setCountryAsin(row, country, asin);
rowAcceptedCount++;
}
if (rowAsinCount == 0) {
int rowAcceptedCount = collectRowCountryAsins(row, rowMap);
if (countCountryCells(row) == 0) {
skippedCount++;
}
if (rowAcceptedCount > 0) {
@@ -754,7 +763,7 @@ public class QueryAsinService {
}
private Long resolveGroupId(Map<Integer, String> rowMap) {
Integer groupIndex = findGroupIndex();
Integer groupIndex = resolveGroupHeaderIndex();
String groupText = cell(rowMap, groupIndex);
if (groupText.isEmpty()) {
return fallbackGroupId != null && fallbackGroupId > 0 ? fallbackGroupId : null;
@@ -767,6 +776,51 @@ public class QueryAsinService {
return id;
}
private int collectRowCountryAsins(QueryAsinEntity row, Map<Integer, String> rowMap) {
int acceptedCount = 0;
for (String country : SUPPORTED_COUNTRIES) {
Integer index = resolveCountryHeaderIndex(country);
if (index == null) {
continue;
}
acceptedCount += acceptCountryAsin(row, country, cell(rowMap, index));
}
Integer singleCountryIndex = findSingleCountryIndex();
Integer singleAsinIndex = findSingleAsinIndex();
if (singleCountryIndex != null && singleAsinIndex != null) {
acceptedCount += acceptCountryAsin(
row,
normalizeCountryAlias(cell(rowMap, singleCountryIndex)),
cell(rowMap, singleAsinIndex));
}
return acceptedCount;
}
private int acceptCountryAsin(QueryAsinEntity row, String country, String asin) {
String normalizedAsin = normalizeBlank(asin).toUpperCase(Locale.ROOT);
if (normalizedAsin.isEmpty()) {
return 0;
}
asinCount++;
if (country == null || country.isEmpty() || !SUPPORTED_COUNTRIES.contains(country)) {
skippedCount++;
return 0;
}
if (normalizedAsin.length() > 64) {
skippedCount++;
return 0;
}
String existing = getCountryAsin(row, country);
if (!existing.isEmpty()) {
if (!existing.equals(normalizedAsin)) {
skippedCount++;
}
return 0;
}
setCountryAsin(row, country, normalizedAsin);
return 1;
}
private void loadGroups() {
if (groupsLoaded) {
return;
@@ -801,6 +855,88 @@ public class QueryAsinService {
};
}
private Integer resolveGroupHeaderIndex() {
Integer index = findGroupIndex();
if (index != null) {
return index;
}
index = firstHeader("groupname", "\u5206\u7ec4id", "\u5206\u7ec4");
if (index != null) {
return index;
}
return findHeaderByPredicate(key -> key.contains("\u5206\u7ec4") || key.contains("group"));
}
private Integer resolveShopHeaderIndex() {
Integer index = findShopIndex();
if (index != null) {
return index;
}
index = firstHeader(
"\u5e97\u94fa\u540d", "\u5e97\u94fa\u540d\u79f0", "\u5e97\u94fa\u8d26\u53f7", "\u8d26\u53f7",
"account", "selleraccount", "storeaccount", "storename");
if (index != null) {
return index;
}
return findHeaderByPredicate(key ->
(key.contains("\u5e97\u94fa") && (key.contains("\u540d") || key.contains("\u8d26\u53f7")))
|| (key.contains("shop") && (key.contains("name") || key.contains("account")))
|| key.contains("selleraccount")
|| key.contains("storeaccount"));
}
private Integer resolveCountryHeaderIndex(String country) {
Integer index = findCountryIndex(country);
if (index != null) {
return index;
}
return switch (country) {
case "DE" -> firstHeader("\u5fb7\u56fd", "\u5fb7", "\u5fb7\u56fdasin");
case "UK" -> firstHeader("\u82f1\u56fd", "\u82f1", "\u82f1\u56fdasin");
case "FR" -> firstHeader("\u6cd5\u56fd", "\u6cd5", "\u6cd5\u56fdasin");
case "IT" -> firstHeader("\u610f\u5927\u5229", "\u610f", "\u610f\u5927\u5229asin");
case "ES" -> firstHeader("\u897f\u73ed\u7259", "\u897f", "\u897f\u73ed\u7259asin");
default -> null;
};
}
private Integer findSingleCountryIndex() {
Integer index = firstHeader(
"\u4e2d\u6587\u56fd\u5bb6\u540d", "\u56fd\u5bb6", "\u56fd\u5bb6\u540d", "\u56fd\u5bb6\u7ad9\u70b9",
"\u7ad9\u70b9", "country", "countryname", "marketplace", "site");
if (index != null) {
return index;
}
return findHeaderByPredicate(key ->
key.contains("\u56fd\u5bb6")
|| key.contains("\u7ad9\u70b9")
|| key.contains("country")
|| key.contains("marketplace")
|| key.equals("site"));
}
private Integer findSingleAsinIndex() {
return firstHeader("asin", "\u67e5\u8be2asin", "\u4ea7\u54c1asin", "\u94fe\u63a5asin");
}
private boolean hasWideCountryColumns() {
return SUPPORTED_COUNTRIES.stream().anyMatch(country -> resolveCountryHeaderIndex(country) != null);
}
private boolean hasSingleCountryAsinColumns() {
return findSingleCountryIndex() != null && findSingleAsinIndex() != null;
}
private Integer findHeaderByPredicate(Predicate<String> predicate) {
for (Map.Entry<String, Integer> entry : headerIndexByKey.entrySet()) {
String key = entry.getKey();
if (key != null && predicate.test(key)) {
return entry.getValue();
}
}
return null;
}
private Integer firstHeader(String... names) {
for (String name : names) {
Integer index = headerIndexByKey.get(normalizeHeaderKey(name));

31
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultItemService.java
View File

@@ -125,15 +125,18 @@ public class TaskResultItemService {
String status,
Object payload) {
String payloadJson = writeJson(payload);
String payloadHash = hash(payloadJson);
String normalizedScopeKey = firstNonBlank(scopeKey, "task:" + taskId).trim();
String normalizedItemKey = firstNonBlank(itemKey, "result:" + resultId).trim();
String scopeHash = hash(normalizedScopeKey);
String storedPayload = transientPayloadStorageService.storeResultItemPayload(
moduleType, taskId, scopeHash, normalizedItemKey, payloadJson);
String payloadHash = hash(payloadJson);
LocalDateTime now = LocalDateTime.now();
TaskResultItemEntity existing = find(taskId, moduleType, scopeHash, normalizedItemKey);
if (isUnchanged(existing, resultId, normalizedScopeKey, countryCode, asin, status, payloadHash)) {
return;
}
String storedPayload = transientPayloadStorageService.storeResultItemPayload(
moduleType, taskId, scopeHash, normalizedItemKey, payloadJson);
if (existing == null) {
TaskResultItemEntity entity = new TaskResultItemEntity();
entity.setTaskId(taskId);
@@ -159,6 +162,10 @@ public class TaskResultItemService {
if (existing == null) {
throw new BusinessException("保存任务结果明细失败");
}
if (isUnchanged(existing, resultId, normalizedScopeKey, countryCode, asin, status, payloadHash)) {
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(storedPayload, existing.getPayloadJson());
return;
}
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(existing.getPayloadJson(), storedPayload);
taskResultItemMapper.update(null, new LambdaUpdateWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getId, existing.getId())
@@ -172,6 +179,24 @@ public class TaskResultItemService {
.set(TaskResultItemEntity::getUpdatedAt, now));
}
private boolean isUnchanged(TaskResultItemEntity existing,
Long resultId,
String scopeKey,
String countryCode,
String asin,
String status,
String payloadHash) {
if (existing == null) {
return false;
}
return java.util.Objects.equals(existing.getResultId(), resultId)
&& java.util.Objects.equals(existing.getScopeKey(), scopeKey)
&& java.util.Objects.equals(existing.getCountryCode(), countryCode)
&& java.util.Objects.equals(existing.getAsin(), asin)
&& java.util.Objects.equals(existing.getStatus(), status)
&& java.util.Objects.equals(existing.getPayloadHash(), payloadHash);
}
private TaskResultItemEntity find(Long taskId, String moduleType, String scopeHash, String itemKey) {
return taskResultItemMapper.selectOne(new LambdaQueryWrapper<TaskResultItemEntity>()
.eq(TaskResultItemEntity::getTaskId, taskId)

1
backend-java/src/main/resources/application-local.example.yml
View File

@@ -31,6 +31,7 @@ AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH=/v1/workflow/run
AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID=7632683471312355338
AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN=
AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE=10
AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS=60000
AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES=20
AIIMAGE_ROCKETMQ_NAME_SERVER=121.196.149.225:9876

2
backend-java/src/main/resources/application.yml
View File

@@ -136,6 +136,8 @@ aiimage:
coze-workflow-id: ${AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID:7632683471312355338}
coze-token: ${AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN:}
coze-batch-size: ${AIIMAGE_APPEARANCE_PATENT_COZE_BATCH_SIZE:10}
coze-connect-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_CONNECT_TIMEOUT_MILLIS:10000}
coze-read-timeout-millis: ${AIIMAGE_APPEARANCE_PATENT_COZE_READ_TIMEOUT_MILLIS:60000}
stale-timeout-minutes: ${AIIMAGE_APPEARANCE_PATENT_STALE_TIMEOUT_MINUTES:20}
stale-finalize-cron: ${AIIMAGE_APPEARANCE_PATENT_STALE_FINALIZE_CRON:0 */2 * * * *}
security: