package com.xiaopeng.operation.service.impl;
|
|
|
import cn.hutool.core.bean.BeanUtil;
|
import com.alibaba.fastjson.JSONObject;
|
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
|
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
|
import com.xiaopeng.api.dto.pushMessage.AliPushMessageDTO;
|
import com.xiaopeng.api.feignClient.account.XpOperationUserClient;
|
import com.xiaopeng.api.feignClient.account.XpOperationUserNotificationClient;
|
import com.xiaopeng.api.feignClient.pushMessage.XpAliPushMessageClient;
|
import com.xiaopeng.api.model.account.XpOperationsUser;
|
import com.xiaopeng.api.model.account.XpOperationsUserNotification;
|
import com.xiaopeng.api.model.operation.XpAgreement;
|
import com.xiaopeng.common.core.domain.R;
|
import com.xiaopeng.common.core.enums.NotificationTypeEnum;
|
import com.xiaopeng.operation.dto.XpAgreementDTO;
|
import com.xiaopeng.operation.mapper.XpAgreementMapper;
|
import com.xiaopeng.operation.service.XpAgreementService;
|
import com.xiaopeng.operation.vo.XpAgreementVo;
|
import lombok.RequiredArgsConstructor;
|
import lombok.extern.slf4j.Slf4j;
|
import org.springframework.context.annotation.Lazy;
|
import org.springframework.stereotype.Service;
|
|
import javax.annotation.Resource;
|
import java.time.LocalDate;
|
import java.time.format.DateTimeFormatter;
|
import java.util.*;
|
import java.util.concurrent.LinkedBlockingQueue;
|
import java.util.concurrent.ThreadPoolExecutor;
|
import java.util.concurrent.TimeUnit;
|
import java.util.stream.Collectors;
|
|
/**
|
* <p>
|
* 协议设置 服务实现类
|
* </p>
|
*
|
* @author liu
|
* @since 2026-02-03
|
*/
|
@Slf4j
|
@Service
|
@RequiredArgsConstructor(onConstructor_ = @Lazy)
|
public class XpAgreementServiceImpl extends ServiceImpl<XpAgreementMapper, XpAgreement> implements XpAgreementService {
|
|
private final XpOperationUserClient xpOperationUserClient;
|
|
private final XpOperationUserNotificationClient xpOperationUserNotificationClient;
|
|
private final XpAliPushMessageClient xpAliPushMessageClient;
|
|
@Override
|
public XpAgreementVo getContentByType(Integer type) {
|
XpAgreement xpAgreement = this.getOne(new LambdaQueryWrapper<XpAgreement>().eq(XpAgreement::getContentType,type));
|
if (xpAgreement != null) {
|
XpAgreementVo xpAgreementVo = new XpAgreementVo();
|
xpAgreementVo.setId(xpAgreement.getId());
|
xpAgreementVo.setContent(xpAgreement.getContent());
|
xpAgreementVo.setType(xpAgreement.getContentType());
|
xpAgreementVo.setVersion(xpAgreement.getVersion());
|
return xpAgreementVo;
|
}
|
return null;
|
}
|
|
@Override
|
public List<XpAgreementVo> getContentByTypeAll() {
|
List<XpAgreement> list = this.list(new LambdaQueryWrapper<XpAgreement>().in(XpAgreement::getContentType, Arrays.asList(1,2,3,4)));
|
List<XpAgreementVo> xpAgreementVos = new ArrayList<>();
|
for (XpAgreement xpAgreement : list) {
|
XpAgreementVo xpAgreementVo = new XpAgreementVo();
|
xpAgreementVo.setId(xpAgreement.getId());
|
xpAgreementVo.setContent(xpAgreement.getContent());
|
xpAgreementVo.setType(xpAgreement.getContentType());
|
xpAgreementVo.setVersion(xpAgreement.getVersion());
|
xpAgreementVos.add(xpAgreementVo);
|
}
|
return xpAgreementVos;
|
}
|
|
/**
|
* 根据类型获取配置
|
* @param type
|
* @return
|
*/
|
@Override
|
public XpAgreement getAgreementContent(Integer type) {
|
return this.lambdaQuery().eq(XpAgreement::getContentType, type).last("limit 1").one();
|
}
|
|
@Override
|
public void saveAgreement(XpAgreementDTO dto) {
|
//根据类型查询协议
|
Optional<XpAgreement> opt = this.lambdaQuery().eq(XpAgreement::getContentType, dto.getContentType()).last("limit 1").oneOpt();
|
|
//如果存在则更新，不存在则新增
|
if (opt.isPresent()) {
|
XpAgreement xpAgreement = opt.get();
|
xpAgreement.setContent(dto.getContent());
|
updateById(xpAgreement);
|
} else {
|
XpAgreement xpAgreement = BeanUtil.copyProperties(dto, XpAgreement.class);
|
this.save(xpAgreement);
|
}
|
|
// 判断是否需要发送通知（用户服务协议=1 或 平台隐私政策=2）
|
if (dto.getContentType() != null && (dto.getContentType() == 1 || dto.getContentType() == 2)) {
|
// 发送协议更新提示通知（type=9）
|
sendAgreementUpdateNotification(dto.getContentType());
|
// 发送协议授权通知（type=10）
|
sendAgreementAuthNotification(dto.getContentType());
|
}
|
}
|
|
/**
|
* 发送协议更新通知
|
* @param contentType 协议类型
|
*/
|
private void sendAgreementUpdateNotification(Integer contentType) {
|
// 异步处理，避免阻塞主流程
|
ThreadPoolExecutor executor = new ThreadPoolExecutor(
|
1, 1, 0L, TimeUnit.MILLISECONDS,
|
new LinkedBlockingQueue<Runnable>()
|
);
|
|
executor.execute(() -> {
|
try {
|
// 1. 获取所有启用状态的运维用户
|
R<List<XpOperationsUser>> result = xpOperationUserClient.listAllEnabled();
|
if (result == null || result.getCode() != 200 || result.getData() == null || result.getData().isEmpty()) {
|
log.warn("没有需要通知的启用状态用户");
|
return;
|
}
|
|
List<XpOperationsUser> enabledUsers = result.getData();
|
log.info("协议更新通知，共需通知 {} 位用户", enabledUsers.size());
|
|
// 2. 构建通知内容
|
String agreementTypeName = getAgreementTypeName(contentType);
|
String currentDate = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
String message = String.format("尊敬的用户，你好！%s已于%s更新了内容，可前往个人中心-协议与政策查看详情", agreementTypeName, currentDate);
|
String bodyJson = buildAgreementPushBodyJson(message);
|
|
// 3. 构建并保存通知记录
|
List<XpOperationsUserNotification> notifications = enabledUsers.stream()
|
.map(user -> buildAgreementNotification(message, bodyJson, user))
|
.collect(Collectors.toList());
|
|
// 批量保存通知记录
|
xpOperationUserNotificationClient.addNotificationBatch(notifications);
|
log.info("协议更新通知记录保存成功，共 {} 条", notifications.size());
|
|
// 4. 按设备类型分组并批量推送
|
pushMessageByDeviceType(enabledUsers, message);
|
|
} catch (Exception e) {
|
log.error("发送协议更新通知失败", e);
|
}
|
});
|
|
executor.shutdown();
|
}
|
|
/**
|
* 获取协议类型名称
|
* @param contentType 协议类型
|
* @return 协议类型名称
|
*/
|
private String getAgreementTypeName(Integer contentType) {
|
if (contentType == 1) {
|
return "用户服务协议";
|
} else if (contentType == 2) {
|
return "平台隐私政策";
|
}
|
return "协议";
|
}
|
|
/**
|
* 构建推送 body JSON
|
* @param message 消息内容
|
* @return JSON 字符串
|
*/
|
private String buildAgreementPushBodyJson(String message) {
|
JSONObject body = new JSONObject();
|
body.put("content", message);
|
body.put("strategy", NotificationTypeEnum.POLICY_CONTENT_CHANGED.getCode()); // 9
|
body.put("type", 0);
|
body.put("id", 0);
|
return body.toJSONString();
|
}
|
|
/**
|
* 构建通知实体
|
* @param message 消息内容
|
* @param bodyJson 推送body内容
|
* @param user 运维用户
|
* @return 通知实体
|
*/
|
private XpOperationsUserNotification buildAgreementNotification(String message, String bodyJson, XpOperationsUser user) {
|
XpOperationsUserNotification notification = new XpOperationsUserNotification();
|
notification.setNotificationType(NotificationTypeEnum.POLICY_CONTENT_CHANGED.getCode()); // 9
|
notification.setNotificationContent(message);
|
notification.setOperationsUserId(user.getId());
|
notification.setDataId(0L);
|
notification.setReadStatus(0); // 未读
|
notification.setNotificationExt(bodyJson);
|
return notification;
|
}
|
|
/**
|
* 按设备类型分组推送消息
|
* @param users 用户列表
|
* @param message 消息内容
|
*/
|
private void pushMessageByDeviceType(List<XpOperationsUser> users, String message) {
|
// 按设备类型分组 (1=Android, 2=iOS)
|
Map<Integer, List<XpOperationsUser>> usersByDeviceType = users.stream()
|
.filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
|
.collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
|
|
// 分别推送 Android 和 iOS
|
for (Map.Entry<Integer, List<XpOperationsUser>> entry : usersByDeviceType.entrySet()) {
|
Integer deviceType = entry.getKey();
|
List<XpOperationsUser> deviceUsers = entry.getValue();
|
|
// 按 1000 个手机号一批进行推送
|
List<List<XpOperationsUser>> batches = partition(deviceUsers, 1000);
|
|
for (List<XpOperationsUser> batch : batches) {
|
String accounts = batch.stream()
|
.map(XpOperationsUser::getPhoneNumber)
|
.collect(Collectors.joining(","));
|
|
AliPushMessageDTO pushDTO = buildPushMessageDTO(deviceType, accounts, message);
|
|
try {
|
xpAliPushMessageClient.pushMessage(pushDTO);
|
log.info("协议更新推送成功，设备类型={}，用户数={}",
|
deviceType == 1 ? "Android" : "iOS", batch.size());
|
} catch (Exception e) {
|
log.error("协议更新推送失败，设备类型={}", deviceType, e);
|
}
|
}
|
}
|
}
|
|
/**
|
* 构建推送消息 DTO
|
* @param deviceType 设备类型
|
* @param accounts 手机号（逗号分隔）
|
* @param message 消息内容
|
* @return 推送 DTO
|
*/
|
private AliPushMessageDTO buildPushMessageDTO(Integer deviceType, String accounts, String message) {
|
AliPushMessageDTO dto = new AliPushMessageDTO();
|
dto.setDeviceType(deviceType);
|
dto.setAccount(accounts);
|
dto.setTitle("协议政策内容更新提示");
|
dto.setBody(buildAgreementPushBodyJson(message));
|
dto.setPushType("NOTICE");
|
return dto;
|
}
|
|
/**
|
* 将列表按指定大小分批
|
* @param list 原始列表
|
* @param batchSize 批次大小
|
* @return 分批后的列表
|
*/
|
private <T> List<List<T>> partition(List<T> list, int batchSize) {
|
List<List<T>> batches = new ArrayList<>();
|
for (int i = 0; i < list.size(); i += batchSize) {
|
batches.add(list.subList(i, Math.min(i + batchSize, list.size())));
|
}
|
return batches;
|
}
|
|
/**
|
* 发送协议更新授权通知
|
* @param contentType 协议类型
|
*/
|
private void sendAgreementAuthNotification(Integer contentType) {
|
// 异步处理，避免阻塞主流程
|
ThreadPoolExecutor executor = new ThreadPoolExecutor(
|
1, 1, 0L, TimeUnit.MILLISECONDS,
|
new LinkedBlockingQueue<Runnable>()
|
);
|
|
executor.execute(() -> {
|
try {
|
// 1. 获取所有启用状态的运维用户
|
R<List<XpOperationsUser>> result = xpOperationUserClient.listAllEnabled();
|
if (result == null || result.getCode() != 200 || result.getData() == null || result.getData().isEmpty()) {
|
log.warn("没有需要通知的启用状态用户");
|
return;
|
}
|
|
List<XpOperationsUser> enabledUsers = result.getData();
|
log.info("协议授权通知，共需通知 {} 位用户", enabledUsers.size());
|
|
// 2. 构建通知内容
|
String agreementTypeName = getAgreementTypeName(contentType);
|
String currentDate = LocalDate.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd"));
|
String message = String.format("尊敬的用户，你好！%s已于%s更新了内容，需要你前往授权", agreementTypeName, currentDate);
|
String bodyJson = buildAuthPushBodyJson(message);
|
|
// 3. 构建并保存通知记录
|
List<XpOperationsUserNotification> notifications = enabledUsers.stream()
|
.map(user -> buildAuthNotification(message, bodyJson, user))
|
.collect(Collectors.toList());
|
|
// 批量保存通知记录
|
xpOperationUserNotificationClient.addNotificationBatch(notifications);
|
log.info("协议授权通知记录保存成功，共 {} 条", notifications.size());
|
|
// 4. 按设备类型分组并批量推送
|
pushAuthMessageByDeviceType(enabledUsers, message);
|
|
} catch (Exception e) {
|
log.error("发送协议授权通知失败", e);
|
}
|
});
|
|
executor.shutdown();
|
}
|
|
/**
|
* 构建授权推送 body JSON
|
* @param message 消息内容
|
* @return JSON 字符串
|
*/
|
private String buildAuthPushBodyJson(String message) {
|
JSONObject body = new JSONObject();
|
body.put("content", message);
|
body.put("strategy", NotificationTypeEnum.AGREEMENT_CHANGED_NEED_AUTH.getCode()); // 10
|
body.put("type", 0);
|
body.put("id", 0);
|
return body.toJSONString();
|
}
|
|
/**
|
* 构建授权通知实体
|
* @param message 消息内容
|
* @param bodyJson 推送body内容
|
* @param user 运维用户
|
* @return 通知实体
|
*/
|
private XpOperationsUserNotification buildAuthNotification(String message, String bodyJson, XpOperationsUser user) {
|
XpOperationsUserNotification notification = new XpOperationsUserNotification();
|
notification.setNotificationType(NotificationTypeEnum.AGREEMENT_CHANGED_NEED_AUTH.getCode()); // 10
|
notification.setNotificationContent(message);
|
notification.setOperationsUserId(user.getId());
|
notification.setDataId(0L);
|
notification.setReadStatus(0); // 未读
|
notification.setNotificationExt(bodyJson);
|
return notification;
|
}
|
|
/**
|
* 按设备类型分组推送授权消息
|
* @param users 用户列表
|
* @param message 消息内容
|
*/
|
private void pushAuthMessageByDeviceType(List<XpOperationsUser> users, String message) {
|
// 按设备类型分组 (1=Android, 2=iOS)
|
Map<Integer, List<XpOperationsUser>> usersByDeviceType = users.stream()
|
.filter(u -> u.getDeviceType() != null && u.getPhoneNumber() != null)
|
.collect(Collectors.groupingBy(XpOperationsUser::getDeviceType));
|
|
// 分别推送 Android 和 iOS
|
for (Map.Entry<Integer, List<XpOperationsUser>> entry : usersByDeviceType.entrySet()) {
|
Integer deviceType = entry.getKey();
|
List<XpOperationsUser> deviceUsers = entry.getValue();
|
|
// 按 1000 个手机号一批进行推送
|
List<List<XpOperationsUser>> batches = partition(deviceUsers, 1000);
|
|
for (List<XpOperationsUser> batch : batches) {
|
String accounts = batch.stream()
|
.map(XpOperationsUser::getPhoneNumber)
|
.collect(Collectors.joining(","));
|
|
AliPushMessageDTO pushDTO = buildAuthPushMessageDTO(deviceType, accounts, message);
|
|
try {
|
xpAliPushMessageClient.pushMessage(pushDTO);
|
log.info("协议授权推送成功，设备类型={}，用户数={}",
|
deviceType == 1 ? "Android" : "iOS", batch.size());
|
} catch (Exception e) {
|
log.error("协议授权推送失败，设备类型={}", deviceType, e);
|
}
|
}
|
}
|
}
|
|
/**
|
* 构建授权推送消息 DTO
|
* @param deviceType 设备类型
|
* @param accounts 手机号（逗号分隔）
|
* @param message 消息内容
|
* @return 推送 DTO
|
*/
|
private AliPushMessageDTO buildAuthPushMessageDTO(Integer deviceType, String accounts, String message) {
|
AliPushMessageDTO dto = new AliPushMessageDTO();
|
dto.setDeviceType(deviceType);
|
dto.setAccount(accounts);
|
dto.setTitle("协议政策内容更新授权");
|
dto.setBody(buildAuthPushBodyJson(message));
|
dto.setPushType("NOTICE");
|
return dto;
|
}
|
}
|
