package cn.iocoder.yudao.module.mdm.service.agent;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.framework.common.exception.ServiceException;
|
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportCommitReqVO;
|
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportPreflightReqVO;
|
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportPreflightRespVO;
|
import cn.iocoder.yudao.module.mdm.controller.admin.agent.vo.MdmAgentImportReceiptRespVO;
|
import cn.iocoder.yudao.module.mdm.controller.admin.item.vo.MdmItemImportExcelVO;
|
import cn.iocoder.yudao.module.mdm.controller.admin.item.vo.MdmItemSaveReqVO;
|
import cn.iocoder.yudao.module.mdm.dal.dataobject.agent.MdmAgentImportCommandDO;
|
import cn.iocoder.yudao.module.mdm.dal.dataobject.agent.MdmAgentImportCommandRowDO;
|
import cn.iocoder.yudao.module.mdm.dal.dataobject.item.MdmItemDO;
|
import cn.iocoder.yudao.module.mdm.dal.mysql.agent.MdmAgentImportCommandMapper;
|
import cn.iocoder.yudao.module.mdm.dal.mysql.agent.MdmAgentImportCommandRowMapper;
|
import cn.iocoder.yudao.module.mdm.service.item.MdmItemImportContext;
|
import cn.iocoder.yudao.module.mdm.service.item.MdmItemRowResolution;
|
import cn.iocoder.yudao.module.mdm.service.item.MdmItemService;
|
import cn.iocoder.yudao.module.system.framework.agent.AgentBridgeException;
|
import cn.iocoder.yudao.module.system.framework.agent.AgentPrincipal;
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import jakarta.annotation.Resource;
|
import org.springframework.dao.DuplicateKeyException;
|
import org.springframework.http.HttpStatus;
|
import org.springframework.stereotype.Service;
|
import org.springframework.transaction.PlatformTransactionManager;
|
import org.springframework.transaction.support.TransactionTemplate;
|
import org.springframework.validation.annotation.Validated;
|
|
import java.security.MessageDigest;
|
import java.security.NoSuchAlgorithmException;
|
import java.time.Duration;
|
import java.time.LocalDateTime;
|
import java.util.*;
|
import java.util.concurrent.atomic.AtomicReference;
|
|
/**
|
* 物料导入幂等提交：同一 commandId/idempotencyKey 重复提交返回原回执，绝不重复建物料。
|
*/
|
@Service
|
@Validated
|
public class MdmAgentImportCommandService {
|
|
public static final String STATUS_RUNNING = "RUNNING";
|
public static final String STATUS_SUCCEEDED = "SUCCEEDED";
|
public static final String STATUS_PARTIAL_FAILED = "PARTIAL_FAILED";
|
public static final String STATUS_FAILED = "FAILED";
|
public static final String STATUS_UNKNOWN = "UNKNOWN";
|
|
private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
|
private static final Duration LEASE_DURATION = Duration.ofMinutes(5);
|
|
@Resource
|
private MdmAgentImportPreflightService preflightService;
|
@Resource
|
private MdmItemService itemService;
|
@Resource
|
private MdmAgentImportPolicy policy;
|
@Resource
|
private MdmAgentImportCommandMapper commandMapper;
|
@Resource
|
private MdmAgentImportCommandRowMapper commandRowMapper;
|
@Resource
|
private PlatformTransactionManager transactionManager;
|
|
public MdmAgentImportReceiptRespVO commit(MdmAgentImportCommitReqVO reqVO, AgentPrincipal principal) {
|
if (!policy.isWriteEnabled()) {
|
throw new AgentBridgeException(HttpStatus.SERVICE_UNAVAILABLE, "WRITE_DISABLED",
|
"智能体写入开关未打开，提交已拒绝");
|
}
|
if (Boolean.TRUE.equals(reqVO.getSyncMes()) && !policy.isAllowMesSync()) {
|
throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "MES_SYNC_DISABLED",
|
"服务未开放同步 MES，请关闭 syncMes 后重试");
|
}
|
String requestDigest = buildRequestDigest(reqVO);
|
Long tenantId = principal.tenantId();
|
Long ownerUserId = principal.userId();
|
|
// 1. 幂等命中：命令 ID 优先，其次幂等键
|
MdmAgentImportCommandDO existing = commandMapper.selectByCommandId(tenantId, ownerUserId, reqVO.getCommandId());
|
if (existing != null) {
|
if (!existing.getRequestDigest().equals(requestDigest)) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "COMMAND_ID_CONFLICT",
|
"同一命令 ID 携带了不同的请求内容，请检查任务状态后重试");
|
}
|
return toReceipt(existing);
|
}
|
existing = commandMapper.selectByIdempotencyKey(tenantId, ownerUserId, reqVO.getIdempotencyKey());
|
if (existing != null) {
|
if (!existing.getRequestDigest().equals(requestDigest)) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
|
"同一幂等键携带了不同的请求内容，请检查任务状态后重试");
|
}
|
return toReceipt(existing);
|
}
|
|
// 2. 重新预检并比对计划摘要，防止确认后数据已变化
|
MdmAgentImportPreflightRespVO preflight = preflightService.preflight(toPreflightRequest(reqVO), principal);
|
if (!reqVO.getPreflightHash().equalsIgnoreCase(preflight.getPreflightHash())) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "PLAN_STALE",
|
"预检计划已变化（编码状态或主数据被改动），请重新预检并确认", false,
|
Map.of("preflight", preflight));
|
}
|
if (!Boolean.TRUE.equals(preflight.getCanCommit())) {
|
throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "PREFLIGHT_BLOCKED",
|
"预检存在阻断项，不能提交", false, Map.of("preflight", preflight));
|
}
|
|
// 3. 登记命令（唯一键兜住并发重复提交）
|
MdmAgentImportCommandDO command = MdmAgentImportCommandDO.builder()
|
.commandId(reqVO.getCommandId())
|
.idempotencyKey(reqVO.getIdempotencyKey())
|
.requestDigest(requestDigest)
|
.tenantId(tenantId)
|
.ownerUserId(ownerUserId)
|
.sessionId(reqVO.getSessionId())
|
.attachmentId(reqVO.getAttachmentId())
|
.sourceSha256(reqVO.getSourceSha256().toLowerCase(Locale.ROOT))
|
.preflightHash(reqVO.getPreflightHash().toLowerCase(Locale.ROOT))
|
.proposalHash(reqVO.getProposalHash())
|
.syncMes(Boolean.TRUE.equals(reqVO.getSyncMes()))
|
.status(STATUS_RUNNING)
|
.totalCount(reqVO.getRows().size())
|
.successCount(0).failedCount(0).skippedCount(0)
|
.leaseExpiresAt(LocalDateTime.now().plus(LEASE_DURATION))
|
.build();
|
try {
|
commandMapper.insert(command);
|
} catch (DuplicateKeyException ex) {
|
MdmAgentImportCommandDO raced = commandMapper.selectByCommandId(tenantId, ownerUserId, reqVO.getCommandId());
|
if (raced == null) {
|
raced = commandMapper.selectByIdempotencyKey(tenantId, ownerUserId, reqVO.getIdempotencyKey());
|
}
|
if (raced == null) {
|
throw ex;
|
}
|
if (!raced.getRequestDigest().equals(requestDigest)) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
|
"同一命令/幂等键携带了不同的请求内容，请检查任务状态后重试");
|
}
|
return toReceipt(raced);
|
}
|
|
// 4. 逐行独立事务执行；任一行失败不影响其他行
|
int success = 0;
|
int failed = 0;
|
int skipped = 0;
|
TransactionTemplate rowTxTemplate = new TransactionTemplate(transactionManager);
|
List<MdmItemImportExcelVO> excelRows = reqVO.getRows().stream()
|
.map(MdmAgentImportCommandService::toExcelVO).toList();
|
MdmItemImportContext context = itemService.buildImportContext(excelRows);
|
try {
|
for (int index = 0; index < reqVO.getRows().size(); index++) {
|
MdmAgentImportPreflightReqVO.Row row = reqVO.getRows().get(index);
|
MdmAgentImportPreflightRespVO.Row planned = preflight.getRows().get(index);
|
if ("SKIP".equals(planned.getAction())) {
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("SKIP").status("SKIPPED")
|
.targetItemId(planned.getTargetItemId())
|
.retryable(false).build());
|
skipped++;
|
continue;
|
}
|
if (!"CREATE".equals(planned.getAction())) {
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("CREATE").status("FAILED")
|
.errorCode("PREFLIGHT_BLOCKED").errorMessage("该行预检未通过").retryable(false).build());
|
failed++;
|
continue;
|
}
|
MdmItemRowResolution resolution = itemService.resolveImportRow(excelRows.get(index), context);
|
if (!resolution.isSuccess()) {
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("CREATE").status("FAILED")
|
.errorCode("ROW_INVALID").errorMessage(trim(resolution.failureMessage(), 512))
|
.retryable(false).build());
|
failed++;
|
continue;
|
}
|
AtomicReference<Throwable> rowError = new AtomicReference<>();
|
boolean[] concurrentSkip = {false};
|
try {
|
rowTxTemplate.executeWithoutResult(status -> {
|
try {
|
// 提交前重查，缩小并发重复建物料的窗口
|
List<MdmItemDO> current = itemService.getItemListByCode(row.getCode().trim());
|
if (!current.isEmpty()) {
|
concurrentSkip[0] = true;
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("SKIP").status("SKIPPED")
|
.targetItemId(current.get(0).getId()).retryable(false).build());
|
return;
|
}
|
MdmItemSaveReqVO saveReqVO = resolution.saveReqVO();
|
saveReqVO.setSyncMes(Boolean.TRUE.equals(reqVO.getSyncMes()));
|
Long itemId = itemService.createItem(saveReqVO);
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("CREATE").status("SUCCESS").targetItemId(itemId)
|
.retryable(false).build());
|
} catch (Throwable t) {
|
rowError.set(t);
|
throw t;
|
}
|
});
|
if (concurrentSkip[0]) {
|
skipped++;
|
} else {
|
success++;
|
}
|
} catch (Exception ex) {
|
Throwable cause = rowError.get() != null ? rowError.get() : ex;
|
String message = cause instanceof ServiceException
|
? ((ServiceException) cause).getMessage() : cause.getMessage();
|
String errorCode = cause instanceof ServiceException ? "ITEM_CREATE_REJECTED" : "ITEM_CREATE_FAILED";
|
commandRowMapper.insert(baseRow(command.getCommandId(), row)
|
.action("CREATE").status("FAILED")
|
.errorCode(errorCode).errorMessage(trim(message, 512))
|
.retryable(!(cause instanceof ServiceException)).build());
|
failed++;
|
}
|
}
|
} catch (Exception ex) {
|
// 基础设施异常中断：保留已完成行回执，命令标记 UNKNOWN 供对账
|
command.setStatus(STATUS_UNKNOWN);
|
command.setErrorCode("COMMIT_INTERRUPTED");
|
command.setErrorMessage(trim(ex.getMessage(), 512));
|
command.setSuccessCount(success);
|
command.setFailedCount(failed);
|
command.setSkippedCount(skipped);
|
commandMapper.updateById(command);
|
return toReceipt(command);
|
}
|
|
command.setSuccessCount(success);
|
command.setFailedCount(failed);
|
command.setSkippedCount(skipped);
|
command.setStatus(failed == 0 ? STATUS_SUCCEEDED
|
: (success + skipped == 0 ? STATUS_FAILED : STATUS_PARTIAL_FAILED));
|
command.setLeaseExpiresAt(LocalDateTime.now());
|
commandMapper.updateById(command);
|
return toReceipt(command);
|
}
|
|
public MdmAgentImportReceiptRespVO getCommand(String commandId, AgentPrincipal principal) {
|
MdmAgentImportCommandDO command = commandMapper.selectByCommandId(
|
principal.tenantId(), principal.userId(), commandId);
|
if (command == null) {
|
throw new AgentBridgeException(HttpStatus.NOT_FOUND, "COMMAND_NOT_FOUND", "导入命令不存在");
|
}
|
return toReceipt(command);
|
}
|
|
private MdmAgentImportReceiptRespVO toReceipt(MdmAgentImportCommandDO command) {
|
String status = command.getStatus();
|
if (STATUS_RUNNING.equals(status) && command.getLeaseExpiresAt() != null
|
&& command.getLeaseExpiresAt().isBefore(LocalDateTime.now())) {
|
// 租约过期说明执行进程可能已中断；不改库，仅如实告知调用方结果未知
|
status = STATUS_UNKNOWN;
|
}
|
List<MdmAgentImportCommandRowDO> rowDOs = commandRowMapper.selectListByCommandId(command.getCommandId());
|
List<MdmAgentImportReceiptRespVO.Row> rows = rowDOs.stream()
|
.map(row -> MdmAgentImportReceiptRespVO.Row.builder()
|
.candidateId(row.getCandidateId())
|
.rowNo(row.getSourceRow())
|
.code(row.getCode())
|
.name(row.getName())
|
.action(row.getAction())
|
.status(row.getStatus())
|
.targetItemId(row.getTargetItemId())
|
.targetVersion(row.getTargetVersion())
|
.errorCode(row.getErrorCode())
|
.errorMessage(row.getErrorMessage())
|
.retryable(row.getRetryable())
|
.build())
|
.toList();
|
int success = (int) rowDOs.stream().filter(row -> "SUCCESS".equals(row.getStatus())).count();
|
int failed = (int) rowDOs.stream().filter(row -> "FAILED".equals(row.getStatus())).count();
|
int skipped = (int) rowDOs.stream().filter(row -> "SKIPPED".equals(row.getStatus())).count();
|
int unknown = (int) rowDOs.stream().filter(row -> "UNKNOWN".equals(row.getStatus())).count();
|
return MdmAgentImportReceiptRespVO.builder()
|
.commandId(command.getCommandId())
|
.idempotencyKey(command.getIdempotencyKey())
|
.status(status)
|
.attachmentId(command.getAttachmentId())
|
.sessionId(command.getSessionId())
|
.sourceSha256(command.getSourceSha256())
|
.preflightHash(command.getPreflightHash())
|
.proposalHash(command.getProposalHash())
|
.summary(MdmAgentImportReceiptRespVO.Summary.builder()
|
.total(command.getTotalCount())
|
.success(success).failed(failed).skipped(skipped).unknown(unknown)
|
.build())
|
.errorCode(command.getErrorCode())
|
.errorMessage(command.getErrorMessage())
|
.rows(rows)
|
.createTime(command.getCreateTime())
|
.completedTime(STATUS_RUNNING.equals(command.getStatus()) ? null : command.getUpdateTime())
|
.build();
|
}
|
|
private static MdmAgentImportCommandRowDO.MdmAgentImportCommandRowDOBuilder baseRow(
|
String commandId, MdmAgentImportPreflightReqVO.Row row) {
|
return MdmAgentImportCommandRowDO.builder()
|
.commandId(commandId)
|
.candidateId(row.getCandidateId())
|
.sourceRow(row.getRowNo())
|
.code(row.getCode())
|
.name(row.getName());
|
}
|
|
private static MdmItemImportExcelVO toExcelVO(MdmAgentImportPreflightReqVO.Row row) {
|
return MdmItemImportExcelVO.builder()
|
.code(row.getCode())
|
.name(row.getName())
|
.specification(row.getSpecification())
|
.categoryCode(row.getCategoryCode())
|
.unitMeasureCode(row.getUnitMeasureCode())
|
.build();
|
}
|
|
private static MdmAgentImportPreflightReqVO toPreflightRequest(MdmAgentImportCommitReqVO reqVO) {
|
MdmAgentImportPreflightReqVO preflightReqVO = new MdmAgentImportPreflightReqVO();
|
preflightReqVO.setAttachmentId(reqVO.getAttachmentId());
|
preflightReqVO.setSessionId(reqVO.getSessionId());
|
preflightReqVO.setSourceSha256(reqVO.getSourceSha256());
|
preflightReqVO.setProposalRevision(reqVO.getProposalRevision());
|
preflightReqVO.setSyncMes(reqVO.getSyncMes());
|
preflightReqVO.setRows(reqVO.getRows());
|
return preflightReqVO;
|
}
|
|
/**
|
* 请求摘要只包含业务内容，不含 commandId/idempotencyKey，便于识别“同键不同内容”。
|
*/
|
private static String buildRequestDigest(MdmAgentImportCommitReqVO reqVO) {
|
Map<String, Object> canonical = new LinkedHashMap<>();
|
canonical.put("attachment_id", reqVO.getAttachmentId());
|
canonical.put("session_id", reqVO.getSessionId());
|
canonical.put("source_sha256", StrUtil.nullToEmpty(reqVO.getSourceSha256()).toLowerCase(Locale.ROOT));
|
canonical.put("preflight_hash", StrUtil.nullToEmpty(reqVO.getPreflightHash()).toLowerCase(Locale.ROOT));
|
canonical.put("proposal_hash", StrUtil.nullToEmpty(reqVO.getProposalHash()));
|
canonical.put("proposal_revision", StrUtil.nullToEmpty(reqVO.getProposalRevision()));
|
canonical.put("sync_mes", Boolean.TRUE.equals(reqVO.getSyncMes()));
|
List<Map<String, Object>> rows = new ArrayList<>();
|
for (MdmAgentImportPreflightReqVO.Row row : reqVO.getRows()) {
|
Map<String, Object> item = new LinkedHashMap<>();
|
item.put("candidate_id", row.getCandidateId());
|
item.put("row_no", row.getRowNo());
|
item.put("code", StrUtil.nullToEmpty(row.getCode()));
|
item.put("name", StrUtil.nullToEmpty(row.getName()));
|
item.put("specification", StrUtil.nullToEmpty(row.getSpecification()));
|
item.put("category_code", StrUtil.nullToEmpty(row.getCategoryCode()));
|
item.put("unit_measure_code", StrUtil.nullToEmpty(row.getUnitMeasureCode()));
|
rows.add(item);
|
}
|
canonical.put("rows", rows);
|
try {
|
byte[] payload = OBJECT_MAPPER.writeValueAsBytes(canonical);
|
return HexFormat.of().formatHex(MessageDigest.getInstance("SHA-256").digest(payload));
|
} catch (NoSuchAlgorithmException ex) {
|
throw new IllegalStateException("SHA-256 不可用", ex);
|
} catch (JsonProcessingException ex) {
|
throw new IllegalStateException("请求摘要生成失败", ex);
|
}
|
}
|
|
private static String trim(String value, int maxLength) {
|
if (value == null) {
|
return null;
|
}
|
return value.length() <= maxLength ? value : value.substring(0, maxLength);
|
}
|
}
|
