fix: 人员数据治理迁移6
This commit is contained in:
+1
-1
@@ -30,4 +30,4 @@ src/main/resources/application-dev.yaml
|
||||
.vscode
|
||||
bin
|
||||
node-data-center-mock/
|
||||
.workbuddy
|
||||
.workbuddy/*
|
||||
+6
@@ -68,6 +68,12 @@ public class PersonApiConfigController {
|
||||
return Result.success(personApiConfigService.detail(id));
|
||||
}
|
||||
|
||||
@At
|
||||
@SaCheckPermission("data.person.governance.api")
|
||||
public Result syncOptionChoices() {
|
||||
return Result.success(personApiConfigService.syncOptionChoices());
|
||||
}
|
||||
|
||||
/**
|
||||
* 保存API配置。
|
||||
*
|
||||
|
||||
+21
@@ -0,0 +1,21 @@
|
||||
package com.budwk.app.zhgh.personGovernance.apiConfig.model;
|
||||
|
||||
import lombok.Data;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
@Data
|
||||
public class PersonApiSyncOptions {
|
||||
private Boolean autoCreateDict;
|
||||
private String unitConfigId;
|
||||
private List<FieldRule> rules = new ArrayList<>();
|
||||
|
||||
@Data
|
||||
public static class FieldRule {
|
||||
private String sourceField;
|
||||
private String targetField;
|
||||
private Integer length;
|
||||
private String dictCode;
|
||||
}
|
||||
}
|
||||
+5
@@ -4,6 +4,7 @@ import com.budwk.app.base.page.Pagination;
|
||||
import com.budwk.app.base.param.PageForm;
|
||||
import com.budwk.app.base.service.BaseService;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiConfig;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiSyncOptions;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiFieldMapping;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiRequestParam;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiTestCase;
|
||||
@@ -28,6 +29,10 @@ public interface PersonApiConfigService extends BaseService<PersonApiConfig> {
|
||||
*/
|
||||
PersonApiConfig saveConfig(String data);
|
||||
|
||||
PersonApiSyncOptions syncOptions(String configId);
|
||||
|
||||
NutMap syncOptionChoices();
|
||||
|
||||
void deleteConfig(String id);
|
||||
|
||||
List<NutMap> tableNames();
|
||||
|
||||
+85
-2
@@ -11,11 +11,14 @@ import com.budwk.app.base.dao.CndPlus;
|
||||
import com.budwk.app.base.param.PageForm;
|
||||
import com.budwk.app.base.service.impl.BaseServiceImpl;
|
||||
import com.budwk.app.sys.models.Sys_task;
|
||||
import com.budwk.app.sys.models.Sys_config;
|
||||
import com.budwk.app.sys.models.Sys_dict;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.job.PersonGovernanceSyncJob;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiConfig;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiFieldMapping;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiRequestParam;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiTestCase;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.model.PersonApiSyncOptions;
|
||||
import com.budwk.app.zhgh.personGovernance.apiConfig.service.PersonApiConfigService;
|
||||
import com.budwk.app.zhgh.personGovernance.memberMatch.model.PersonMemberMatchConfig;
|
||||
import org.nutz.aop.interceptor.ioc.TransAop;
|
||||
@@ -103,6 +106,7 @@ public class PersonApiConfigServiceImpl extends BaseServiceImpl<PersonApiConfig>
|
||||
});
|
||||
return NutMap.NEW()
|
||||
.addv("config", config)
|
||||
.addv("syncOptions", syncOptions(id))
|
||||
.addv("mappings", dao().query(PersonApiFieldMapping.class, Cnd.where("configId", "=", id).asc("sort")))
|
||||
.addv("params", params)
|
||||
.addv("testCases", testCases);
|
||||
@@ -157,7 +161,14 @@ public class PersonApiConfigServiceImpl extends BaseServiceImpl<PersonApiConfig>
|
||||
Assert.isTrue(mappings.stream().filter(v -> Boolean.TRUE.equals(v.getIdentityFlag())).count() == 1
|
||||
&& identity != null && Strings.isNotBlank(identity.getSourceField())
|
||||
&& Strings.isNotBlank(identity.getTargetField())
|
||||
&& !List.of("id", "pullTime").contains(identity.getTargetField()), "请选择一个有效的业务唯一标识映射");
|
||||
&& !"pullTime".equals(identity.getTargetField())
|
||||
&& (!"id".equals(identity.getTargetField()) || isUnitConfig(config)), "请选择一个有效的业务唯一标识映射");
|
||||
}
|
||||
if ("sys_unit".equalsIgnoreCase(config.getTargetTable())) {
|
||||
Assert.isTrue(isUnitConfig(config) && "UPSERT".equals(config.getSyncMode())
|
||||
&& "id".equals(config.getIdentityField()), "单位API配置必须使用存在即修改模式,并将来源单位编码映射到id作为唯一标识");
|
||||
Assert.isTrue(mappings.stream().anyMatch(mapping -> "name".equals(mapping.getTargetField())
|
||||
&& Strings.isNotBlank(mapping.getSourceField())), "单位API配置必须映射单位名称name");
|
||||
}
|
||||
if ("HTTP_API".equals(config.getSourceType())) pagination(config, params);
|
||||
|
||||
@@ -251,7 +262,78 @@ public class PersonApiConfigServiceImpl extends BaseServiceImpl<PersonApiConfig>
|
||||
PersonApiFieldMapping.class, Json.toJson(payload.get("mappings")));
|
||||
List<PersonApiRequestParam> params = Json.fromJsonAsList(
|
||||
PersonApiRequestParam.class, Json.toJson(payload.get("params")));
|
||||
return saveConfig(config, mappings, params);
|
||||
PersonApiSyncOptions options = payload.containsKey("syncOptions")
|
||||
? Json.fromJson(PersonApiSyncOptions.class, Json.toJson(payload.get("syncOptions"))) : null;
|
||||
if (options != null) {
|
||||
validateSyncOptions(config, mappings, options);
|
||||
}
|
||||
PersonApiConfig saved = saveConfig(config, mappings, params);
|
||||
if (options != null) {
|
||||
Sys_config settings = new Sys_config();
|
||||
settings.setConfigKey(syncOptionsKey(saved.getId()));
|
||||
settings.setConfigValue(Json.toJson(options));
|
||||
settings.setNote("人员治理API同步选项");
|
||||
dao().insertOrUpdate(settings);
|
||||
}
|
||||
return saved;
|
||||
}
|
||||
|
||||
private String syncOptionsKey(String configId) {
|
||||
return "personGovernance.syncOptions." + configId;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PersonApiSyncOptions syncOptions(String configId) {
|
||||
Sys_config settings = dao().fetch(Sys_config.class, syncOptionsKey(configId));
|
||||
return settings == null ? new PersonApiSyncOptions()
|
||||
: Json.fromJson(PersonApiSyncOptions.class, settings.getConfigValue());
|
||||
}
|
||||
|
||||
@Override
|
||||
public NutMap syncOptionChoices() {
|
||||
return NutMap.NEW()
|
||||
.addv("dicts", dao().query(Sys_dict.class, Cnd.where("parentId", "=", "").or("parentId", "is", null).asc("name")))
|
||||
.addv("unitConfigs", dao().query(PersonApiConfig.class,
|
||||
Cnd.where("sourceType", "=", "HTTP_API").and("targetTable", "=", "sys_unit")
|
||||
.and("syncMode", "=", "UPSERT").and("identityField", "=", "id")
|
||||
.and(Cnd.exps("delFlag", "=", false).or("delFlag", "is", null)).asc("configName")).stream()
|
||||
.map(item -> NutMap.NEW().addv("id", item.getId()).addv("configName", item.getConfigName())).toList());
|
||||
}
|
||||
|
||||
private boolean isUnitConfig(PersonApiConfig config) {
|
||||
return "HTTP_API".equals(config.getSourceType()) && "sys_unit".equalsIgnoreCase(config.getTargetTable());
|
||||
}
|
||||
|
||||
private void validateSyncOptions(PersonApiConfig config, List<PersonApiFieldMapping> mappings, PersonApiSyncOptions options) {
|
||||
if (Strings.isNotBlank(options.getUnitConfigId())) {
|
||||
PersonApiConfig unitConfig = fetch(options.getUnitConfigId());
|
||||
Assert.isTrue(!"sys_unit".equalsIgnoreCase(config.getTargetTable()) && unitConfig != null
|
||||
&& !Boolean.TRUE.equals(unitConfig.getDelFlag()) && isUnitConfig(unitConfig)
|
||||
&& "UPSERT".equals(unitConfig.getSyncMode()) && "id".equals(unitConfig.getIdentityField()),
|
||||
"请选择有效的单位API配置");
|
||||
Assert.isTrue(mappings != null && mappings.stream().anyMatch(mapping -> "unitId".equals(mapping.getTargetField())
|
||||
&& Strings.isNotBlank(mapping.getSourceField())), "自动同步单位需要配置unitId映射");
|
||||
}
|
||||
Assert.notNull(options.getRules(), "字段配置不能为空");
|
||||
java.util.Set<List<String>> ruleFields = new java.util.HashSet<>();
|
||||
for (PersonApiSyncOptions.FieldRule rule : options.getRules()) {
|
||||
Assert.notNull(rule, "字段配置不能为空");
|
||||
Assert.hasText(rule.getSourceField(), "请填写配置规则的来源字段");
|
||||
Assert.isTrue(ruleFields.add(List.of(rule.getSourceField(), Strings.sNull(rule.getTargetField()))),
|
||||
"同一来源字段和目标字段只能配置一条规则");
|
||||
Assert.isTrue(rule.getLength() == null || rule.getLength() >= 0, "字段长度必须为非负整数");
|
||||
Assert.isTrue(mappings != null && mappings.stream().anyMatch(mapping ->
|
||||
Objects.equals(mapping.getSourceField(), rule.getSourceField())
|
||||
&& Objects.equals(Strings.sNull(mapping.getTargetField()), Strings.sNull(rule.getTargetField()))),
|
||||
"字段规则必须关联一条字段映射");
|
||||
if (Strings.isNotBlank(rule.getDictCode())) {
|
||||
Assert.hasText(rule.getTargetField(), "选择字典前请填写目标字段");
|
||||
Assert.isTrue(!"pullTime".equals(rule.getTargetField())
|
||||
&& (!"id".equals(rule.getTargetField()) || isUnitConfig(config)), "系统生成字段不能配置字典");
|
||||
Assert.notNull(dao().fetch(Sys_dict.class, Cnd.where("code", "=", rule.getDictCode())
|
||||
.and(Cnd.exps("parentId", "=", "").or("parentId", "is", null))), "字典分类不存在");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -268,6 +350,7 @@ public class PersonApiConfigServiceImpl extends BaseServiceImpl<PersonApiConfig>
|
||||
PersonApiConfig config = fetch(id);
|
||||
config.setEnabled(false);
|
||||
reconcileTask(config);
|
||||
dao().delete(Sys_config.class, syncOptionsKey(id));
|
||||
}
|
||||
|
||||
@Override
|
||||
|
||||
+121
-10
@@ -5,6 +5,9 @@ import cn.hutool.core.date.DateUtil;
|
||||
import cn.hutool.crypto.digest.DigestUtil;
|
||||
import com.budwk.app.base.constant.RedisConstant;
|
||||
import com.budwk.app.sys.models.Sys_task;
|
||||
import com.budwk.app.sys.models.Sys_dict;
|
||||
import com.budwk.app.sys.models.Sys_unit;
|
||||
import com.budwk.app.sys.services.SysDictService;
|
||||
import com.budwk.app.sys.models.Sys_user;
|
||||
import com.budwk.app.sys.models.Sys_user_history;
|
||||
import com.budwk.app.sys.models.Sys_user_source;
|
||||
@@ -55,6 +58,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
private RedisService redisService;
|
||||
@Inject
|
||||
private PersonMemberMatchService personMemberMatchService;
|
||||
@Inject
|
||||
private SysDictService sysDictService;
|
||||
|
||||
/**
|
||||
* configId为保存的配置,triggerType为MANUAL/SCHEDULED;返回批次,FAILED时errorMessage为原因。
|
||||
@@ -88,9 +93,22 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
});
|
||||
List<PersonApiFieldMapping> mappings = dao.query(PersonApiFieldMapping.class,
|
||||
Cnd.where("configId", "=", configId).asc("sort"));
|
||||
PersonApiSyncOptions options = personApiConfigService.syncOptions(configId);
|
||||
boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable());
|
||||
if (unitTarget) {
|
||||
// 单位写入可能随父批次一起提交,行锁一直持有到外层事务完成。
|
||||
Sql lock = Sqls.create("SELECT id FROM person_api_config WHERE id=@id FOR UPDATE").setParam("id", configId);
|
||||
lock.setCallback(Sqls.callback.str());
|
||||
dao.execute(lock);
|
||||
}
|
||||
Map<String, Set<String>> dictionaryValues = new TreeMap<>();
|
||||
Set<String> unitIds = new LinkedHashSet<>();
|
||||
// 页面会加载整张表的列;批次字段由系统生成,未配置的非必填列交给数据库默认值。
|
||||
mappings = mappings.stream()
|
||||
.filter(m -> !List.of("id", "pullTime").contains(Strings.sNull(m.getTargetField())))
|
||||
.filter(m -> !"pullTime".equals(m.getTargetField()) && (unitTarget || !"id".equals(m.getTargetField())))
|
||||
.filter(m -> Strings.isNotBlank(m.getTargetField()) || options.getRules().stream().noneMatch(rule ->
|
||||
rule.getLength() != null && Strings.isBlank(rule.getTargetField())
|
||||
&& Objects.equals(rule.getSourceField(), m.getSourceField())))
|
||||
.filter(m -> Boolean.TRUE.equals(m.getRequired())
|
||||
|| (Strings.isNotBlank(m.getSourceField()) && Strings.isNotBlank(m.getTargetField())))
|
||||
.toList();
|
||||
@@ -136,7 +154,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
}
|
||||
log.info("[人员同步] 开始字段映射转换及目标表批量写入,configId={},batchNo={},sourceCount={},mappingCount={}",
|
||||
configId, batch.getBatchNo(), rows.size(), mappings.size());
|
||||
insertPage(config, mappings, columns, rows, batch, lockKey, owner);
|
||||
insertPage(config, mappings, columns, rows, batch, lockKey, owner, options, dictionaryValues, unitIds);
|
||||
applySyncOptions(config, options, dictionaryValues, unitIds);
|
||||
log.info("[人员同步] 目标表批量写入完成,configId={},batchNo={},successCount={},newCount={},changeCount={},skipCount={}",
|
||||
configId, batch.getBatchNo(), batch.getSuccessCount(), batch.getNewCount(),
|
||||
batch.getChangeCount(), batch.getSourceCount() - batch.getSuccessCount());
|
||||
@@ -175,12 +194,13 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
if (!rows.isEmpty() && pageSize > 0) {
|
||||
Assert.isTrue(fingerprints.add(DigestUtil.sha256Hex(Json.toJson(rows))), "接口重复返回同一页,已终止并回滚本批次");
|
||||
}
|
||||
insertPage(config, mappings, columns, rows, batch, lockKey, owner);
|
||||
insertPage(config, mappings, columns, rows, batch, lockKey, owner, options, dictionaryValues, unitIds);
|
||||
// 按接口原始条数判定,已存在而跳过的记录仍属于这一页。
|
||||
// 仅数量模式是单次限量查询,即使正好返回配置数量也不能继续构造下一页请求。
|
||||
if (singleLimitRequest || pageSize == 0 || rows.size() < pageSize) break;
|
||||
page = Math.addExact(page, paging.getLong("step"));
|
||||
}
|
||||
applySyncOptions(config, options, dictionaryValues, unitIds);
|
||||
// 第三方接口同步同样以配置开关为准,分页全部写入完成后只执行一次会员规则匹配。
|
||||
if (shouldMatchMemberRules(config)) {
|
||||
personMemberMatchService.reconcileForApiConfig(configId);
|
||||
@@ -222,6 +242,57 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
}
|
||||
|
||||
/** 工会会员与福利会员匹配只受配置开关控制,两类来源均可在同步完成后执行。 */
|
||||
private void applySyncOptions(PersonApiConfig config, PersonApiSyncOptions options,
|
||||
Map<String, Set<String>> dictionaryValues, Set<String> unitIds) {
|
||||
if (!unitIds.isEmpty() && Strings.isNotBlank(options.getUnitConfigId())) {
|
||||
Set<String> missing = missingUnitIds(unitIds);
|
||||
if (!missing.isEmpty()) {
|
||||
PersonApiConfig unitConfig = dao.fetch(PersonApiConfig.class, options.getUnitConfigId());
|
||||
Assert.isTrue(!"sys_unit".equalsIgnoreCase(config.getTargetTable()) && unitConfig != null
|
||||
&& "sys_unit".equalsIgnoreCase(unitConfig.getTargetTable())
|
||||
&& "HTTP_API".equals(unitConfig.getSourceType()), "关联的单位API配置不可用");
|
||||
PersonSyncBatch unitBatch = execute(unitConfig.getId(), "LINKED");
|
||||
Assert.isTrue("SUCCESS".equals(unitBatch.getStatus()), "自动同步单位失败:" + unitBatch.getErrorMessage());
|
||||
Set<String> stillMissing = missingUnitIds(missing);
|
||||
Assert.isTrue(stillMissing.isEmpty(), "单位API未返回所需单位:" + String.join("、",
|
||||
stillMissing.stream().limit(5).toList()));
|
||||
}
|
||||
}
|
||||
boolean created = false;
|
||||
for (Map.Entry<String, Set<String>> entry : dictionaryValues.entrySet()) {
|
||||
Sys_dict parent = dao.fetch(Sys_dict.class, Cnd.where("code", "=", entry.getKey())
|
||||
.and(Cnd.exps("parentId", "=", "").or("parentId", "is", null)));
|
||||
Assert.notNull(parent, "字典分类不存在:" + entry.getKey());
|
||||
// 同一字典分类下串行检查和插入,避免并发同步重复创建子项。
|
||||
Sql lock = Sqls.create("SELECT id FROM sys_dict WHERE id=@id FOR UPDATE").setParam("id", parent.getId());
|
||||
lock.setCallback(Sqls.callback.str());
|
||||
dao.execute(lock);
|
||||
Set<String> existing = new HashSet<>();
|
||||
dao.query(Sys_dict.class, Cnd.where("parentId", "=", parent.getId()))
|
||||
.forEach(item -> existing.add(item.getCode()));
|
||||
for (String value : entry.getValue()) {
|
||||
if (existing.add(value)) {
|
||||
Sys_dict item = new Sys_dict();
|
||||
item.setCode(value);
|
||||
item.setName(value);
|
||||
sysDictService.save(item, parent.getId());
|
||||
created = true;
|
||||
}
|
||||
}
|
||||
}
|
||||
if (created) sysDictService.clearCache();
|
||||
}
|
||||
|
||||
private Set<String> missingUnitIds(Set<String> unitIds) {
|
||||
Set<String> missing = new LinkedHashSet<>(unitIds);
|
||||
List<String> ids = new ArrayList<>(unitIds);
|
||||
for (int start = 0; start < ids.size(); start += WRITE_BATCH_SIZE) {
|
||||
List<String> chunk = ids.subList(start, Math.min(start + WRITE_BATCH_SIZE, ids.size()));
|
||||
dao.query(Sys_unit.class, Cnd.where("id", "in", chunk)).forEach(unit -> missing.remove(unit.getId()));
|
||||
}
|
||||
return missing;
|
||||
}
|
||||
|
||||
private boolean shouldMatchMemberRules(PersonApiConfig config) {
|
||||
return Boolean.TRUE.equals(config.getMatchMemberRules());
|
||||
}
|
||||
@@ -301,7 +372,14 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
List<NutMap> columns = personApiConfigService.columns(config.getTargetTable());
|
||||
Map<String, NutMap> fields = new HashMap<>();
|
||||
columns.forEach(c -> fields.put(c.getString("columnName"), c));
|
||||
Assert.isTrue(fields.containsKey("id") && fields.containsKey("pullTime"), "目标中间表必须包含id和pullTime字段");
|
||||
boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable());
|
||||
Assert.isTrue(fields.containsKey("id") && (unitTarget || fields.containsKey("pullTime")), "目标中间表必须包含id和pullTime字段");
|
||||
if (unitTarget) {
|
||||
Assert.isTrue("HTTP_API".equals(config.getSourceType()) && "UPSERT".equals(config.getSyncMode())
|
||||
&& "id".equals(config.getIdentityField()), "单位API必须按来源单位id新增或更新");
|
||||
Assert.isTrue(mappings.stream().anyMatch(mapping -> "name".equals(mapping.getTargetField())
|
||||
&& Strings.isNotBlank(mapping.getSourceField())), "单位API配置必须映射单位名称name");
|
||||
}
|
||||
Assert.isTrue(fields.get("id").getString("columnType").toLowerCase(Locale.ROOT).contains("char"), "中间表id须支持字符串主键");
|
||||
Assert.notEmpty(mappings, "请配置字段映射");
|
||||
Set<String> targets = new HashSet<>();
|
||||
@@ -315,7 +393,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
if (!"APPEND".equals(mode)) {
|
||||
Assert.hasText(config.getIdentityField(), "去重同步必须配置唯一标识");
|
||||
Assert.isTrue(targets.contains(config.getIdentityField())
|
||||
&& !List.of("id", "pullTime").contains(config.getIdentityField()), "唯一标识必须是有效的业务映射字段");
|
||||
&& !"pullTime".equals(config.getIdentityField())
|
||||
&& (unitTarget || !"id".equals(config.getIdentityField())), "唯一标识必须是有效的业务映射字段");
|
||||
}
|
||||
return columns;
|
||||
}
|
||||
@@ -333,12 +412,18 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
*/
|
||||
@Aop(TransAop.READ_COMMITTED)
|
||||
private void insertPage(PersonApiConfig config, List<PersonApiFieldMapping> mappings, List<NutMap> columns,
|
||||
List<?> rows, PersonSyncBatch batch, String lockKey, String owner) {
|
||||
List<?> rows, PersonSyncBatch batch, String lockKey, String owner,
|
||||
PersonApiSyncOptions options, Map<String, Set<String>> dictionaryValues, Set<String> unitIds) {
|
||||
Assert.isTrue((long) batch.getSourceCount() + rows.size() <= MAX_ROWS, "超过单批最大行数,已终止同步");
|
||||
batch.setSourceCount(batch.getSourceCount() + rows.size());
|
||||
Map<String, String> types = new HashMap<>();
|
||||
columns.forEach(c -> types.put(c.getString("columnName"), c.getString("columnType")));
|
||||
String mode = syncMode(config);
|
||||
boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable());
|
||||
List<PersonApiSyncOptions.FieldRule> lengthRules = options.getRules().stream()
|
||||
.filter(rule -> rule.getLength() != null).toList();
|
||||
List<PersonApiSyncOptions.FieldRule> dictionaryRules = Boolean.TRUE.equals(options.getAutoCreateDict())
|
||||
? options.getRules().stream().filter(rule -> Strings.isNotBlank(rule.getDictCode())).toList() : List.of();
|
||||
PersonApiFieldMapping identityMapping = mappings.stream()
|
||||
.filter(mapping -> Strings.isNotBlank(config.getIdentityField())
|
||||
&& Objects.equals(config.getIdentityField(), mapping.getTargetField()))
|
||||
@@ -349,6 +434,11 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
int chunkSourceCount = Math.min(offset + WRITE_BATCH_SIZE, rows.size()) - offset;
|
||||
for (Object row : rows.subList(offset, Math.min(offset + WRITE_BATCH_SIZE, rows.size()))) {
|
||||
Assert.isTrue(row instanceof Map<?, ?>, "数据集行必须为JSON对象");
|
||||
boolean matchesLength = lengthRules.stream().allMatch(rule -> {
|
||||
Object value = personApiConfigService.readResponsePath(row, rule.getSourceField());
|
||||
return value != null && String.valueOf(value).length() == rule.getLength();
|
||||
});
|
||||
if (!matchesLength) continue;
|
||||
// 先过滤空业务标识,再校验其他必填字段及转换类型;未配置标识的全量追加保持原行为。
|
||||
// 不改变原始rows和sourceCount,避免整页被过滤时提前结束分页,跳过数量由原始数减成功数统计。
|
||||
if (identityMapping != null) {
|
||||
@@ -373,8 +463,13 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
Object identity = values.get(config.getIdentityField());
|
||||
Assert.isTrue(identity != null && Strings.isNotBlank(String.valueOf(identity)), "唯一标识不能为空:" + config.getIdentityField());
|
||||
}
|
||||
values.put("id", R.UU32());
|
||||
values.put("pullTime", batch.getStartedAt());
|
||||
if (unitTarget) {
|
||||
Assert.hasText(values.getString("name"), "单位名称不能为空");
|
||||
values.putIfAbsent("unitcode", values.get("id"));
|
||||
} else {
|
||||
values.put("id", R.UU32());
|
||||
values.put("pullTime", batch.getStartedAt());
|
||||
}
|
||||
valuesList.add(values);
|
||||
}
|
||||
// 当前500条全部被过滤时不查询目标表,也不提交空批次。
|
||||
@@ -390,7 +485,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
// 保护标识只约束定时任务;人工同步仍保持现有行为,便于管理员主动修复数据。
|
||||
boolean protectSysUser = "SCHEDULED".equals(batch.getTriggerType())
|
||||
&& "sys_user".equalsIgnoreCase(config.getTargetTable());
|
||||
dao.run(connection -> writeChunk(connection, config, valuesList, mode, protectSysUser, counts));
|
||||
dao.run(connection -> writeChunk(connection, config, valuesList, mode, protectSysUser, counts,
|
||||
dictionaryRules, dictionaryValues, Strings.isNotBlank(options.getUnitConfigId()) ? unitIds : null));
|
||||
batch.setNewCount(batch.getNewCount() + counts[0]);
|
||||
batch.setChangeCount(batch.getChangeCount() + counts[1]);
|
||||
batch.setSuccessCount(batch.getSuccessCount() + counts[0] + counts[1]);
|
||||
@@ -406,7 +502,9 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
*/
|
||||
@Aop(TransAop.READ_COMMITTED)
|
||||
private void writeChunk(Connection connection, PersonApiConfig config, List<NutMap> rows,
|
||||
String mode, boolean protectSysUser, int[] counts) throws SQLException {
|
||||
String mode, boolean protectSysUser, int[] counts,
|
||||
List<PersonApiSyncOptions.FieldRule> dictionaryRules,
|
||||
Map<String, Set<String>> dictionaryValues, Set<String> unitIds) throws SQLException {
|
||||
String table = quoteIdentifier(config.getTargetTable());
|
||||
Map<Integer, String> existing = new HashMap<>();
|
||||
Set<Integer> protectedRows = new HashSet<>();
|
||||
@@ -514,6 +612,19 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
|
||||
}
|
||||
}
|
||||
if (!updates.isEmpty()) executeWriteBatch(connection, table, new ArrayList<>(updates.values()), true);
|
||||
for (Collection<NutMap> written : List.of(inserts.values(), updates.values())) {
|
||||
for (NutMap row : written) {
|
||||
for (PersonApiSyncOptions.FieldRule rule : dictionaryRules) {
|
||||
String value = row.getString(rule.getTargetField());
|
||||
if (Strings.isNotBlank(value)) {
|
||||
dictionaryValues.computeIfAbsent(rule.getDictCode(), key -> new LinkedHashSet<>()).add(value);
|
||||
}
|
||||
}
|
||||
if (unitIds != null && Strings.isNotBlank(row.getString("unitId"))) {
|
||||
unitIds.add(row.getString("unitId"));
|
||||
}
|
||||
}
|
||||
}
|
||||
if (protectSysUser) {
|
||||
String changeReason = syncHistoryReason(config);
|
||||
recordSyncHistories(updateSnapshots, updateChangeTypes, changeReason);
|
||||
|
||||
@@ -133,6 +133,15 @@
|
||||
<el-form-item label="是否匹配会员规则" class="person-governance-system-member-match">
|
||||
<el-switch v-model="form.matchMemberRules" :disabled="formLoading"></el-switch>
|
||||
</el-form-item>
|
||||
<el-form-item label="自动创建字典子项" class="person-governance-system-member-match">
|
||||
<el-switch v-model="syncOptions.autoCreateDict" :disabled="formLoading"></el-switch>
|
||||
</el-form-item>
|
||||
<el-form-item v-if="form.targetTable !== 'sys_unit'" label="自动同步单位" class="person-governance-system-member-match">
|
||||
<el-switch v-model="autoSyncUnit" :disabled="formLoading" @change="handleAutoSyncUnit"></el-switch>
|
||||
<el-select v-if="autoSyncUnit" v-model="syncOptions.unitConfigId" clearable filterable placeholder="请选择单位API配置">
|
||||
<el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option>
|
||||
</el-select>
|
||||
</el-form-item>
|
||||
<div class="person-governance-mapping-actions">
|
||||
<el-button
|
||||
class="person-governance-clear-mappings"
|
||||
@@ -181,6 +190,17 @@
|
||||
<label>是否匹配会员规则</label>
|
||||
<el-switch v-model="form.matchMemberRules" :disabled="formLoading"></el-switch>
|
||||
</div>
|
||||
<div class="person-governance-mapping-toolbar-item person-governance-mapping-switch">
|
||||
<label>自动创建字典子项</label>
|
||||
<el-switch v-model="syncOptions.autoCreateDict" :disabled="formLoading"></el-switch>
|
||||
</div>
|
||||
<div v-if="form.targetTable !== 'sys_unit'" class="person-governance-mapping-toolbar-item">
|
||||
<label>自动同步单位</label>
|
||||
<el-switch v-model="autoSyncUnit" :disabled="formLoading" @change="handleAutoSyncUnit"></el-switch>
|
||||
<el-select v-if="autoSyncUnit" v-model="syncOptions.unitConfigId" clearable filterable placeholder="请选择单位API配置">
|
||||
<el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option>
|
||||
</el-select>
|
||||
</div>
|
||||
<div class="person-governance-mapping-actions">
|
||||
<el-button
|
||||
class="person-governance-clear-mappings"
|
||||
@@ -269,8 +289,9 @@
|
||||
:disabled="!isIdentityMapping(row)" @input="identityRow=mappings[$event]">设为标识</el-radio>
|
||||
</template>
|
||||
</el-table-column>
|
||||
<el-table-column label="操作" width="80">
|
||||
<template v-slot="{$index}">
|
||||
<el-table-column label="操作" width="130">
|
||||
<template v-slot="{row,$index}">
|
||||
<el-button type="text" :disabled="formLoading" @click="openFieldRule(row)">配置</el-button>
|
||||
<el-button type="text" class="text-danger" @click="removeMapping($index)">
|
||||
删除
|
||||
</el-button>
|
||||
@@ -418,6 +439,27 @@
|
||||
@click="save"
|
||||
>保存配置</el-button>
|
||||
</div>
|
||||
<el-dialog title="字段配置" :visible.sync="fieldRuleVisible" append-to-body width="520px"
|
||||
:close-on-click-modal="false" custom-class="person-governance-dialog">
|
||||
<el-form :model="fieldRuleDraft" label-width="120px">
|
||||
<el-form-item label="来源字段">
|
||||
<el-input :value="fieldRuleRow ? fieldRuleRow.sourceField : ''" disabled></el-input>
|
||||
</el-form-item>
|
||||
<el-form-item label="长度等于">
|
||||
<el-input v-model="fieldRuleDraft.length" clearable placeholder="留空不过滤,例如 10"></el-input>
|
||||
</el-form-item>
|
||||
<el-form-item label="对应字典">
|
||||
<el-select v-model="fieldRuleDraft.dictCode" clearable filterable placeholder="请选择字典分类">
|
||||
<el-option v-for="item in dictChoices" :key="item.code"
|
||||
:label="item.name + '(' + item.code + ')'" :value="item.code"></el-option>
|
||||
</el-select>
|
||||
</el-form-item>
|
||||
</el-form>
|
||||
<div slot="footer">
|
||||
<el-button @click="fieldRuleVisible=false">取消</el-button>
|
||||
<el-button type="primary" @click="applyFieldRule">确定</el-button>
|
||||
</div>
|
||||
</el-dialog>
|
||||
<el-dialog title="快捷映射" :visible.sync="quickMappingVisible" append-to-body
|
||||
width="720px" :close-on-click-modal="false" custom-class="person-governance-dialog">
|
||||
<el-alert title="输入 JSON 数组,以系统字段名为键。仅更新填写的字段,确认后回填,保存配置后生效。"
|
||||
@@ -481,6 +523,13 @@
|
||||
sourceColumns: [],
|
||||
targetColumns: [],
|
||||
mappings: [],
|
||||
syncOptions: { autoCreateDict: false, unitConfigId: '' },
|
||||
autoSyncUnit: false,
|
||||
unitConfigs: [],
|
||||
dictChoices: [],
|
||||
fieldRuleVisible: false,
|
||||
fieldRuleRow: null,
|
||||
fieldRuleDraft: { length: '', dictCode: '' },
|
||||
params: [],
|
||||
testCases: [],
|
||||
identityRow: null,
|
||||
@@ -596,6 +645,47 @@
|
||||
},
|
||||
},
|
||||
methods: {
|
||||
loadSyncOptionChoices() {
|
||||
return $.post(apiBase + '/syncOptionChoices').then((res) => {
|
||||
if (res.code === 0) {
|
||||
this.$set(this, 'dictChoices', res.data.dicts || [])
|
||||
this.$set(this, 'unitConfigs', res.data.unitConfigs || [])
|
||||
} else {
|
||||
this.notifyWarning(res.msg || '加载同步配置选项失败')
|
||||
}
|
||||
}).fail(() => this.notifyWarning('加载同步配置选项失败'))
|
||||
},
|
||||
handleAutoSyncUnit(enabled) {
|
||||
if (!enabled) this.$set(this.syncOptions, 'unitConfigId', '')
|
||||
},
|
||||
openFieldRule(row) {
|
||||
this.$set(this, 'fieldRuleRow', row)
|
||||
this.$set(this, 'fieldRuleDraft', {
|
||||
length: row.filterLength == null ? '' : String(row.filterLength),
|
||||
dictCode: row.dictCode || '',
|
||||
})
|
||||
this.$set(this, 'fieldRuleVisible', true)
|
||||
},
|
||||
applyFieldRule() {
|
||||
const row = this.fieldRuleRow
|
||||
const length = this.fieldRuleDraft.length.trim()
|
||||
const dictCode = this.fieldRuleDraft.dictCode
|
||||
if ((length || dictCode) && !(row.sourceField || '').trim()) {
|
||||
this.notifyWarning('请先填写来源字段')
|
||||
return
|
||||
}
|
||||
if (length && (!/^\d+$/.test(length) || Number(length) > 2147483647)) {
|
||||
this.notifyWarning('字段长度必须为非负整数')
|
||||
return
|
||||
}
|
||||
if (dictCode && !row.targetField) {
|
||||
this.notifyWarning('选择字典前请填写目标字段')
|
||||
return
|
||||
}
|
||||
this.$set(row, 'filterLength', length ? Number(length) : null)
|
||||
this.$set(row, 'dictCode', dictCode)
|
||||
this.$set(this, 'fieldRuleVisible', false)
|
||||
},
|
||||
openQuickMapping() {
|
||||
this.$set(this, 'quickMappingText', '')
|
||||
this.$set(this, 'quickMappingVisible', true)
|
||||
@@ -659,7 +749,8 @@
|
||||
isIdentityMapping(row) {
|
||||
return Boolean(row && row.sourceField && row.sourceField.trim()
|
||||
&& row.targetField && row.targetField.trim()
|
||||
&& !['id', 'pullTime'].includes(row.targetField))
|
||||
&& row.targetField !== 'pullTime'
|
||||
&& (row.targetField !== 'id' || this.form.targetTable === 'sys_unit'))
|
||||
},
|
||||
// 弹窗使用独立草稿;启停状态不限制周期编辑,取消不修改配置,未设置时预选每5分钟。
|
||||
openCronDialog() {
|
||||
@@ -879,6 +970,8 @@
|
||||
targetComment: c.columnComment || '',
|
||||
required: old ? Boolean(old.required) : c.isNullable === 'NO',
|
||||
identityFlag: old ? Boolean(old.identityFlag) : false,
|
||||
filterLength: old ? old.filterLength : null,
|
||||
dictCode: old ? old.dictCode : '',
|
||||
}
|
||||
})
|
||||
this.$set(this, 'mappings', rows)
|
||||
@@ -1013,6 +1106,17 @@
|
||||
}
|
||||
},
|
||||
save() {
|
||||
const unitTarget = this.form.targetTable === 'sys_unit'
|
||||
if (!unitTarget && this.autoSyncUnit && !this.syncOptions.unitConfigId) {
|
||||
this.notifyWarning('请选择单位API配置')
|
||||
return
|
||||
}
|
||||
if (unitTarget && (this.form.sourceType !== 'HTTP_API' || this.form.syncMode !== 'UPSERT'
|
||||
|| !this.identityRow || this.identityRow.targetField !== 'id')) {
|
||||
this.notifyWarning('单位API配置请选择全量同步(存在即修改),并将单位编码映射到 id 作为唯一标识')
|
||||
this.$set(this, 'activeTab', 'mapping')
|
||||
return
|
||||
}
|
||||
if (this.form.syncMode && this.form.syncMode !== 'APPEND'
|
||||
&& (!this.mappings.includes(this.identityRow) || !this.isIdentityMapping(this.identityRow))) {
|
||||
this.notifyWarning('请在唯一标识列选择完整的业务字段映射(例如工号 gh);目标字段不能是 id 或 pullTime')
|
||||
@@ -1031,9 +1135,19 @@
|
||||
})
|
||||
this.$set(this.form, 'tokenQueryResult', this.testResults.TOKEN)
|
||||
this.$set(this.form, 'dataQueryResult', this.testResults.DATA)
|
||||
const syncOptions = {
|
||||
autoCreateDict: this.syncOptions.autoCreateDict,
|
||||
unitConfigId: !unitTarget && this.autoSyncUnit ? this.syncOptions.unitConfigId : '',
|
||||
rules: this.mappings.filter((row) => row.filterLength != null || row.dictCode).map((row) => ({
|
||||
sourceField: row.sourceField,
|
||||
targetField: row.targetField,
|
||||
length: row.filterLength,
|
||||
dictCode: row.dictCode,
|
||||
})),
|
||||
}
|
||||
this.$set(this, 'formLoading', true)
|
||||
$.post(apiBase + '/save', {
|
||||
data: JSON.stringify({ config: this.form, mappings: this.mappings, params: this.params }),
|
||||
data: JSON.stringify({ config: this.form, mappings: this.mappings, params: this.params, syncOptions }),
|
||||
})
|
||||
.then((res) => {
|
||||
if (res.code === 0) {
|
||||
@@ -1124,6 +1238,18 @@
|
||||
this.$set(this, 'activeTab', 'mapping')
|
||||
}
|
||||
this.$set(this, 'mappings', res.data.mappings || [])
|
||||
const syncOptions = res.data.syncOptions || {}
|
||||
this.$set(this.syncOptions, 'autoCreateDict', Boolean(syncOptions.autoCreateDict))
|
||||
this.$set(this.syncOptions, 'unitConfigId', syncOptions.unitConfigId || '')
|
||||
this.$set(this, 'autoSyncUnit', Boolean(syncOptions.unitConfigId))
|
||||
;(syncOptions.rules || []).forEach((rule) => {
|
||||
const row = this.mappings.find((item) => item.sourceField === rule.sourceField
|
||||
&& (item.targetField || '') === (rule.targetField || ''))
|
||||
if (row) {
|
||||
this.$set(row, 'filterLength', rule.length)
|
||||
this.$set(row, 'dictCode', rule.dictCode || '')
|
||||
}
|
||||
})
|
||||
const params = (res.data.params || []).map((v) => {
|
||||
if (!v.requestScope) {
|
||||
this.$set(v, 'requestScope', 'DATA')
|
||||
@@ -1165,6 +1291,7 @@
|
||||
this.$set(this.form, 'sourceType', sourceType)
|
||||
this.$set(this, 'activeEndpoint', sourceType === 'HTTP_API' ? 'TOKEN' : 'DATA')
|
||||
this.$set(this, 'activeTab', sourceType === 'HTTP_API' ? 'params' : 'mapping')
|
||||
this.loadSyncOptionChoices()
|
||||
this.loadTables()
|
||||
.then(() => {
|
||||
if (q.get('id')) {
|
||||
|
||||
Reference in New Issue
Block a user