fix: 人员数据治理迁移7

This commit is contained in:
2026-11-11 11:54:47 +08:00
parent 756b2d7f14
commit 6aabaf7e52
2 changed files with 60 additions and 11 deletions
@@ -4,13 +4,18 @@ import cn.hutool.core.bean.BeanUtil;
import cn.hutool.core.date.DateUtil; import cn.hutool.core.date.DateUtil;
import cn.hutool.crypto.digest.DigestUtil; import cn.hutool.crypto.digest.DigestUtil;
import com.budwk.app.base.constant.RedisConstant; import com.budwk.app.base.constant.RedisConstant;
import com.budwk.app.base.constant.RoleConstant;
import com.budwk.app.base.utils.PwdUtil;
import com.budwk.app.sys.models.Sys_task; import com.budwk.app.sys.models.Sys_task;
import com.budwk.app.sys.models.Sys_dict; import com.budwk.app.sys.models.Sys_dict;
import com.budwk.app.sys.models.Sys_unit; import com.budwk.app.sys.models.Sys_unit;
import com.budwk.app.sys.services.SysDictService; import com.budwk.app.sys.services.SysDictService;
import com.budwk.app.sys.models.Sys_user; 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_history;
import com.budwk.app.sys.models.Sys_user_role;
import com.budwk.app.sys.models.Sys_user_source; import com.budwk.app.sys.models.Sys_user_source;
import com.budwk.app.sys.models.Sys_role;
import com.budwk.app.sys.services.SysRoleService;
import com.budwk.app.zhgh.staffmanage.member.constant.MemberChangeOrigin; import com.budwk.app.zhgh.staffmanage.member.constant.MemberChangeOrigin;
import com.budwk.app.zhgh.staffmanage.member.constant.MemberChangeType; import com.budwk.app.zhgh.staffmanage.member.constant.MemberChangeType;
import com.budwk.app.zhgh.personGovernance.apiConfig.model.*; import com.budwk.app.zhgh.personGovernance.apiConfig.model.*;
@@ -60,6 +65,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
private PersonMemberMatchService personMemberMatchService; private PersonMemberMatchService personMemberMatchService;
@Inject @Inject
private SysDictService sysDictService; private SysDictService sysDictService;
@Inject
private SysRoleService sysRoleService;
/** /**
* configId为保存的配置,triggerType为MANUAL/SCHEDULED;返回批次,FAILED时errorMessage为原因。 * configId为保存的配置,triggerType为MANUAL/SCHEDULED;返回批次,FAILED时errorMessage为原因。
@@ -373,7 +380,9 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
Map<String, NutMap> fields = new HashMap<>(); Map<String, NutMap> fields = new HashMap<>();
columns.forEach(c -> fields.put(c.getString("columnName"), c)); columns.forEach(c -> fields.put(c.getString("columnName"), c));
boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable()); boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable());
Assert.isTrue(fields.containsKey("id") && (unitTarget || fields.containsKey("pullTime")), "目标中间表必须包含id和pullTime字段"); boolean userTarget = "sys_user".equalsIgnoreCase(config.getTargetTable());
Assert.isTrue(fields.containsKey("id") && (unitTarget || userTarget || fields.containsKey("pullTime")),
"目标中间表必须包含id和pullTime字段");
if (unitTarget) { if (unitTarget) {
Assert.isTrue("HTTP_API".equals(config.getSourceType()) && "UPSERT".equals(config.getSyncMode()) Assert.isTrue("HTTP_API".equals(config.getSourceType()) && "UPSERT".equals(config.getSyncMode())
&& "id".equals(config.getIdentityField()), "单位API必须按来源单位id新增或更新"); && "id".equals(config.getIdentityField()), "单位API必须按来源单位id新增或更新");
@@ -420,6 +429,10 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
columns.forEach(c -> types.put(c.getString("columnName"), c.getString("columnType"))); columns.forEach(c -> types.put(c.getString("columnName"), c.getString("columnType")));
String mode = syncMode(config); String mode = syncMode(config);
boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable()); boolean unitTarget = "sys_unit".equalsIgnoreCase(config.getTargetTable());
boolean userTarget = "sys_user".equalsIgnoreCase(config.getTargetTable());
boolean preserveUserNulls = userTarget && "UPSERT".equals(mode);
List<PersonApiFieldMapping> requiredUserMappings = preserveUserNulls
? mappings.stream().filter(mapping -> Boolean.TRUE.equals(mapping.getRequired())).toList() : List.of();
List<PersonApiSyncOptions.FieldRule> lengthRules = options.getRules().stream() List<PersonApiSyncOptions.FieldRule> lengthRules = options.getRules().stream()
.filter(rule -> rule.getLength() != null).toList(); .filter(rule -> rule.getLength() != null).toList();
List<PersonApiSyncOptions.FieldRule> dictionaryRules = Boolean.TRUE.equals(options.getAutoCreateDict()) List<PersonApiSyncOptions.FieldRule> dictionaryRules = Boolean.TRUE.equals(options.getAutoCreateDict())
@@ -450,7 +463,7 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
NutMap values = NutMap.NEW(); NutMap values = NutMap.NEW();
for (PersonApiFieldMapping mapping : mappings) { for (PersonApiFieldMapping mapping : mappings) {
Object value = personApiConfigService.readResponsePath(row, mapping.getSourceField()); Object value = personApiConfigService.readResponsePath(row, mapping.getSourceField());
if (Boolean.TRUE.equals(mapping.getRequired())) { if (Boolean.TRUE.equals(mapping.getRequired()) && !(preserveUserNulls && value == null)) {
Assert.isTrue(value != null && !Strings.isBlank(String.valueOf(value)), "必填来源字段为空:" + mapping.getSourceField()); Assert.isTrue(value != null && !Strings.isBlank(String.valueOf(value)), "必填来源字段为空:" + mapping.getSourceField());
} }
try { try {
@@ -468,8 +481,10 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
values.putIfAbsent("unitcode", values.get("id")); values.putIfAbsent("unitcode", values.get("id"));
} else { } else {
values.put("id", R.UU32()); values.put("id", R.UU32());
if (types.containsKey("pullTime")) {
values.put("pullTime", batch.getStartedAt()); values.put("pullTime", batch.getStartedAt());
} }
}
valuesList.add(values); valuesList.add(values);
} }
// 当前500条全部被过滤时不查询目标表,也不提交空批次。 // 当前500条全部被过滤时不查询目标表,也不提交空批次。
@@ -486,7 +501,8 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
boolean protectSysUser = "SCHEDULED".equals(batch.getTriggerType()) boolean protectSysUser = "SCHEDULED".equals(batch.getTriggerType())
&& "sys_user".equalsIgnoreCase(config.getTargetTable()); && "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)); dictionaryRules, dictionaryValues, Strings.isNotBlank(options.getUnitConfigId()) ? unitIds : null,
requiredUserMappings));
batch.setNewCount(batch.getNewCount() + counts[0]); batch.setNewCount(batch.getNewCount() + counts[0]);
batch.setChangeCount(batch.getChangeCount() + counts[1]); batch.setChangeCount(batch.getChangeCount() + counts[1]);
batch.setSuccessCount(batch.getSuccessCount() + counts[0] + counts[1]); batch.setSuccessCount(batch.getSuccessCount() + counts[0] + counts[1]);
@@ -504,8 +520,10 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
private void writeChunk(Connection connection, PersonApiConfig config, List<NutMap> rows, private void writeChunk(Connection connection, PersonApiConfig config, List<NutMap> rows,
String mode, boolean protectSysUser, int[] counts, String mode, boolean protectSysUser, int[] counts,
List<PersonApiSyncOptions.FieldRule> dictionaryRules, List<PersonApiSyncOptions.FieldRule> dictionaryRules,
Map<String, Set<String>> dictionaryValues, Set<String> unitIds) throws SQLException { Map<String, Set<String>> dictionaryValues, Set<String> unitIds,
List<PersonApiFieldMapping> requiredUserMappings) throws SQLException {
String table = quoteIdentifier(config.getTargetTable()); String table = quoteIdentifier(config.getTargetTable());
boolean userTarget = "sys_user".equalsIgnoreCase(config.getTargetTable());
Map<Integer, String> existing = new HashMap<>(); Map<Integer, String> existing = new HashMap<>();
Set<Integer> protectedRows = new HashSet<>(); Set<Integer> protectedRows = new HashSet<>();
Map<Integer, Integer> representatives = new HashMap<>(); Map<Integer, Integer> representatives = new HashMap<>();
@@ -607,11 +625,13 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
List<Sys_user_source> entities = inserts.values().stream() List<Sys_user_source> entities = inserts.values().stream()
.map(row -> BeanUtil.toBean(row, Sys_user_source.class)).toList(); .map(row -> BeanUtil.toBean(row, Sys_user_source.class)).toList();
dao.insert(entities); dao.insert(entities);
} else if (userTarget) {
insertSystemUsers(connection, table, new ArrayList<>(inserts.values()), requiredUserMappings);
} else { } else {
executeWriteBatch(connection, table, new ArrayList<>(inserts.values()), false); executeWriteBatch(connection, table, new ArrayList<>(inserts.values()), false, false);
} }
} }
if (!updates.isEmpty()) executeWriteBatch(connection, table, new ArrayList<>(updates.values()), true); if (!updates.isEmpty()) executeWriteBatch(connection, table, new ArrayList<>(updates.values()), true, userTarget);
for (Collection<NutMap> written : List.of(inserts.values(), updates.values())) { for (Collection<NutMap> written : List.of(inserts.values(), updates.values())) {
for (NutMap row : written) { for (NutMap row : written) {
for (PersonApiSyncOptions.FieldRule rule : dictionaryRules) { for (PersonApiSyncOptions.FieldRule rule : dictionaryRules) {
@@ -642,6 +662,26 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
counts[1] = updates.size(); counts[1] = updates.size();
} }
private void insertSystemUsers(Connection connection, String table, List<NutMap> rows,
List<PersonApiFieldMapping> requiredMappings) throws SQLException {
Sys_role publicRole = sysRoleService.getByCode(RoleConstant.PUBLIC);
List<Sys_user_role> roles = new ArrayList<>();
for (NutMap row : rows) {
String salt = R.UU32();
row.put("salt", salt);
row.put("password", PwdUtil.getPassword(PwdUtil.generate(12), salt));
for (PersonApiFieldMapping mapping : requiredMappings) {
Assert.notNull(row.get(mapping.getTargetField()), "必填来源字段为空:" + mapping.getSourceField());
}
Sys_user_role role = new Sys_user_role();
role.setUserId(row.getString("id"));
role.setRoleId(publicRole.getId());
roles.add(role);
}
executeWriteBatch(connection, table, rows, false, false);
dao.insert(roles);
}
/** /**
* 比较更新前人员快照与本次映射值,返回 MemberChangeType 编码;空集合表示没有实际业务变化。 * 比较更新前人员快照与本次映射值,返回 MemberChangeType 编码;空集合表示没有实际业务变化。
* id、pullTime属于主键或同步批次字段,不参与人员变更判断。 * id、pullTime属于主键或同步批次字段,不参与人员变更判断。
@@ -653,6 +693,7 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
for (Map.Entry<String, Object> entry : incomingValues.entrySet()) { for (Map.Entry<String, Object> entry : incomingValues.entrySet()) {
String field = entry.getKey(); String field = entry.getKey();
if (List.of("id", "pullTime").contains(field) if (List.of("id", "pullTime").contains(field)
|| entry.getValue() == null
|| valuesEqual(currentValues.get(field), entry.getValue())) { || valuesEqual(currentValues.get(field), entry.getValue())) {
continue; continue;
} }
@@ -703,6 +744,10 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
} }
Sys_user_history history = new Sys_user_history(); Sys_user_history history = new Sys_user_history();
BeanUtil.copyProperties(user, history); BeanUtil.copyProperties(user, history);
if (changeTypes.contains(MemberChangeType.NEW.name())) {
history.setPassword(null);
history.setSalt(null);
}
history.setId(R.UU32()); history.setId(R.UU32());
history.setChangeTime(new Date()); history.setChangeTime(new Date());
history.setChangeOrigin(MemberChangeOrigin.SYSTEM.name()); history.setChangeOrigin(MemberChangeOrigin.SYSTEM.name());
@@ -738,11 +783,15 @@ public class PersonGovernanceSyncServiceImpl implements PersonGovernanceSyncServ
/** JDBC executeBatch真正批量提交;更新只覆盖映射字段和pullTime,不覆盖目标主键及未映射字段。 */ /** JDBC executeBatch真正批量提交;更新只覆盖映射字段和pullTime,不覆盖目标主键及未映射字段。 */
@Aop(TransAop.READ_COMMITTED) @Aop(TransAop.READ_COMMITTED)
private void executeWriteBatch(Connection connection, String table, List<NutMap> rows, boolean update) throws SQLException { private void executeWriteBatch(Connection connection, String table, List<NutMap> rows,
boolean update, boolean preserveNulls) throws SQLException {
List<String> fields = new ArrayList<>(rows.get(0).keySet()); List<String> fields = new ArrayList<>(rows.get(0).keySet());
if (update) fields.remove("id"); if (update) fields.remove("id");
String sql = update String sql = update
? "UPDATE " + table + " SET " + fields.stream().map(f -> quoteIdentifier(f) + "=?") ? "UPDATE " + table + " SET " + fields.stream().map(field -> {
String column = quoteIdentifier(field);
return column + (preserveNulls ? "=COALESCE(?," + column + ")" : "=?");
})
.collect(java.util.stream.Collectors.joining(",")) + " WHERE `id`=?" .collect(java.util.stream.Collectors.joining(",")) + " WHERE `id`=?"
: "INSERT INTO " + table + " (" + fields.stream().map(this::quoteIdentifier) : "INSERT INTO " + table + " (" + fields.stream().map(this::quoteIdentifier)
.collect(java.util.stream.Collectors.joining(",")) + ") VALUES (" .collect(java.util.stream.Collectors.joining(",")) + ") VALUES ("
@@ -138,7 +138,7 @@
</el-form-item> </el-form-item>
<el-form-item v-if="form.targetTable !== 'sys_unit'" label="自动同步单位" class="person-governance-system-member-match"> <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-switch v-model="autoSyncUnit" :disabled="formLoading" @change="handleAutoSyncUnit"></el-switch>
<el-select v-if="autoSyncUnit" v-model="syncOptions.unitConfigId" clearable filterable placeholder="请选择单位API配置"> <el-select v-if="autoSyncUnit" v-model="syncOptions.unitConfigId" clearable filterable placeholder="请选择单位API配置" style="margin-left: 10px">
<el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option> <el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option>
</el-select> </el-select>
</el-form-item> </el-form-item>
@@ -197,7 +197,7 @@
<div v-if="form.targetTable !== 'sys_unit'" class="person-governance-mapping-toolbar-item"> <div v-if="form.targetTable !== 'sys_unit'" class="person-governance-mapping-toolbar-item">
<label>自动同步单位</label> <label>自动同步单位</label>
<el-switch v-model="autoSyncUnit" :disabled="formLoading" @change="handleAutoSyncUnit"></el-switch> <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-select v-if="autoSyncUnit" v-model="syncOptions.unitConfigId" clearable filterable placeholder="请选择单位API配置" style="margin-left: 10px">
<el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option> <el-option v-for="item in unitConfigs" :key="item.id" :label="item.configName" :value="item.id"></el-option>
</el-select> </el-select>
</div> </div>