java/Robotaxi.git

			@@ -2,16 +2,35 @@


			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.api.model.operation.XpAgreement;
			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>
			@@ -21,8 +40,16 @@
			* @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) {
			@@ -77,5 +104,311 @@
			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;
			}
			}