增加rufts
This commit is contained in:
super
+250
-21
@@ -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
@@ -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;
|
||||
|
||||
+4
-4
@@ -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) {
|
||||
}
|
||||
}
|
||||
|
||||
+368
-47
@@ -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) {
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user