481 lines
21 KiB
Java
481 lines
21 KiB
Java
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.dto.entity.TUser;
|
||
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.ArrayList;
|
||
import java.util.Date;
|
||
import java.util.HashMap;
|
||
import java.util.List;
|
||
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;
|
||
}
|
||
|
||
// 第一遍：收集所有用户名，批量查询用户ID（替代 N+1 逐条查询）
|
||
List<String> usernames = new ArrayList<>(dataArray.size());
|
||
for (int i = 0; i < dataArray.size(); i++) {
|
||
JSONObject dataObject = dataArray.getJSONObject(i);
|
||
if (dataObject == null) continue;
|
||
String u = getStringValue(dataObject, "username");
|
||
if (!u.isEmpty()) {
|
||
usernames.add(u);
|
||
}
|
||
}
|
||
Map<String, Long> userIdCache = new HashMap<>();
|
||
if (!usernames.isEmpty()) {
|
||
List<TUser> tUsers = userMapper.batchFindUserIdByUsernames(usernames);
|
||
for (TUser tUser : tUsers) {
|
||
if (tUser.getUserNo() != null) {
|
||
userIdCache.put(tUser.getUserNo(), tUser.getId());
|
||
}
|
||
}
|
||
}
|
||
|
||
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 = userIdCache.get(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;
|
||
}
|
||
}
|