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 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 rowError = new AtomicReference<>(); boolean[] concurrentSkip = {false}; try { rowTxTemplate.executeWithoutResult(status -> { try { // 提交前重查,缩小并发重复建物料的窗口 List 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 rowDOs = commandRowMapper.selectListByCommandId(command.getCommandId()); List 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 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> rows = new ArrayList<>(); for (MdmAgentImportPreflightReqVO.Row row : reqVO.getRows()) { Map 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); } }