using System.Diagnostics; using System.Linq.Expressions; using System.Text; using BSI.Stack.Rabbit; using BSI.Stack.Rabbit.Handler; using DotNetCore.CAP; using Microsoft.Extensions.Logging; using Newtonsoft.Json; using Newtonsoft.Json.Linq; using RabbitMQ.Client.Events; using SJZY.Expand.ABP.Core.PublicMethords; using SJZY.InterfaceRelay.Application.Contracts; using SJZY.Track.Application.Contract.Dtos.TrackPull; 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.Enums; using SJZY.Track.Domain.Queries; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.Rabbit; public class TrackRefreshEventHandler : ITransientDependency, IRabbitHandler { private readonly ICapPublisher _capPublisher; private readonly ILogger _logger; private readonly IOrderRepository _orderRepository; private readonly ITimezoneQueries _timezoneQueries; private readonly ITrackDetailsRepository _trackDetailsRepository; private readonly ITrackPullCacheService _trackPullCacheService; public TrackRefreshEventHandler(IOrderRepository orderRepository, ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService, ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries, ILogger logger) { _orderRepository = orderRepository; _capPublisher = capPublisher; _trackPullCacheService = trackPullCacheService; _trackDetailsRepository = trackDetailsRepository; _timezoneQueries = timezoneQueries; _logger = logger; } public bool Enable(ConsumeOptions options) { return options.Queue == GlobalConst.TrackRefreshSubscribe; } public async Task Handle(IServiceProvider sp, BasicDeliverEventArgs args, ConsumeOptions options) { var msg = Encoding.UTF8.GetString(args.Body.ToArray()); var @event = System.Text.Json.JsonSerializer.Deserialize(msg); if (@event == null) { _logger.LogWarning( $"rabbit on queue({options.Queue}) received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } await ConsumerAsync(@event); } public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event) { if (!@event.IsSuccess) { _logger.LogWarning("请求id: {@RequestId},轨迹刷新失败", @event.RequestId); return; } var attach = Base64Helper.DeSerialize(@event.Attach); if (attach is null || @event.Data == null) { _logger.LogWarning("请求id: {@RequestId},轨迹消费失败,attach或data解析为空", @event.RequestId); return; } var config = await _trackPullCacheService.GetPullParamsWithContentAsync(attach.SubSystemId, attach.ServiceProviderId); if (config == null) { _logger.LogWarning("请求id: {@RequestId},轨迹消费失败,找不到拉取配置:{@Attach}", attach.ToJson(), @event.RequestId); return; } var data = JObject.Parse(@event.Data); var orderToken = data.SelectToken(config.ParamConfig.OrderPath); if (orderToken is JArray orderTokens) { foreach (var token in orderTokens) { await HandleOrderAsync(token, config, attach); } } else if (orderToken is JObject) { await HandleOrderAsync(orderToken, config, attach); } } private async Task HandleOrderAsync(JToken orderToken, TrackPullParamsWithContentDto paramsWithContent, TrackRefreshAttach attach) { var paramConfig = paramsWithContent.ParamConfig; var orderNo = orderToken[paramConfig.OrderNoName]?.Value(); if (orderNo.IsNullOrWhiteSpace()) return; var trackToken = orderToken.SelectToken(paramConfig.TrackDetailsPath); if (trackToken is null) return; var order = await _orderRepository.GetFirstAsync(x => x.OrderType == attach.OrderType && x.OrderNo == orderNo, x => new { x.Id, x.LatestPushTrackCode, x.CustomId }); if (order == default) return; var context = new TrackRefreshContext { OrderId = order.Id, OrderType = attach.OrderType, OrderNo = orderNo!, ServiceProviderName = attach.ServiceProviderName, CustomId = order.CustomId, SubSystemId = attach.SubSystemId }; if (trackToken is JArray trackTokens) { foreach (var item in trackTokens) { await HandleTrackAsync(item, paramsWithContent, context); } } else if (trackToken is JObject) { await HandleTrackAsync(trackToken, paramsWithContent, context); } } private async Task HandleTrackAsync(JToken trackToken, TrackPullParamsWithContentDto paramsWithContent, TrackRefreshContext context) { var originTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)); // 如果已经存在一模一样的报文了说明处理过了,直接return if (await _trackDetailsRepository.AnyAsync(context.OrderId, originTrackDataHash)) { return; } var paramConfig = paramsWithContent.ParamConfig; var contentConfigs = paramsWithContent.ContentConfigs; TrackPullContentConfigDto? matchedContentConfig = null; // 1、匹配轨迹代码 // 2、匹配轨迹状态 if (!paramConfig.TrackStatusName.IsNullOrWhiteSpace()) { var trackStatus = trackToken[paramConfig.TrackStatusName!]?.Value(); if (!trackStatus.IsNullOrWhiteSpace()) { matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValue == trackStatus); } } // 3、匹配轨迹内容 else if (!paramConfig.TrackContentName.IsNullOrWhiteSpace()) { var trackContent = trackToken[paramConfig.TrackContentName!]?.Value(); if (!trackContent.IsNullOrWhiteSpace()) { Debug.Assert(trackContent != null, nameof(trackContent) + " != null"); // 3.1 按内容直接匹配 // 3.2 按关键字分析轨迹内容 if (matchedContentConfig is null) { foreach (var contentConfig in contentConfigs) { foreach (var keyword in contentConfig.TrackKeyword?.Split(',') ?? Enumerable.Empty()) { if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase)) { matchedContentConfig = contentConfig; break; } } if (matchedContentConfig is null) continue; foreach (var keyword in contentConfig.TrackKeywordReverse?.Split(',') ?? Enumerable.Empty()) { if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase)) { matchedContentConfig = null; break; } } } } } } // 轨迹发生时间 var trackTime = paramConfig.TrackTimeName == null ? null : trackToken[paramConfig.TrackTimeName]?.Value(); // 轨迹时区 var (trackTimeZone, hasTimezoneBottomLine) = await GetTimezone(paramConfig, matchedContentConfig, context, trackToken); // 轨迹发生地 var (trackPlace, hasPlaceBottomLine) = await GetTrackPlace(paramConfig, matchedContentConfig, context, trackToken); if (matchedContentConfig is null) { _logger.LogDebug("订单号:{OrderNo},未匹配到轨迹内容,轨迹报文:{TrackContent}", context.OrderNo, trackToken.ToString()); await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime, trackTimeZone, paramConfig); return; } var deficiencyTagValue = CalcDeficiencyTagValue(hasPlaceBottomLine, hasTimezoneBottomLine, trackToken, context); // 新增/更新轨迹 var trackUpsertEvent = new TrackUpsertEvent { OrderIdentifiers = new[] { new OrderIdentifier { OrderId = context.OrderId, OrderNo = context.OrderNo, OrderType = context.OrderType, } }, TrackRefreshData = new TrackRefreshStruct { ServiceProviderName = context.ServiceProviderName, OriginTrackDataHash = originTrackDataHash, OriginTrackContent = paramConfig.TrackContentName != null ? trackToken[paramConfig.TrackContentName]?.Value() : GlobalConst.EmptyString }, SubSystemId = context.SubSystemId, TrackCode = matchedContentConfig.TrackContentCode, TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackTime = trackTime ?? DateTime.Now, TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, DeficiencyTagValue = deficiencyTagValue, IsAdd = true, }; await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } /// /// 触发了兜底,但是没有兜底住,对订单进行标记 /// 1表示时区,2表示发生地,3表示两种 /// /// /// /// /// /// private int CalcDeficiencyTagValue(bool hasPlaceBottomLine, bool hasTimezoneBottomLine, JToken trackToken, TrackRefreshContext context) { var deficiencyTagValue = 0; if (hasTimezoneBottomLine) { _logger.LogWarning("订单号:{@OrderNo}, 轨迹发生时区无法解析,触发了兜底,但是没有兜底住:{@Data}", context.OrderNo, trackToken.ToString()); deficiencyTagValue += 1; } if (hasPlaceBottomLine) { _logger.LogWarning("订单号:{@OrderNo}, 轨迹发生地无法解析,触发了兜底,但是没有兜底住:{@Data}", context.OrderNo, trackToken.ToString()); deficiencyTagValue += 2; } return deficiencyTagValue; } private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, DateTime? trackTime, float? trackTimeZone, TrackPullParamConfigDto paramConfig) { // 新增服务商轨迹到es var track = new TrackDetailsIndex { IndexName = $"{GlobalConst.TrackIndex}-{DateTime.Now:yyyy.MM.dd}", SubSystemId = context.SubSystemId, OrderId = context.OrderId, OrderType = context.OrderType, OrderNo = context.OrderNo, TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackTime = trackTime ?? DateTime.Now, TrackTimeZone = trackTimeZone ?? GlobalConst.NoneTrackTimeZone, ServiceProviderName = context.ServiceProviderName, OriginTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)), CreateUser = GlobalConst.SystemUser, CreateTime = DateTime.Now }; if (!paramConfig.TrackContentName.IsNullOrWhiteSpace()) { track.OriginTrackContent = trackToken[paramConfig.TrackContentName!]?.Value() ?? GlobalConst.EmptyString; } track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); await _trackDetailsRepository.AddAsync(track); } /// /// 获取轨迹时区 /// /// 配置信息 /// 配置信息 /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(float? trackTimeZone, bool hasTimezoneBottomLine)> GetTimezone( TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生时区 var trackTimeZone = paramConfig.TrackTimeZoneName == null ? null : trackToken[paramConfig.TrackTimeZoneName]?.Value(); bool hasTimezoneBottomLine = false; //执行了兜底但是没有兜底住,需要记录 if (!trackTimeZone.HasValue && matchedContentConfig?.TrackZonePlaceValue is not null) { //兜底没配置的就不需要兜底,拉取的发生地没就没 var timezone = await GetTimezone(matchedContentConfig, context.OrderId); trackTimeZone = timezone; hasTimezoneBottomLine = !timezone.HasValue; } return (trackTimeZone, hasTimezoneBottomLine); } /// /// 获得轨迹地址 /// /// 配置信息 /// 配置信息 /// 订单id /// /// 时区,是否执行了兜底轨迹时区 private async Task<(string? trackPlace, bool hasPlaceBottomLine)> GetTrackPlace(TrackPullParamConfigDto paramConfig, TrackPullContentConfigDto? matchedContentConfig, TrackRefreshContext context, JToken trackToken) { // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName == null ? null : trackToken[paramConfig.TrackPlaceName]?.Value(); var hasPlaceBottomLine = false; //执行了兜底但是没有兜底住,需要记录 if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue is not null) { //兜底没配置的就不需要兜底,拉取的发生地没就没 trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value, context.OrderId); hasPlaceBottomLine = string.IsNullOrWhiteSpace(trackPlace); } return (trackPlace, hasPlaceBottomLine); } /// /// 根据订单id获取兜底位置 /// /// /// /// private Task GetPlaceByOrder(int trackValue, int orderId) { Expression> resultExpression = (TrackLocationValue) trackValue switch { TrackLocationValue.ArrivalPort => x => x.ArrivalPort, TrackLocationValue.ArriveCity => x => x.ArrivalCity, TrackLocationValue.ArrivalCountry => x => x.ArrivalCountry, TrackLocationValue.LeavePort => x => x.LeavePort, TrackLocationValue.LeaveCity => x => x.LeaveCity, TrackLocationValue.LeaveCountry => x => x.LeaveCountry, _ => x => x.LeaveCountry }; return _orderRepository.GetFirstAsync(x => x.Id == orderId, resultExpression); } /// /// 获取时区 /// /// /// /// private async Task GetTimezone(TrackPullContentConfigDto matchedContentConfig, int orderId) { var placeOrCode = await GetPlaceByOrder(matchedContentConfig!.TrackZonePlaceValue.Value, orderId); string timezone = null; if (!string.IsNullOrEmpty(placeOrCode)) { if (matchedContentConfig!.TrackZonePlaceValue.Value == (int) TrackLocationValue.ArrivalPort || matchedContentConfig!.TrackZonePlaceValue.Value == (int) TrackLocationValue.LeavePort) { timezone = await _timezoneQueries.GetPortTimezone(placeOrCode); return DateTimeHelper.ConvertTimeZoneToFloat(timezone); } timezone = await _timezoneQueries.GetCityTimezone(placeOrCode); return DateTimeHelper.ConvertTimeZoneToFloat(timezone); } return null; } }