321 lines
13 KiB
C#
321 lines
13 KiB
C#
using System.Diagnostics;
|
||
using System.Linq.Expressions;
|
||
using DotNetCore.CAP;
|
||
using Newtonsoft.Json;
|
||
using Newtonsoft.Json.Linq;
|
||
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.Consts;
|
||
using SJZY.Track.Domain.Enums;
|
||
using SJZY.Track.Domain.Queries;
|
||
using Volo.Abp.DependencyInjection;
|
||
|
||
namespace SJZY.Track.Application.CAP.EventHandlers;
|
||
|
||
public class TrackRefreshEventHandler : ITransientDependency, ICapSubscribe
|
||
{
|
||
private readonly IOrderRepository _orderRepository;
|
||
private readonly ICapPublisher _capPublisher;
|
||
private readonly ITrackPullCacheService _trackPullCacheService;
|
||
private readonly ITrackDetailsRepository _trackDetailsRepository;
|
||
private readonly ITimezoneQueries _timezoneQueries;
|
||
|
||
public TrackRefreshEventHandler(IOrderRepository orderRepository,
|
||
ICapPublisher capPublisher, ITrackPullCacheService trackPullCacheService,
|
||
ITrackDetailsRepository trackDetailsRepository, ITimezoneQueries timezoneQueries)
|
||
{
|
||
_orderRepository = orderRepository;
|
||
_capPublisher = capPublisher;
|
||
_trackPullCacheService = trackPullCacheService;
|
||
_trackDetailsRepository = trackDetailsRepository;
|
||
_timezoneQueries = timezoneQueries;
|
||
}
|
||
|
||
[CapSubscribe(GlobalConst.TrackRefreshSubscribe, Group = GlobalConst.TrackRefreshSubscribe)]
|
||
public async Task ConsumerAsync(InterfaceRelayForwardReplayEvent @event)
|
||
{
|
||
if (!@event.IsSuccess) return;
|
||
var attach = Base64Helper.DeSerialize<TrackRefreshAttach>(@event.Attach);
|
||
if (attach is null) return;
|
||
|
||
var config = await _trackPullCacheService.GetPullParamWithContentConfigAsync(attach.SubSystemId,
|
||
attach.ServiceProviderId);
|
||
if (@event.Data == null || config == null) return;
|
||
|
||
var data = JObject.Parse(@event.Data);
|
||
var orderToken = data[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, TrackPullParamWithContentConfigDto paramWithContentConfig,
|
||
TrackRefreshAttach attach)
|
||
{
|
||
var paramConfig = paramWithContentConfig.ParamConfig;
|
||
var orderNo = orderToken[paramConfig.OrderNoName]?.Value<string>();
|
||
if (orderNo.IsNullOrWhiteSpace()) return;
|
||
|
||
var trackToken = orderToken[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, paramWithContentConfig, context);
|
||
}
|
||
}
|
||
else if (trackToken is JObject)
|
||
{
|
||
await HandleTrackAsync(trackToken, paramWithContentConfig, context);
|
||
}
|
||
}
|
||
|
||
private async Task HandleTrackAsync(JToken trackToken, TrackPullParamWithContentConfigDto paramWithContentConfig,
|
||
TrackRefreshContext context)
|
||
{
|
||
var originTrackData = JsonConvert.SerializeObject(trackToken);
|
||
// 如果已经存在一模一样的报文了说明处理过了，直接return
|
||
if (await _trackDetailsRepository.AnyAsync(context.OrderId, originTrackData)) return;
|
||
|
||
var paramConfig = paramWithContentConfig.ParamConfig;
|
||
var contentConfigs = paramWithContentConfig.ContentConfigs;
|
||
TrackPullContentConfigDto? matchedContentConfig = null;
|
||
|
||
// 1、匹配轨迹代码
|
||
if (paramConfig.TrackContentCodeName != null)
|
||
{
|
||
var trackCode = trackToken[paramConfig.TrackContentCodeName]?.Value<string>();
|
||
if (!trackCode.IsNullOrWhiteSpace())
|
||
{
|
||
matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackContentCodeValue == trackCode);
|
||
}
|
||
}
|
||
// 2、匹配轨迹状态
|
||
else if (paramConfig.TrackStatusName != null)
|
||
{
|
||
var trackStatus = trackToken[paramConfig.TrackStatusName]?.Value<string>();
|
||
if (!trackStatus.IsNullOrWhiteSpace())
|
||
{
|
||
matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackStatusValue == trackStatus);
|
||
}
|
||
}
|
||
// 3、匹配轨迹内容
|
||
else if (paramConfig.TrackContentName != null)
|
||
{
|
||
var trackContent = trackToken[paramConfig.TrackContentName]?.Value<string>();
|
||
if (!trackContent.IsNullOrWhiteSpace())
|
||
{
|
||
Debug.Assert(trackContent != null, nameof(trackContent) + " != null");
|
||
matchedContentConfig = contentConfigs.FirstOrDefault(x => x.TrackContentValue == trackContent);
|
||
|
||
// 按关键字分析轨迹内容
|
||
if (matchedContentConfig is null)
|
||
{
|
||
foreach (var contentConfig in contentConfigs)
|
||
{
|
||
foreach (var keyword in contentConfig.TrackKeyword?.Split(',') ?? Enumerable.Empty<string>())
|
||
{
|
||
if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase))
|
||
{
|
||
matchedContentConfig = contentConfig;
|
||
break;
|
||
}
|
||
}
|
||
|
||
if (matchedContentConfig is null) continue;
|
||
|
||
foreach (var keyword in contentConfig.TrackKeywordReverse?.Split(',') ??
|
||
Enumerable.Empty<string>())
|
||
{
|
||
if (trackContent.Contains(keyword, StringComparison.OrdinalIgnoreCase))
|
||
{
|
||
matchedContentConfig = null;
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// 轨迹发生地
|
||
var trackPlace = paramConfig.TrackPlaceName == null
|
||
? null
|
||
: trackToken[paramConfig.TrackPlaceName]?.Value<string>();
|
||
var hasPlaceBottomLine = false; //执行了兜底但是没有兜底住，需要记录
|
||
if (trackPlace.IsNullOrWhiteSpace() && matchedContentConfig?.TrackPlaceValue != null && matchedContentConfig.TrackPlaceValue.HasValue)
|
||
{
|
||
//兜底没配置的就不需要兜底，拉取的发生地没就没
|
||
trackPlace = await GetPlaceByOrder(matchedContentConfig!.TrackPlaceValue.Value,context.OrderId);
|
||
hasPlaceBottomLine = string.IsNullOrWhiteSpace(trackPlace);
|
||
}
|
||
|
||
// 轨迹发生时间
|
||
var trackTime = paramConfig.TrackTimeName == null
|
||
? null
|
||
: trackToken[paramConfig.TrackTimeName]?.Value<DateTime>();
|
||
// 轨迹发生时区
|
||
var trackTimeZone = paramConfig.TrackTimeZoneName == null
|
||
? null
|
||
: trackToken[paramConfig.TrackTimeZoneName]?.Value<float>();
|
||
|
||
//兜底时区
|
||
var hasTimezoneBottomLine = false; //执行了兜底但是没有兜底住，需要记录
|
||
if (!trackTimeZone.HasValue && matchedContentConfig?.TrackZonePlaceValue != null && matchedContentConfig.TrackZonePlaceValue.HasValue)
|
||
{
|
||
//兜底没配置的就不需要兜底，拉取的发生地没就没
|
||
var timezone= await GetTimezone(matchedContentConfig,context.OrderId);
|
||
|
||
trackTimeZone = timezone;
|
||
hasTimezoneBottomLine = !timezone.HasValue;
|
||
}
|
||
|
||
if (matchedContentConfig is null)
|
||
{
|
||
var notMappedEvent = new TrackInsertForNotMappedEvent
|
||
{
|
||
OrderId = context.OrderId,
|
||
OrderType = context.OrderType,
|
||
OrderNo = context.OrderNo,
|
||
ServiceProviderName = context.ServiceProviderName,
|
||
OriginTrackData = JsonConvert.SerializeObject(trackToken),
|
||
TrackPlace = trackPlace ?? GlobalConst.EmptyString,
|
||
TrackTime = trackTime ?? DateTime.Now,
|
||
TrackTimeZone = trackTimeZone ?? 8
|
||
};
|
||
if (paramConfig.TrackContentName != null)
|
||
{
|
||
notMappedEvent.OriginTrackContent = trackToken[paramConfig.TrackContentName]?.Value<string>();
|
||
}
|
||
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackInsertForNotMappedEventSubscribe, notMappedEvent);
|
||
return;
|
||
}
|
||
|
||
int deficiencyTagValue = 0;
|
||
// 触发了兜底，但是没有兜底住，对订单进行标记
|
||
if (hasPlaceBottomLine || hasTimezoneBottomLine)
|
||
{
|
||
//1表示时区，2表示发生地，3表示两种
|
||
var placeTagValue = hasPlaceBottomLine ? 2 : 0;
|
||
var timezoneTagValue = hasTimezoneBottomLine ? 1 : 0;
|
||
deficiencyTagValue = placeTagValue + timezoneTagValue;
|
||
}
|
||
|
||
// 新增/更新轨迹
|
||
var trackUpsertEvent = new TrackUpsertEvent
|
||
{
|
||
OrderIdentifiers = new[]
|
||
{
|
||
new OrderIdentifier
|
||
{
|
||
OrderId = context.OrderId,
|
||
OrderNo = context.OrderNo,
|
||
OrderType = context.OrderType,
|
||
}
|
||
},
|
||
|
||
TrackRefreshData = new TrackRefreshStruct
|
||
{
|
||
ServiceProviderName = context.ServiceProviderName,
|
||
OriginTrackData = originTrackData,
|
||
IsCompleted = matchedContentConfig.IsCompleted,
|
||
OriginTrackContent = paramConfig.TrackContentName != null
|
||
? trackToken[paramConfig.TrackContentName]?.Value<string>()
|
||
: GlobalConst.EmptyString
|
||
},
|
||
SubSystemId = context.SubSystemId,
|
||
TrackCode = matchedContentConfig.TrackContentCode,
|
||
TrackPlace = trackPlace ?? GlobalConst.EmptyString,
|
||
TrackTime = trackTime ?? DateTime.Now,
|
||
TrackTimeZone = trackTimeZone ?? 8,
|
||
DeficiencyTagValue = deficiencyTagValue
|
||
};
|
||
|
||
trackUpsertEvent.IsAdd =
|
||
!await _trackDetailsRepository.AnyByCodeAsync(context.OrderId, matchedContentConfig.TrackContentCode);
|
||
await _capPublisher.PublishAsync(GlobalConst.TrackUpsertForRefreshSubscribe, trackUpsertEvent);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 根据订单id获取兜底位置
|
||
/// </summary>
|
||
/// <param name="trackValue"></param>
|
||
/// <param name="orderId"></param>
|
||
/// <returns></returns>
|
||
private Task<string?> GetPlaceByOrder(int trackValue,int orderId)
|
||
{
|
||
Expression<Func<OrderEntity, string?>> 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);
|
||
}
|
||
|
||
/// <summary>
|
||
/// 获取时区
|
||
/// </summary>
|
||
/// <param name="matchedContentConfig"></param>
|
||
/// <param name="orderId"></param>
|
||
/// <returns></returns>
|
||
private async Task<float?> 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;
|
||
}
|
||
} |