first commit
This commit is contained in:
8324144
@@ -0,0 +1,457 @@
|
||||
package cn.wepact.dfm.training.service;
|
||||
|
||||
import cn.hutool.json.JSONArray;
|
||||
import cn.hutool.json.JSONException;
|
||||
import cn.hutool.json.JSONObject;
|
||||
import cn.wepact.dfm.training.dto.SyncRequest;
|
||||
import cn.wepact.dfm.training.dto.entity.SyncTaskLog;
|
||||
import cn.wepact.dfm.training.dto.entity.SyncUserInfoLog;
|
||||
import cn.wepact.dfm.training.mapper.SyncUserInfoLogMapper;
|
||||
import cn.wepact.dfm.training.mapper.TUserMapper;
|
||||
import cn.wepact.dfm.training.util.HttpClientUtil;
|
||||
import com.alibaba.fastjson.JSON;
|
||||
// 修改 import 语句
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
// 删除这两行
|
||||
// import com.itextpdf.text.log.Logger;
|
||||
// import com.itextpdf.text.log.LoggerFactory;
|
||||
import org.apache.http.HttpResponse;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.impl.client.HttpClients;
|
||||
import org.apache.http.util.EntityUtils;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.beans.factory.annotation.Value;
|
||||
import org.springframework.dao.DataAccessException;
|
||||
import org.springframework.http.HttpStatus;
|
||||
import org.springframework.http.ResponseEntity;
|
||||
import org.springframework.stereotype.Service;
|
||||
|
||||
import javax.transaction.Transactional;
|
||||
import java.math.BigDecimal;
|
||||
import java.sql.Timestamp;
|
||||
import java.text.ParseException;
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Date;
|
||||
import java.util.HashMap;
|
||||
import java.util.Map;
|
||||
|
||||
@Service
|
||||
public class SyncUserInfoService {
|
||||
|
||||
private static final Logger logger = LoggerFactory.getLogger(SyncUserInfoService.class);
|
||||
@Autowired
|
||||
private SyncUserInfoLogMapper syncUserInfoMapper;
|
||||
@Autowired
|
||||
private TUserMapper userMapper;
|
||||
@Value("${api.All_API_USER_URL}")
|
||||
private String All_API_USER_URL;
|
||||
@Value("${api.Incr_API_USER_URL}")
|
||||
private String Incr_API_USER_URL;
|
||||
@Value("${api.getAPITokenUrl}")
|
||||
private String getAPITokenUrl;
|
||||
|
||||
@Transactional(rollbackOn = Exception.class)
|
||||
public ResponseEntity<String> syncUserInfos(SyncRequest syncRequest) {
|
||||
try {
|
||||
// 执行同步逻辑
|
||||
// 如果成功
|
||||
int offset =syncRequest.getOffset();
|
||||
int limit = syncRequest.getLimit();
|
||||
String searchStartTime = syncRequest.getSearchStartTime();
|
||||
String searchEndTime = syncRequest.getSearchEndTime();
|
||||
String syncType = syncRequest.getSyncType();
|
||||
int duration = syncRequest.getDuration();
|
||||
if ("full".equalsIgnoreCase(syncType)) {
|
||||
// 全量同步，使用请求中传入的时间区间
|
||||
fullSync(offset, limit, searchStartTime, searchEndTime);
|
||||
} else if ("incremental".equalsIgnoreCase(syncType)) {
|
||||
// 增量同步，通常使用最后一次同步的时间戳
|
||||
incrementalSync(offset, limit,duration);
|
||||
}else {
|
||||
throw new IllegalArgumentException("Invalid syncType: " + syncType);
|
||||
}
|
||||
return ResponseEntity.ok("同步用户成功");
|
||||
} catch (Exception e) {
|
||||
// 如果发生异常
|
||||
return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR).body("同步用户失败: " + e.getMessage());
|
||||
}
|
||||
}
|
||||
private void fullSync(int offset, int limit, String searchStartTime, String searchEndTime) {
|
||||
while (true) {
|
||||
// 请求数据
|
||||
Map<String, Object> requestParams = new HashMap<>();
|
||||
requestParams.put("offset", offset);
|
||||
requestParams.put("limit", limit);
|
||||
requestParams.put("searchStartTime", searchStartTime);
|
||||
requestParams.put("searchEndTime", searchEndTime);
|
||||
|
||||
// 调用API获取数据
|
||||
String response = callApi(All_API_USER_URL, requestParams);
|
||||
|
||||
// 如果返回数据为空，则结束同步
|
||||
if (response == null || response.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
|
||||
try {
|
||||
// 将返回的数据解析为 JSONObject
|
||||
JSONObject responseJson = new JSONObject(response);
|
||||
|
||||
// 检查返回的 "data" 是否为空
|
||||
if (responseJson.containsKey("data") && responseJson.getJSONArray("data").size() == 0) {
|
||||
System.out.println("Data is empty, exiting the loop.");
|
||||
break; // 如果 "data" 为空，跳出循环
|
||||
}
|
||||
|
||||
// 在这里处理 "data" 非空的情况
|
||||
// 你可以继续进行数据处理
|
||||
// ...
|
||||
|
||||
} catch (JSONException e) {
|
||||
// 如果 JSON 解析出错，打印异常并跳出循环
|
||||
System.err.println("Error parsing JSON response: " + e.getMessage());
|
||||
break;
|
||||
}
|
||||
// 解析返回数据并保存到数据库
|
||||
saveSyncUserData(offset, limit, requestParams, response);
|
||||
|
||||
// 更新offset值，进行下一次同步
|
||||
offset += limit;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
|
||||
private void incrementalSync(int offset, int limit,int duration) {
|
||||
// 获取上次同步的时间点，通常从数据库中获取
|
||||
String lastSyncTime = getLastSyncTimeFromDatabase();
|
||||
|
||||
// 增量同步，使用上次同步的时间点作为查询的起始时间
|
||||
String searchStartTime = lastSyncTime;
|
||||
String searchEndTime = getCurrentTime();
|
||||
|
||||
// 请求数据
|
||||
while (true) {
|
||||
Map<String, Object> requestParams = new HashMap<>();
|
||||
requestParams.put("offset", offset);
|
||||
requestParams.put("limit", limit);
|
||||
requestParams.put("searchStartTime", searchStartTime);
|
||||
requestParams.put("searchEndTime", searchEndTime);
|
||||
requestParams.put("duration", duration);
|
||||
// 调用API获取数据
|
||||
String response = callApi(Incr_API_USER_URL, requestParams);
|
||||
// 如果返回数据为空，则结束同步
|
||||
if (response == null || response.isEmpty()) {
|
||||
break;
|
||||
}
|
||||
|
||||
try {
|
||||
// 将返回的数据解析为 JSONObject
|
||||
JSONObject responseJson = new JSONObject(response);
|
||||
|
||||
// 检查返回的 "data" 是否为空
|
||||
if (responseJson.containsKey("data") && responseJson.getJSONArray("data").size() == 0) {
|
||||
System.out.println("Data is empty, exiting the loop.");
|
||||
break; // 如果 "data" 为空，跳出循环
|
||||
}
|
||||
|
||||
// 在这里处理 "data" 非空的情况
|
||||
// 你可以继续进行数据处理
|
||||
// ...
|
||||
|
||||
} catch (JSONException e) {
|
||||
// 如果 JSON 解析出错，打印异常并跳出循环
|
||||
System.err.println("Error parsing JSON response: " + e.getMessage());
|
||||
break;
|
||||
}
|
||||
// 解析返回数据并保存到数据库
|
||||
saveSyncUserData(offset, limit, requestParams, response);
|
||||
|
||||
// 更新offset值，进行下一次同步
|
||||
offset += limit;
|
||||
}
|
||||
|
||||
// 增量同步后，更新最后同步时间
|
||||
updateLastSyncTimeToDatabase(getCurrentTime());
|
||||
}
|
||||
private void saveSyncUserData(int offset, int limit, Map<String, Object> params, String responseData) {
|
||||
// 检查入参
|
||||
if (responseData == null || responseData.isEmpty()) {
|
||||
logger.error("响应数据为空");
|
||||
return;
|
||||
}
|
||||
|
||||
JSONObject responseJson = null;
|
||||
JSONArray dataArray = null;
|
||||
try {
|
||||
responseJson = new JSONObject(responseData);
|
||||
// 检查data字段是否存在且不为空
|
||||
if (!responseJson.containsKey("data")) {
|
||||
logger.error("响应数据中没有data字段");
|
||||
return;
|
||||
}
|
||||
|
||||
dataArray = responseJson.getJSONArray("data");
|
||||
if (dataArray == null || dataArray.size() == 0) {
|
||||
logger.info("没有需要同步的数据");
|
||||
return;
|
||||
}
|
||||
|
||||
for (int i = 0; i < dataArray.size(); i++) {
|
||||
JSONObject dataObject = dataArray.getJSONObject(i);
|
||||
if (dataObject == null) {
|
||||
logger.warn("第{}条数据为空，跳过处理", i);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 获取用户名，进行空值检查
|
||||
String username = getStringValue(dataObject, "username");
|
||||
if (username == null || username.trim().isEmpty()) {
|
||||
logger.warn("用户名为空，跳过处理第{}条数据", i);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 获取系统内部的user_id
|
||||
Long userId = userMapper.findUserIdByUsername(username);
|
||||
if (userId == null) {
|
||||
logger.warn("未找到用户: {}", username);
|
||||
continue;
|
||||
}
|
||||
|
||||
// 后续代码保持不变
|
||||
SyncUserInfoLog userInfo = new SyncUserInfoLog();
|
||||
// 填充 SyncTaskLog 对象的字段
|
||||
userInfo.setSyncTime(new Timestamp(System.currentTimeMillis()));
|
||||
userInfo.setOffset(offset);
|
||||
userInfo.setSyncLimit(limit);
|
||||
userInfo.setSearchStartTime(parseTimestamp((String) params.get("searchStartTime")));
|
||||
userInfo.setSearchEndTime(parseTimestamp((String) params.get("searchEndTime")));
|
||||
userInfo.setStatus(1); // 假设同步成功
|
||||
|
||||
userInfo.setId(getStringValue(dataObject, "id"));
|
||||
userInfo.setUserId(getStringValue(dataObject, "userId"));
|
||||
userInfo.setUsername(getStringValue(dataObject, "username"));
|
||||
userInfo.setFullname(getStringValue(dataObject, "fullname"));
|
||||
userInfo.setGender(getIntValue(dataObject, "gender"));
|
||||
userInfo.setIdNo(getStringValue(dataObject, "idNo"));
|
||||
userInfo.setBirthday(getDateValue(dataObject, "birthday"));
|
||||
userInfo.setDeptId(getStringValue(dataObject, "deptId"));
|
||||
userInfo.setDeptName(getStringValue(dataObject, "deptName"));
|
||||
userInfo.setPositionId(getStringValue(dataObject, "positionId"));
|
||||
userInfo.setPositionName(getStringValue(dataObject, "positionName"));
|
||||
userInfo.setGradeId(getStringValue(dataObject, "gradeId"));
|
||||
userInfo.setGradeName(getStringValue(dataObject, "gradeName"));
|
||||
userInfo.setManagerId(getStringValue(dataObject, "managerId"));
|
||||
userInfo.setManagerFullname(getStringValue(dataObject, "managerFullname"));
|
||||
userInfo.setDeptManagerId(getStringValue(dataObject, "deptManagerId"));
|
||||
userInfo.setDeptManagerFullname(getStringValue(dataObject, "deptManagerFullname"));
|
||||
userInfo.setMobile(getStringValue(dataObject, "mobile"));
|
||||
userInfo.setEmail(getStringValue(dataObject, "email"));
|
||||
userInfo.setUserNo(getStringValue(dataObject, "userNo"));
|
||||
userInfo.setHireDate(getDateValue(dataObject, "hireDate"));
|
||||
userInfo.setExpiredTime(getDateValue(dataObject, "expiredTime"));
|
||||
userInfo.setAreaCode(getStringValue(dataObject, "areaCode"));
|
||||
userInfo.setUserStatus(getIntValue(dataObject, "userStatus"));
|
||||
userInfo.setAvatarUrl(getStringValue(dataObject, "avatarUrl"));
|
||||
userInfo.setAdmin(getIntValue(dataObject, "admin"));
|
||||
userInfo.setDeleted(getIntValue(dataObject, "deleted"));
|
||||
userInfo.setThirdUserId(getStringValue(dataObject, "thirdUserId"));
|
||||
userInfo.setCreateTime(getDateValue(dataObject, "createTime"));
|
||||
userInfo.setUpdateTime(getDateValue(dataObject, "updateTime"));
|
||||
userInfo.setUserType(getIntValue(dataObject, "userType"));
|
||||
// 额外存储API响应的原始数据和错误信息
|
||||
userInfo.setResponseData(responseData);
|
||||
userInfo.setErrorMessage(optString(dataObject, "errorMessage", ""));
|
||||
// Check if the user already exists in the database
|
||||
try {
|
||||
SyncUserInfoLog existingUser = syncUserInfoMapper.getUserInfoByUserId(userInfo.getUserId());
|
||||
if (existingUser == null) {
|
||||
// New user, insert into database
|
||||
try {
|
||||
syncUserInfoMapper.insertUserInfo(userInfo);
|
||||
logger.info("Successfully inserted new user: {}", userInfo.getUserId());
|
||||
} catch (DataAccessException e) {
|
||||
logger.error("Failed to insert new user: {}, error: {}", userInfo.getUserId(), e.getMessage());
|
||||
userInfo.setStatus(0); // 设置同步状态为失败
|
||||
userInfo.setErrorMessage(e.getMessage());
|
||||
syncUserInfoMapper.insertUserInfo(userInfo); // 保存失败记录
|
||||
continue; // 继续处理下一条记录
|
||||
}
|
||||
} else {
|
||||
// Existing user, update the user info
|
||||
try {
|
||||
syncUserInfoMapper.updateUserInfo(userInfo);
|
||||
logger.info("Successfully updated user: {}", userInfo.getUserId());
|
||||
} catch (DataAccessException e) {
|
||||
logger.error("Failed to update user: {}, error: {}", userInfo.getUserId(), e.getMessage());
|
||||
userInfo.setStatus(0); // 设置同步状态为失败
|
||||
userInfo.setErrorMessage(e.getMessage());
|
||||
syncUserInfoMapper.updateUserInfo(userInfo); // 保存失败记录
|
||||
continue; // 继续处理下一条记录
|
||||
}
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error("Error processing user: {}, error: {}", userInfo.getUserId(), e.getMessage());
|
||||
continue; // 继续处理下一条记录
|
||||
}
|
||||
|
||||
}
|
||||
} catch (Exception e) {
|
||||
logger.error("同步数据处理失败：", e);
|
||||
throw e; // 抛出异常，跳出循环
|
||||
}
|
||||
|
||||
}
|
||||
public String getLastSyncTimeFromDatabase() {
|
||||
String formattedLastSyncTime = null;
|
||||
try {
|
||||
// 调用Mapper方法获取时间
|
||||
Date lastSyncTime = syncUserInfoMapper.getLastSyncTime();
|
||||
// Check if lastSyncTime is not null
|
||||
if (lastSyncTime != null) {
|
||||
// Format the date to a readable string (e.g., "yyyy-MM-dd HH:mm:ss")
|
||||
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
||||
formattedLastSyncTime = dateFormat.format(lastSyncTime);
|
||||
} else {
|
||||
formattedLastSyncTime = "No sync time recorded.";
|
||||
}
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
formattedLastSyncTime = "Error retrieving last sync time.";
|
||||
}
|
||||
return formattedLastSyncTime;
|
||||
}
|
||||
|
||||
private String getCurrentTime() {
|
||||
// 获取当前时间，用作增量同步结束的时间
|
||||
// 创建日期格式化对象
|
||||
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
||||
|
||||
// 获取当前系统时间
|
||||
Date now = new Date();
|
||||
|
||||
// 格式化当前时间并返回字符串
|
||||
return dateFormat.format(now);
|
||||
}
|
||||
|
||||
private void updateLastSyncTimeToDatabase(String lastSyncTime) {
|
||||
// 更新最后同步时间到数据库
|
||||
try {
|
||||
// 将字符串转换为 Date 对象
|
||||
SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
|
||||
Date date = dateFormat.parse(lastSyncTime);
|
||||
|
||||
// 调用 Mapper 方法更新最后同步时间
|
||||
syncUserInfoMapper.updateLastSyncTime(date);
|
||||
|
||||
System.out.println("最后同步时间已更新: " + lastSyncTime);
|
||||
} catch (ParseException e) {
|
||||
e.printStackTrace();
|
||||
System.err.println("时间格式错误: " + lastSyncTime);
|
||||
} catch (Exception e) {
|
||||
e.printStackTrace();
|
||||
System.err.println("更新数据库时发生错误");
|
||||
}
|
||||
}
|
||||
// 调用接口
|
||||
private String callApi(String apiUrl, Map<String, Object> params) {
|
||||
// 假设 accessToken 是从某个地方获取的，比如从配置文件、Session 或者前端传递
|
||||
String accessToken = getAccessToken(); // 你可以根据自己的需求获取 token
|
||||
|
||||
// 调用 HttpClientUtil 的 doPost 方法
|
||||
String response = HttpClientUtil.doPost(apiUrl, params, accessToken);
|
||||
|
||||
// 如果需要，你可以进一步处理响应结果
|
||||
if (response != null && !response.isEmpty()) {
|
||||
// 处理响应数据，比如解析 JSON 或者记录日志
|
||||
return response;
|
||||
} else {
|
||||
// 错误处理
|
||||
System.out.println("API 请求失败，未收到响应");
|
||||
return null;
|
||||
}
|
||||
}
|
||||
// 假设这是获取 accessToken 的方法，根据需要修改
|
||||
private String getAccessToken() {
|
||||
String apiUrl =getAPITokenUrl;
|
||||
String accessToken = null;
|
||||
|
||||
try (CloseableHttpClient client = HttpClients.createDefault()) {
|
||||
HttpGet getRequest = new HttpGet(apiUrl);
|
||||
|
||||
// 执行请求
|
||||
HttpResponse response = client.execute(getRequest);
|
||||
|
||||
// 获取响应内容
|
||||
String responseString = EntityUtils.toString(response.getEntity());
|
||||
|
||||
// 解析响应 JSON
|
||||
com.alibaba.fastjson.JSONObject responseObject = JSON.parseObject(responseString);
|
||||
accessToken = responseObject.getString("accessToken"); // 假设返回的是 {"accessToken": "your-token"}
|
||||
} catch (Exception e) {
|
||||
System.out.println("获取 accessToken 失败: " + e.getMessage());
|
||||
}
|
||||
|
||||
return accessToken;
|
||||
}
|
||||
public static void main(String[] args) {
|
||||
|
||||
// String token = getAccessToken();
|
||||
// System.out.println("Access Token: " + token);
|
||||
}
|
||||
|
||||
// 辅助方法：转换字符串为时间戳
|
||||
private Timestamp parseTimestamp(String value) {
|
||||
try {
|
||||
return value != null && !value.isEmpty() ? Timestamp.valueOf(value) : null;
|
||||
} catch (IllegalArgumentException e) {
|
||||
// logger.error("时间转换失败: " + value, e);
|
||||
return null;
|
||||
}
|
||||
}
|
||||
private Date getDateValue(JSONObject json, String key) {
|
||||
String value = getStringValue(json, key);
|
||||
try {
|
||||
return value != null && !value.isEmpty() ? Timestamp.valueOf(value) : null;
|
||||
} catch (IllegalArgumentException e) {
|
||||
return null; // Invalid date format, return null
|
||||
}
|
||||
}
|
||||
// 获取 Long 值，若为空则返回 null
|
||||
private Long getLongValue(JSONObject json, String key) {
|
||||
return json.containsKey(key) && json.get(key) != null ? json.getLong(key) : null;
|
||||
}
|
||||
|
||||
// 获取 String 值，若为空则返回空字符串
|
||||
private String getStringValue(JSONObject json, String key) {
|
||||
return json.containsKey(key) && json.get(key) != null ? json.getStr(key) : "";
|
||||
}
|
||||
|
||||
// 获取 Integer 值，若为空则返回 0
|
||||
private Integer getIntValue(JSONObject json, String key) {
|
||||
if (json != null && json.containsKey(key) && json.get(key) != null) {
|
||||
Object value = json.get(key);
|
||||
if (value instanceof Integer) {
|
||||
return (Integer) value; // 如果值是 Integer，直接返回
|
||||
} else {
|
||||
try {
|
||||
return Integer.parseInt(value.toString()); // 如果是其他类型，转换为 Integer
|
||||
} catch (NumberFormatException e) {
|
||||
return 0; // 转换失败，返回 0
|
||||
}
|
||||
}
|
||||
}
|
||||
return 0; // 如果 json 为 null，或者 key 不存在，或者值是 JSONObject.NULL，返回 0
|
||||
}
|
||||
|
||||
// 获取 BigDecimal 值，若为空则返回 null
|
||||
private BigDecimal getBigDecimalValue(JSONObject json, String key) {
|
||||
return json.containsKey(key) && json.get(key) != null ? json.getBigDecimal(key) : null;
|
||||
}
|
||||
// 实现类似于 org.json.JSONObject 的 optString 方法
|
||||
public String optString(JSONObject json, String key, String defaultValue) {
|
||||
return json.containsKey(key) && json.get(key) != null ? json.getStr(key) : defaultValue;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user