using System.Diagnostics; using System.Text; using DotNetCore.CAP; using Fake.UnitOfWork; using InterfaceForward.Domain.Shared.Helpers; using Microsoft.Extensions.Logging; using Microsoft.Extensions.Options; using Newtonsoft.Json; using Newtonsoft.Json.Linq; using RabbitMQ.Client.Events; using TrackCenter.Application.Dtos.TrackPull; using TrackCenter.Application.Events; using TrackCenter.Application.Helpers; using TrackCenter.Application.Options; using TrackCenter.Application.Services.TrackPull; using TrackCenter.Domain.Aggregates.OrderAggregate; using TrackCenter.Domain.Aggregates.TrackDetailsAggregate; namespace TrackCenter.Application.Rabbit; [ExposeServices(typeof(IRabbitHandler))] public class TrackRefreshEventHandler( IOrderRepository orderRepository, ICapPublisher capPublisher, TrackPullCacheService trackPullCacheService, ITrackDetailsEsRepository trackDetailsEsRepository, ILogger logger, IOptionsMonitor options) : ITransientDependency, IRabbitHandler { public bool Enable(ConsumeOptions coptions) { return coptions.Queue.StartsWith(GlobalConst.TrackRefreshSubscribe); } [UnitOfWork] public virtual async Task Handle(BasicDeliverEventArgs args) { var msg = string.Empty; try { msg = Encoding.UTF8.GetString(args.Body.ToArray()); var @event = System.Text.Json.JsonSerializer.Deserialize(msg); if (@event == null) { logger.LogWarning( $"rabbit on queue received msg can't convert to {nameof(InterfaceRelayForwardReplayEvent)}"); return; } await ConsumerAsync(@event); } catch (Exception ex) { Log.Error($"TrackRefreshEventHandler出错,错误信息:{ex}\r\n-----\r\n消息:{msg}"); } } 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.IsNullOrWhiteSpace()) { logger.LogWarning("请求id: {@RequestId},轨迹消费失败,attach或data解析为空", @event.RequestId); return; } // 最优先级,最匹配的 权重 sort // 客户、渠道、产品 // */*/* 全部 // 客户A/渠道A/产品A // 客户A // 客户A 渠道B 产品A var config = await trackPullCacheService.GetPullParamsWithContentAsync(attach.AppKey, attach.ServiceProviderCode); if (config is null) { logger.LogWarning("请求id: {@RequestId},轨迹消费失败,找不到拉取配置:{@Attach}", attach.ToJson(), @event.RequestId); return; } var data = JToken.Parse(@event.Data); var orderToken = data.SelectToken(options.CurrentValue.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 = options.CurrentValue; var orderNo = orderToken.SelectToken(paramConfig.OrderNoName)?.Value(); if (orderNo.IsNullOrWhiteSpace()) return; var trackToken = orderToken.SelectToken(paramConfig.TrackDetailsPath); if (trackToken is null) return; //回传单号可能是主单也可能是外部单号 //通过平台code和(外部单号/主单号)查询主单 var order = await orderRepository.GetTrackOrderByProviderAndOrderNo(attach.ServiceProviderCode, orderNo!); if (order == default) { logger.LogWarning($"{attach.ToJson()},轨迹消费失败,找不到主单,平台code:{attach.ServiceProviderCode},单号:{orderNo}"); return; } var context = new TrackRefreshContext { OrderId = order.MainOrderId, OrderNo = order.OrderNo, Customer = order.Customer, OrderType = attach.OrderType, AppKey = attach.AppKey, ServiceProviderCode = attach.ServiceProviderCode }; 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 trackDetailsEsRepository.AnyAsync(context.OrderId, originTrackDataHash)) { return; } var paramConfig = options.CurrentValue; var contentConfigs = paramsWithContent.ContentConfigs; TrackPullContentConfigDto? matchedContentConfig = null; // 轨迹内容 var trackContent = paramConfig.TrackContentName.IsNullOrWhiteSpace() ? null : trackToken.SelectToken(paramConfig.TrackContentName)?.Value(); // 1、匹配轨迹代码 // 2、匹配轨迹状态 if (matchedContentConfig == null && !paramConfig.TrackStatusName.IsNullOrWhiteSpace()) { var trackStatus = trackToken.SelectToken(paramConfig.TrackStatusName)?.Value(); if (!trackStatus.IsNullOrWhiteSpace()) { matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValues.Contains(trackStatus!)); } } // 3、匹配轨迹内容 if (matchedContentConfig == null && !paramConfig.TrackContentName.IsNullOrWhiteSpace()) { if (!trackContent.IsNullOrWhiteSpace()) { Debug.Assert(trackContent != null, nameof(trackContent) + " != null"); // 3.1 按内容直接匹配 // 3.2 按关键字分析轨迹内容 foreach (var contentConfig in contentConfigs) { foreach (var keyword in contentConfig.TrackKeywords) { if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase)) { matchedContentConfig = contentConfig; break; } } if (matchedContentConfig is null) continue; foreach (var keyword in contentConfig.TrackKeywordReverses) { if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase)) { matchedContentConfig = null; break; } } if (matchedContentConfig is not null) break; // 找到匹配的,跳出循环 } } } // 轨迹发生ISO时间 var trackIsoTime = paramConfig.TrackIsoTimeName.IsNullOrWhiteSpace() ? null : trackToken.SelectToken(paramConfig.TrackIsoTimeName)?.Value(); // 轨迹发生UTC时间 var trackUtcTime = string.IsNullOrWhiteSpace(paramConfig.TrackUtcTimeName) ? null : trackToken.SelectToken(paramConfig.TrackUtcTimeName)?.ToObject(); // 轨迹发生时间 var trackTime = trackIsoTime?.DateTime ?? (paramConfig.TrackTimeName.IsNullOrWhiteSpace() ? null : trackToken.SelectToken(paramConfig.TrackTimeName)?.Value()); // 轨迹发生时区 var trackTimeZone = (float?) trackIsoTime?.Offset.TotalHours ?? (paramConfig.TrackTimeZoneName.IsNullOrWhiteSpace() ? null : trackToken.SelectToken(paramConfig.TrackTimeZoneName)?.Value()); // 轨迹发生地 var trackPlace = paramConfig.TrackPlaceName.IsNullOrWhiteSpace() ? null : trackToken.SelectToken(paramConfig.TrackPlaceName)?.Value(); // 轨迹扩展信息,根据输入的多个字段提取报文内容,构建一个新的对象的json形式 var trackExtra = paramConfig.TrackExtraNames == null ? null : ExtraBuild(trackToken, paramConfig.TrackExtraNames); // 计算时间和时区信息 (trackTime, trackTimeZone) = CalcTimeAndTimezone(trackIsoTime, trackUtcTime, trackTime, trackTimeZone); if (matchedContentConfig is null) { logger.LogDebug("订单号:{OrderNo},未匹配到轨迹内容,轨迹报文:{TrackContent}", context.OrderNo, trackToken.ToString()); await InsertNotMappedTrackAsync(trackToken, context, trackPlace, trackTime.Value, trackTimeZone.Value, trackContent, trackExtra); return; } // 新增/更新轨迹 var trackUpsertEvent = new TrackUpsertEvent { OrderIdentifiers = new[] { new OrderIdentifier { OrderId = context.OrderId, OrderNo = context.OrderNo, OrderType = context.OrderType, } }, TrackRefreshData = new TrackRefreshStruct { ServiceProviderCode = context.ServiceProviderCode, OriginTrackDataHash = originTrackDataHash, OriginTrackContent = trackContent ?? GlobalConst.EmptyString, }, AppKey = context.AppKey, Customer = context.Customer, TrackCode = matchedContentConfig.TrackContentCode, TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackTime = trackTime.Value, TrackTimeZone = trackTimeZone.Value, IsAdd = true, //S06新增 //服务商轨迹默认不对外 IsExternal = false, Source = (int) TrackSourceEnum.ExternalServiceProvide, Extra = trackExtra }; await capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent); } private async Task InsertNotMappedTrackAsync(JToken trackToken, TrackRefreshContext context, string? trackPlace, DateTime trackTime, float trackTimeZone, string? trackContent, string? extra) { // 新增服务商轨迹到es var track = new TrackDetailsIndex { IndexName = $"{GlobalConst.TrackDetailsIndex}-{DateTime.Now:yyyy.MM}", AppKey = context.AppKey, OrderId = context.OrderId, OrderType = context.OrderType, OrderNo = context.OrderNo, TrackPlace = trackPlace ?? GlobalConst.EmptyString, TrackTime = trackTime, TrackTimeZone = trackTimeZone, ServiceProviderCode = context.ServiceProviderCode, OriginTrackDataHash = StringHelper.GetHash(JsonConvert.SerializeObject(trackToken)), OriginTrackContent = trackContent ?? GlobalConst.EmptyString, CreateUser = GlobalConst.SystemUser, CreateTime = DateTime.Now, //来源轨迹内容,第一次新增的时候使用轨迹内容表提供的 TrackStatusId = GlobalConst.NotExistId, TrackStatus = GlobalConst.EmptyString, TrackStatusCode = GlobalConst.EmptyString, TrackContentId = GlobalConst.NotExistId, TrackContent = GlobalConst.EmptyString, TrackEnContent = GlobalConst.EmptyString, //S06新增轨迹明细字段 //服务商轨迹默认不对外 IsExternal = false, Source = (int) TrackSourceEnum.ExternalServiceProvide, Extra = extra }; track.ZeroTrackTime = Math.Abs(track.TrackTimeZone - GlobalConst.NoneTrackTimeZone) < 1 ? track.TrackTime : track.TrackTime.Subtract(TimeSpan.FromHours(track.TrackTimeZone)); await trackDetailsEsRepository.AddAsync(track); } private static string? ExtraBuild(JToken jToken, string trackExtraNames) { const string separator = ","; var trackExtraNamesArray = trackExtraNames.Split(separator); if (trackExtraNamesArray.Length == 0) return null; var jObject = new JObject(); foreach (var fieldName in trackExtraNamesArray) { var token = jToken.SelectToken(fieldName); if (token == null) continue; jObject.Add(fieldName, token); } return jObject.ToString(); } // 原始时间(utc0+时区),时区 private static (DateTime Time, float TimeZone) CalcTimeAndTimezone(DateTimeOffset? isoTime, DateTime? utcTime, DateTime? unknownTime, float? timezone) { // 存在iso时间 if (isoTime.HasValue) { return (isoTime.Value.DateTime, (float) isoTime.Value.Offset.TotalHours); } // 存在时区 if (timezone.HasValue) { if (utcTime.HasValue) return (utcTime.Value.AddHours(timezone.Value), timezone.Value); if (unknownTime.HasValue) return (unknownTime.Value, timezone.Value); // tips:不存在时间信息,使用一个非常小的时间来代表不存在! return (GlobalConst.MinDateTime, timezone.Value); } if (utcTime.HasValue && unknownTime.HasValue) { // 可计算出时区信息 if (utcTime != unknownTime) { var timeZone = (float) (unknownTime - utcTime).Value.TotalHours; return (unknownTime.Value, timeZone); } // 只有utc时间,存为0时区 return (utcTime.Value, 0); } if (utcTime.HasValue) { return (utcTime.Value, 0); } if (unknownTime.HasValue) { return (unknownTime.Value, GlobalConst.NoneTrackTimeZone); } return (GlobalConst.MinDateTime, GlobalConst.NoneTrackTimeZone); } }