package cn.iocoder.yudao.module.system.service.agent.imports;
|
|
import cn.hutool.core.util.StrUtil;
|
import cn.iocoder.yudao.framework.common.exception.ServiceException;
|
import cn.iocoder.yudao.framework.security.core.service.SecurityFrameworkService;
|
import cn.iocoder.yudao.module.system.api.agent.imports.*;
|
import cn.iocoder.yudao.module.system.dal.dataobject.agent.AgentImportCommandDO;
|
import cn.iocoder.yudao.module.system.dal.dataobject.agent.AgentImportCommandRowDO;
|
import cn.iocoder.yudao.module.system.dal.mysql.agent.AgentImportCommandMapper;
|
import cn.iocoder.yudao.module.system.dal.mysql.agent.AgentImportCommandRowMapper;
|
import cn.iocoder.yudao.module.system.framework.agent.AgentBridgeException;
|
import cn.iocoder.yudao.module.system.framework.agent.AgentPrincipal;
|
import cn.iocoder.yudao.module.system.service.storage.SystemAgentImportBridgeService;
|
import com.fasterxml.jackson.core.JsonProcessingException;
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
import jakarta.annotation.Resource;
|
import org.springframework.beans.factory.ObjectProvider;
|
import org.springframework.beans.factory.annotation.Value;
|
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.nio.charset.StandardCharsets;
|
import java.security.MessageDigest;
|
import java.security.NoSuchAlgorithmException;
|
import java.time.Duration;
|
import java.time.LocalDateTime;
|
import java.util.*;
|
|
/**
|
* 通用业务导入编排：适配器只负责单行校验/写入，本类统一处理权限、幂等、逐行事务与回执。
|
*/
|
@Service
|
@Validated
|
public class SystemAgentGenericImportService {
|
|
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 ObjectProvider<AgentImportModuleAdapter> adapterProvider;
|
@Resource
|
private SecurityFrameworkService securityFrameworkService;
|
@Resource
|
private SystemAgentImportBridgeService attachmentService;
|
@Resource
|
private AgentImportCommandMapper commandMapper;
|
@Resource
|
private AgentImportCommandRowMapper commandRowMapper;
|
@Resource
|
private PlatformTransactionManager transactionManager;
|
|
@Value("${yudao.agent.bridge.write-enabled:false}")
|
private boolean writeEnabled;
|
|
@Value("${yudao.agent.bridge.max-rows:5000}")
|
private int maxRows;
|
|
public List<Map<String, String>> modules() {
|
return adapters().values().stream()
|
.map(adapter -> Map.of("moduleId", adapter.moduleId(), "permission", adapter.permission()))
|
.toList();
|
}
|
|
public AgentImportPreflightRespVO preflight(String moduleId, AgentImportPreflightReqVO request,
|
AgentPrincipal principal) {
|
AgentImportModuleAdapter adapter = requireAdapter(moduleId);
|
requirePermission(adapter);
|
validateAttachment(request.getAttachmentId(), request.getSourceSha256());
|
validateRows(request);
|
AgentImportPreflightRespVO result = adapter.preflight(principal, request);
|
validatePlan(result, request);
|
return result;
|
}
|
|
public AgentImportReceiptRespVO commit(String moduleId, AgentImportCommitReqVO request,
|
AgentPrincipal principal) {
|
if (!writeEnabled) {
|
throw new AgentBridgeException(HttpStatus.SERVICE_UNAVAILABLE, "WRITE_DISABLED",
|
"智能体写入开关未打开，提交已拒绝");
|
}
|
AgentImportModuleAdapter adapter = requireAdapter(moduleId);
|
requirePermission(adapter);
|
validateAttachment(request.getAttachmentId(), request.getSourceSha256());
|
validateRows(request);
|
|
Long tenantId = principal.tenantId();
|
Long ownerUserId = principal.userId();
|
String digest = requestDigest(moduleId, request);
|
|
// 1. 幂等命中：命令 ID 优先，其次幂等键
|
AgentImportCommandDO existing = commandMapper.selectByCommandId(
|
tenantId, ownerUserId, moduleId, request.getCommandId());
|
if (existing != null) {
|
requireSameDigest(existing, digest);
|
return toReceipt(existing);
|
}
|
existing = commandMapper.selectByIdempotencyKey(
|
tenantId, ownerUserId, moduleId, request.getIdempotencyKey());
|
if (existing != null) {
|
requireSameDigest(existing, digest);
|
return toReceipt(existing);
|
}
|
|
// 2. 重新预检并比对计划摘要，防止确认后主数据变化
|
AgentImportPreflightRespVO plan = adapter.preflight(principal, request);
|
validatePlan(plan, request);
|
if (!StrUtil.equalsIgnoreCase(request.getPreflightHash(), plan.getPreflightHash())) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "PLAN_STALE",
|
"预检计划已变化，请重新预检并确认", false, Map.of("preflight", plan));
|
}
|
if (!Boolean.TRUE.equals(plan.getCanCommit())) {
|
throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "PREFLIGHT_BLOCKED",
|
"预检存在阻断项，不能提交", false, Map.of("preflight", plan));
|
}
|
|
// 3. 登记命令（唯一键兜住并发重复提交）
|
AgentImportCommandDO command = AgentImportCommandDO.builder()
|
.commandId(request.getCommandId())
|
.idempotencyKey(request.getIdempotencyKey())
|
.requestDigest(digest)
|
.moduleId(moduleId)
|
.tenantId(tenantId)
|
.ownerUserId(ownerUserId)
|
.sessionId(request.getSessionId())
|
.attachmentId(request.getAttachmentId())
|
.sourceSha256(request.getSourceSha256().toLowerCase(Locale.ROOT))
|
.preflightHash(request.getPreflightHash().toLowerCase(Locale.ROOT))
|
.proposalHash(request.getProposalHash())
|
.status(STATUS_RUNNING)
|
.totalCount(request.getRows().size())
|
.successCount(0).failedCount(0).skippedCount(0)
|
.leaseExpiresAt(LocalDateTime.now().plus(LEASE_DURATION))
|
.build();
|
try {
|
commandMapper.insert(command);
|
} catch (DuplicateKeyException ex) {
|
AgentImportCommandDO raced = commandMapper.selectByCommandId(
|
tenantId, ownerUserId, moduleId, request.getCommandId());
|
if (raced == null) {
|
raced = commandMapper.selectByIdempotencyKey(
|
tenantId, ownerUserId, moduleId, request.getIdempotencyKey());
|
}
|
if (raced == null) {
|
throw ex;
|
}
|
requireSameDigest(raced, digest);
|
return toReceipt(raced);
|
}
|
|
// 4. 逐行独立事务执行；单行失败不影响其他行
|
int success = 0;
|
int failed = 0;
|
int skipped = 0;
|
TransactionTemplate rowTx = new TransactionTemplate(transactionManager);
|
try {
|
for (int index = 0; index < request.getRows().size(); index++) {
|
AgentImportPreflightReqVO.Row row = request.getRows().get(index);
|
AgentImportPreflightRespVO.Row planned = plan.getRows().get(index);
|
if ("SKIP".equals(planned.getAction())) {
|
commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
|
.action("SKIP").status("SKIPPED")
|
.targetItemId(planned.getTargetId()).retryable(false).build());
|
skipped++;
|
continue;
|
}
|
if (!"CREATE".equals(planned.getAction())) {
|
commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
|
.action("CREATE").status("FAILED")
|
.errorCode("PREFLIGHT_BLOCKED").errorMessage("该行预检未通过").retryable(false).build());
|
failed++;
|
continue;
|
}
|
try {
|
AgentImportRowReceipt receipt = rowTx.execute(status ->
|
adapter.commitRow(principal, request, planned, row));
|
if (receipt == null) {
|
throw new IllegalStateException("适配器未返回回执");
|
}
|
commandRowMapper.insert(baseRow(moduleId, request.getCommandId(), row)
|
.action(StrUtil.blankToDefault(receipt.action(), "CREATE"))
|
.status("SUCCESS")
|
.targetItemId(receipt.targetId())
|
.targetVersion(receipt.targetVersion())
|
.code(StrUtil.blankToDefault(receipt.code(), row.getValues().get("code")))
|
.name(StrUtil.blankToDefault(receipt.name(), row.getValues().get("name")))
|
.retryable(false).build());
|
success++;
|
} catch (Exception ex) {
|
recordFailure(moduleId, request.getCommandId(), row, ex);
|
failed++;
|
}
|
}
|
} catch (Exception ex) {
|
command.setStatus(STATUS_UNKNOWN);
|
command.setErrorCode("COMMIT_INTERRUPTED");
|
command.setErrorMessage(trim(ex.getMessage(), 512));
|
command.setSuccessCount(success);
|
command.setFailedCount(failed);
|
command.setSkippedCount(skipped);
|
command.setUpdateTime(LocalDateTime.now());
|
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());
|
command.setUpdateTime(LocalDateTime.now());
|
commandMapper.updateById(command);
|
return toReceipt(command);
|
}
|
|
public AgentImportReceiptRespVO getCommand(String moduleId, String commandId, AgentPrincipal principal) {
|
requireAdapter(moduleId);
|
AgentImportCommandDO command = commandMapper.selectByCommandId(
|
principal.tenantId(), principal.userId(), moduleId, commandId);
|
if (command == null) {
|
throw new AgentBridgeException(HttpStatus.NOT_FOUND, "COMMAND_NOT_FOUND", "导入命令不存在");
|
}
|
return toReceipt(command);
|
}
|
|
private void recordFailure(String moduleId, String commandId, AgentImportPreflightReqVO.Row row,
|
Exception ex) {
|
String errorCode = "ROW_WRITE_FAILED";
|
boolean retryable = true;
|
String message = ex.getMessage();
|
if (ex instanceof AgentBridgeException bridge) {
|
errorCode = bridge.getErrorCode();
|
retryable = bridge.isRetryable();
|
message = bridge.getReason();
|
} else if (ex instanceof ServiceException service) {
|
errorCode = "ROW_WRITE_REJECTED";
|
retryable = false;
|
message = service.getMessage();
|
}
|
commandRowMapper.insert(baseRow(moduleId, commandId, row)
|
.action("CREATE").status("FAILED")
|
.errorCode(errorCode).errorMessage(trim(message, 512)).retryable(retryable).build());
|
}
|
|
private AgentImportPreflightRespVO.Row requirePlanRow(AgentImportPreflightRespVO plan, int index) {
|
if (plan.getRows() == null || index >= plan.getRows().size()) {
|
throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
|
"适配器返回的预检行数与请求不一致");
|
}
|
return plan.getRows().get(index);
|
}
|
|
private void validatePlan(AgentImportPreflightRespVO plan, AgentImportPreflightReqVO request) {
|
if (plan == null || StrUtil.isBlank(plan.getPreflightHash())) {
|
throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
|
"适配器未返回预检摘要");
|
}
|
if (plan.getRows() == null || plan.getRows().size() != request.getRows().size()) {
|
throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
|
"适配器返回的预检行数与请求不一致");
|
}
|
for (int index = 0; index < plan.getRows().size(); index++) {
|
if (!request.getRows().get(index).getCandidateId()
|
.equals(plan.getRows().get(index).getCandidateId())) {
|
throw new AgentBridgeException(HttpStatus.BAD_GATEWAY, "ADAPTER_CONTRACT_INVALID",
|
"适配器返回的预检行顺序与请求不一致");
|
}
|
}
|
}
|
|
private void validateAttachment(Long attachmentId, String sourceSha256) {
|
String actual = attachmentService.sha256OfAttachment(attachmentId);
|
if (!StrUtil.equalsIgnoreCase(actual, sourceSha256)) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "SOURCE_CHANGED",
|
"附件内容已变化，请重新上传后再试", true, null);
|
}
|
}
|
|
private void validateRows(AgentImportPreflightReqVO request) {
|
if (request.getRows() == null || request.getRows().isEmpty()) {
|
throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "ROWS_EMPTY", "没有可导入的数据行");
|
}
|
if (request.getRows().size() > maxRows) {
|
throw new AgentBridgeException(HttpStatus.UNPROCESSABLE_ENTITY, "ROWS_TOO_MANY",
|
"单次导入超过 " + maxRows + " 行，请拆分文件");
|
}
|
}
|
|
private void requirePermission(AgentImportModuleAdapter adapter) {
|
if (!securityFrameworkService.hasPermission(adapter.permission())) {
|
throw new AgentBridgeException(HttpStatus.FORBIDDEN, "FORBIDDEN",
|
"当前用户缺少该模块的导入权限（" + adapter.permission() + "）");
|
}
|
}
|
|
private AgentImportModuleAdapter requireAdapter(String moduleId) {
|
AgentImportModuleAdapter adapter = adapters().get(moduleId);
|
if (adapter == null) {
|
throw new AgentBridgeException(HttpStatus.NOT_FOUND, "MODULE_NOT_SUPPORTED",
|
"不支持的导入模块：" + moduleId);
|
}
|
return adapter;
|
}
|
|
private Map<String, AgentImportModuleAdapter> adapters() {
|
Map<String, AgentImportModuleAdapter> result = new LinkedHashMap<>();
|
for (AgentImportModuleAdapter adapter : adapterProvider) {
|
result.put(adapter.moduleId(), adapter);
|
}
|
return result;
|
}
|
|
private static AgentImportCommandRowDO.AgentImportCommandRowDOBuilder baseRow(
|
String moduleId, String commandId, AgentImportPreflightReqVO.Row row) {
|
return AgentImportCommandRowDO.builder()
|
.commandId(commandId)
|
.moduleId(moduleId)
|
.candidateId(row.getCandidateId())
|
.sourceRow(row.getRowNo())
|
.code(row.getValues().get("code"))
|
.name(row.getValues().get("name"));
|
}
|
|
private static void requireSameDigest(AgentImportCommandDO command, String digest) {
|
if (!command.getRequestDigest().equals(digest)) {
|
throw new AgentBridgeException(HttpStatus.CONFLICT, "IDEMPOTENCY_CONFLICT",
|
"同一命令或幂等键携带了不同的请求内容，请检查任务状态后重试");
|
}
|
}
|
|
private static String requestDigest(String moduleId, AgentImportCommitReqVO request) {
|
Map<String, Object> canonical = new LinkedHashMap<>();
|
canonical.put("module_id", moduleId);
|
canonical.put("attachment_id", request.getAttachmentId());
|
canonical.put("session_id", request.getSessionId());
|
canonical.put("source_sha256", StrUtil.nullToEmpty(request.getSourceSha256()).toLowerCase(Locale.ROOT));
|
canonical.put("preflight_hash", StrUtil.nullToEmpty(request.getPreflightHash()).toLowerCase(Locale.ROOT));
|
canonical.put("proposal_hash", StrUtil.nullToEmpty(request.getProposalHash()));
|
canonical.put("proposal_revision", StrUtil.nullToEmpty(request.getProposalRevision()));
|
canonical.put("options", request.getOptions() == null ? Map.of() : new TreeMap<>(request.getOptions()));
|
List<Map<String, Object>> rows = new ArrayList<>();
|
for (AgentImportPreflightReqVO.Row row : request.getRows()) {
|
Map<String, Object> item = new LinkedHashMap<>();
|
item.put("candidate_id", row.getCandidateId());
|
item.put("row_no", row.getRowNo());
|
item.put("values", row.getValues() == null ? Map.of() : new TreeMap<>(row.getValues()));
|
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);
|
}
|
}
|
|
/**
|
* 适配器计算预检摘要的统一入口，避免各模块摘要口径漂移。
|
*/
|
public static String preflightHash(String moduleId, AgentImportPreflightReqVO request,
|
Object resolvedPayload) {
|
Map<String, Object> canonical = new LinkedHashMap<>();
|
canonical.put("module_id", moduleId);
|
canonical.put("attachment_id", request.getAttachmentId());
|
canonical.put("source_sha256", StrUtil.nullToEmpty(request.getSourceSha256()).toLowerCase(Locale.ROOT));
|
canonical.put("proposal_revision", StrUtil.nullToEmpty(request.getProposalRevision()));
|
canonical.put("options", request.getOptions() == null ? Map.of() : new TreeMap<>(request.getOptions()));
|
List<Map<String, Object>> rows = new ArrayList<>();
|
for (AgentImportPreflightReqVO.Row row : request.getRows()) {
|
Map<String, Object> item = new LinkedHashMap<>();
|
item.put("candidate_id", row.getCandidateId());
|
item.put("row_no", row.getRowNo());
|
item.put("values", row.getValues() == null ? Map.of() : new TreeMap<>(row.getValues()));
|
rows.add(item);
|
}
|
canonical.put("rows", rows);
|
canonical.put("resolved", resolvedPayload);
|
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 AgentImportReceiptRespVO toReceipt(AgentImportCommandDO command) {
|
String status = command.getStatus();
|
if (STATUS_RUNNING.equals(status) && command.getLeaseExpiresAt() != null
|
&& command.getLeaseExpiresAt().isBefore(LocalDateTime.now())) {
|
status = STATUS_UNKNOWN;
|
}
|
List<AgentImportCommandRowDO> rowDOs = commandRowMapper.selectListByCommandId(command.getCommandId());
|
List<AgentImportReceiptRespVO.Row> rows = rowDOs.stream()
|
.map(row -> AgentImportReceiptRespVO.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 AgentImportReceiptRespVO.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(AgentImportReceiptRespVO.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 String trim(String value, int maxLength) {
|
if (value == null) {
|
return null;
|
}
|
return value.length() <= maxLength ? value : value.substring(0, maxLength);
|
}
|
}
|
