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

处理一些合并冲突

Browse Source
This commit is contained in:
super
2026-05-05 14:11:35 +08:00
parent 710b953120
commit 203937335d
19 changed files with 1075 additions and 198 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
app/amazon/detail_spider.py
View File

@@ -111,30 +111,14 @@ class ChromeAmzone(ChromeAmzoneBase):
<<<<<<< HEAD
class SpiderTask(TaskBase):
task_name = "亚马逊采集"
=======
class SpiderTask:
mark_name = "亚马逊采集"
def __init__(self, user_info: dict = None):
"""初始化审批任务处理器
Args:
user_info: 用户信息字典，包含 company, username, password
"""
self.user_info = user_info or {}
self.running = True
super().__init__(user_info)
self.result_api_module = "appearance-patent"
def log(self, message: str, level: str = "INFO"):
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
if level == "ERROR":
show_notification(message, "error")
print(f"[{timestamp}] [PriceTask] [{level}] {message}")
>>>>>>> 6d398b66bc24634b13f1de162e8f68836e64441d
@staticmethod
def group_by_id_prefix(data):
"""
@@ -186,17 +170,12 @@ class SpiderTask:
Args:
task_data: 任务数据
"""
task_id = None
try:
<<<<<<< HEAD
data = task_data.get("data", {})
task_id = data.get("taskId")
groups = self.normalize_groups(data)
=======
data = task_data.get("data", {})
self.result_api_module = "similar-asin" if task_data.get("type") == "similar-asin-run" else "appearance-patent"
task_id = data.get("taskId")
groups = self.normalize_groups(data)
>>>>>>> 6d398b66bc24634b13f1de162e8f68836e64441d
# 用于测试
limit = data.get("limit", None)
@@ -330,7 +309,7 @@ class SpiderTask:
runing_task[task_id]["error"] = str(e)
def post_result(self, task_id: int, chunkIndex:int,chunkTotal: int, asin: str, error:str="",
item_data:dict={},
item_data=None,
is_done: bool = False):
"""回传处理结果到API
"""
@@ -342,7 +321,7 @@ class SpiderTask:
"chunkIndex": chunkIndex,
"chunkTotal": chunkTotal,
"error": error,
"groups": item_data,
"groups": item_data or [],
"done": is_done
}
@@ -356,8 +335,8 @@ class SpiderTask:
self.log(f"回传数据: {payload}")
response = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=request_timeout,
verify=False
)

4
app/amazon/similar_asin.py
View File

@@ -468,8 +468,8 @@ class SimilarAsinTask(TaskBase):
self.log(f"回传数据: {payload}")
response = requests.post(
url,
json=payload,
headers={"Content-Type": "application/json"},
data=json.dumps(payload, ensure_ascii=False).encode("utf-8"),
headers={"Content-Type": "application/json; charset=utf-8"},
timeout=request_timeout,
verify=False
)

6
backend-java/src/main/java/com/nanri/aiimage/AiImageApplication.java
View File

@@ -4,11 +4,17 @@ import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.scheduling.annotation.EnableScheduling;
import java.time.ZoneId;
import java.util.TimeZone;
@SpringBootApplication
@EnableScheduling
public class AiImageApplication {
private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
public static void main(String[] args) {
TimeZone.setDefault(TimeZone.getTimeZone(BUSINESS_ZONE));
SpringApplication.run(AiImageApplication.class, args);
}
}

25
backend-java/src/main/java/com/nanri/aiimage/common/service/DistributedJobLockService.java
View File

@@ -20,6 +20,12 @@ public class DistributedJobLockService {
+ "else return 0 end",
Long.class
);
private static final DefaultRedisScript<Long> RENEW_SCRIPT = new DefaultRedisScript<>(
"if redis.call('get', KEYS[1]) == ARGV[1] then "
+ "return redis.call('pexpire', KEYS[1], ARGV[2]) "
+ "else return 0 end",
Long.class
);
private final StringRedisTemplate stringRedisTemplate;
private final String ownerPrefix;
@@ -74,6 +80,25 @@ public class DistributedJobLockService {
log.warn("[job-lock] release failed jobName={} key={} msg={}", jobName, key, ex.getMessage());
}
}
public boolean renew(Duration ttl) {
if (released) {
return false;
}
Duration actualTtl = (ttl == null || ttl.isNegative() || ttl.isZero()) ? Duration.ofMinutes(10) : ttl;
try {
Long renewed = stringRedisTemplate.execute(
RENEW_SCRIPT,
Collections.singletonList(key),
token,
String.valueOf(actualTtl.toMillis())
);
return Long.valueOf(1L).equals(renewed);
} catch (Exception ex) {
log.warn("[job-lock] renew failed jobName={} key={} msg={}", jobName, key, ex.getMessage());
return false;
}
}
}
private String buildLockKey(String jobName) {

6
backend-java/src/main/java/com/nanri/aiimage/config/SchedulingConfig.java
View File

@@ -6,15 +6,21 @@ import org.springframework.context.annotation.Configuration;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import java.time.Clock;
import java.time.ZoneId;
@Configuration
@Slf4j
public class SchedulingConfig {
private static final ZoneId BUSINESS_ZONE = ZoneId.of("Asia/Shanghai");
@Bean
public TaskScheduler taskScheduler() {
ThreadPoolTaskScheduler scheduler = new ThreadPoolTaskScheduler();
scheduler.setPoolSize(4);
scheduler.setThreadNamePrefix("aiimage-scheduling-");
scheduler.setClock(Clock.system(BUSINESS_ZONE));
scheduler.setWaitForTasksToCompleteOnShutdown(true);
scheduler.setAwaitTerminationSeconds(30);
scheduler.setErrorHandler(ex -> log.warn("[scheduling] task execution failed: {}", ex.getMessage(), ex));

78
backend-java/src/main/java/com/nanri/aiimage/config/TaskOperationLockConfig.java Normal file
View File

@@ -0,0 +1,78 @@
package com.nanri.aiimage.config;
import com.nanri.aiimage.common.exception.BusinessException;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import jakarta.servlet.http.HttpServletRequest;
import jakarta.servlet.http.HttpServletResponse;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.annotation.Configuration;
import org.springframework.web.servlet.HandlerInterceptor;
import org.springframework.web.servlet.config.annotation.InterceptorRegistry;
import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
import java.util.Locale;
import java.util.Set;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
@Configuration
@RequiredArgsConstructor
public class TaskOperationLockConfig implements WebMvcConfigurer {
private final TaskDistributedLockService taskDistributedLockService;
@Override
public void addInterceptors(InterceptorRegistry registry) {
registry.addInterceptor(new TaskOperationLockInterceptor(taskDistributedLockService))
.addPathPatterns("/api/**");
}
@Slf4j
private static final class TaskOperationLockInterceptor implements HandlerInterceptor {
private static final Pattern TASK_PATH = Pattern.compile(".*/api/([^/]+)/tasks/(\\d+)(?:/.*)?$");
private static final Set<String> MUTATING_METHODS = Set.of("POST", "PUT", "PATCH", "DELETE");
private static final String LOCK_ATTRIBUTE = TaskOperationLockInterceptor.class.getName() + ".LOCK";
private final TaskDistributedLockService taskDistributedLockService;
private TaskOperationLockInterceptor(TaskDistributedLockService taskDistributedLockService) {
this.taskDistributedLockService = taskDistributedLockService;
}
@Override
public boolean preHandle(HttpServletRequest request, HttpServletResponse response, Object handler) {
if (!MUTATING_METHODS.contains(request.getMethod())) {
return true;
}
String uri = request.getRequestURI();
Matcher matcher = TASK_PATH.matcher(uri == null ? "" : uri);
if (!matcher.matches()) {
return true;
}
String moduleType = matcher.group(1).trim().toUpperCase(Locale.ROOT).replace('-', '_');
Long taskId = Long.valueOf(matcher.group(2));
TaskDistributedLockService.LockHandle lockHandle =
taskDistributedLockService.acquire(moduleType, taskId, TaskDistributedLockService.DEFAULT_WAIT_MILLIS);
if (lockHandle == null) {
log.info("[task-operation-lock] rejected busy task operation method={} uri={} moduleType={} taskId={}",
request.getMethod(), uri, moduleType, taskId);
throw new BusinessException(40902, "TASK_LOCK_BUSY");
}
request.setAttribute(LOCK_ATTRIBUTE, lockHandle);
return true;
}
@Override
public void afterCompletion(HttpServletRequest request,
HttpServletResponse response,
Object handler,
Exception ex) {
Object lockHandle = request.getAttribute(LOCK_ATTRIBUTE);
if (lockHandle instanceof TaskDistributedLockService.LockHandle handle) {
handle.close();
}
}
}
}

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

@@ -6,6 +6,7 @@ import com.nanri.aiimage.config.AppearancePatentProperties;
import com.nanri.aiimage.modules.appearancepatent.model.dto.AppearancePatentResultRowDto;
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;
@@ -24,6 +25,8 @@ import java.util.Map;
@Slf4j
public class AppearancePatentCozeClient {
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
private final AppearancePatentProperties properties;
private final ObjectMapper objectMapper;
@@ -216,7 +219,8 @@ public class AppearancePatentCozeClient {
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
});
request.body(body);
return request.exchange((clientRequest, clientResponse) -> {
@@ -237,7 +241,8 @@ public class AppearancePatentCozeClient {
.uri(joinUrl(properties.getCozeBaseUrl(), path))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
})
.exchange((clientRequest, clientResponse) -> {
byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());

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

@@ -38,6 +38,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
@@ -71,6 +72,7 @@ import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.LinkedHashMap;
import java.util.LinkedHashSet;
import java.util.List;
@@ -98,6 +100,9 @@ public class AppearancePatentTaskService {
private static final String CONTENT_TYPE_XLSX = "application/vnd.openxmlformats-officedocument.spreadsheetml.sheet";
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);
private static final long TASK_LOCK_WAIT_MILLIS = 10000L;
private static final long TASK_LOCK_RETRY_DELAY_MILLIS = 200L;
private static final List<String> RESULT_HEADERS = List.of(
"id",
"asin",
@@ -125,6 +130,7 @@ public class AppearancePatentTaskService {
private final TransientPayloadStorageService transientPayloadStorageService;
private final PlatformTransactionManager transactionManager;
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
@Autowired
@Qualifier("cozeTaskExecutor")
private TaskExecutor cozeTaskExecutor;
@@ -296,24 +302,41 @@ public class AppearancePatentTaskService {
.eq(FileResultEntity::getUserId, userId)
.orderByDesc(FileResultEntity::getCreatedAt)
.last("limit " + safeLimit));
Map<Long, String> statusMap = new LinkedHashMap<>();
Map<Long, FileTaskEntity> taskMap = new LinkedHashMap<>();
List<Long> taskIds = rows.stream().map(FileResultEntity::getTaskId).filter(Objects::nonNull).distinct().toList();
if (!taskIds.isEmpty()) {
for (FileTaskEntity task : fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
.select(FileTaskEntity::getId, FileTaskEntity::getStatus)
.select(FileTaskEntity::getId,
FileTaskEntity::getStatus,
FileTaskEntity::getCreatedAt,
FileTaskEntity::getUpdatedAt,
FileTaskEntity::getFinishedAt)
.in(FileTaskEntity::getId, taskIds))) {
statusMap.put(task.getId(), task.getStatus());
taskMap.put(task.getId(), task);
}
}
Map<Long, TaskFileJobEntity> jobMap = taskFileJobService.findAssembleJobsByResultIds(MODULE_TYPE, rows.stream()
.map(FileResultEntity::getId)
.filter(Objects::nonNull)
.toList());
List<FileResultEntity> sortedRows = new ArrayList<>();
for (FileResultEntity row : rows) {
String taskStatus = statusMap.get(row.getTaskId());
FileTaskEntity task = taskMap.get(row.getTaskId());
String taskStatus = task == null ? null : task.getStatus();
if (STATUS_PENDING.equals(taskStatus)) {
continue;
}
sortedRows.add(row);
}
sortedRows.sort(Comparator
.comparingInt((FileResultEntity row) -> historyPriority(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId())))
.thenComparing((FileResultEntity row) -> historyActivityTime(row, taskMap.get(row.getTaskId()), jobMap.get(row.getId())),
Comparator.nullsLast(Comparator.reverseOrder()))
.thenComparing(FileResultEntity::getCreatedAt, Comparator.nullsLast(Comparator.reverseOrder()))
.thenComparing(FileResultEntity::getId, Comparator.nullsLast(Comparator.reverseOrder())));
for (FileResultEntity row : sortedRows) {
FileTaskEntity task = taskMap.get(row.getTaskId());
String taskStatus = task == null ? null : task.getStatus();
vo.getItems().add(toHistoryItem(row, taskStatus, jobMap.get(row.getId())));
}
return vo;
@@ -384,6 +407,16 @@ public class AppearancePatentTaskService {
}
public void submitResult(Long taskId, AppearancePatentSubmitResultRequest request) {
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, TASK_LOCK_WAIT_MILLIS);
if (lockHandle == null) {
throw new BusinessException(40902, "任务正在处理上一批结果，请稍后重试");
}
try (lockHandle) {
submitResultLocked(taskId, request);
}
}
private void submitResultLocked(Long taskId, AppearancePatentSubmitResultRequest request) {
if (transactionManager != null) {
SubmitContext context = inNewTransaction(() -> persistSubmittedChunk(taskId, request));
inNewTransaction(() -> {
@@ -423,7 +456,7 @@ public class AppearancePatentTaskService {
chunk.setScopeHash(scopeHash);
chunk.setChunkIndex(chunkIndex);
chunk.setChunkTotal(chunkTotal);
String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson);
chunk.setPayloadJson(storedPayload);
chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson));
chunk.setCreatedAt(LocalDateTime.now());
@@ -536,10 +569,16 @@ public class AppearancePatentTaskService {
if (heartbeatMillis > thresholdMillis) {
continue;
}
inNewTransaction(() -> {
finalizeStaleTask(task.getId(), "Python interrupted before uploading final appearance patent result");
return null;
});
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L);
if (lockHandle == null) {
continue;
}
try (lockHandle) {
inNewTransaction(() -> {
finalizeStaleTask(task.getId(), "Python interrupted before uploading final appearance patent result");
return null;
});
}
}
return;
}
@@ -559,7 +598,16 @@ public class AppearancePatentTaskService {
if (heartbeatMillis > thresholdMillis) {
continue;
}
finalizeTask(task, "Python interrupted before uploading final appearance patent result", allRowCount(task), true);
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(task.getId(), 0L);
if (lockHandle == null) {
continue;
}
try (lockHandle) {
FileTaskEntity latestTask = fileTaskMapper.selectById(task.getId());
if (latestTask != null) {
finalizeTask(latestTask, "Python interrupted before uploading final appearance patent result", allRowCount(latestTask), true);
}
}
}
}
@@ -596,7 +644,7 @@ public class AppearancePatentTaskService {
chunk.setScopeHash(scopeHash);
chunk.setChunkIndex(chunkIndex);
chunk.setChunkTotal(chunkTotal);
String storedPayload = transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
String storedPayload = storeSharedChunkPayload(taskId, scopeHash, chunkIndex, payloadJson);
chunk.setPayloadJson(storedPayload);
chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson));
chunk.setCreatedAt(LocalDateTime.now());
@@ -784,7 +832,7 @@ public class AppearancePatentTaskService {
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 = transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
String storedPayload = storeSharedChunkPayloadVersioned(taskId, scopeHash, chunkIndex, payloadJson);
chunk.setPayloadJson(storedPayload);
chunk.setPayloadHash(DigestUtil.sha256Hex(payloadJson));
chunk.setUpdatedAt(LocalDateTime.now());
@@ -876,7 +924,7 @@ public class AppearancePatentTaskService {
}
List<AppearancePatentResultRowDto> representatives = new ArrayList<>();
for (List<AppearancePatentResultRowDto> siblings : groupedRows.values()) {
boolean alreadyResolved = siblings.stream().anyMatch(this::hasResolvedCozeFields);
boolean alreadyResolved = siblings.stream().anyMatch(this::hasCompleteCozeResult);
if (alreadyResolved) {
continue;
}
@@ -1064,6 +1112,18 @@ public class AppearancePatentTaskService {
if (job == null || job.getTaskId() == null || job.getResultId() == null) {
throw new BusinessException("result file job arguments are incomplete");
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(job.getTaskId(), TASK_LOCK_WAIT_MILLIS);
if (lockHandle == null) {
taskFileJobService.requeue(job.getId(), "Task is busy, waiting for appearance patent result merge");
touchJavaSideTaskActivity(job.getTaskId());
return false;
}
try (lockHandle) {
return processResultFileJobLocked(job);
}
}
private boolean processResultFileJobLocked(TaskFileJobEntity job) {
FileTaskEntity task = fileTaskMapper.selectById(job.getTaskId());
if (task == null || !MODULE_TYPE.equals(task.getModuleType())) {
throw new BusinessException("task not found");
@@ -1076,9 +1136,15 @@ public class AppearancePatentTaskService {
.eq(TaskChunkEntity::getTaskId, task.getId())
.eq(TaskChunkEntity::getModuleType, MODULE_TYPE)
.orderByAsc(TaskChunkEntity::getChunkIndex));
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
int cozeWorkUnits = countCozeWorkUnits(chunks, Math.max(1, properties.getCozeBatchSize()));
int totalProgressUnits = Math.max(3, cozeWorkUnits + 3);
if (countPendingCozeStates(task.getId()) > 0) {
taskFileJobService.touchRunning(job.getId());
touchJavaSideTaskActivity(task.getId());
saveFileBuildProgress(task, job, totalProgressUnits, 1, "Coze submitted, waiting for result");
return false;
}
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
saveFileBuildProgress(task, job, totalProgressUnits, 0, "Submitting Coze");
boolean pendingCoze = submitCozeBatches(task, result, job, chunks, allRowsByBaseId);
if (pendingCoze) {
@@ -1113,7 +1179,18 @@ public class AppearancePatentTaskService {
if (state == null || state.getId() == null) {
continue;
}
cozeTaskExecutor.execute(() -> pollPendingCozeState(state.getId()));
Long stateId = state.getId();
try {
cozeTaskExecutor.execute(() -> {
log.info("[appearance-patent] coze poll worker entered stateId={} taskId={} executeId={}",
stateId, state.getTaskId(), state.getCozeExecuteId());
pollPendingCozeState(stateId);
});
} catch (Exception ex) {
log.warn("[appearance-patent] coze poll dispatch failed stateId={} taskId={} executeId={} err={}",
stateId, state.getTaskId(), state.getCozeExecuteId(),
firstNonBlank(ex.getMessage(), ex.getClass().getSimpleName()), ex);
}
}
}
}
@@ -1231,8 +1308,7 @@ public class AppearancePatentTaskService {
batchTotal
);
String batchPayload = writeJson(batchRows, "serialize coze batch payload failed");
String storedBatchPayload = transientPayloadStorageService.storeParsedPayloadFast(
MODULE_TYPE, task.getId(), batchScopeHash, batchPayload, true);
String storedBatchPayload = storeSharedCozeBatchPayload(task.getId(), batchScopeHash, batchPayload);
TaskScopeStateEntity state = new TaskScopeStateEntity();
state.setTaskId(task.getId());
state.setModuleType(MODULE_TYPE);
@@ -1260,25 +1336,53 @@ public class AppearancePatentTaskService {
}
private void pollPendingCozeState(Long stateId) {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
Long taskIdForLock = null;
TaskScopeStateEntity lockState = taskScopeStateMapper.selectById(stateId);
if (lockState != null) {
taskIdForLock = lockState.getTaskId();
}
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskIdForLock, 0L);
if (lockHandle == null) {
if (taskIdForLock != null) {
log.info("[appearance-patent] coze poll skipped because task is locked taskId={} stateId={}",
taskIdForLock, stateId);
}
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
return;
try (lockHandle) {
pollPendingCozeStateLocked(stateId);
}
if (!tryClaimCozeStateForPoll(state)) {
return;
}
CozeBatchContext context = readCozeBatchContext(state);
if (context == null || context.jobId() == null || context.resultId() == null) {
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing");
return;
}
taskFileJobService.touchRunning(context.jobId());
}
private void pollPendingCozeStateLocked(Long stateId) {
try {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
return;
}
if (!tryClaimCozeStateForPoll(state)) {
log.info("[appearance-patent] coze poll skipped by claim guard taskId={} stateId={} executeId={} lastPolledAt={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), state.getCozeLastPolledAt());
return;
}
CozeBatchContext context = readCozeBatchContext(state);
if (context == null || context.jobId() == null || context.resultId() == null) {
log.warn("[appearance-patent] coze poll aborted because batch context is missing taskId={} stateId={} executeId={} stateJson={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), abbreviate(state.getStateJson(), 300));
markCozeStateTerminal(state, COZE_STATUS_FAILED, "Coze batch context missing");
return;
}
taskFileJobService.touchRunning(context.jobId());
log.info("[appearance-patent] coze poll start taskId={} stateId={} executeId={} jobId={} chunk={} batch={}/{}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(),
context.jobId(), context.chunkIndex(), context.batchIndex(), context.batchTotal());
AppearancePatentCozeClient.CozePollResponse poll = cozeClient.pollWorkflow(state.getCozeExecuteId());
if (!poll.hasPayload() && !poll.isFinished() && !isCozeStateTimedOut(state)) {
log.info("[appearance-patent] coze poll pending taskId={} stateId={} executeId={} status={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), poll.status());
updateCozeStateRunning(state, null);
return;
}
@@ -1291,6 +1395,9 @@ public class AppearancePatentTaskService {
: "Coze async workflow completed without output";
}
List<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
if (batchRows.isEmpty() && failureMessage.isBlank()) {
failureMessage = "Coze batch payload missing";
}
List<AppearancePatentResultRowDto> cozeRows = failureMessage.isBlank()
? cozeClient.mergeRowsFromDataText(batchRows, poll.resolvedPayloadText())
: cozeClient.markRowsFailed(batchRows, failureMessage);
@@ -1302,27 +1409,38 @@ public class AppearancePatentTaskService {
markCozeStateTerminal(state,
failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED,
failureMessage.isBlank() ? null : failureMessage);
maybeFinalizeCozeJob(state.getTaskId(), context);
log.info("[appearance-patent] coze poll completed taskId={} stateId={} executeId={} status={} batchRows={} mergedRows={} failure={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(),
failureMessage.isBlank() ? COZE_STATUS_DONE : COZE_STATUS_FAILED,
batchRows.size(), cozeRows.size(), firstNonBlank(failureMessage, "-"));
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
} catch (Exception ex) {
TaskScopeStateEntity state = taskScopeStateMapper.selectById(stateId);
String message = firstNonBlank(ex.getMessage(), "Coze poll failed");
if (isCozeStateTimedOut(state)) {
List<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
if (task != null) {
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
mergeCozeRowsIntoChunk(task,
context.chunkScopeHash(),
context.chunkIndex(),
cozeClient.markRowsFailed(batchRows, message),
allRowsByBaseId);
if (state != null) {
CozeBatchContext context = readCozeBatchContext(state);
if (isCozeStateTimedOut(state) && context != null) {
List<AppearancePatentResultRowDto> batchRows = readCozeBatchRows(state);
FileTaskEntity task = fileTaskMapper.selectById(state.getTaskId());
if (task != null) {
Map<String, List<AppearancePatentParsedRowVo>> allRowsByBaseId = loadAllRowsByBaseId(task);
mergeCozeRowsIntoChunk(task,
context.chunkScopeHash(),
context.chunkIndex(),
cozeClient.markRowsFailed(batchRows, message),
allRowsByBaseId);
}
markCozeStateTerminal(state, COZE_STATUS_FAILED, message);
maybeFinalizeCozeJobLocked(state.getTaskId(), context);
return;
}
markCozeStateTerminal(state, COZE_STATUS_FAILED, message);
maybeFinalizeCozeJob(state.getTaskId(), context);
log.warn("[appearance-patent] coze poll failed taskId={} stateId={} executeId={} err={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), message);
updateCozeStateRunning(state, message);
return;
}
log.warn("[appearance-patent] coze poll failed taskId={} stateId={} executeId={} err={}",
state.getTaskId(), state.getId(), state.getCozeExecuteId(), message);
updateCozeStateRunning(state, message);
log.warn("[appearance-patent] coze poll crashed before state refresh stateId={} err={}",
stateId, message, ex);
}
}
@@ -1376,6 +1494,19 @@ public class AppearancePatentTaskService {
}
private void maybeFinalizeCozeJob(Long taskId, CozeBatchContext context) {
if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) {
return;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(taskId, 0L);
if (taskLockHandle == null) {
return;
}
try (taskLockHandle) {
maybeFinalizeCozeJobLocked(taskId, context);
}
}
private void maybeFinalizeCozeJobLocked(Long taskId, CozeBatchContext context) {
if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) {
return;
}
@@ -1388,19 +1519,15 @@ public class AppearancePatentTaskService {
if (countPendingCozeStates(taskId) > 0) {
return;
}
FileTaskEntity task = fileTaskMapper.selectById(taskId);
FileResultEntity result = fileResultMapper.selectById(context.resultId());
TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId());
if (task == null || result == null || job == null || "SUCCESS".equals(job.getStatus())) {
if (job == null || "SUCCESS".equals(job.getStatus())) {
return;
}
int cozeWorkUnits = countCompletedCozeStates(taskId);
int totalProgressUnits = Math.max(3, cozeWorkUnits + 3);
completeCozeFileJob(task, result, job, totalProgressUnits, cozeWorkUnits);
taskFileJobService.markSuccess(job, result.getResultFileUrl());
cleanupResultFileJob(job);
log.info("[appearance-patent] coze async job finalized taskId={} jobId={} resultId={} resultFileUrl={}",
taskId, job.getId(), result.getId(), result.getResultFileUrl());
boolean requeued = taskFileJobService.requeue(job.getId(), "Coze results ready, assembling xlsx");
if (requeued) {
log.info("[appearance-patent] coze async results ready, result file job requeued taskId={} jobId={} resultId={}",
taskId, job.getId(), context.resultId());
}
} catch (Exception ex) {
TaskFileJobEntity job = taskFileJobService.findAssembleJob(taskId, MODULE_TYPE, context.resultId());
if (job != null) {
@@ -1416,6 +1543,9 @@ public class AppearancePatentTaskService {
TaskFileJobEntity job,
int totalProgressUnits,
int cozeWorkUnits) {
if (countPendingCozeStates(task.getId()) > 0) {
throw new BusinessException("Coze 结果仍在处理中，暂不能生成结果文件");
}
int assembleProgress = Math.max(1, Math.min(totalProgressUnits - 2, cozeWorkUnits));
saveFileBuildProgress(task, job, totalProgressUnits, assembleProgress, "Assembling xlsx");
assembleResultWorkbook(task, result);
@@ -1534,6 +1664,20 @@ public class AppearancePatentTaskService {
+ ":batch:" + batchIndex;
}
private String abbreviate(String value, int maxLength) {
if (value == null) {
return "";
}
int normalizedMaxLength = Math.max(0, maxLength);
if (value.length() <= normalizedMaxLength) {
return value;
}
if (normalizedMaxLength <= 3) {
return value.substring(0, normalizedMaxLength);
}
return value.substring(0, normalizedMaxLength - 3) + "...";
}
private int countCozeWorkUnits(List<TaskChunkEntity> chunks, int batchSize) {
if (chunks == null || chunks.isEmpty()) {
return 0;
@@ -1579,6 +1723,11 @@ public class AppearancePatentTaskService {
if (job == null || job.getTaskId() == null) {
return;
}
if (countPendingCozeStates(job.getTaskId()) > 0) {
log.warn("[appearance-patent] skip cleanup because coze is still pending taskId={} jobId={}",
job.getTaskId(), job.getId());
return;
}
deleteTransientTaskPayloads(job.getTaskId());
taskChunkMapper.delete(new LambdaQueryWrapper<TaskChunkEntity>()
.eq(TaskChunkEntity::getTaskId, job.getTaskId())
@@ -1591,17 +1740,19 @@ public class AppearancePatentTaskService {
private void assembleResultWorkbook(FileTaskEntity task, FileResultEntity result) {
AppearancePatentParsedPayloadDto parsed = readParsedPayload(task);
Map<String, AppearancePatentResultRowDto> resultMap = loadPersistedResultRowsWithRetry(task.getId(), parsed.getAllItems().size());
long resolvedRows = parsed.getAllItems().stream()
.filter(row -> findResultRow(row, resultMap) != null || findResultRowByAsin(row.getAsin(), resultMap) != null)
List<AppearancePatentParsedRowVo> receivedRows = filterReceivedParsedRows(parsed.getAllItems(), resultMap);
long resolvedRows = receivedRows.stream()
.filter(row -> findResultRow(row, resultMap) != null)
.count();
long reasonRows = resultMap.values().stream()
.filter(this::hasReasonFields)
.count();
log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} resultRows={} resolvedRows={} reasonRows={}",
task.getId(), parsed.getAllItems().size(), resultMap.size(), resolvedRows, reasonRows);
if (!parsed.getAllItems().isEmpty() && resultMap.isEmpty()) {
log.info("[appearance-patent] assemble workbook taskId={} parsedRows={} receivedRows={} resultRows={} resolvedRows={} reasonRows={}",
task.getId(), parsed.getAllItems().size(), receivedRows.size(), resultMap.size(), resolvedRows, reasonRows);
if (!parsed.getAllItems().isEmpty() && receivedRows.isEmpty()) {
throw new BusinessException("外观专利检测结果为空，请稍后重试生成结果文件");
}
validateCompleteCozeCoverage(task.getId(), receivedRows, resultMap);
File outputDir = new File(storageProperties.getLocalTempDir(), "appearance-patent-result");
if (!outputDir.exists() && !outputDir.mkdirs()) {
throw new BusinessException("创建结果目录失败");
@@ -1614,13 +1765,13 @@ public class AppearancePatentTaskService {
+ "-result.xlsx";
File xlsx = new File(outputDir, tempFilename);
try {
writeResultWorkbook(xlsx, parsed, resultMap);
writeResultWorkbook(xlsx, parsed, receivedRows, resultMap);
String objectKey = ossStorageService.uploadResultFile(xlsx, MODULE_TYPE);
result.setResultFilename(filename);
result.setResultFileUrl(objectKey);
result.setResultFileSize(xlsx.length());
result.setResultContentType(CONTENT_TYPE_XLSX);
result.setRowCount(parsed.getAllItems().size());
result.setRowCount(receivedRows.size());
} finally {
if (xlsx.exists() && !xlsx.delete()) {
log.warn("[appearance-patent] delete temp xlsx failed file={}", xlsx);
@@ -1628,6 +1779,56 @@ public class AppearancePatentTaskService {
}
}
private List<AppearancePatentParsedRowVo> filterReceivedParsedRows(List<AppearancePatentParsedRowVo> parsedRows,
Map<String, AppearancePatentResultRowDto> resultMap) {
if (parsedRows == null || parsedRows.isEmpty() || resultMap == null || resultMap.isEmpty()) {
return List.of();
}
List<AppearancePatentParsedRowVo> receivedRows = new ArrayList<>();
for (AppearancePatentParsedRowVo parsedRow : parsedRows) {
if (findResultRow(parsedRow, resultMap) != null) {
receivedRows.add(parsedRow);
}
}
return receivedRows;
}
private void validateCompleteCozeCoverage(Long taskId,
List<AppearancePatentParsedRowVo> receivedRows,
Map<String, AppearancePatentResultRowDto> resultMap) {
if (receivedRows == null || receivedRows.isEmpty()) {
return;
}
int expectedRows = 0;
int missingRows = 0;
List<String> sampleAsins = new ArrayList<>();
for (AppearancePatentParsedRowVo parsedRow : receivedRows) {
if (!hasPromptFields(parsedRow)) {
continue;
}
expectedRows++;
AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap);
if (resultRow == null || !hasCompleteCozeResult(resultRow)) {
missingRows++;
if (sampleAsins.size() < 5) {
sampleAsins.add(firstNonBlank(parsedRow.getAsin(), firstNonBlank(parsedRow.getDisplayId(), "")));
}
}
}
if (expectedRows > 0 && missingRows > 0) {
log.warn("[appearance-patent] incomplete coze coverage taskId={} expectedRows={} missingRows={} samples={}",
taskId, expectedRows, missingRows, sampleAsins);
throw new BusinessException("Coze 结果不完整：缺少 " + missingRows + "/" + expectedRows + " 条检测结果，请等待重试或重新运行任务");
}
}
private boolean hasCompleteCozeResult(AppearancePatentResultRowDto row) {
return hasUsableCozeField(row.getTitleRisk())
&& hasUsableCozeField(row.getAppearanceRisk())
&& hasUsableCozeField(row.getPatentRisk())
&& hasUsableCozeField(row.getConclusion());
}
private Map<String, AppearancePatentResultRowDto> loadPersistedResultRows(Long taskId) {
Map<String, AppearancePatentResultRowDto> result = new LinkedHashMap<>();
List<TaskChunkEntity> chunks = taskChunkMapper.selectList(new LambdaQueryWrapper<TaskChunkEntity>()
@@ -1667,7 +1868,10 @@ public class AppearancePatentTaskService {
}
}
private void writeResultWorkbook(File xlsx, AppearancePatentParsedPayloadDto parsed, Map<String, AppearancePatentResultRowDto> resultMap) {
private void writeResultWorkbook(File xlsx,
AppearancePatentParsedPayloadDto parsed,
List<AppearancePatentParsedRowVo> receivedRows,
Map<String, AppearancePatentResultRowDto> resultMap) {
try (SXSSFWorkbook workbook = new SXSSFWorkbook(200); FileOutputStream fos = new FileOutputStream(xlsx)) {
Sheet sheet = workbook.createSheet("外观专利检测结果");
CellStyle headerStyle = workbook.createCellStyle();
@@ -1686,11 +1890,9 @@ public class AppearancePatentTaskService {
}
int rowIndex = 1;
for (AppearancePatentParsedRowVo parsedRow : parsed.getAllItems()) {
List<AppearancePatentParsedRowVo> rowsToWrite = receivedRows == null ? List.of() : receivedRows;
for (AppearancePatentParsedRowVo parsedRow : rowsToWrite) {
AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap);
if (resultRow == null) {
resultRow = findResultRowByAsin(parsedRow.getAsin(), resultMap);
}
String missingReason = "";
if (resultRow == null) {
missingReason = hasPromptFields(parsedRow) ? "未匹配到检测结果" : "未送检：缺少标题或图片";
@@ -1708,7 +1910,7 @@ public class AppearancePatentTaskService {
row.createCell(col++).setCellValue(resultRow == null ? missingReason : userFacingCozeCellValue(resultRow, resultRow.getPatentRisk()));
row.createCell(col).setCellValue(resultRow == null ? "未送检" : userFacingConclusion(resultRow));
}
writeReasonSheet(workbook, headerStyle, parsed, resultMap);
writeReasonSheet(workbook, headerStyle, rowsToWrite, resultMap);
workbook.write(fos);
workbook.dispose();
} catch (Exception ex) {
@@ -1718,7 +1920,7 @@ public class AppearancePatentTaskService {
private void writeReasonSheet(SXSSFWorkbook workbook,
CellStyle headerStyle,
AppearancePatentParsedPayloadDto parsed,
List<AppearancePatentParsedRowVo> rowsToWrite,
Map<String, AppearancePatentResultRowDto> resultMap) {
Sheet sheet = workbook.createSheet("原因");
Row header = sheet.createRow(0);
@@ -1731,15 +1933,12 @@ public class AppearancePatentTaskService {
Set<String> writtenAsins = new LinkedHashSet<>();
int rowIndex = 1;
for (AppearancePatentParsedRowVo parsedRow : parsed.getAllItems()) {
for (AppearancePatentParsedRowVo parsedRow : rowsToWrite == null ? List.<AppearancePatentParsedRowVo>of() : rowsToWrite) {
String asin = normalize(parsedRow.getAsin()).toUpperCase(Locale.ROOT);
if (asin.isBlank() || !writtenAsins.add(asin)) {
continue;
}
AppearancePatentResultRowDto resultRow = findResultRow(parsedRow, resultMap);
if (resultRow == null) {
resultRow = findResultRowByAsin(asin, resultMap);
}
Row row = sheet.createRow(rowIndex++);
row.createCell(0).setCellValue(asin);
row.createCell(1).setCellValue(resultRow == null ? "" : firstNonBlank(resultRow.getAppearanceReason(), ""));
@@ -2072,6 +2271,36 @@ public class AppearancePatentTaskService {
return vo;
}
private int historyPriority(FileResultEntity row, FileTaskEntity task, TaskFileJobEntity job) {
String taskStatus = task == null ? null : task.getStatus();
if (STATUS_RUNNING.equals(taskStatus) || isHistoryFileBuilding(row, taskStatus, job)) {
return 0;
}
return 1;
}
private LocalDateTime historyActivityTime(FileResultEntity row, FileTaskEntity task, TaskFileJobEntity job) {
LocalDateTime latest = latestTime(
task == null ? null : task.getUpdatedAt(),
job == null ? null : job.getUpdatedAt(),
task == null ? null : task.getFinishedAt(),
row == null ? null : row.getCreatedAt(),
task == null ? null : task.getCreatedAt());
return latest == null && row != null ? row.getCreatedAt() : latest;
}
private boolean isHistoryFileBuilding(FileResultEntity row, String taskStatus, TaskFileJobEntity job) {
if (!STATUS_SUCCESS.equals(taskStatus)) {
return false;
}
boolean fileReady = row != null && row.getResultFileUrl() != null && !row.getResultFileUrl().isBlank();
if (fileReady) {
return false;
}
String fileStatus = job == null ? null : job.getStatus();
return !STATUS_SUCCESS.equals(fileStatus) && !STATUS_FAILED.equals(fileStatus);
}
private void attachFileJobState(AppearancePatentHistoryItemVo vo, FileResultEntity row, TaskFileJobEntity job) {
vo.setFileReady(row.getResultFileUrl() != null && !row.getResultFileUrl().isBlank());
if (job == null) {
@@ -2126,6 +2355,22 @@ public class AppearancePatentTaskService {
return t == null ? null : t.toString();
}
private LocalDateTime latestTime(LocalDateTime... values) {
LocalDateTime latest = null;
if (values == null) {
return null;
}
for (LocalDateTime value : values) {
if (value == null) {
continue;
}
if (latest == null || value.isAfter(latest)) {
latest = value;
}
}
return latest;
}
private String firstNonBlank(String preferred, String fallback) {
return preferred == null || preferred.isBlank() ? fallback : preferred.trim();
}
@@ -2157,9 +2402,35 @@ public class AppearancePatentTaskService {
}
private String storeParsedPayload(Long taskId, String scopeHash, String parsedPayloadJson) {
requireSharedTransientPayloadStorage("parsed payload");
return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, parsedPayloadJson, false);
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
return taskDistributedLockService.acquire(MODULE_TYPE, taskId, TASK_LOCK_TTL, waitMillis);
}
private String storeSharedChunkPayload(Long taskId, String scopeHash, Integer chunkIndex, String payloadJson) {
requireSharedTransientPayloadStorage("chunk payload");
return transientPayloadStorageService.storeChunkPayload(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
}
private String storeSharedChunkPayloadVersioned(Long taskId, String scopeHash, Integer chunkIndex, String payloadJson) {
requireSharedTransientPayloadStorage("merged chunk payload");
return transientPayloadStorageService.storeChunkPayloadVersioned(MODULE_TYPE, taskId, scopeHash, chunkIndex, payloadJson);
}
private String storeSharedCozeBatchPayload(Long taskId, String scopeHash, String payloadJson) {
requireSharedTransientPayloadStorage("coze batch payload");
return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, payloadJson, true);
}
private void requireSharedTransientPayloadStorage(String payloadType) {
if (!transientPayloadStorageService.isSharedWriteEnabled()) {
throw new BusinessException("外观专利" + payloadType + "必须写入共享临时存储 RustFS，请检查 aiimage.transient-storage 配置");
}
}
private long elapsedMs(long start, long end) {
return (end - start) / 1_000_000L;
}

16
backend-java/src/main/java/com/nanri/aiimage/modules/brand/service/BrandTaskService.java
View File

@@ -32,6 +32,7 @@ import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskItemVo;
import com.nanri.aiimage.modules.brand.model.vo.LegacyBrandTaskListVo;
import com.nanri.aiimage.modules.file.service.oss.OssStorageService;
import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
import lombok.RequiredArgsConstructor;
@@ -89,6 +90,7 @@ public class BrandTaskService {
private final BrandTaskProgressCacheService brandTaskProgressCacheService;
private final BrandTaskStorageService brandTaskStorageService;
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
private final ObjectMapper objectMapper;
private final TaskFileJobService taskFileJobService;
private final TaskProgressSnapshotService taskProgressSnapshotService;
@@ -1211,6 +1213,11 @@ public class BrandTaskService {
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
.lt(BrandCrawlTaskEntity::getUpdatedAt, threshold));
for (BrandCrawlTaskEntity task : runningTasks) {
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L);
if (taskLockHandle == null) {
continue;
}
try (taskLockHandle) {
if (tryRecoverCompletedRunningTask(task)) {
continue;
}
@@ -1236,6 +1243,15 @@ public class BrandTaskService {
}
}
}
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis);
if (lockHandle == null) {
log.info("[brand-task-lock] skip task because lock is busy taskId={}", taskId);
}
return lockHandle;
}
private boolean tryRecoverCompletedRunningTask(BrandCrawlTaskEntity task) {
if (task == null || task.getId() == null) {

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

@@ -14,6 +14,7 @@ import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskCacheService;
import com.nanri.aiimage.modules.shopmatch.service.ShopMatchTaskService;
import com.nanri.aiimage.modules.task.mapper.FileTaskMapper;
import com.nanri.aiimage.modules.task.model.entity.FileTaskEntity;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Value;
@@ -58,6 +59,7 @@ public class DeleteBrandStaleTaskService {
private final PatrolDeleteTaskCacheService patrolDeleteTaskCacheService;
private final DeleteBrandProgressProperties deleteBrandProgressProperties;
private final DistributedJobLockService distributedJobLockService;
private final TaskDistributedLockService taskDistributedLockService;
@Value("${aiimage.temp-dir.retention-hours:24}")
private long tempDirRetentionHours;
@@ -148,6 +150,11 @@ public class DeleteBrandStaleTaskService {
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_DELETE_BRAND, task.getId());
if (taskLockHandle == null) {
continue;
}
try (taskLockHandle) {
try {
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
FileTaskEntity refreshed = fileTaskMapper.selectById(task.getId());
@@ -184,6 +191,7 @@ public class DeleteBrandStaleTaskService {
deleteBrandTaskCacheService.delete(task.getId());
log.warn("[stale-check] delete-brand failed taskId={} reason=timeout", task.getId());
}
}
}
}
@@ -229,6 +237,12 @@ public class DeleteBrandStaleTaskService {
task.getId(), task.getCreatedAt(), initialThreshold);
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRODUCT_RISK, task.getId());
if (taskLockHandle == null) {
stats.skippedTaskCount++;
continue;
}
try (taskLockHandle) {
try {
if (productRiskTaskService.tryFinalizeTask(task.getId(), true)) {
stats.finalizedTaskCount++;
@@ -252,6 +266,7 @@ public class DeleteBrandStaleTaskService {
} else {
stats.skippedTaskCount++;
}
}
}
return stats;
}
@@ -294,6 +309,12 @@ public class DeleteBrandStaleTaskService {
stats.skippedTaskCount++;
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PRICE_TRACK, task.getId());
if (taskLockHandle == null) {
stats.skippedTaskCount++;
continue;
}
try (taskLockHandle) {
try {
if (priceTrackTaskService.tryFinalizeTask(task.getId(), true)) {
stats.finalizedTaskCount++;
@@ -318,6 +339,7 @@ public class DeleteBrandStaleTaskService {
} else {
stats.skippedTaskCount++;
}
}
}
return stats;
}
@@ -364,6 +386,12 @@ public class DeleteBrandStaleTaskService {
task.getId(), task.getCreatedAt(), initialThreshold);
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_SHOP_MATCH, task.getId());
if (taskLockHandle == null) {
stats.skippedTaskCount++;
continue;
}
try (taskLockHandle) {
try {
if (shopMatchTaskService.tryFinalizeTask(task.getId(), true)) {
stats.finalizedTaskCount++;
@@ -388,6 +416,7 @@ public class DeleteBrandStaleTaskService {
} else {
stats.skippedTaskCount++;
}
}
}
return stats;
}
@@ -428,6 +457,12 @@ public class DeleteBrandStaleTaskService {
stats.skippedTaskCount++;
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_PATROL_DELETE, task.getId());
if (taskLockHandle == null) {
stats.skippedTaskCount++;
continue;
}
try (taskLockHandle) {
try {
if (patrolDeleteTaskService.tryFinalizeTask(task.getId(), true)) {
stats.finalizedTaskCount++;
@@ -451,6 +486,7 @@ public class DeleteBrandStaleTaskService {
} else {
stats.skippedTaskCount++;
}
}
}
return stats;
}
@@ -463,6 +499,14 @@ public class DeleteBrandStaleTaskService {
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) {
log.info("[task-lock] skip stale task because task lock is busy moduleType={} taskId={}", moduleType, taskId);
}
return lockHandle;
}
@Scheduled(cron = "${aiimage.delete-brand-progress.finalize-check-cron:30 */2 * * * *}")
public void finalizeCompletedRunningTasks() {
DistributedJobLockService.LockHandle lockHandle =
@@ -483,11 +527,17 @@ public class DeleteBrandStaleTaskService {
.last("limit 100"));
for (FileTaskEntity task : runningTasks) {
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(MODULE_TYPE_DELETE_BRAND, task.getId());
if (taskLockHandle == null) {
continue;
}
try (taskLockHandle) {
try {
deleteBrandRunService.tryFinalizeTask(task.getId(), true);
} catch (Exception ignored) {
// Keep compensation best-effort; the next schedule can retry.
}
}
}
}
}

9
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/client/SimilarAsinCozeClient.java
View File

@@ -6,6 +6,7 @@ import com.nanri.aiimage.config.SimilarAsinProperties;
import com.nanri.aiimage.modules.similarasin.model.dto.SimilarAsinResultRowDto;
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;
@@ -24,6 +25,8 @@ import java.util.Map;
@Slf4j
public class SimilarAsinCozeClient {
private static final MediaType APPLICATION_JSON_UTF8 = new MediaType("application", "json", StandardCharsets.UTF_8);
private final SimilarAsinProperties properties;
private final ObjectMapper objectMapper;
@@ -216,7 +219,8 @@ public class SimilarAsinCozeClient {
.uri(joinUrl(properties.getCozeBaseUrl(), properties.getCozeWorkflowPath()))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
});
request.body(body);
return request.exchange((clientRequest, clientResponse) -> {
@@ -237,7 +241,8 @@ public class SimilarAsinCozeClient {
.uri(joinUrl(properties.getCozeBaseUrl(), path))
.headers(headers -> {
headers.setBearerAuth(stripBearer(properties.getCozeToken()));
headers.setContentType(MediaType.APPLICATION_JSON);
headers.setContentType(APPLICATION_JSON_UTF8);
headers.set(HttpHeaders.ACCEPT_CHARSET, StandardCharsets.UTF_8.name());
})
.exchange((clientRequest, clientResponse) -> {
byte[] responseBytes = StreamUtils.copyToByteArray(clientResponse.getBody());

38
backend-java/src/main/java/com/nanri/aiimage/modules/similarasin/service/SimilarAsinTaskService.java
View File

@@ -39,6 +39,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskChunkEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskProgressSnapshotEntity;
import com.nanri.aiimage.modules.task.model.entity.TaskScopeStateEntity;
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
import com.nanri.aiimage.modules.task.service.TaskDistributedLockService;
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
import com.nanri.aiimage.modules.task.service.TransientPayloadStorageService;
import lombok.RequiredArgsConstructor;
@@ -121,6 +122,7 @@ public class SimilarAsinTaskService {
private final SimilarAsinTaskCacheService taskCacheService;
private final SimilarAsinProperties properties;
private final TaskFileJobService taskFileJobService;
private final TaskDistributedLockService taskDistributedLockService;
private final TaskProgressSnapshotService taskProgressSnapshotService;
private final TransientPayloadStorageService transientPayloadStorageService;
private final PlatformTransactionManager transactionManager;
@@ -536,10 +538,16 @@ public class SimilarAsinTaskService {
if (heartbeatMillis > thresholdMillis) {
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L);
if (taskLockHandle == null) {
continue;
}
try (taskLockHandle) {
inNewTransaction(() -> {
finalizeStaleTask(task.getId(), "Python interrupted before uploading final similar ASIN result");
return null;
});
}
}
return;
}
@@ -559,7 +567,13 @@ public class SimilarAsinTaskService {
if (heartbeatMillis > thresholdMillis) {
continue;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(task.getId(), 0L);
if (taskLockHandle == null) {
continue;
}
try (taskLockHandle) {
finalizeTask(task, "Python interrupted before uploading final similar ASIN result", allRowCount(task), true);
}
}
}
@@ -1271,6 +1285,18 @@ public class SimilarAsinTaskService {
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
return;
}
TaskDistributedLockService.LockHandle taskLockHandle = acquireTaskLock(state.getTaskId(), 0L);
if (taskLockHandle == null) {
return;
}
try (taskLockHandle) {
state = taskScopeStateMapper.selectById(stateId);
if (state == null || state.getCozeExecuteId() == null || state.getCozeExecuteId().isBlank()) {
return;
}
if (!(COZE_STATUS_SUBMITTED.equals(state.getCozeStatus()) || COZE_STATUS_RUNNING.equals(state.getCozeStatus()))) {
return;
}
if (!tryClaimCozeStateForPoll(state)) {
return;
}
@@ -1328,6 +1354,7 @@ public class SimilarAsinTaskService {
state.getTaskId(), state.getId(), state.getCozeExecuteId(), message);
updateCozeStateRunning(state, message);
}
}
}
private void updateCozeStateRunning(TaskScopeStateEntity state, String error) {
@@ -1383,8 +1410,7 @@ public class SimilarAsinTaskService {
if (taskId == null || context == null || countPendingCozeStates(taskId) > 0) {
return;
}
DistributedJobLockService.LockHandle lockHandle =
distributedJobLockService.tryLock("similar-asin:coze-finalize:" + taskId, Duration.ofMinutes(5));
TaskDistributedLockService.LockHandle lockHandle = acquireTaskLock(taskId, 0L);
if (lockHandle == null) {
return;
}
@@ -1415,6 +1441,14 @@ public class SimilarAsinTaskService {
}
}
private TaskDistributedLockService.LockHandle acquireTaskLock(Long taskId, long waitMillis) {
TaskDistributedLockService.LockHandle lockHandle = taskDistributedLockService.acquire(MODULE_TYPE, taskId, waitMillis);
if (lockHandle == null) {
log.info("[similar-asin] skip task operation because task lock is busy taskId={}", taskId);
}
return lockHandle;
}
private void completeCozeFileJob(FileTaskEntity task,
FileResultEntity result,
TaskFileJobEntity job,

160
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskDistributedLockService.java Normal file
View File

@@ -0,0 +1,160 @@
package com.nanri.aiimage.modules.task.service;
import com.nanri.aiimage.common.service.DistributedJobLockService;
import jakarta.annotation.PreDestroy;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.Locale;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
@Service
@RequiredArgsConstructor
@Slf4j
public class TaskDistributedLockService {
public static final Duration DEFAULT_LOCK_TTL = Duration.ofMinutes(5);
public static final long DEFAULT_WAIT_MILLIS = 10000L;
private static final long RETRY_DELAY_MILLIS = 200L;
private final DistributedJobLockService distributedJobLockService;
private final ThreadLocal<Map<String, HeldLock>> localLocks = ThreadLocal.withInitial(LinkedHashMap::new);
private final ScheduledExecutorService renewalExecutor = Executors.newSingleThreadScheduledExecutor(runnable -> {
Thread thread = new Thread(runnable, "task-lock-renewal");
thread.setDaemon(true);
return thread;
});
public LockHandle acquire(String moduleType, Long taskId) {
return acquire(moduleType, taskId, DEFAULT_WAIT_MILLIS);
}
public LockHandle acquire(String moduleType, Long taskId, long waitMillis) {
return acquire(moduleType, taskId, DEFAULT_LOCK_TTL, waitMillis);
}
public LockHandle acquire(String moduleType, Long taskId, Duration ttl, long waitMillis) {
String lockName = taskLockName(moduleType, taskId);
if (lockName == null) {
return null;
}
Map<String, HeldLock> heldLocks = localLocks.get();
HeldLock heldLock = heldLocks.get(lockName);
if (heldLock != null) {
heldLock.depth++;
return new LockHandle(lockName);
}
long deadline = System.nanoTime() + Math.max(0L, waitMillis) * 1_000_000L;
do {
DistributedJobLockService.LockHandle delegate = distributedJobLockService.tryLock(lockName, ttl);
if (delegate != null) {
HeldLock newHeldLock = new HeldLock(delegate, ttl);
scheduleRenewal(lockName, newHeldLock, ttl);
heldLocks.put(lockName, newHeldLock);
return new LockHandle(lockName);
}
if (waitMillis <= 0L || System.nanoTime() >= deadline) {
return null;
}
try {
Thread.sleep(RETRY_DELAY_MILLIS);
} catch (InterruptedException ex) {
Thread.currentThread().interrupt();
return null;
}
} while (true);
}
public String taskLockName(String moduleType, Long taskId) {
if (moduleType == null || moduleType.isBlank() || taskId == null || taskId <= 0) {
return null;
}
String normalizedModule = moduleType.trim().toLowerCase(Locale.ROOT).replace('_', '-');
return "task:" + normalizedModule + ":" + taskId;
}
private void scheduleRenewal(String lockName, HeldLock heldLock, Duration ttl) {
long ttlMillis = Math.max(1000L, ttl.toMillis());
long renewalDelayMillis = Math.max(1000L, ttlMillis / 3L);
heldLock.renewalFuture = renewalExecutor.scheduleWithFixedDelay(() -> {
if (heldLock.closed) {
return;
}
boolean renewed = heldLock.delegate.renew(heldLock.ttl);
if (!renewed) {
heldLock.closed = true;
ScheduledFuture<?> future = heldLock.renewalFuture;
if (future != null) {
future.cancel(false);
}
log.warn("[task-lock] renewal failed lockName={}", lockName);
}
}, renewalDelayMillis, renewalDelayMillis, TimeUnit.MILLISECONDS);
}
@PreDestroy
public void shutdownRenewalExecutor() {
renewalExecutor.shutdownNow();
}
private static final class HeldLock {
private final DistributedJobLockService.LockHandle delegate;
private final Duration ttl;
private ScheduledFuture<?> renewalFuture;
private int depth = 1;
private volatile boolean closed;
private HeldLock(DistributedJobLockService.LockHandle delegate, Duration ttl) {
this.delegate = delegate;
this.ttl = ttl;
}
private void close() {
closed = true;
if (renewalFuture != null) {
renewalFuture.cancel(false);
}
delegate.close();
}
}
public final class LockHandle implements AutoCloseable {
private final String lockName;
private boolean closed;
private LockHandle(String lockName) {
this.lockName = lockName;
}
@Override
public void close() {
if (closed) {
return;
}
closed = true;
Map<String, HeldLock> heldLocks = localLocks.get();
HeldLock heldLock = heldLocks.get(lockName);
if (heldLock == null) {
return;
}
heldLock.depth--;
if (heldLock.depth <= 0) {
heldLocks.remove(lockName);
if (heldLocks.isEmpty()) {
localLocks.remove();
}
heldLock.close();
}
}
}
}

20
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskFileJobService.java
View File

@@ -168,6 +168,26 @@ public class TaskFileJobService {
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now()));
}
@Transactional
public boolean requeue(Long jobId, String message) {
if (jobId == null || jobId <= 0) {
return false;
}
int updated = taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
.eq(TaskFileJobEntity::getId, jobId)
.ne(TaskFileJobEntity::getStatus, "SUCCESS")
.set(TaskFileJobEntity::getStatus, "PENDING")
.set(TaskFileJobEntity::getErrorMessage, message)
.set(TaskFileJobEntity::getUpdatedAt, LocalDateTime.now())
.set(TaskFileJobEntity::getFinishedAt, null));
if (updated <= 0) {
return false;
}
TaskFileJobEntity refreshed = taskFileJobMapper.selectById(jobId);
publishDispatchEvent(refreshed);
return true;
}
public void markSuccess(TaskFileJobEntity job, String resultFileUrl) {
taskFileJobMapper.update(null, new LambdaUpdateWrapper<TaskFileJobEntity>()
.eq(TaskFileJobEntity::getId, job.getId())

35
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TaskResultFileJobWorker.java
View File

@@ -29,6 +29,7 @@ import java.util.List;
public class TaskResultFileJobWorker {
private final TaskFileJobService taskFileJobService;
private final TaskDistributedLockService taskDistributedLockService;
private final TaskResultPayloadService taskResultPayloadService;
private final FileResultMapper fileResultMapper;
private final TaskFileJobLocalDispatcher taskFileJobLocalDispatcher;
@@ -103,20 +104,30 @@ public class TaskResultFileJobWorker {
private void processInternal(TaskFileJobEntity job) {
long startedAt = System.currentTimeMillis();
try {
boolean completed = dispatch(job);
if (!completed) {
taskFileJobService.touchRunning(job.getId());
log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
TaskDistributedLockService.LockHandle lockHandle =
taskDistributedLockService.acquire(job.getModuleType(), job.getTaskId(), TaskDistributedLockService.DEFAULT_WAIT_MILLIS);
if (lockHandle == null) {
taskFileJobService.requeue(job.getId(), "Task is busy, waiting for previous task operation");
log.info("[task-file-job] process requeued because task lock is busy jobId={} taskId={} moduleType={} resultId={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId());
return;
}
String resultFileUrl = resolveResultFileUrl(job);
taskFileJobService.markSuccess(job, resultFileUrl);
cleanupAfterSuccess(job);
log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt, resultFileUrl);
try (lockHandle) {
boolean completed = dispatch(job);
if (!completed) {
taskFileJobService.touchRunning(job.getId());
log.info("[task-file-job] process deferred jobId={} taskId={} moduleType={} resultId={} elapsedMs={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt);
return;
}
String resultFileUrl = resolveResultFileUrl(job);
taskFileJobService.markSuccess(job, resultFileUrl);
cleanupAfterSuccess(job);
log.info("[task-file-job] process success jobId={} taskId={} moduleType={} resultId={} elapsedMs={} resultFileUrl={}",
job.getId(), job.getTaskId(), job.getModuleType(), job.getResultId(),
System.currentTimeMillis() - startedAt, resultFileUrl);
}
} catch (Exception ex) {
String message = ex.getMessage() == null ? "结果文件生成失败" : ex.getMessage();
log.warn("[task-file-job] process failed jobId={} taskId={} moduleType={} resultId={} msg={}",

16
backend-java/src/main/java/com/nanri/aiimage/modules/task/service/TransientPayloadStorageService.java
View File

@@ -33,7 +33,12 @@ public class TransientPayloadStorageService {
private final ObjectMapper objectMapper;
public boolean isWriteEnabled() {
return properties.isEnabled();
return properties.isEnabled()
&& (rustfsObjectStorageService.isConfigured() || storageProperties.getLocalTempDir() != null);
}
public boolean isSharedWriteEnabled() {
return properties.isEnabled() && rustfsObjectStorageService.isConfigured();
}
public String storeScopePayload(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) {
@@ -170,16 +175,19 @@ public class TransientPayloadStorageService {
return content;
}
String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey);
String pointer = storeLocal(objectKey, content);
if (pointer == null && rustfsObjectStorageService.isConfigured()) {
String pointer = null;
if (rustfsObjectStorageService.isConfigured()) {
try {
pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, content, verifyAfterUpload);
} catch (Exception ex) {
log.warn("[transient-payload] rustfs upload failed and local fallback unavailable objectKey={} err={}",
log.warn("[transient-payload] rustfs upload failed objectKey={} err={}",
objectKey, ex.getMessage());
throw ex;
}
}
if (pointer == null) {
pointer = storeLocal(objectKey, content);
}
if (pointer == null) {
return content;
}

37
backend-java/src/main/java/com/nanri/aiimage/modules/ziniao/service/ZiniaoShopIndexService.java
View File

@@ -341,7 +341,8 @@ public class ZiniaoShopIndexService {
cursor.setMessage(ex.getMessage());
cursor.setLastFinishedAt(now);
if (isIpWhitelistRefreshFailure(ex)) {
markExistingEntriesRefreshBlocked(now, REFRESH_BLOCKED_REASON_IP_WHITELIST, ex.getMessage());
log.warn("[ziniao-index] refresh blocked reason={} existing shop index rows left untouched",
REFRESH_BLOCKED_REASON_IP_WHITELIST);
}
ziniaoTransientCacheService.put(CACHE_TYPE_SHOP_INDEX_REFRESH_CURSOR, "global", cursor, DEFAULT_CURSOR_TTL);
log.warn("[ziniao-index] refresh failed status=FAILED msg={}", ex.getMessage());
@@ -589,40 +590,6 @@ public class ZiniaoShopIndexService {
return message != null && message.contains("白名单");
}
private void markExistingEntriesRefreshBlocked(long now, String reason, String message) {
List<ZiniaoMemoryStoreEntity> entities = ziniaoMemoryStoreService.listAliveEntitiesByType(
ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY,
SHOP_INDEX_LIST_LIMIT
);
if (entities.isEmpty()) {
return;
}
int updated = 0;
for (ZiniaoMemoryStoreEntity entity : entities) {
String rowKey = entity.getCacheKey();
ZiniaoShopIndexEntryDto existingEntry;
try {
existingEntry = objectMapper.readValue(entity.getPayloadJson(), ZiniaoShopIndexEntryDto.class);
} catch (Exception ex) {
log.warn("[ziniao-index] skip refresh-block mark corrupt shop_index row cacheKey={}", rowKey);
continue;
}
if (existingEntry == null || !STATUS_ACTIVE.equals(existingEntry.getStatus())) {
continue;
}
existingEntry.setLastRefreshBlockedAt(now);
existingEntry.setRefreshBlockedReason(reason);
if (message != null && !message.isBlank()) {
existingEntry.setMessage(message);
}
ziniaoMemoryStoreService.put(ZiniaoMemoryStoreService.CACHE_TYPE_SHOP_INDEX_ENTRY, rowKey, existingEntry, resolveEntryTtl());
updated++;
}
if (updated > 0) {
log.warn("[ziniao-index] refresh blocked reason={} existing active rows kept usable count={}", reason, updated);
}
}
private Duration resolveEntryTtl() {
Integer ttlHours = ziniaoProperties.getShopIndexEntryTtlHours();
if (ttlHours == null || ttlHours <= 0) {

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

@@ -10,6 +10,8 @@ spring:
multipart:
max-file-size: 200MB
max-request-size: 500MB
jackson:
time-zone: Asia/Shanghai
datasource:
driver-class-name: com.mysql.cj.jdbc.Driver
url: ${AIIMAGE_DB_URL:jdbc:mysql://127.0.0.1:3306/aiimage?useUnicode=true&characterEncoding=UTF-8&serverTimezone=Asia/Shanghai&useSSL=false&allowPublicKeyRetrieval=true}

234
scripts/coze_diag.py Normal file
View File

@@ -0,0 +1,234 @@
#!/usr/bin/env python3
import argparse
import json
import os
import sys
import time
from typing import Any, Dict, Optional
import requests
DEFAULT_BASE_URL = "https://api.coze.cn"
DEFAULT_RUN_PATH = "/v1/workflow/run"
DEFAULT_HISTORY_PATH = "/v1/workflows/{workflow_id}/run_histories/{execute_id}"
DEFAULT_SAMPLE_PROMPT = (
"请帮我排查以下亚马逊商品在英国及欧洲地区是否存在知识产权侵权风险。"
"请直接给出明确结论（有侵权风险 或 未发现明显侵权风险）。"
)
def build_headers(token: str) -> Dict[str, str]:
normalized = (token or "").strip()
if normalized.lower().startswith("bearer "):
normalized = normalized[7:].strip()
return {
"Authorization": f"Bearer {normalized}",
"Content-Type": "application/json; charset=utf-8",
"Accept-Charset": "utf-8",
}
def join_url(base_url: str, path: str) -> str:
return base_url.rstrip("/") + "/" + path.lstrip("/")
def print_block(title: str, payload: Any) -> None:
print(f"\n=== {title} ===")
if isinstance(payload, (dict, list)):
print(json.dumps(payload, ensure_ascii=False, indent=2))
else:
print(payload)
def request_json(method: str, url: str, headers: Dict[str, str], **kwargs: Any) -> Dict[str, Any]:
started = time.time()
response = requests.request(method, url, headers=headers, timeout=kwargs.pop("timeout", 60), **kwargs)
elapsed_ms = int((time.time() - started) * 1000)
body_text = response.text
print_block(
f"{method.upper()} {url} -> HTTP {response.status_code} ({elapsed_ms} ms)",
body_text if len(body_text) < 5000 else body_text[:5000] + "\n...<truncated>...",
)
response.raise_for_status()
try:
return response.json()
except Exception as exc:
raise RuntimeError(f"response is not valid JSON: {exc}") from exc
def extract_execute_id(payload: Dict[str, Any]) -> str:
for key in ("execute_id", "executeId"):
value = payload.get(key)
if isinstance(value, str) and value.strip():
return value.strip()
raise RuntimeError("execute_id not found in Coze response")
def payload_data_text(payload: Dict[str, Any]) -> str:
data = payload.get("data")
if isinstance(data, str):
return data
if isinstance(data, dict):
inner = data.get("data")
if isinstance(inner, str):
return inner
return ""
def workflow_status(payload: Dict[str, Any]) -> str:
for key in ("status", "run_status", "workflow_status"):
value = payload.get(key)
if isinstance(value, str) and value.strip():
return value.strip().upper()
data = payload.get("data")
if isinstance(data, dict):
for key in ("status", "run_status", "workflow_status"):
value = data.get(key)
if isinstance(value, str) and value.strip():
return value.strip().upper()
return ""
def looks_finished(payload: Dict[str, Any]) -> bool:
status = workflow_status(payload)
return status in {"SUCCESS", "SUCCEEDED", "DONE", "FAILED", "ERROR", "CANCELLED"}
def build_sample_run_body(workflow_id: str, prompt: str) -> Dict[str, Any]:
return {
"workflow_id": workflow_id,
"parameters": {
"title_list": ["WIRESTER Green Aluminum Round Bike Bell"],
"url_list": ["https://m.media-amazon.com/images/I/51dETWpSQXL._AC_SY450_.jpg"],
"items": [
{
"group_key": "diag::1",
"row_id": "1",
"asin": "B0D14L34S7",
"country": "英国",
"title": "WIRESTER Green Aluminum Round Bike Bell",
"url": "https://m.media-amazon.com/images/I/51dETWpSQXL._AC_SY450_.jpg",
}
],
"prompt": prompt,
},
"is_async": True,
}
def poll_history(
base_url: str,
workflow_id: str,
execute_id: str,
token: str,
history_path: str,
poll_interval: float,
max_polls: int,
) -> int:
headers = build_headers(token)
history_url = join_url(
base_url,
history_path.replace("{workflow_id}", workflow_id).replace("{execute_id}", execute_id),
)
for idx in range(1, max_polls + 1):
print(f"\n--- poll #{idx} execute_id={execute_id} ---")
payload = request_json("GET", history_url, headers)
code = payload.get("code")
status = workflow_status(payload)
data_text = payload_data_text(payload)
print(f"code={code!r} status={status or '-'} has_data={bool(data_text)}")
if data_text:
print_block("Resolved Data", data_text)
return 0
if looks_finished(payload):
print("workflow reached terminal status but no data was returned")
return 2
time.sleep(max(0.2, poll_interval))
print("history polling reached max attempts without terminal result")
return 3
def main() -> int:
parser = argparse.ArgumentParser(description="Diagnose Coze run/history connectivity from server.")
parser.add_argument("--base-url", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_BASE_URL", DEFAULT_BASE_URL))
parser.add_argument("--workflow-id", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID") or os.getenv("workflow_id"))
parser.add_argument("--token", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN") or os.getenv("coze_token"))
parser.add_argument("--run-path", default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_PATH", DEFAULT_RUN_PATH))
parser.add_argument(
"--history-path",
default=os.getenv("AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_HISTORY_PATH", DEFAULT_HISTORY_PATH),
)
parser.add_argument("--execute-id", help="Only test history/poll for an existing execute_id.")
parser.add_argument("--poll-interval", type=float, default=2.0)
parser.add_argument("--max-polls", type=int, default=20)
parser.add_argument("--prompt", default=DEFAULT_SAMPLE_PROMPT)
parser.add_argument("--skip-run", action="store_true", help="Do not create a new run; only poll execute_id.")
args = parser.parse_args()
if not args.workflow_id:
print("missing workflow id: pass --workflow-id or set AIIMAGE_APPEARANCE_PATENT_COZE_WORKFLOW_ID/workflow_id", file=sys.stderr)
return 4
if not args.token:
print("missing token: pass --token or set AIIMAGE_APPEARANCE_PATENT_COZE_TOKEN/coze_token", file=sys.stderr)
return 4
print_block(
"Config",
{
"base_url": args.base_url,
"workflow_id": args.workflow_id,
"run_path": args.run_path,
"history_path": args.history_path,
"has_token": bool(args.token),
"execute_id": args.execute_id,
"skip_run": args.skip_run,
},
)
if args.execute_id:
return poll_history(
args.base_url,
args.workflow_id,
args.execute_id,
args.token,
args.history_path,
args.poll_interval,
args.max_polls,
)
if args.skip_run:
print("--skip-run requires --execute-id", file=sys.stderr)
return 4
headers = build_headers(args.token)
run_url = join_url(args.base_url, args.run_path)
body = build_sample_run_body(args.workflow_id, args.prompt)
print_block("Run Request Body", body)
payload = request_json("POST", run_url, headers, json=body)
execute_id = extract_execute_id(payload)
print(f"\nexecute_id={execute_id}")
return poll_history(
args.base_url,
args.workflow_id,
execute_id,
args.token,
args.history_path,
args.poll_interval,
args.max_polls,
)
if __name__ == "__main__":
try:
raise SystemExit(main())
except requests.HTTPError as exc:
response = exc.response
if response is not None:
print(f"HTTP error: {response.status_code} {response.text}", file=sys.stderr)
else:
print(f"HTTP error: {exc}", file=sys.stderr)
raise
except Exception as exc:
print(f"fatal: {exc}", file=sys.stderr)
raise