ObjectDigitalMigration.java 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374
  1. package com.persagy.proxy.migration.handler;
  2. import cn.hutool.core.collection.CollUtil;
  3. import cn.hutool.core.util.StrUtil;
  4. import com.fasterxml.jackson.databind.node.JsonNodeFactory;
  5. import com.fasterxml.jackson.databind.node.ObjectNode;
  6. import com.persagy.dmp.basic.model.QueryCriteria;
  7. import com.persagy.dmp.basic.utils.JsonNodeUtils;
  8. import com.persagy.dmp.common.constant.ResponseCode;
  9. import com.persagy.dmp.digital.client.DigitalObjectFacade;
  10. import com.persagy.dmp.digital.entity.ObjectDigital;
  11. import com.persagy.dmp.rwd.migrate.entity.SyncData;
  12. import com.persagy.proxy.adm.request.AdmResponse;
  13. import com.persagy.proxy.common.entity.InstanceUrlParam;
  14. import com.persagy.proxy.migration.constant.MigrationConstant;
  15. import com.persagy.proxy.migration.constant.MigrationObjType;
  16. import com.persagy.proxy.migration.constant.MigrationTable;
  17. import com.persagy.proxy.migration.constant.MigrationType;
  18. import com.persagy.proxy.migration.model.DataMigrationExcel;
  19. import com.persagy.proxy.migration.model.DataMigrationResponse;
  20. import com.persagy.proxy.migration.model.MigrationInfo;
  21. import com.persagy.proxy.migration.service.Impl.MigrationAbstractServiceImpl;
  22. import lombok.extern.slf4j.Slf4j;
  23. import java.util.*;
  24. import java.util.stream.Collectors;
  25. /**
  26. * @author lvxianyun
  27. * @title: ObjectMigration
  28. * @projectName adm-middleware
  29. * @description: 迁移adm交付的数据到现场,覆盖现场数据,提供增加、删除、修改 记录放入excel; dt_object 数据迁移 adm -> target project
  30. * @date 2021/12/8 19:53
  31. */
  32. @Slf4j
  33. public class ObjectDigitalMigration extends MigrationAbstractServiceImpl<ObjectDigital> implements IDataMigration, IDataMigrationProcess<ObjectDigital> {
  34. private static ObjectDigitalMigration instance;
  35. private ObjectDigitalMigration() {}
  36. // 静态方法返回该实例
  37. public static ObjectDigitalMigration getInstance() {
  38. if(instance == null) {
  39. synchronized (ObjectDigitalMigration.class) {
  40. if (instance == null) {
  41. instance = new ObjectDigitalMigration();
  42. }
  43. }
  44. }
  45. return instance;
  46. }
  47. /**
  48. * 数据迁移前,数据对比信息 汇总
  49. *
  50. * @param context
  51. * @param migrationInfo
  52. * @return List<DataMigrationExcel>
  53. */
  54. @Override
  55. public List<DataMigrationExcel> migrateForExcel(InstanceUrlParam context, MigrationInfo migrationInfo) {
  56. log.info("######################### dt_object 数据迁移内容获取开始 #########################");
  57. long start = System.currentTimeMillis();
  58. List<DataMigrationExcel> dataMigrationExcels = new ArrayList<>();
  59. for(MigrationObjType objType : MigrationObjType.values()){
  60. log.info("######################### dt_object "+objType.getCode()+"数据迁移内容获取开始 #########################");
  61. QueryCriteria queryCriteria = getQueryCriteria(objType.getCode());
  62. dataMigrationExcels.addAll(startMigrateForExcel(context,migrationInfo,queryCriteria));
  63. log.info("######################### dt_object "+objType.getCode()+"数据迁移内容获取结束 #########################");
  64. }
  65. long end = System.currentTimeMillis();
  66. log.info("######################### dt_object 数数据迁移内容获取已结束 时间:"+(end-start)+" #########################");
  67. return dataMigrationExcels;
  68. }
  69. /**
  70. * 迁移数据,汇总记录日志
  71. * @param context
  72. * @param migrationInfo
  73. * @return
  74. */
  75. @Override
  76. public AdmResponse migrateForLog(InstanceUrlParam context, MigrationInfo migrationInfo) {
  77. log.info("######################### dt_object 同步开始 #########################");
  78. long start = System.currentTimeMillis();
  79. for(MigrationObjType objType : MigrationObjType.values()){
  80. log.info("######################### dt_object "+objType.getCode()+"同步开始 #########################");
  81. QueryCriteria queryCriteria = getQueryCriteria(objType.getCode());
  82. List<SyncData> syncDataList = startMigrateForLog(context,migrationInfo,queryCriteria);
  83. log.info("######################### dt_object "+objType.getCode()+"同步结束 共计:"+syncDataList.size()+"条 #########################");
  84. }
  85. long end = System.currentTimeMillis();
  86. log.info("######################### dt_object 数据迁移已结束 时间:"+(end-start)+" #########################");
  87. return AdmResponse.success();
  88. }
  89. private QueryCriteria getQueryCriteria(String typeCode){
  90. QueryCriteria queryCriteria = new QueryCriteria();
  91. queryCriteria.setSize(-1L);
  92. ObjectNode criteria = JsonNodeFactory.instance.objectNode();
  93. criteria.put("valid", 1);
  94. criteria.put("objType", typeCode);
  95. queryCriteria.setCriteria(criteria);
  96. return queryCriteria;
  97. }
  98. /**
  99. * 迁移数据,记录日志
  100. *
  101. */
  102. private List<SyncData> startMigrateForLog(InstanceUrlParam context, MigrationInfo migrationInfo, QueryCriteria queryCriteria) {
  103. List<SyncData> syncDataList = new ArrayList<>();
  104. List<ObjectNode> admData = getAdmData(context,queryCriteria);
  105. String queryUrl = requestUrl(context, migrationInfo, MigrationType.QUERY.getCode());
  106. List<ObjectDigital> projectData = getProjectData(queryCriteria, ObjectDigital.class, queryUrl);
  107. List<ObjectNode> objectNodeList = JsonNodeUtils.toListNode(projectData, null, null);
  108. if(CollUtil.isEmpty(admData)){
  109. log.error(" ######################### dt_object adm已交付:"+ ResponseCode.C0320.getDesc() +" 同步结束 #########################");
  110. }
  111. String insertUrl = requestUrl(context, migrationInfo, MigrationType.CREATE.getCode());
  112. if(CollUtil.isEmpty(objectNodeList)){
  113. DataMigrationResponse dataMigrationResponse = insertBatch(admData, ObjectDigital.class, insertUrl);
  114. syncDataList = processDataForLog(dataMigrationResponse, MigrationType.CREATE.getCode());
  115. }
  116. Map<String,Object> projectMap = toEntityMap(objectNodeList, ObjectDigital.class);
  117. Map<String,Object> admMap = toEntityMap(admData, ObjectDigital.class);
  118. //差集 新增
  119. Map<String,Object> doSubtractFromInsert = doSubtractAdd(admMap, projectMap);
  120. if(!CollUtil.isEmpty(doSubtractFromInsert)){
  121. List<ObjectNode> insertData = toList(doSubtractFromInsert, admData);
  122. DataMigrationResponse dataMigrationResponse = insertBatch(insertData, ObjectDigital.class, insertUrl);
  123. syncDataList.addAll(processDataForLog(dataMigrationResponse, MigrationType.CREATE.getCode()));
  124. }
  125. //差集 删除
  126. Map<String,Object> doSubtractFromDelete = doSubtractDel(admMap, projectMap);
  127. if(!CollUtil.isEmpty(doSubtractFromDelete)){
  128. List<String> deleteIds = doSubtractFromDelete.entrySet().stream().map(en -> en.getKey()).collect(Collectors.toList());
  129. String delUrl = requestUrl(context, migrationInfo, MigrationType.DELETE.getCode());
  130. //处理删除的数据
  131. DataMigrationResponse dataMigrationResponse = deleteBatch(deleteIds, ObjectDigital.class, delUrl);
  132. List<String> successIds = (List<String>) dataMigrationResponse.getData();
  133. List<ObjectNode> delObjs = toListByIds(successIds, admData);
  134. dataMigrationResponse.setData(delObjs);
  135. syncDataList.addAll(processDataForLog(dataMigrationResponse, MigrationType.DELETE.getCode()));
  136. }
  137. //交集更新
  138. Map<String,Object> intersectionUpdateFrom = doIntersectionGetFrom(admMap, projectMap);
  139. Map<String,Object> intersectionUpdateTo = doIntersectionGetTo(admMap, projectMap);
  140. if(!CollUtil.isEmpty(intersectionUpdateFrom) && !CollUtil.isEmpty(intersectionUpdateTo)){
  141. Map<String, Object> compareData = compareData(intersectionUpdateFrom, intersectionUpdateTo);
  142. if(CollUtil.isNotEmpty(compareData) && compareData.size() > 0){
  143. List<ObjectNode> updateData = toList(compareData, admData);
  144. String updateUrl = requestUrl(context, migrationInfo, MigrationType.UPDATE.getCode());
  145. DataMigrationResponse dataMigrationResponse = updateBatch(updateData, ObjectDigital.class, updateUrl);
  146. syncDataList.addAll(processDataForLog(dataMigrationResponse, MigrationType.UPDATE.getCode()));
  147. }
  148. }
  149. return super.addSynLog(context, syncDataList);
  150. }
  151. /**
  152. * 迁移数据,记录excel数据
  153. *
  154. */
  155. private List<DataMigrationExcel> startMigrateForExcel(InstanceUrlParam context, MigrationInfo migrationInfo, QueryCriteria queryCriteria) {
  156. List<DataMigrationExcel> dataMigrationExcels = new ArrayList<>();
  157. List<ObjectNode> admData = getAdmData(context,queryCriteria);
  158. String queryUrl = requestUrl(context, migrationInfo, MigrationType.QUERY.getCode());
  159. List<ObjectDigital> projectData = getProjectData(queryCriteria, ObjectDigital.class, queryUrl);
  160. List<ObjectNode> objectNodeList = JsonNodeUtils.toListNode(projectData, null, null);
  161. if(CollUtil.isEmpty(admData)){
  162. log.error(" ######################### dt_object adm已交付:"+ ResponseCode.C0320.getDesc() +" 同步结束 #########################");
  163. return Collections.emptyList();
  164. }
  165. //String insertUrl = requestUrl(context, migrationInfo, MigrationType.INSERT.getCode());
  166. if(CollUtil.isEmpty(objectNodeList)){
  167. //DataMigrationResponse dataMigrationResponse = insertBatch(admData, ObjectDigital.class, insertUrl);
  168. dataMigrationExcels.addAll(processDataForExcel(admData, MigrationType.CREATE.getCode()));
  169. return dataMigrationExcels;
  170. }
  171. Map<String,Object> projectMap = toEntityMap(objectNodeList, ObjectDigital.class);
  172. Map<String,Object> admMap = toEntityMap(admData, ObjectDigital.class);
  173. //差集 新增
  174. Map<String,Object> doSubtractFromInsert = doSubtractAdd(admMap, projectMap);
  175. if(!CollUtil.isEmpty(doSubtractFromInsert)){
  176. List<ObjectNode> insertData = toList(doSubtractFromInsert, admData);
  177. //DataMigrationResponse dataMigrationResponse = insertBatch(insertData, ObjectDigital.class, insertUrl);
  178. dataMigrationExcels.addAll(processDataForExcel(insertData, MigrationType.CREATE.getCode()));
  179. }
  180. //差集 删除
  181. Map<String,Object> doSubtractFromDelete = doSubtractDel(admMap, projectMap);
  182. if(!CollUtil.isEmpty(doSubtractFromDelete)){
  183. List<String> deleteIds = doSubtractFromDelete.entrySet().stream().map(en -> en.getKey()).collect(Collectors.toList());
  184. List<ObjectNode> deleteData = toListByIds(deleteIds, admData);
  185. //String delUrl = requestUrl(context, migrationInfo, MigrationType.DELETE.getCode());
  186. //DataMigrationResponse dataMigrationResponse = deleteBatch(deleteIds, ObjectDigital.class, delUrl);
  187. dataMigrationExcels.addAll(processDataForExcel(deleteData, MigrationType.DELETE.getCode()));
  188. }
  189. //交集更新
  190. Map<String,Object> intersectionUpdateFrom = doIntersectionGetFrom(admMap, projectMap);
  191. Map<String,Object> intersectionUpdateTo = doIntersectionGetTo(admMap, projectMap);
  192. if(!CollUtil.isEmpty(intersectionUpdateFrom) && !CollUtil.isEmpty(intersectionUpdateTo)){
  193. Map<String, Object> compareData = compareData(intersectionUpdateFrom, intersectionUpdateTo);
  194. if(CollUtil.isNotEmpty(compareData) && compareData.size() > 0){
  195. List<ObjectNode> updateData = toList(compareData, admData);
  196. //String updateUrl = requestUrl(context, migrationInfo, MigrationType.UPDATE.getCode());
  197. //DataMigrationResponse dataMigrationResponse = updateBatch(updateData, ObjectDigital.class, updateUrl);
  198. dataMigrationExcels.addAll(processDataForExcel(updateData, MigrationType.UPDATE.getCode()));
  199. }
  200. }
  201. return dataMigrationExcels;
  202. }
  203. /**
  204. * 获取adm交付的数据,默认获取有效数据
  205. *
  206. * @param context
  207. * @return
  208. */
  209. @Override
  210. public List<ObjectNode> getAdmData(InstanceUrlParam context, QueryCriteria queryCriteria) {
  211. List<ObjectNode> digitalList = DigitalObjectFacade.query(context.getGroupCode(),context.getProjectId(),context.getUserId(),context.getUserId(),queryCriteria);
  212. if(CollUtil.isEmpty(digitalList)){
  213. return Collections.emptyList();
  214. }
  215. return digitalList;
  216. }
  217. /**
  218. * 根据map转为需要处理的对象集合
  219. * @param fromMap
  220. * @param fromList
  221. * @return
  222. */
  223. public List<ObjectNode> toList(Map<String, Object> fromMap, List<ObjectNode> fromList) {
  224. if(CollUtil.isEmpty(fromMap) || CollUtil.isEmpty(fromList)){
  225. return Collections.emptyList();
  226. }
  227. return toListByIds(fromMap.keySet(), fromList);
  228. }
  229. /**
  230. * 根据对象ids转为需要处理的对象集合
  231. * @param ids
  232. * @param fromList
  233. * @return
  234. */
  235. public List<ObjectNode> toListByIds(Collection<String> ids, List<ObjectNode> fromList) {
  236. if(CollUtil.isEmpty(ids) || CollUtil.isEmpty(fromList)){
  237. return Collections.emptyList();
  238. }
  239. List<ObjectNode> results = new ArrayList<>(ids.size());
  240. Map<String, ObjectNode> objectDigitalMap = new HashMap<>(fromList.size());
  241. for(ObjectNode node : fromList){
  242. if(!node.get("id").isTextual()){
  243. continue;
  244. }
  245. objectDigitalMap.put(node.get("id").asText(), node);
  246. }
  247. for(String key : ids){
  248. results.add(objectDigitalMap.get(key));
  249. }
  250. return results;
  251. }
  252. /**
  253. * 拼接访问链接
  254. *
  255. * @param context
  256. * @param migrationInfo
  257. * @param type
  258. * @return
  259. */
  260. @Override
  261. public String requestUrl(InstanceUrlParam context, MigrationInfo migrationInfo, String type) {
  262. if(StrUtil.isBlank(migrationInfo.getTargetUrl())){
  263. return null;
  264. }
  265. StringBuilder url = new StringBuilder(migrationInfo.getTargetUrl());
  266. if(type.equals(MigrationType.QUERY.getCode())) {
  267. url.append(MigrationConstant.DT_OBJECT);
  268. }else if(type.equals(MigrationType.CREATE.getCode())){
  269. url.append(MigrationConstant.DT_OBJECT_INSERT);
  270. }else if(type.equals(MigrationType.UPDATE.getCode())){
  271. url.append(MigrationConstant.DT_OBJECT_UPDATE);
  272. }else{
  273. url.append(MigrationConstant.DT_OBJECT_DELETE);
  274. }
  275. return commonUrl(context, url.toString(), migrationInfo);
  276. }
  277. /**
  278. * 将目标集合处理成excel格式的集合 使用与insert/delete
  279. *
  280. * @param list
  281. * @return
  282. */
  283. @Override
  284. public List<DataMigrationExcel> processDataForExcel(Collection list, String type) {
  285. if(CollUtil.isEmpty(list)){
  286. return Collections.emptyList();
  287. }
  288. List<DataMigrationExcel> dataMigrationExcels = new ArrayList<>();
  289. for(Object obj : list){
  290. ObjectDigital digital = JsonNodeUtils.toEntity((ObjectNode) obj, ObjectDigital.class, null);
  291. DataMigrationExcel excel = new DataMigrationExcel();
  292. excel.setId(digital.getId());
  293. excel.setName(String.join(StrUtil.DASHED, digital.getLocalId(), digital.getLocalName()));
  294. excel.setType(type);
  295. dataMigrationExcels.add(excel);
  296. }
  297. return dataMigrationExcels;
  298. }
  299. /**
  300. * 将目标集合处理成SyncData(日志)格式的集合 入库
  301. *
  302. * @param dataMigrationResponse
  303. * @param type insert/delete
  304. * @return
  305. */
  306. @Override
  307. public List<SyncData> processDataForLog(DataMigrationResponse dataMigrationResponse, String type) {
  308. if(CollUtil.isEmpty(dataMigrationResponse.getData()) && CollUtil.isEmpty(dataMigrationResponse.getErrorMsg())){
  309. return Collections.emptyList();
  310. }
  311. List<SyncData> syncDataList = new ArrayList<>();
  312. if(CollUtil.isNotEmpty(dataMigrationResponse.getData())){
  313. for(Object obj : dataMigrationResponse.getData()){
  314. ObjectDigital digital = (ObjectDigital) obj;
  315. SyncData syncData = new SyncData();
  316. syncData.setTableName(MigrationTable.DTOBJECT.getCode());
  317. syncData.setName(digital.getName());
  318. syncData.setObjectType(digital.getObjType());
  319. syncData.setType(type);//操作
  320. syncData.setClassCode(digital.getClassCode());
  321. syncData.setTargetId(digital.getId());
  322. syncData.setLocalId(digital.getLocalId());
  323. syncData.setLocalName(digital.getLocalName());
  324. syncDataList.add(syncData);
  325. }
  326. }
  327. if(CollUtil.isNotEmpty(dataMigrationResponse.getErrorMsg())){
  328. syncDataList.addAll(addErrorInfoForLog(dataMigrationResponse.getErrorMsg(), MigrationTable.DTOBJECT.getCode()));
  329. }
  330. return syncDataList;
  331. }
  332. }