using System.Globalization; using DotNetCore.CAP; using Newtonsoft.Json; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.TrackPush; using SJZY.Track.Application.Contract.Events; using SJZY.Track.Application.Contract.Services; using SJZY.Track.Application.Helpers; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate; using SJZY.Track.Domain.Aggregates.TrackPushAggregate; using SJZY.Track.Domain.Consts; using SJZY.Track.Domain.Enums; using Volo.Abp.DependencyInjection; using Volo.Abp.Http.Client.DynamicProxying; namespace SJZY.Track.Application.CAP.EventHandlers; public class TrackPushEventHandler : ITransientDependency, ICapSubscribe { private readonly IHttpClientProxy _httpClientProxy; private readonly IOrderRepository _orderRepository; private readonly ITrackPushCacheService _trackPushCacheService; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ITrackPushRecordsRepository _trackPushRecordsRepository; private List _pushConfigs = null!; private List _orders = null!; public TrackPushEventHandler(IHttpClientProxy httpClientProxy, IOrderRepository orderRepository, ITrackPushCacheService trackPushCacheService, ITrackDetailsRepository trackDetailsRepository, ITrackPushRecordsRepository trackPushRecordsRepository) { _httpClientProxy = httpClientProxy; _orderRepository = orderRepository; _trackPushCacheService = trackPushCacheService; _trackDetailsRepository = trackDetailsRepository; _trackPushRecordsRepository = trackPushRecordsRepository; } [CapSubscribe(GlobalConst.TrackPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] [CapSubscribe(GlobalConst.TrackManualPushSubscribe, Group = GlobalConst.TrackPushSubscribe)] public async Task ConsumerAsync(TrackPushEvent @event) { var pushConfigs = await _trackPushCacheService.GetPushContentConfigsAsync(@event.SubSystemId, @event.CustomId); if (pushConfigs == null || pushConfigs.Count == 0) return; _pushConfigs = pushConfigs; var orders = await _orderRepository.GetListAsync(x => @event.OrderIds.Contains(x.Id), x => new OrderEntity { Id = x.Id, OrderNo = x.OrderNo, LatestPushTrackCode = x.LatestPushTrackCode, OrderType = x.OrderType }); if (orders.Count == 0) return; _orders = orders; switch (@event.OptionType) { case OptionType.无: return; case OptionType.新增: await DoPushAsync(@event); break; case OptionType.删除: break; case OptionType.修改: await DoUpdateAsync(@event); break; default: throw new ArgumentOutOfRangeException(); } } private async Task DoPushAsync(TrackPushEvent @event) { var items = new List(); TrackPushContentConfigDto? config; if (@event.OperationMan == null) { foreach (var order in _orders) { if (order.QuestionType == (int)TrackProblemType.Error) continue; // 系统不推异常单,只能人工操作 // 推送新增,需要根据上一次的推送 推断 本次推送 是否可行 config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); if (config == null) continue; if (!config.IsAutoPush) continue; if (config.OrderType != order.OrderType) continue; var log = new TrackPushRecordIndex { OptionType = @event.OptionType.ToString(), OrderId = order.Id, OrderNo = order.OrderNo, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackCode = @event.TrackCode, TrackContent = @event.SystemPushInfo.TrackContent, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CreateUser = GlobalConst.SystemUser, }; var item = await BuiltPushItem(config, log); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } } else //手工推送 { config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null) return; var tracks = await _trackDetailsRepository.GetListAsync(@event.OrderIds, @event.TrackCode); tracks = tracks.DistinctBy(x => x.OrderId).ToList(); // 防止重复推送 var temp=this.DateTimeFormat(DateTime.Now, -8.0, config); if (tracks.Count == 0) return; // 一个能推的轨迹都不存在 // tips:人工推送只要能推就推了,不会校验已推和顺序问题,因为可能是更新 foreach (var track in tracks) { var order = _orders.First(x => x.Id == track.OrderId); if (config is { IsAllPush: true }) { // 推断 本次推送 是否可行 config = GetValidPushConfigAsync(_pushConfigs, order.LatestPushTrackCode, @event.TrackCode); } if (config == null) continue; var log = new TrackPushRecordIndex { OptionType = @event.OptionType.ToString(), OrderId = track.OrderId, OrderNo = order.OrderNo, TrackStatus = track.TrackStatus, TrackCode = @event.TrackCode, TrackContent = track.TrackContent, TrackPlace = track.TrackPlace, TrackTime = track.TrackTime, TrackTimeZone = track.TrackTimeZone, CreateUser = @event.OperationMan, }; var item = await BuiltPushItem(config, log); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } } if (items.Count == 0) return; string asciiString = Base64Helper.SerializeObject(OptionType.新增.ToString()); var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( InterfaceRelayConst.PushTrackInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); if (r.Code != InterfaceRelayConst.SuccessCode) { throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}"); } } private async Task DoUpdateAsync(TrackPushEvent @event) { var config = _pushConfigs.FirstOrDefault(x => x.TrackCode == @event.TrackCode); if (config == null || !config.IsCanUpdate) return; var items = new List(); foreach (var order in _orders) { var log = new TrackPushRecordIndex { OptionType = @event.OptionType.ToString(), OrderId = order.Id, OrderNo = order.OrderNo, TrackStatus = @event.SystemPushInfo.TrackStatus, TrackCode = @event.TrackCode, TrackContent = @event.SystemPushInfo.TrackContent, TrackPlace = @event.SystemPushInfo.TrackPlace, TrackTime = @event.SystemPushInfo.TrackTime, TrackTimeZone = @event.SystemPushInfo.TrackTimeZone, CreateUser = GlobalConst.SystemUser, }; var item = BuiltUpdateItem(config, log); item.Log = JsonConvert.SerializeObject(log); items.Add(item); } if (items.Count == 0) return; string asciiString = Base64Helper.SerializeObject(OptionType.修改.ToString()); var r = await _httpClientProxy.Service.BatchForwardThenPushToQueueAsync( InterfaceRelayConst.TrackUpdateInterfaceCode, _pushConfigs.First().ServiceProviderCode, items, GlobalConst.TrackPushCompletedSubscribe, attach: asciiString); if (r.Code != InterfaceRelayConst.SuccessCode) { throw new BusinessException(message: $"requestId: {r.RequestId}; {r.Msg}"); } } private TrackUpdateInput BuiltUpdateItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) { var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone); var trackDateStr = DateTimeFormat(trackDatetime,config.TrackTimeZone ?? log.TrackTimeZone,config); var track = new TrackItem { // 推真实的发生地和发生时间(时间是计算后的) Location = log.TrackPlace, Date = trackDateStr, TimeZone = config.TrackTimeZone ?? log.TrackTimeZone, Description = config.TrackEnContentValue, DescriptionCn = config.TrackContentValue, Code = config.TrackContentCodeValue, Status = config.TrackStatusValue, Extra = config.ExpandFieldsValue }; var body = new TrackUpdateInput { OrderNo = log.OrderNo, }; body.Data.Add(track); log.PushContent = body.ToJson(); return body; } private async Task BuiltPushItem(TrackPushContentConfigDto config, TrackPushRecordIndex log) { var trackDatetime = log.TrackTime.AddHours(config.TrackTimeZone ?? log.TrackTimeZone - log.TrackTimeZone); var trackDateStr = DateTimeFormat(trackDatetime,config.TrackTimeZone ?? log.TrackTimeZone,config); var track = new TrackItem { // 推真实的发生地和发生时间(时间是计算后的) Location = log.TrackPlace, Date = trackDateStr, TimeZone = config.TrackTimeZone ?? log.TrackTimeZone, Description = config.TrackEnContentValue, DescriptionCn = config.TrackContentValue, Code = config.TrackContentCodeValue, Status = config.TrackStatusValue, Extra = config.ExpandFieldsValue }; var body = new TrackPushInput { OrderNo = log.OrderNo, }; body.Data.Add(track); if (config.IsAllPush) { var history = await _trackPushRecordsRepository.GetLatestTrackPushRecordAsync(log.OrderId); if (history != default) { body = JsonConvert.DeserializeObject(history.PushContent); body!.Data.Add(track); } } log.PushContent = body.ToJson(); return body; } /// /// 根据轨迹代码推断下一个(升序)要推送的内容,如果找不到或者已经是最后一个了 /// 则查询有没有可交换的推送,没有就返回null /// /// /// /// /// TrackPushContentConfigDto? GetValidPushConfigAsync(List cache, string? latestPushTrackCode, string? trackCode) { var idx = cache.FindIndex(x => x.TrackCode == latestPushTrackCode); if (idx + 1 >= cache.Count) return default; // 推完了 // find next var matched = cache[idx + 1].TrackCode == trackCode ? cache[idx + 1] : // find swap cache.FirstOrDefault(x => x.TrackCode == cache[idx + 1].SwapTrackCode); return matched != null && matched.TrackCode == trackCode ? matched : default; } /// /// 转换推送时间格式 /// /// /// /// /// private string DateTimeFormat(DateTime dateTime,double timezone, TrackPushContentConfigDto config) { if (config.DateTimeFormatType == DateTimeFormatEnum.Timestamp) return DateTimeHelper.ConvertToUnixTimestamp(dateTime).ToString(); if (config.DateTimeFormatType == DateTimeFormatEnum.DateTimeAndTimezone && !string.IsNullOrEmpty(config.DateTimeFormat) && !string.IsNullOrEmpty(config.TimezoneFormat)) { var zoneStr = DateTimeHelper.TimezoneFormat(timezone, config.TimezoneFormat); var datetimeStr = DateTimeHelper.FormatDateTime(dateTime, config.DateTimeFormat); return $"{datetimeStr} {zoneStr}"; } return dateTime.ToString("yyyy-MM-ddTHH:mm:ss.fffK"); } }