提交更新
This commit is contained in:
4
app/.env
4
app/.env
@@ -12,8 +12,8 @@ client_name=ShuFuAI
|
|||||||
|
|
||||||
|
|
||||||
# java_api_base=http://47.111.163.154:18080
|
# java_api_base=http://47.111.163.154:18080
|
||||||
java_api_base=http://127.0.0.1:18080
|
# java_api_base=http://127.0.0.1:18080
|
||||||
# java_api_base=http://8.136.19.173:18080
|
# java_api_base=http://8.136.19.173:18080
|
||||||
# java_api_base=http://121.196.149.225:18080
|
java_api_base=http://121.196.149.225:18080
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
File diff suppressed because one or more lines are too long
@@ -149,9 +149,16 @@ public class AppearancePatentCozeClient {
|
|||||||
}
|
}
|
||||||
|
|
||||||
String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
|
String status = normalize(resolveWorkflowStatus(pollRoot)).toUpperCase(Locale.ROOT);
|
||||||
if (status.contains("FAIL") || status.contains("ERROR") || status.contains("CANCEL")) {
|
if (isFailedWorkflowStatus(status)) {
|
||||||
throw new IllegalStateException(firstNonBlank(resolveFailureMessage(pollRoot), "Coze async workflow failed"));
|
throw new IllegalStateException(firstNonBlank(resolveFailureMessage(pollRoot), "Coze async workflow failed"));
|
||||||
}
|
}
|
||||||
|
if (isSuccessfulWorkflowStatus(status)) {
|
||||||
|
String outputText = extractWorkflowOutputText(pollRoot);
|
||||||
|
if (!outputText.isBlank()) {
|
||||||
|
return wrapDataPayload(outputText);
|
||||||
|
}
|
||||||
|
throw new IllegalStateException("Coze async workflow completed without output");
|
||||||
|
}
|
||||||
sleepQuietly(Math.max(200, properties.getCozePollIntervalMillis()));
|
sleepQuietly(Math.max(200, properties.getCozePollIntervalMillis()));
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -251,7 +258,7 @@ public class AppearancePatentCozeClient {
|
|||||||
return List.of();
|
return List.of();
|
||||||
}
|
}
|
||||||
JsonNode dataRoot = objectMapper.readTree(dataText);
|
JsonNode dataRoot = objectMapper.readTree(dataText);
|
||||||
JsonNode array = dataRoot.path("data");
|
JsonNode array = dataRoot.isArray() ? dataRoot : dataRoot.path("data");
|
||||||
List<CozeResult> results = new ArrayList<>();
|
List<CozeResult> results = new ArrayList<>();
|
||||||
if (array.isArray()) {
|
if (array.isArray()) {
|
||||||
for (JsonNode node : array) {
|
for (JsonNode node : array) {
|
||||||
@@ -502,6 +509,23 @@ public class AppearancePatentCozeClient {
|
|||||||
return discoverEmbeddedData(root);
|
return discoverEmbeddedData(root);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private String extractWorkflowOutputText(JsonNode root) {
|
||||||
|
JsonNode outputNode = findFirstField(root, "output");
|
||||||
|
if (outputNode == null || outputNode.isNull() || outputNode.isMissingNode()) {
|
||||||
|
return "";
|
||||||
|
}
|
||||||
|
if (!outputNode.isTextual()) {
|
||||||
|
return outputNode.toString();
|
||||||
|
}
|
||||||
|
String output = normalize(outputNode.asText(""));
|
||||||
|
if (output.isBlank()) {
|
||||||
|
return "";
|
||||||
|
}
|
||||||
|
JsonNode parsedOutput = parseJsonOrMissing(output);
|
||||||
|
String nestedOutput = text(firstNonNull(parsedOutput.get("Output"), parsedOutput.get("output")));
|
||||||
|
return nestedOutput == null || nestedOutput.isBlank() ? output : nestedOutput;
|
||||||
|
}
|
||||||
|
|
||||||
private String discoverEmbeddedData(JsonNode node) {
|
private String discoverEmbeddedData(JsonNode node) {
|
||||||
if (node == null || node.isNull() || node.isMissingNode()) {
|
if (node == null || node.isNull() || node.isMissingNode()) {
|
||||||
return "";
|
return "";
|
||||||
@@ -577,6 +601,22 @@ public class AppearancePatentCozeClient {
|
|||||||
|| item.has("patent reason"));
|
|| item.has("patent reason"));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private boolean isSuccessfulWorkflowStatus(String status) {
|
||||||
|
String normalized = normalize(status).toUpperCase(Locale.ROOT);
|
||||||
|
return normalized.equals("SUCCESS")
|
||||||
|
|| normalized.equals("SUCCEEDED")
|
||||||
|
|| normalized.equals("COMPLETED")
|
||||||
|
|| normalized.equals("COMPLETE")
|
||||||
|
|| normalized.equals("DONE");
|
||||||
|
}
|
||||||
|
|
||||||
|
private boolean isFailedWorkflowStatus(String status) {
|
||||||
|
String normalized = normalize(status).toUpperCase(Locale.ROOT);
|
||||||
|
return normalized.contains("FAIL")
|
||||||
|
|| normalized.contains("ERROR")
|
||||||
|
|| normalized.contains("CANCEL");
|
||||||
|
}
|
||||||
|
|
||||||
private JsonNode parseJsonOrMissing(String value) {
|
private JsonNode parseJsonOrMissing(String value) {
|
||||||
String normalized = normalize(value);
|
String normalized = normalize(value);
|
||||||
if (!(normalized.startsWith("{") || normalized.startsWith("["))) {
|
if (!(normalized.startsWith("{") || normalized.startsWith("["))) {
|
||||||
@@ -611,6 +651,38 @@ public class AppearancePatentCozeClient {
|
|||||||
return findTextByFieldName(root, "execute_id", "executeId");
|
return findTextByFieldName(root, "execute_id", "executeId");
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private JsonNode findFirstField(JsonNode node, String name) {
|
||||||
|
if (node == null || node.isMissingNode() || node.isNull() || name == null || name.isBlank()) {
|
||||||
|
return objectMapper.missingNode();
|
||||||
|
}
|
||||||
|
if (node.isTextual()) {
|
||||||
|
return findFirstField(parseJsonOrMissing(node.asText("")), name);
|
||||||
|
}
|
||||||
|
if (node.isArray()) {
|
||||||
|
for (JsonNode child : node) {
|
||||||
|
JsonNode found = findFirstField(child, name);
|
||||||
|
if (found != null && !found.isMissingNode() && !found.isNull()) {
|
||||||
|
return found;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return objectMapper.missingNode();
|
||||||
|
}
|
||||||
|
if (node.isObject()) {
|
||||||
|
JsonNode direct = node.get(name);
|
||||||
|
if (direct != null && !direct.isMissingNode() && !direct.isNull()) {
|
||||||
|
return direct;
|
||||||
|
}
|
||||||
|
for (java.util.Iterator<Map.Entry<String, JsonNode>> it = node.fields(); it.hasNext(); ) {
|
||||||
|
Map.Entry<String, JsonNode> entry = it.next();
|
||||||
|
JsonNode found = findFirstField(entry.getValue(), name);
|
||||||
|
if (found != null && !found.isMissingNode() && !found.isNull()) {
|
||||||
|
return found;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return objectMapper.missingNode();
|
||||||
|
}
|
||||||
|
|
||||||
private String findTextByFieldName(JsonNode node, String... names) {
|
private String findTextByFieldName(JsonNode node, String... names) {
|
||||||
if (node == null || node.isMissingNode() || node.isNull()) {
|
if (node == null || node.isMissingNode() || node.isNull()) {
|
||||||
return "";
|
return "";
|
||||||
|
|||||||
@@ -113,8 +113,8 @@ public class AppearancePatentTaskService {
|
|||||||
private final TransientPayloadStorageService transientPayloadStorageService;
|
private final TransientPayloadStorageService transientPayloadStorageService;
|
||||||
private final PlatformTransactionManager transactionManager;
|
private final PlatformTransactionManager transactionManager;
|
||||||
|
|
||||||
@Transactional
|
|
||||||
public AppearancePatentParseVo parseAndCreateTask(AppearancePatentParseRequest request) {
|
public AppearancePatentParseVo parseAndCreateTask(AppearancePatentParseRequest request) {
|
||||||
|
long startedAt = System.nanoTime();
|
||||||
if (request.getUserId() == null || request.getUserId() <= 0) {
|
if (request.getUserId() == null || request.getUserId() <= 0) {
|
||||||
throw new BusinessException("user_id 不合法");
|
throw new BusinessException("user_id 不合法");
|
||||||
}
|
}
|
||||||
@@ -131,6 +131,7 @@ public class AppearancePatentTaskService {
|
|||||||
List<String> mergedHeaders = new ArrayList<>();
|
List<String> mergedHeaders = new ArrayList<>();
|
||||||
int totalRows = 0;
|
int totalRows = 0;
|
||||||
int droppedRows = 0;
|
int droppedRows = 0;
|
||||||
|
long parseStartedAt = System.nanoTime();
|
||||||
for (AppearancePatentSourceFileDto source : sourceFiles) {
|
for (AppearancePatentSourceFileDto source : sourceFiles) {
|
||||||
if (source.getFileKey() == null || source.getFileKey().isBlank()) {
|
if (source.getFileKey() == null || source.getFileKey().isBlank()) {
|
||||||
throw new BusinessException("fileKey 不能为空");
|
throw new BusinessException("fileKey 不能为空");
|
||||||
@@ -149,7 +150,9 @@ public class AppearancePatentTaskService {
|
|||||||
if (allRows.isEmpty()) {
|
if (allRows.isEmpty()) {
|
||||||
throw new BusinessException("未解析到有效 ASIN 数据");
|
throw new BusinessException("未解析到有效 ASIN 数据");
|
||||||
}
|
}
|
||||||
|
long parsedAt = System.nanoTime();
|
||||||
List<AppearancePatentParsedGroupVo> groups = buildParsedGroups(allRows);
|
List<AppearancePatentParsedGroupVo> groups = buildParsedGroups(allRows);
|
||||||
|
long groupedAt = System.nanoTime();
|
||||||
|
|
||||||
FileTaskEntity task = new FileTaskEntity();
|
FileTaskEntity task = new FileTaskEntity();
|
||||||
task.setTaskNo(MODULE_TYPE + "-" + IdUtil.getSnowflakeNextIdStr());
|
task.setTaskNo(MODULE_TYPE + "-" + IdUtil.getSnowflakeNextIdStr());
|
||||||
@@ -170,11 +173,14 @@ public class AppearancePatentTaskService {
|
|||||||
throw new BusinessException("序列化解析结果失败");
|
throw new BusinessException("序列化解析结果失败");
|
||||||
}
|
}
|
||||||
fileTaskMapper.insert(task);
|
fileTaskMapper.insert(task);
|
||||||
|
long taskInsertedAt = System.nanoTime();
|
||||||
|
|
||||||
String aggregateScopeKey = buildAggregateScopeKey(sourceFiles);
|
String aggregateScopeKey = buildAggregateScopeKey(sourceFiles);
|
||||||
String sourceScopeHash = DigestUtil.sha256Hex(aggregateScopeKey);
|
String sourceScopeHash = DigestUtil.sha256Hex(aggregateScopeKey);
|
||||||
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), sourceFiles, mergedHeaders, groups, allRows);
|
String parsedPayload = buildParsedPayloadJson(request.getAiPrompt(), sourceFiles, mergedHeaders, groups, allRows);
|
||||||
|
long payloadBuiltAt = System.nanoTime();
|
||||||
String parsedPayloadPointer = storeParsedPayload(task.getId(), sourceScopeHash, parsedPayload);
|
String parsedPayloadPointer = storeParsedPayload(task.getId(), sourceScopeHash, parsedPayload);
|
||||||
|
long payloadStoredAt = System.nanoTime();
|
||||||
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), sourceFiles, parsedPayloadPointer));
|
task.setResultJson(buildTaskResultJson(request.getAiPrompt(), sourceFiles, parsedPayloadPointer));
|
||||||
task.setUpdatedAt(LocalDateTime.now());
|
task.setUpdatedAt(LocalDateTime.now());
|
||||||
fileTaskMapper.updateById(task);
|
fileTaskMapper.updateById(task);
|
||||||
@@ -203,6 +209,7 @@ public class AppearancePatentTaskService {
|
|||||||
scope.setCreatedAt(LocalDateTime.now());
|
scope.setCreatedAt(LocalDateTime.now());
|
||||||
scope.setUpdatedAt(LocalDateTime.now());
|
scope.setUpdatedAt(LocalDateTime.now());
|
||||||
taskScopeStateMapper.insert(scope);
|
taskScopeStateMapper.insert(scope);
|
||||||
|
long persistedAt = System.nanoTime();
|
||||||
|
|
||||||
AppearancePatentParseVo vo = new AppearancePatentParseVo();
|
AppearancePatentParseVo vo = new AppearancePatentParseVo();
|
||||||
vo.setTaskId(task.getId());
|
vo.setTaskId(task.getId());
|
||||||
@@ -215,6 +222,20 @@ public class AppearancePatentTaskService {
|
|||||||
vo.setAiPrompt(normalize(request.getAiPrompt()));
|
vo.setAiPrompt(normalize(request.getAiPrompt()));
|
||||||
vo.setItems(allRows);
|
vo.setItems(allRows);
|
||||||
vo.setGroups(groups);
|
vo.setGroups(groups);
|
||||||
|
long finishedAt = System.nanoTime();
|
||||||
|
log.info("[appearance-patent] parse timing taskId={} files={} rows={} groups={} totalMs={} parseMs={} groupMs={} taskInsertMs={} payloadJsonMs={} payloadStoreMs={} persistMs={} responseMs={}",
|
||||||
|
task.getId(),
|
||||||
|
sourceFiles.size(),
|
||||||
|
allRows.size(),
|
||||||
|
groups.size(),
|
||||||
|
elapsedMs(startedAt, finishedAt),
|
||||||
|
elapsedMs(parseStartedAt, parsedAt),
|
||||||
|
elapsedMs(parsedAt, groupedAt),
|
||||||
|
elapsedMs(groupedAt, taskInsertedAt),
|
||||||
|
elapsedMs(taskInsertedAt, payloadBuiltAt),
|
||||||
|
elapsedMs(payloadBuiltAt, payloadStoredAt),
|
||||||
|
elapsedMs(payloadStoredAt, persistedAt),
|
||||||
|
elapsedMs(persistedAt, finishedAt));
|
||||||
return vo;
|
return vo;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1615,9 +1636,9 @@ public class AppearancePatentTaskService {
|
|||||||
payload.setAiPrompt(normalize(aiPrompt));
|
payload.setAiPrompt(normalize(aiPrompt));
|
||||||
payload.setSourceFiles(sourceFiles == null ? List.of() : sourceFiles);
|
payload.setSourceFiles(sourceFiles == null ? List.of() : sourceFiles);
|
||||||
payload.setHeaders(headers == null ? List.of() : headers);
|
payload.setHeaders(headers == null ? List.of() : headers);
|
||||||
payload.setItems(allRows == null ? List.of() : allRows);
|
payload.setItems(List.of());
|
||||||
payload.setGroups(groups == null ? List.of() : groups);
|
payload.setGroups(groups == null ? List.of() : groups);
|
||||||
payload.setAllItems(allRows == null ? List.of() : allRows);
|
payload.setAllItems(List.of());
|
||||||
return writeJson(payload, "保存解析结果失败");
|
return writeJson(payload, "保存解析结果失败");
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1633,7 +1654,11 @@ public class AppearancePatentTaskService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
private String storeParsedPayload(Long taskId, String scopeHash, String parsedPayloadJson) {
|
private String storeParsedPayload(Long taskId, String scopeHash, String parsedPayloadJson) {
|
||||||
return transientPayloadStorageService.storeParsedPayload(MODULE_TYPE, taskId, scopeHash, parsedPayloadJson, true);
|
return transientPayloadStorageService.storeParsedPayloadFast(MODULE_TYPE, taskId, scopeHash, parsedPayloadJson, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
private long elapsedMs(long start, long end) {
|
||||||
|
return (end - start) / 1_000_000L;
|
||||||
}
|
}
|
||||||
|
|
||||||
private AppearancePatentParsedPayloadDto readParsedPayload(FileTaskEntity task) {
|
private AppearancePatentParsedPayloadDto readParsedPayload(FileTaskEntity task) {
|
||||||
@@ -1642,15 +1667,41 @@ public class AppearancePatentTaskService {
|
|||||||
String pointer = root.path("parsedPayloadRef").asText("");
|
String pointer = root.path("parsedPayloadRef").asText("");
|
||||||
if (!pointer.isBlank()) {
|
if (!pointer.isBlank()) {
|
||||||
String json = transientPayloadStorageService.resolvePayload(pointer, "read appearance patent parsed payload failed");
|
String json = transientPayloadStorageService.resolvePayload(pointer, "read appearance patent parsed payload failed");
|
||||||
return objectMapper.readValue(json, AppearancePatentParsedPayloadDto.class);
|
return hydrateParsedPayloadRows(objectMapper.readValue(json, AppearancePatentParsedPayloadDto.class));
|
||||||
}
|
}
|
||||||
if (root.path("allItems").isArray()) {
|
if (root.path("allItems").isArray()) {
|
||||||
return objectMapper.treeToValue(root, AppearancePatentParsedPayloadDto.class);
|
return hydrateParsedPayloadRows(objectMapper.treeToValue(root, AppearancePatentParsedPayloadDto.class));
|
||||||
}
|
}
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
throw new BusinessException("读取解析载荷失败");
|
throw new BusinessException("读取解析载荷失败");
|
||||||
}
|
}
|
||||||
return new AppearancePatentParsedPayloadDto();
|
return hydrateParsedPayloadRows(new AppearancePatentParsedPayloadDto());
|
||||||
|
}
|
||||||
|
|
||||||
|
private AppearancePatentParsedPayloadDto hydrateParsedPayloadRows(AppearancePatentParsedPayloadDto payload) {
|
||||||
|
if (payload == null) {
|
||||||
|
return new AppearancePatentParsedPayloadDto();
|
||||||
|
}
|
||||||
|
List<AppearancePatentParsedRowVo> rows = payload.getAllItems();
|
||||||
|
if (rows == null || rows.isEmpty()) {
|
||||||
|
rows = payload.getItems();
|
||||||
|
}
|
||||||
|
if ((rows == null || rows.isEmpty()) && payload.getGroups() != null) {
|
||||||
|
rows = payload.getGroups().stream()
|
||||||
|
.filter(Objects::nonNull)
|
||||||
|
.flatMap(group -> group.getItems() == null ? java.util.stream.Stream.empty() : group.getItems().stream())
|
||||||
|
.toList();
|
||||||
|
}
|
||||||
|
if (rows == null) {
|
||||||
|
rows = List.of();
|
||||||
|
}
|
||||||
|
if (payload.getAllItems() == null || payload.getAllItems().isEmpty()) {
|
||||||
|
payload.setAllItems(rows);
|
||||||
|
}
|
||||||
|
if (payload.getItems() == null || payload.getItems().isEmpty()) {
|
||||||
|
payload.setItems(rows);
|
||||||
|
}
|
||||||
|
return payload;
|
||||||
}
|
}
|
||||||
|
|
||||||
private Map<String, AppearancePatentResultRowDto> readChunkRows(TaskChunkEntity chunk) {
|
private Map<String, AppearancePatentResultRowDto> readChunkRows(TaskChunkEntity chunk) {
|
||||||
|
|||||||
@@ -614,6 +614,9 @@ public class BrandTaskService {
|
|||||||
if (STATUS_FAILED.equalsIgnoreCase(freshStatus)) {
|
if (STATUS_FAILED.equalsIgnoreCase(freshStatus)) {
|
||||||
throw new BusinessException("任务已结束,不能继续组装结果");
|
throw new BusinessException("任务已结束,不能继续组装结果");
|
||||||
}
|
}
|
||||||
|
for (BrandSourceFileDto sourceFile : sourceFiles) {
|
||||||
|
brandTaskStorageService.refreshFileAggregate(taskId, sourceFile.getFileUrl());
|
||||||
|
}
|
||||||
Map<String, BrandFileAggregateCacheDto> aggregates = brandTaskStorageService.getAllFileAggregates(taskId);
|
Map<String, BrandFileAggregateCacheDto> aggregates = brandTaskStorageService.getAllFileAggregates(taskId);
|
||||||
log.info("[brand-finalize] taskId={} aggregateFiles={} expectedFiles={} thread={}",
|
log.info("[brand-finalize] taskId={} aggregateFiles={} expectedFiles={} thread={}",
|
||||||
taskId,
|
taskId,
|
||||||
@@ -621,12 +624,12 @@ public class BrandTaskService {
|
|||||||
totalCount,
|
totalCount,
|
||||||
Thread.currentThread().getName());
|
Thread.currentThread().getName());
|
||||||
if (aggregates.size() < totalCount) {
|
if (aggregates.size() < totalCount) {
|
||||||
return;
|
throw new BusinessException("品牌检测结果分片尚未收齐,请稍后重试");
|
||||||
}
|
}
|
||||||
for (BrandSourceFileDto sourceFile : sourceFiles) {
|
for (BrandSourceFileDto sourceFile : sourceFiles) {
|
||||||
BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl());
|
BrandFileAggregateCacheDto aggregate = aggregates.get(sourceFile.getFileUrl());
|
||||||
if (aggregate == null || !Boolean.TRUE.equals(aggregate.getCompleted())) {
|
if (aggregate == null || !Boolean.TRUE.equals(aggregate.getCompleted())) {
|
||||||
return;
|
throw new BusinessException("品牌检测结果分片尚未收齐,请稍后重试");
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1208,6 +1211,9 @@ public class BrandTaskService {
|
|||||||
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
|
.eq(BrandCrawlTaskEntity::getStatus, STATUS_RUNNING)
|
||||||
.lt(BrandCrawlTaskEntity::getUpdatedAt, threshold));
|
.lt(BrandCrawlTaskEntity::getUpdatedAt, threshold));
|
||||||
for (BrandCrawlTaskEntity task : runningTasks) {
|
for (BrandCrawlTaskEntity task : runningTasks) {
|
||||||
|
if (tryRecoverCompletedRunningTask(task)) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
Map<Object, Object> progress = brandTaskProgressCacheService.getProgress(task.getId());
|
Map<Object, Object> progress = brandTaskProgressCacheService.getProgress(task.getId());
|
||||||
if (progress.isEmpty()) {
|
if (progress.isEmpty()) {
|
||||||
failStaleRunningTask(task.getId(), "任务进度缓存已过期,任务已自动失败");
|
failStaleRunningTask(task.getId(), "任务进度缓存已过期,任务已自动失败");
|
||||||
@@ -1231,6 +1237,34 @@ public class BrandTaskService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private boolean tryRecoverCompletedRunningTask(BrandCrawlTaskEntity task) {
|
||||||
|
if (task == null || task.getId() == null) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
try {
|
||||||
|
List<BrandSourceFileDto> sourceFiles = parseSourceFiles(task.getFilePaths());
|
||||||
|
if (sourceFiles.isEmpty()) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
List<BrandParsedFileCacheDto> cachedFiles = brandTaskStorageService.getParsedPayload(task.getId());
|
||||||
|
Map<String, BrandParsedFileCacheDto> cachedByUrl = indexCachedFiles(cachedFiles);
|
||||||
|
for (BrandSourceFileDto sourceFile : sourceFiles) {
|
||||||
|
BrandTaskStorageService.ChunkStoreResult result =
|
||||||
|
brandTaskStorageService.refreshFileAggregate(task.getId(), sourceFile.getFileUrl());
|
||||||
|
BrandFileAggregateCacheDto aggregate = result.aggregate();
|
||||||
|
if (aggregate == null || !Boolean.TRUE.equals(aggregate.getCompleted())) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
log.info("[brand-stale-check] recover completed running taskId={} files={}", task.getId(), sourceFiles.size());
|
||||||
|
finalizeTask(task.getId(), normalizeStrategy(task.getStrategy()), sourceFiles, cachedByUrl, sourceFiles.size());
|
||||||
|
return true;
|
||||||
|
} catch (Exception ex) {
|
||||||
|
log.warn("[brand-stale-check] recover completed running taskId={} failed msg={}", task.getId(), ex.getMessage());
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
private void failStaleRunningTask(Long taskId, String message) {
|
private void failStaleRunningTask(Long taskId, String message) {
|
||||||
brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
|
brandCrawlTaskMapper.update(null, new LambdaUpdateWrapper<BrandCrawlTaskEntity>()
|
||||||
.eq(BrandCrawlTaskEntity::getId, taskId)
|
.eq(BrandCrawlTaskEntity::getId, taskId)
|
||||||
|
|||||||
@@ -278,20 +278,30 @@ public class BrandTaskStorageService {
|
|||||||
if (state == null) {
|
if (state == null) {
|
||||||
throw new BusinessException("Brand task aggregate state missing: " + scopeKey);
|
throw new BusinessException("Brand task aggregate state missing: " + scopeKey);
|
||||||
}
|
}
|
||||||
String storedAggregate = transientPayloadStorageService.storeScopePayload(MODULE_TYPE, taskId, scopeHash, aggregateJson, true);
|
String storedAggregate = transientPayloadStorageService.storeScopePayloadVersioned(MODULE_TYPE, taskId, scopeHash, aggregateJson, true);
|
||||||
int chunkTotal = aggregate.getChunkTotal() == null ? 0 : aggregate.getChunkTotal();
|
int chunkTotal = aggregate.getChunkTotal() == null ? 0 : aggregate.getChunkTotal();
|
||||||
int receivedCount = aggregate.getReceivedChunkCount() == null ? 0 : aggregate.getReceivedChunkCount();
|
int receivedCount = aggregate.getReceivedChunkCount() == null ? 0 : aggregate.getReceivedChunkCount();
|
||||||
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(state.getStateJson(), storedAggregate);
|
int completed = Boolean.TRUE.equals(aggregate.getCompleted()) ? 1 : 0;
|
||||||
taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
|
int updated = taskScopeStateMapper.update(null, new LambdaUpdateWrapper<TaskScopeStateEntity>()
|
||||||
.eq(TaskScopeStateEntity::getId, state.getId())
|
.eq(TaskScopeStateEntity::getId, state.getId())
|
||||||
|
.and(wrapper -> wrapper
|
||||||
|
.lt(TaskScopeStateEntity::getReceivedChunkCount, receivedCount)
|
||||||
|
.or()
|
||||||
|
.eq(TaskScopeStateEntity::getReceivedChunkCount, receivedCount)
|
||||||
|
.le(TaskScopeStateEntity::getCompleted, completed))
|
||||||
.set(TaskScopeStateEntity::getScopeKey, scopeKey)
|
.set(TaskScopeStateEntity::getScopeKey, scopeKey)
|
||||||
.set(TaskScopeStateEntity::getStateJson, storedAggregate)
|
.set(TaskScopeStateEntity::getStateJson, storedAggregate)
|
||||||
.set(TaskScopeStateEntity::getChunkTotal, chunkTotal)
|
.set(TaskScopeStateEntity::getChunkTotal, chunkTotal)
|
||||||
.set(TaskScopeStateEntity::getReceivedChunkCount, receivedCount)
|
.set(TaskScopeStateEntity::getReceivedChunkCount, receivedCount)
|
||||||
.set(TaskScopeStateEntity::getCompleted, Boolean.TRUE.equals(aggregate.getCompleted()) ? 1 : 0)
|
.set(TaskScopeStateEntity::getCompleted, completed)
|
||||||
.set(TaskScopeStateEntity::getLastChunkAt, now)
|
.set(TaskScopeStateEntity::getLastChunkAt, now)
|
||||||
.set(TaskScopeStateEntity::getLastError, null)
|
.set(TaskScopeStateEntity::getLastError, null)
|
||||||
.set(TaskScopeStateEntity::getUpdatedAt, now));
|
.set(TaskScopeStateEntity::getUpdatedAt, now));
|
||||||
|
if (updated > 0) {
|
||||||
|
transientPayloadStorageService.deleteReplacedPayloadIfNeeded(state.getStateJson(), storedAggregate);
|
||||||
|
} else {
|
||||||
|
transientPayloadStorageService.deletePayloadIfPresent(storedAggregate);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private TaskScopeStateEntity getScopeState(Long taskId, String scopeHash) {
|
private TaskScopeStateEntity getScopeState(Long taskId, String scopeHash) {
|
||||||
|
|||||||
@@ -32,6 +32,10 @@ public class RustfsObjectStorageService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public String uploadText(String objectKey, String content) {
|
public String uploadText(String objectKey, String content) {
|
||||||
|
return uploadText(objectKey, content, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
public String uploadText(String objectKey, String content, boolean verifyAfterUpload) {
|
||||||
if (!isConfigured()) {
|
if (!isConfigured()) {
|
||||||
throw new IllegalStateException("transient storage is not configured");
|
throw new IllegalStateException("transient storage is not configured");
|
||||||
}
|
}
|
||||||
@@ -46,7 +50,9 @@ public class RustfsObjectStorageService {
|
|||||||
.stream(stream, bytes.length, -1)
|
.stream(stream, bytes.length, -1)
|
||||||
.contentType("application/json")
|
.contentType("application/json")
|
||||||
.build());
|
.build());
|
||||||
verifyObjectVisible(objectKey);
|
if (verifyAfterUpload) {
|
||||||
|
verifyObjectVisible(objectKey);
|
||||||
|
}
|
||||||
return objectKey;
|
return objectKey;
|
||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
last = ex;
|
last = ex;
|
||||||
|
|||||||
@@ -127,6 +127,11 @@ public class PatrolDeleteTaskCacheService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public void deleteTaskCache(Long taskId) {
|
public void deleteTaskCache(Long taskId) {
|
||||||
|
evictTaskCacheOnly(taskId);
|
||||||
|
taskScopePayloadStorageService.deleteTaskScopePayloads(taskId, MODULE_TYPE);
|
||||||
|
}
|
||||||
|
|
||||||
|
public void evictTaskCacheOnly(Long taskId) {
|
||||||
if (taskId == null || taskId <= 0) {
|
if (taskId == null || taskId <= 0) {
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
@@ -137,7 +142,6 @@ public class PatrolDeleteTaskCacheService {
|
|||||||
} catch (Exception ex) {
|
} catch (Exception ex) {
|
||||||
log.warn("[patrol-delete-cache] delete cache degraded taskId={} msg={}", taskId, ex.getMessage());
|
log.warn("[patrol-delete-cache] delete cache degraded taskId={} msg={}", taskId, ex.getMessage());
|
||||||
}
|
}
|
||||||
taskScopePayloadStorageService.deleteTaskScopePayloads(taskId, MODULE_TYPE);
|
|
||||||
}
|
}
|
||||||
|
|
||||||
public void saveTaskCache(FileTaskEntity task) {
|
public void saveTaskCache(FileTaskEntity task) {
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import com.nanri.aiimage.modules.task.model.entity.TaskFileJobEntity;
|
|||||||
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
import com.nanri.aiimage.modules.task.service.TaskFileJobService;
|
||||||
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
import com.nanri.aiimage.modules.task.service.TaskProgressSnapshotService;
|
||||||
import com.nanri.aiimage.modules.task.service.TaskResultItemService;
|
import com.nanri.aiimage.modules.task.service.TaskResultItemService;
|
||||||
|
import com.nanri.aiimage.modules.task.service.TaskScopePayloadStorageService;
|
||||||
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
|
import com.nanri.aiimage.modules.ziniao.service.ZiniaoShopSwitchService;
|
||||||
import lombok.RequiredArgsConstructor;
|
import lombok.RequiredArgsConstructor;
|
||||||
import lombok.extern.slf4j.Slf4j;
|
import lombok.extern.slf4j.Slf4j;
|
||||||
@@ -66,6 +67,7 @@ public class PatrolDeleteTaskService {
|
|||||||
private final TaskFileJobService taskFileJobService;
|
private final TaskFileJobService taskFileJobService;
|
||||||
private final TaskResultItemService taskResultItemService;
|
private final TaskResultItemService taskResultItemService;
|
||||||
private final TaskProgressSnapshotService taskProgressSnapshotService;
|
private final TaskProgressSnapshotService taskProgressSnapshotService;
|
||||||
|
private final TaskScopePayloadStorageService taskScopePayloadStorageService;
|
||||||
|
|
||||||
private FileTaskEntity loadTaskForExecution(Long taskId) {
|
private FileTaskEntity loadTaskForExecution(Long taskId) {
|
||||||
Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId));
|
Map<Long, FileTaskEntity> cachedTasks = taskCacheService.getTaskCacheBatch(List.of(taskId));
|
||||||
@@ -135,32 +137,58 @@ public class PatrolDeleteTaskService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
public PatrolDeleteHistoryVo listHistory(Long userId) {
|
public PatrolDeleteHistoryVo listHistory(Long userId) {
|
||||||
|
long startedAt = System.nanoTime();
|
||||||
validateUserId(userId);
|
validateUserId(userId);
|
||||||
PatrolDeleteHistoryVo vo = new PatrolDeleteHistoryVo();
|
PatrolDeleteHistoryVo vo = new PatrolDeleteHistoryVo();
|
||||||
List<FileResultEntity> entities = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
List<FileResultEntity> entities = fileResultMapper.selectList(new LambdaQueryWrapper<FileResultEntity>()
|
||||||
|
.select(FileResultEntity::getId,
|
||||||
|
FileResultEntity::getTaskId,
|
||||||
|
FileResultEntity::getModuleType,
|
||||||
|
FileResultEntity::getSourceFilename,
|
||||||
|
FileResultEntity::getSourceFileUrl,
|
||||||
|
FileResultEntity::getResultFilename,
|
||||||
|
FileResultEntity::getResultFileUrl,
|
||||||
|
FileResultEntity::getSuccess,
|
||||||
|
FileResultEntity::getErrorMessage,
|
||||||
|
FileResultEntity::getUserId,
|
||||||
|
FileResultEntity::getCreatedAt)
|
||||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
.eq(FileResultEntity::getModuleType, MODULE_TYPE)
|
||||||
.eq(FileResultEntity::getUserId, userId)
|
.eq(FileResultEntity::getUserId, userId)
|
||||||
.orderByDesc(FileResultEntity::getCreatedAt)
|
.orderByDesc(FileResultEntity::getCreatedAt)
|
||||||
.last("limit 100"));
|
.last("limit 100"));
|
||||||
|
long resultRowsLoadedAt = System.nanoTime();
|
||||||
if (entities.isEmpty()) {
|
if (entities.isEmpty()) {
|
||||||
vo.setItems(List.of());
|
vo.setItems(List.of());
|
||||||
|
log.info("[patrol-delete] history timing userId={} rows=0 totalMs={} resultQueryMs={} taskQueryMs=0 jobQueryMs=0 buildMs=0",
|
||||||
|
userId, elapsedMs(startedAt, resultRowsLoadedAt), elapsedMs(startedAt, resultRowsLoadedAt));
|
||||||
return vo;
|
return vo;
|
||||||
}
|
}
|
||||||
|
|
||||||
Map<Long, FileTaskEntity> taskMap = loadTaskMap(entities);
|
Map<Long, FileTaskEntity> taskMap = loadHistoryTaskMap(entities);
|
||||||
Map<Long, Map<Long, PatrolDeleteResultItemVo>> snapshotMap = buildSnapshotMap(taskMap);
|
long tasksLoadedAt = System.nanoTime();
|
||||||
Map<Long, TaskFileJobEntity> jobMap = taskFileJobService.findAssembleJobsByResultIds(MODULE_TYPE, entities.stream()
|
Map<Long, TaskFileJobEntity> jobMap = taskFileJobService.findAssembleJobsByResultIds(MODULE_TYPE, entities.stream()
|
||||||
.map(FileResultEntity::getId)
|
.map(FileResultEntity::getId)
|
||||||
.filter(id -> id != null && id > 0)
|
.filter(id -> id != null && id > 0)
|
||||||
.distinct()
|
.distinct()
|
||||||
.toList());
|
.toList());
|
||||||
|
long jobsLoadedAt = System.nanoTime();
|
||||||
List<PatrolDeleteResultItemVo> items = new ArrayList<>();
|
List<PatrolDeleteResultItemVo> items = new ArrayList<>();
|
||||||
for (FileResultEntity entity : entities) {
|
for (FileResultEntity entity : entities) {
|
||||||
FileTaskEntity task = taskMap.get(entity.getTaskId());
|
FileTaskEntity task = taskMap.get(entity.getTaskId());
|
||||||
PatrolDeleteResultItemVo snapshot = snapshotMap.getOrDefault(entity.getTaskId(), Map.of()).get(entity.getId());
|
items.add(toHistoryItem(entity, task, null, jobMap.get(entity.getId())));
|
||||||
items.add(toHistoryItem(entity, task, snapshot, jobMap.get(entity.getId())));
|
|
||||||
}
|
}
|
||||||
vo.setItems(items);
|
vo.setItems(items);
|
||||||
|
long finishedAt = System.nanoTime();
|
||||||
|
log.info("[patrol-delete] history timing userId={} rows={} tasks={} jobs={} totalMs={} resultQueryMs={} taskQueryMs={} jobQueryMs={} buildMs={}",
|
||||||
|
userId,
|
||||||
|
entities.size(),
|
||||||
|
taskMap.size(),
|
||||||
|
jobMap.size(),
|
||||||
|
elapsedMs(startedAt, finishedAt),
|
||||||
|
elapsedMs(startedAt, resultRowsLoadedAt),
|
||||||
|
elapsedMs(resultRowsLoadedAt, tasksLoadedAt),
|
||||||
|
elapsedMs(tasksLoadedAt, jobsLoadedAt),
|
||||||
|
elapsedMs(jobsLoadedAt, finishedAt));
|
||||||
return vo;
|
return vo;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -417,17 +445,32 @@ public class PatrolDeleteTaskService {
|
|||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
public void deleteTask(Long taskId, Long userId) {
|
public void deleteTask(Long taskId, Long userId) {
|
||||||
|
long startedAt = System.nanoTime();
|
||||||
validateUserId(userId);
|
validateUserId(userId);
|
||||||
FileTaskEntity task = loadTaskForExecution(taskId);
|
FileTaskEntity task = loadTaskForExecution(taskId);
|
||||||
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
|
if (task == null || !MODULE_TYPE.equals(task.getModuleType()) || !userId.equals(task.getUserId())) {
|
||||||
throw new BusinessException("任务不存在");
|
throw new BusinessException("任务不存在");
|
||||||
}
|
}
|
||||||
|
long taskLoadedAt = System.nanoTime();
|
||||||
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
|
fileResultMapper.delete(new LambdaQueryWrapper<FileResultEntity>()
|
||||||
.eq(FileResultEntity::getTaskId, taskId)
|
.eq(FileResultEntity::getTaskId, taskId)
|
||||||
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
|
.eq(FileResultEntity::getModuleType, MODULE_TYPE));
|
||||||
cleanupTaskAuxiliaryData(taskId);
|
long resultsDeletedAt = System.nanoTime();
|
||||||
|
cleanupTaskAuxiliaryDataFast(taskId);
|
||||||
|
long auxiliaryDeletedAt = System.nanoTime();
|
||||||
fileTaskMapper.deleteById(taskId);
|
fileTaskMapper.deleteById(taskId);
|
||||||
taskCacheService.deleteTaskCache(taskId);
|
long taskDeletedAt = System.nanoTime();
|
||||||
|
taskCacheService.evictTaskCacheOnly(taskId);
|
||||||
|
long finishedAt = System.nanoTime();
|
||||||
|
log.info("[patrol-delete] delete task timing taskId={} userId={} totalMs={} loadMs={} resultDeleteMs={} auxiliaryDeleteMs={} taskDeleteMs={} cacheMs={}",
|
||||||
|
taskId,
|
||||||
|
userId,
|
||||||
|
elapsedMs(startedAt, finishedAt),
|
||||||
|
elapsedMs(startedAt, taskLoadedAt),
|
||||||
|
elapsedMs(taskLoadedAt, resultsDeletedAt),
|
||||||
|
elapsedMs(resultsDeletedAt, auxiliaryDeletedAt),
|
||||||
|
elapsedMs(auxiliaryDeletedAt, taskDeletedAt),
|
||||||
|
elapsedMs(taskDeletedAt, finishedAt));
|
||||||
}
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
@@ -439,7 +482,7 @@ public class PatrolDeleteTaskService {
|
|||||||
}
|
}
|
||||||
Long taskId = entity.getTaskId();
|
Long taskId = entity.getTaskId();
|
||||||
fileResultMapper.deleteById(resultId);
|
fileResultMapper.deleteById(resultId);
|
||||||
cleanupResultAuxiliaryData(taskId, resultId);
|
cleanupResultAuxiliaryDataFast(taskId, resultId);
|
||||||
reconcileTaskAfterResultRemoval(taskId);
|
reconcileTaskAfterResultRemoval(taskId);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -453,13 +496,12 @@ public class PatrolDeleteTaskService {
|
|||||||
}
|
}
|
||||||
List<FileResultEntity> rows = listTaskRows(taskId);
|
List<FileResultEntity> rows = listTaskRows(taskId);
|
||||||
if (rows.isEmpty()) {
|
if (rows.isEmpty()) {
|
||||||
cleanupTaskAuxiliaryData(taskId);
|
cleanupTaskAuxiliaryDataFast(taskId);
|
||||||
fileTaskMapper.deleteById(taskId);
|
fileTaskMapper.deleteById(taskId);
|
||||||
taskCacheService.deleteTaskCache(taskId);
|
taskCacheService.evictTaskCacheOnly(taskId);
|
||||||
return;
|
return;
|
||||||
}
|
}
|
||||||
updateTaskStatusFromRows(task, rows);
|
updateTaskStatusFromRows(task, rows);
|
||||||
persistSnapshotJson(task, buildSnapshotFromDb(task, rows));
|
|
||||||
fileTaskMapper.updateById(task);
|
fileTaskMapper.updateById(task);
|
||||||
taskCacheService.saveTaskCache(task);
|
taskCacheService.saveTaskCache(task);
|
||||||
}
|
}
|
||||||
@@ -470,11 +512,23 @@ public class PatrolDeleteTaskService {
|
|||||||
taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE);
|
taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void cleanupTaskAuxiliaryDataFast(Long taskId) {
|
||||||
|
taskResultItemService.deleteTaskItemsRowsOnly(taskId, MODULE_TYPE);
|
||||||
|
taskScopePayloadStorageService.deleteTaskScopePayloadRowsOnly(taskId, MODULE_TYPE);
|
||||||
|
taskProgressSnapshotService.delete(taskId, MODULE_TYPE);
|
||||||
|
taskFileJobService.deleteTaskJobs(taskId, MODULE_TYPE);
|
||||||
|
}
|
||||||
|
|
||||||
private void cleanupResultAuxiliaryData(Long taskId, Long resultId) {
|
private void cleanupResultAuxiliaryData(Long taskId, Long resultId) {
|
||||||
taskResultItemService.deleteResultItem(taskId, MODULE_TYPE, resultId);
|
taskResultItemService.deleteResultItem(taskId, MODULE_TYPE, resultId);
|
||||||
taskFileJobService.deleteResultJobs(taskId, MODULE_TYPE, resultId);
|
taskFileJobService.deleteResultJobs(taskId, MODULE_TYPE, resultId);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private void cleanupResultAuxiliaryDataFast(Long taskId, Long resultId) {
|
||||||
|
taskResultItemService.deleteResultItemRowsOnly(taskId, MODULE_TYPE, resultId);
|
||||||
|
taskFileJobService.deleteResultJobs(taskId, MODULE_TYPE, resultId);
|
||||||
|
}
|
||||||
|
|
||||||
private long countTasks(Long userId, List<String> statuses) {
|
private long countTasks(Long userId, List<String> statuses) {
|
||||||
Long count = fileTaskMapper.selectCount(new LambdaQueryWrapper<FileTaskEntity>()
|
Long count = fileTaskMapper.selectCount(new LambdaQueryWrapper<FileTaskEntity>()
|
||||||
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||||
@@ -524,6 +578,35 @@ public class PatrolDeleteTaskService {
|
|||||||
return taskMap;
|
return taskMap;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private Map<Long, FileTaskEntity> loadHistoryTaskMap(List<FileResultEntity> entities) {
|
||||||
|
List<Long> taskIds = entities.stream()
|
||||||
|
.map(FileResultEntity::getTaskId)
|
||||||
|
.filter(id -> id != null && id > 0)
|
||||||
|
.distinct()
|
||||||
|
.toList();
|
||||||
|
Map<Long, FileTaskEntity> taskMap = new LinkedHashMap<>();
|
||||||
|
if (taskIds.isEmpty()) {
|
||||||
|
return taskMap;
|
||||||
|
}
|
||||||
|
int batchSize = Math.max(1, taskPressureProperties.getDbSelectBatchSize());
|
||||||
|
for (int start = 0; start < taskIds.size(); start += batchSize) {
|
||||||
|
int end = Math.min(start + batchSize, taskIds.size());
|
||||||
|
List<FileTaskEntity> tasks = fileTaskMapper.selectList(new LambdaQueryWrapper<FileTaskEntity>()
|
||||||
|
.select(FileTaskEntity::getId,
|
||||||
|
FileTaskEntity::getModuleType,
|
||||||
|
FileTaskEntity::getStatus,
|
||||||
|
FileTaskEntity::getFinishedAt)
|
||||||
|
.eq(FileTaskEntity::getModuleType, MODULE_TYPE)
|
||||||
|
.in(FileTaskEntity::getId, taskIds.subList(start, end)));
|
||||||
|
for (FileTaskEntity task : tasks) {
|
||||||
|
if (task != null && task.getId() != null) {
|
||||||
|
taskMap.put(task.getId(), task);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
return taskMap;
|
||||||
|
}
|
||||||
|
|
||||||
private Map<Long, Map<Long, PatrolDeleteResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
|
private Map<Long, Map<Long, PatrolDeleteResultItemVo>> buildSnapshotMap(Map<Long, FileTaskEntity> taskMap) {
|
||||||
Map<Long, Map<Long, PatrolDeleteResultItemVo>> out = new LinkedHashMap<>();
|
Map<Long, Map<Long, PatrolDeleteResultItemVo>> out = new LinkedHashMap<>();
|
||||||
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
|
for (Map.Entry<Long, FileTaskEntity> entry : taskMap.entrySet()) {
|
||||||
@@ -925,6 +1008,10 @@ public class PatrolDeleteTaskService {
|
|||||||
return Integer.valueOf(RESULT_SUCCESS).equals(dbValue);
|
return Integer.valueOf(RESULT_SUCCESS).equals(dbValue);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private long elapsedMs(long start, long end) {
|
||||||
|
return (end - start) / 1_000_000L;
|
||||||
|
}
|
||||||
|
|
||||||
private List<PatrolDeleteResultItemVo> parseTaskSnapshots(String json) {
|
private List<PatrolDeleteResultItemVo> parseTaskSnapshots(String json) {
|
||||||
if (blank(json)) {
|
if (blank(json)) {
|
||||||
return new ArrayList<>();
|
return new ArrayList<>();
|
||||||
|
|||||||
@@ -115,6 +115,17 @@ public class TaskResultItemService {
|
|||||||
.eq(TaskResultItemEntity::getResultId, resultId));
|
.eq(TaskResultItemEntity::getResultId, resultId));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Transactional
|
||||||
|
public void deleteResultItemRowsOnly(Long taskId, String moduleType, Long resultId) {
|
||||||
|
if (taskId == null || taskId <= 0 || isBlank(moduleType) || resultId == null || resultId <= 0) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
|
||||||
|
.eq(TaskResultItemEntity::getTaskId, taskId)
|
||||||
|
.eq(TaskResultItemEntity::getModuleType, moduleType)
|
||||||
|
.eq(TaskResultItemEntity::getResultId, resultId));
|
||||||
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
public void deleteTaskItems(Long taskId, String moduleType) {
|
public void deleteTaskItems(Long taskId, String moduleType) {
|
||||||
if (taskId == null || taskId <= 0 || isBlank(moduleType)) {
|
if (taskId == null || taskId <= 0 || isBlank(moduleType)) {
|
||||||
@@ -134,6 +145,16 @@ public class TaskResultItemService {
|
|||||||
.eq(TaskResultItemEntity::getModuleType, moduleType));
|
.eq(TaskResultItemEntity::getModuleType, moduleType));
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Transactional
|
||||||
|
public void deleteTaskItemsRowsOnly(Long taskId, String moduleType) {
|
||||||
|
if (taskId == null || taskId <= 0 || isBlank(moduleType)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
taskResultItemMapper.delete(new LambdaQueryWrapper<TaskResultItemEntity>()
|
||||||
|
.eq(TaskResultItemEntity::getTaskId, taskId)
|
||||||
|
.eq(TaskResultItemEntity::getModuleType, moduleType));
|
||||||
|
}
|
||||||
|
|
||||||
private void upsert(Long taskId,
|
private void upsert(Long taskId,
|
||||||
String moduleType,
|
String moduleType,
|
||||||
Long resultId,
|
Long resultId,
|
||||||
|
|||||||
@@ -188,6 +188,16 @@ public class TaskScopePayloadStorageService {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Transactional
|
||||||
|
public void deleteTaskScopePayloadRowsOnly(Long taskId, String moduleType) {
|
||||||
|
if (taskId == null || taskId <= 0 || isBlank(moduleType)) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
taskScopeStateMapper.delete(new LambdaQueryWrapper<TaskScopeStateEntity>()
|
||||||
|
.eq(TaskScopeStateEntity::getTaskId, taskId)
|
||||||
|
.eq(TaskScopeStateEntity::getModuleType, moduleType));
|
||||||
|
}
|
||||||
|
|
||||||
@Transactional
|
@Transactional
|
||||||
public <T> void saveParsedPayloadMap(Long taskId, String moduleType, Map<String, T> payloadByScope) {
|
public <T> void saveParsedPayloadMap(Long taskId, String moduleType, Map<String, T> payloadByScope) {
|
||||||
if (taskId == null || taskId <= 0 || isBlank(moduleType) || payloadByScope == null || payloadByScope.isEmpty()) {
|
if (taskId == null || taskId <= 0 || isBlank(moduleType) || payloadByScope == null || payloadByScope.isEmpty()) {
|
||||||
|
|||||||
@@ -30,10 +30,18 @@ public class TransientPayloadStorageService {
|
|||||||
return store("task-scope", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString);
|
return store("task-scope", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public String storeScopePayloadVersioned(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) {
|
||||||
|
return store("task-scope", moduleType, taskId, scopeHash, UUID.randomUUID().toString(), content, encodeAsJsonString);
|
||||||
|
}
|
||||||
|
|
||||||
public String storeParsedPayload(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) {
|
public String storeParsedPayload(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) {
|
||||||
return store("task-parsed", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString);
|
return store("task-parsed", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
public String storeParsedPayloadFast(String moduleType, Long taskId, String scopeHash, String content, boolean encodeAsJsonString) {
|
||||||
|
return store("task-parsed", moduleType, taskId, scopeHash, "latest", content, encodeAsJsonString, false);
|
||||||
|
}
|
||||||
|
|
||||||
public String storeResultPayload(String moduleType, Long taskId, String scopeHash, String submissionId, String content) {
|
public String storeResultPayload(String moduleType, Long taskId, String scopeHash, String submissionId, String content) {
|
||||||
return store("task-result-payload", moduleType, taskId, scopeHash, submissionId, content, true);
|
return store("task-result-payload", moduleType, taskId, scopeHash, submissionId, content, true);
|
||||||
}
|
}
|
||||||
@@ -114,11 +122,22 @@ public class TransientPayloadStorageService {
|
|||||||
String entryKey,
|
String entryKey,
|
||||||
String content,
|
String content,
|
||||||
boolean encodeAsJsonString) {
|
boolean encodeAsJsonString) {
|
||||||
|
return store(category, moduleType, taskId, scopeHash, entryKey, content, encodeAsJsonString, true);
|
||||||
|
}
|
||||||
|
|
||||||
|
private String store(String category,
|
||||||
|
String moduleType,
|
||||||
|
Long taskId,
|
||||||
|
String scopeHash,
|
||||||
|
String entryKey,
|
||||||
|
String content,
|
||||||
|
boolean encodeAsJsonString,
|
||||||
|
boolean verifyAfterUpload) {
|
||||||
if (!isWriteEnabled()) {
|
if (!isWriteEnabled()) {
|
||||||
return content;
|
return content;
|
||||||
}
|
}
|
||||||
String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey);
|
String objectKey = buildObjectKey(category, moduleType, taskId, scopeHash, entryKey);
|
||||||
String pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, content);
|
String pointer = RUSTFS_POINTER_PREFIX + rustfsObjectStorageService.uploadText(objectKey, content, verifyAfterUpload);
|
||||||
if (!encodeAsJsonString) {
|
if (!encodeAsJsonString) {
|
||||||
return pointer;
|
return pointer;
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -0,0 +1,35 @@
|
|||||||
|
SET @schema_name = DATABASE();
|
||||||
|
|
||||||
|
SET @idx_file_result_patrol_history_exists := (
|
||||||
|
SELECT COUNT(1)
|
||||||
|
FROM information_schema.STATISTICS
|
||||||
|
WHERE TABLE_SCHEMA = @schema_name
|
||||||
|
AND TABLE_NAME = 'biz_file_result'
|
||||||
|
AND INDEX_NAME = 'idx_file_result_patrol_history'
|
||||||
|
);
|
||||||
|
|
||||||
|
SET @sql_add_file_result_patrol_history_idx := IF(
|
||||||
|
@idx_file_result_patrol_history_exists = 0,
|
||||||
|
'ALTER TABLE biz_file_result ADD INDEX idx_file_result_patrol_history (module_type, user_id, created_at, id)',
|
||||||
|
'SELECT 1'
|
||||||
|
);
|
||||||
|
PREPARE stmt_add_file_result_patrol_history_idx FROM @sql_add_file_result_patrol_history_idx;
|
||||||
|
EXECUTE stmt_add_file_result_patrol_history_idx;
|
||||||
|
DEALLOCATE PREPARE stmt_add_file_result_patrol_history_idx;
|
||||||
|
|
||||||
|
SET @idx_file_task_patrol_history_exists := (
|
||||||
|
SELECT COUNT(1)
|
||||||
|
FROM information_schema.STATISTICS
|
||||||
|
WHERE TABLE_SCHEMA = @schema_name
|
||||||
|
AND TABLE_NAME = 'biz_file_task'
|
||||||
|
AND INDEX_NAME = 'idx_file_task_patrol_history'
|
||||||
|
);
|
||||||
|
|
||||||
|
SET @sql_add_file_task_patrol_history_idx := IF(
|
||||||
|
@idx_file_task_patrol_history_exists = 0,
|
||||||
|
'ALTER TABLE biz_file_task ADD INDEX idx_file_task_patrol_history (module_type, id, status, finished_at)',
|
||||||
|
'SELECT 1'
|
||||||
|
);
|
||||||
|
PREPARE stmt_add_file_task_patrol_history_idx FROM @sql_add_file_task_patrol_history_idx;
|
||||||
|
EXECUTE stmt_add_file_task_patrol_history_idx;
|
||||||
|
DEALLOCATE PREPARE stmt_add_file_task_patrol_history_idx;
|
||||||
Binary file not shown.
@@ -331,8 +331,6 @@ async function parseFiles() {
|
|||||||
}]
|
}]
|
||||||
: [])
|
: [])
|
||||||
queuePayloadText.value = ''
|
queuePayloadText.value = ''
|
||||||
await loadDashboard()
|
|
||||||
await loadHistory({ force: true })
|
|
||||||
ElMessage.success(`解析完成,共 ${res.groupCount || 0} 组 / ${res.acceptedRows} 条`)
|
ElMessage.success(`解析完成,共 ${res.groupCount || 0} 组 / ${res.acceptedRows} 条`)
|
||||||
} catch (e) {
|
} catch (e) {
|
||||||
ElMessage.error(e instanceof Error ? e.message : '解析失败')
|
ElMessage.error(e instanceof Error ? e.message : '解析失败')
|
||||||
@@ -465,7 +463,7 @@ function scheduleNextPoll(immediate = false) {
|
|||||||
if (disposed) return
|
if (disposed) return
|
||||||
if (!pollingTaskIds.value.length && !pendingFileTaskIds.value.length) return
|
if (!pollingTaskIds.value.length && !pendingFileTaskIds.value.length) return
|
||||||
await refreshTaskProgress()
|
await refreshTaskProgress()
|
||||||
if (!disposed && (pollingTaskIds.value.length || pendingFileTaskIds.value.length)) {
|
if (!disposed && pollTimer.value == null && (pollingTaskIds.value.length || pendingFileTaskIds.value.length)) {
|
||||||
pollTimer.value = timers.setTimeout('task-poll', run, getTaskPollIntervalMs())
|
pollTimer.value = timers.setTimeout('task-poll', run, getTaskPollIntervalMs())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user