java/Robotaxi.git

package com.xiaopeng.operation.service.impl;
 
 
import cn.hutool.core.bean.BeanUtil;
import com.alibaba.fastjson2.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.constant.AliPushTypeConstant;
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);
        }
 
        // 判断是否需要发送通知 8=运维APP-小鹏Robotaxi自动驾驶出行运维隐私政策 9=运维APP-第三方共享信息清单 10=运维APP-个人信息收集清单 11=运维APP-权限调用
        if (dto.getContentType() != null && (dto.getContentType() == 8 || dto.getContentType() == 9 || dto.getContentType() == 10 || dto.getContentType() == 11)) {
            // 发送协议更新提示通知（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) {
        switch (contentType) {
            case 8:
                return "小鹏Robotaxi自动驾驶出行运维隐私政策";
            case 9:
                return "第三方共享信息清单";
            case 10:
                return "个人信息收集清单";
            case 11:
                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(AliPushTypeConstant.MESSAGE);
        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(AliPushTypeConstant.NOTICE);
        return dto;
    }
}