hz/java-servers
1
0
Fork 0
mirror of https://codeup.aliyun.com/67762337eccfc218f6110e0e/per-boe/java-servers.git synced 2025-12-06 09:26:48 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: 增加重新上传的批处理

Browse Source
This commit is contained in:
liu.zixi
2025-12-01 19:44:01 +08:00
parent 49d3ad5999
commit 38c2784f51
3 changed files with 73 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
servers/boe-server-all/src/main/java/com/xboe/module/boecase/api/CaseUploadTaskApi.java
View File

@@ -1,5 +1,6 @@
package com.xboe.module.boecase.api;
import com.xboe.module.boecase.task.CaseUploadTask;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.StringRedisTemplate;
@@ -20,14 +21,21 @@ public class CaseUploadTaskApi {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private static final String CASE_UPLOAD_LAST_ID_KEY = "case:upload:last:id";
/**
* 清除处理位置标记，使下次任务从头开始执行
*/
@PostMapping("/reset")
public void resetLastProcessedId() {
stringRedisTemplate.delete(CASE_UPLOAD_LAST_ID_KEY);
stringRedisTemplate.delete(CaseUploadTask.CASE_UPLOAD_LAST_ID_KEY);
log.info("已清除上次处理位置标记");
}
/**
* 清除处理位置标记，使下次任务从头开始执行
*/
@PostMapping("/reload/reset")
public void resetReloadProcessedId() {
stringRedisTemplate.delete(CaseUploadTask.CASE_RELOAD_LAST_ID_KEY);
log.info("已清除上次处理位置标记");
}
}

3
servers/boe-server-all/src/main/java/com/xboe/module/boecase/service/impl/CaseKnowledgeServiceImpl.java
View File

@@ -652,7 +652,8 @@ public class CaseKnowledgeServiceImpl implements ICaseKnowledgeService {
.findList(CaseDocumentLog.class, 1,
OrderCondition.desc("sysCreateTime"),
FieldFilters.eq("caseId", caseId),
FieldFilters.eq("requestUrl", CaseAiConstants.CASE_DOC_UPLOAD_INTERFACE_NAME));
FieldFilters.eq("requestUrl", CaseAiConstants.CASE_DOC_UPLOAD_INTERFACE_NAME),
FieldFilters.eq("caseStatus", CaseDocumentLogCaseStatusEnum.SUCCESS.getCode()));
if (logList.isEmpty()) {
log.info("删除案例文档失败，未找到相关的日志记录，caseId: {}", caseId);

67
servers/boe-server-all/src/main/java/com/xboe/module/boecase/task/CaseUploadTask.java
View File

@@ -1,6 +1,9 @@
package com.xboe.module.boecase.task;
import com.xboe.common.OrderCondition;
import com.xboe.constants.CaseAiConstants;
import com.xboe.core.orm.FieldFilters;
import com.xboe.core.orm.QueryBuilder;
import com.xboe.enums.CaseDocumentLogCaseStatusEnum;
import com.xboe.enums.CaseDocumentLogOptStatusEnum;
import com.xboe.enums.CaseDocumentLogOptTypeEnum;
@@ -43,7 +46,39 @@ public class CaseUploadTask {
@Autowired
private StringRedisTemplate stringRedisTemplate;
private static final String CASE_UPLOAD_LAST_ID_KEY = "case:upload:last:id";
public static final String CASE_UPLOAD_LAST_ID_KEY = "case:upload:last:id";
public static final String CASE_RELOAD_LAST_ID_KEY = "case:reload:last:id";
@XxlJob("reloadJob")
public void reloadJob() {
String currentLastId = null;
try {
// 从Redis获取上次处理的最后一条记录ID
String lastProcessedId = stringRedisTemplate.opsForValue().get(CASE_RELOAD_LAST_ID_KEY);
// 查询需要重新加载的案例
List<CaseDocumentLog> logsToReload = listToReload(lastProcessedId);
if (logsToReload.isEmpty()) {
return;
}
currentLastId = logsToReload.get(logsToReload.size() - 1).getId();
for (CaseDocumentLog log : logsToReload) {
String caseId = log.getCaseId();
Cases cases = casesDao.get(caseId);
if (cases != null && StringUtils.isNotBlank(cases.getFilePath())) {
// 更新
caseAiDocumentAsyncHandler.process(CaseDocumentLogOptTypeEnum.UPDATE, cases);
}
}
} catch (Exception e) {
log.error("[reload]执行重新上传任务时发生异常", e);
} finally {
if (currentLastId != null) {
stringRedisTemplate.opsForValue().set(CASE_RELOAD_LAST_ID_KEY, currentLastId);
log.info("[reload] 已重新上传案例，最后一条caseDocumentLogId 已更新为: {}", currentLastId);
}
}
}
@XxlJob("oldDataUploadJob")
public void oldDataUploadJob() {
@@ -142,25 +177,43 @@ public class CaseUploadTask {
* @return 案例列表
*/
private List<Cases> findCasesToProcess(String lastProcessedId) {
com.xboe.core.orm.QueryBuilder queryBuilder = com.xboe.core.orm.QueryBuilder.from(Cases.class);
queryBuilder.addFilter(com.xboe.core.orm.FieldFilters.eq("deleted", false));
QueryBuilder queryBuilder = QueryBuilder.from(Cases.class);
queryBuilder.addFilter(FieldFilters.eq("deleted", false));
// 只处理有文件路径的案例
queryBuilder.addFilter(com.xboe.core.orm.FieldFilters.isNotNull("filePath"));
queryBuilder.addFilter(com.xboe.core.orm.FieldFilters.ne("filePath", ""));
queryBuilder.addFilter(FieldFilters.isNotNull("filePath"));
queryBuilder.addFilter(FieldFilters.ne("filePath", ""));
// 如果有上次处理的ID，则从该ID之后开始查询
if (lastProcessedId != null && !lastProcessedId.isEmpty()) {
queryBuilder.addFilter(com.xboe.core.orm.FieldFilters.gt("id", lastProcessedId));
queryBuilder.addFilter(FieldFilters.gt("id", lastProcessedId));
}
// 按创建时间升序排序
queryBuilder.addOrder(com.xboe.common.OrderCondition.asc("id"));
queryBuilder.addOrder(OrderCondition.asc("id"));
// 限制每次处理的数量，避免一次性处理太多数据
queryBuilder.setPageSize(100);
return casesDao.findList(queryBuilder.builder());
}
/**
* 获取需要重新加载的案例
* @param lastProcessedId
* @return
*/
private List<CaseDocumentLog> listToReload(String lastProcessedId) {
QueryBuilder queryBuilder = QueryBuilder.from(Cases.class);
queryBuilder.addFilter(FieldFilters.eq("deleted", false));
queryBuilder.addFilter(FieldFilters.eq("requestUrl", CaseAiConstants.CASE_DOC_UPLOAD_INTERFACE_NAME));
queryBuilder.addFilter(FieldFilters.eq("caseStatus", CaseDocumentLogCaseStatusEnum.SUCCESS.getCode()));
if (lastProcessedId != null && !lastProcessedId.isEmpty()) {
queryBuilder.addFilter(FieldFilters.gt("id", lastProcessedId));
}
queryBuilder.addOrder(OrderCondition.asc("id"));
queryBuilder.setPageSize(100);
return caseDocumentLogDao.findList(queryBuilder.builder());
}
private void fixOnLastCase(String currentLastId) {
stringRedisTemplate.opsForValue().set(CASE_UPLOAD_LAST_ID_KEY, currentLastId);
log.info("已处理案例，最后一条记录ID已更新为: {}", currentLastId);