# ActivityEventConsumer.java

    package cloud.wconnect.rateshop.consumer.activityevent;

    import cloud.wconnect.commonlib.cons.BaseConstant;
    import cloud.wconnect.commonlib.pojo.dto.PropertyDTO;
    import cloud.wconnect.commonlib.pojo.dto.UpdateLog;
    import cloud.wconnect.commonlib.service.PropertyInternalService;
    import cloud.wconnect.commonlib.util.DateTimeUtils;
    import cloud.wconnect.commonlib.util.JsonUtils;
    import cloud.wconnect.rateshop.entity.rateshop.Event;
    import cloud.wconnect.rateshop.entity.rateshop.EventSourceSetting;
    import cloud.wconnect.rateshop.mapper.rateshop.EventMapper;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.event.EventDTO;
    import cloud.wconnect.rateshop.pojo.dto.rateshop.consumer.event.EventMessage;
    import cloud.wconnect.rateshop.service.rateshop.IEventSourceSettingService;
    import com.fasterxml.jackson.databind.ObjectMapper;
    import org.apache.commons.collections.CollectionUtils;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.stereotype.Component;
    import org.springframework.transaction.annotation.Transactional;
    import software.amazon.awssdk.services.sqs.model.Message;

    import java.util.*;
    import java.util.stream.Collectors;

    /**
     * Activity Event事件的Consumer
     *
     * @author lichen
     * @date 2023/6/13
     */
    @Component
    public class ActivityEventConsumer {

        @Autowired
        private ObjectMapper objectMapper;
        @Autowired
        private EventMapper eventMapper;
        @Autowired
        private IEventSourceSettingService eventSourceSettingService;
        @Autowired
        private PropertyInternalService propertyInternalService;

        /**
         * 处理ActivityEvent事件消息
         *
         * @param message
         * @return void
         * @author lichen
         * @date 2023/6/13
         */
        @Transactional(rollbackFor = Exception.class)
        public void handleMessage(Message message) {
            //把消息转换成EventMessage对象
            EventMessage eventMessage = JsonUtils.convertStringToObject(objectMapper, message.body(), EventMessage.class);

            List<EventDTO> eventDtos = eventMessage.getEvent();
            if(CollectionUtils.isNotEmpty(eventDtos)){
                //根据sourceId查询资产和地址
                List<EventSourceSetting> eventSourceSettings = eventSourceSettingService.listByUrlId(eventMessage.getSourceId());
                if(CollectionUtils.isEmpty(eventSourceSettings)){
                    return;
                }

                //获取任一资产的时区（用于生成当前日期）
                Integer propertyId = eventSourceSettings.get(0).getPropertyId();
                PropertyDTO propertyDTO = propertyInternalService.getOne(propertyId);
                String timezone = propertyDTO.getTimezone();
                //根据资产时区，获取当前时间(yyyyMMdd)
                String currentDate = DateTimeUtils.timestampToStr(DateTimeUtils.dateNowTimestamp(timezone), BaseConstant.DATE_FORMAT_YYYYMMDD, timezone);

                //因为一个url只能对应一个locaiton，所以获取任意一个locationId
                Integer locationId = eventSourceSettings.get(0).getLocationId();

                //更新log
                UpdateLog updateLog = new UpdateLog(0, eventMessage.getQueryDate());
                List<UpdateLog> updateLogs = Arrays.asList(updateLog);
                String updateLogStr = JsonUtils.convertObjListToString(objectMapper, updateLogs);

                //执行过滤和转化
                List<Event> eventList = eventDtos.stream()
                        //过滤beginDate大于等于当前日期的数据
                        .filter(item -> item.getBeginDate() >= Integer.valueOf(currentDate))
                        .map(dto -> {
                            Event event = new Event();
                            event.setTenantId(0).setPropertyId(0).setLocationId(locationId).setUrlId(eventMessage.getSourceId())
                                    .setCategory("").setTitle(dto.getTitle()).setBeginDate(dto.getBeginDate()).setEndDate(dto.getEndDate())
                                    .setLink(dto.getUrl()).setAddress(dto.getLocation()).setDescription("").setUpdateLog(updateLogStr);
                            return event;
                        })
                        //根据beginDate + title去重
                        .collect(Collectors.collectingAndThen(
                                Collectors.toCollection(() -> new TreeSet<>(Comparator.comparing(Event::beginDateAndTitleKey))), ArrayList::new));
                if(CollectionUtils.isNotEmpty(eventList)){
                    //批量保存事件信息
                    eventMapper.saveOrUpdateByUnique(eventList);
                    //清理不存在的规则为：删除beginDate为未来的且本次抓取中不存在的记录（title+beginDate）
                    eventMapper.removeByUrlIdAndBeginDateAndTitle(eventMessage.getSourceId(),eventList);
                }
            }
        }
    }