diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs index 5e35d483..38a3a3f2 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/CompleteWarningHandler.cs @@ -7,6 +7,7 @@ using SJZY.Track.Domain.Aggregates.WarningAggregate.Signals; using SJZY.Track.Domain.Dtos.TrackDetails; using SJZY.Track.Domain.Dtos.Warning; using SJZY.Track.Domain.Enums; +using SJZY.Track.Domain.Extensions; using Volo.Abp.DependencyInjection; namespace SJZY.Track.Application.Services.Warning.Handlers; @@ -35,44 +36,34 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient // 1.只处理未删除的订单 var orderIdSet = deletedSignals.Select(x => x.OrderId).ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id)); - if (orders.Count == 0) return; //不存在符合条件的订单，直接提交信号丢弃 + if (orders.Count == 0) return; - // 2.查出订单指定状态的轨迹（tips:为了指定状态的轨迹是真的被清空了） + // 2.查出订单轨迹 var orderIds = orders.Select(x => x.Id).ToHashSet(); - var orderIdTrackStatusIdsDict = deletedSignals - .Where(x => orderIds.Contains(x.OrderId)) // 过滤掉不必要的信号 - .GroupBy(x => x.OrderId) - .ToDictionary( - x => x.Key, - y => y.Select(x => x.DeletedTrack.TrackStatusId) - .ToHashSet() - ); var orderTracksDict = await _trackDetailsEsRepository.GetOrderTracksDictAsync(new GetOrderTracksDictVO { - OrderIds = orderIdSet + OrderIds = orderIds }); - // 构建k:orderId,v:[k:trackStatusId,v:true] - var orderIdTrackStatusGroupsDict = orderTracksDict - .ToDictionary(x => x.Key, - y => y.Value - .GroupBy(x => x.TrackStatusId) - .ToDictionary(a => a.Key, _ => true) - ); + // 构建订单id和轨迹状态ids字典 + var orderIdTrackStatusIdsDict = orderTracksDict.ToDictionary( + x => x.Key, + y => y.Value.Select(x => x.TrackStatusId).ToHashSet() + ); // 3.查出预警明细(待处理) var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO { - OrderIds = orderIdSet, + OrderIds = orderIds, }); if (warningDetails.Count == 0) return; - // 查询规则项目 + // 4.查询规则项目 var ruleItemIds = warningDetails.Select(x => x.CompleteWarningRuleItemId).ToHashSet(); var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds); var ruleItemDict = ruleItems.ToDictionary(x => x.Id); - // 4.处理明细 + // 5.处理明细 var closeDetailIds = new HashSet(); var updateDetails = new List(); foreach (var detail in warningDetails) @@ -85,16 +76,15 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient continue; } - var trackStatusIdDict = orderIdTrackStatusGroupsDict.GetValueOrDefault(detail.OrderId); + var trackStatusIds = orderIdTrackStatusIdsDict.GetValueOrDefault(detail.OrderId, Empty.Set); // 订单无任何轨迹,或者不存在检查点状态轨迹 - if (trackStatusIdDict == null || !trackStatusIdDict.ContainsKey(detail.CheckPointTrackStatusId)) + if (!trackStatusIds.Contains(detail.CheckPointTrackStatusId)) { closeDetailIds.Add(detail.Id); continue; } // 查看缺失轨迹状态表达式是否有变化,如果有则更新 - var trackStatusIds = trackStatusIdDict.Keys.ToHashSet(); var expression = ruleItem.BuildMissingTrackStatusIdExpression(trackStatusIds); if (!string.Equals(detail.MissingTrackStatusIds, expression)) { @@ -102,21 +92,23 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient updateDetails.Add(detail); } - // todo:需要处理要求轨迹完整的情况？ + // tips:需要处理要求轨迹完整的情况？ } + // 6.批量更新轨迹 if (updateDetails.Count > 0) await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails); + // 没有需要关闭的预警明细 if (closeDetailIds.Count == 0) return; - // 5.预警信息状态修改为【已关闭】 + // 7.预警信息状态修改为【已关闭】 await _warningDetailsEsRepository.BulkCloseCompleteWarningDetailsAsync(new BulkCloseCompleteWarningDetailsVO { DetailsIds = closeDetailIds }); - // 6.关闭预警通知任务 + // 8.关闭预警通知任务 await _warningTaskRepository.CloseWarningNotifyTaskAsync(x => !x.IsClosed && closeDetailIds.Contains(x.WarningDetailsId) @@ -133,92 +125,125 @@ public sealed class CompleteWarningHandler : ICompleteWarningHandler, ITransient // 1.只处理未删除的订单 var orderIdSet = addedSignals.Select(x => x.OrderId).ToHashSet(); var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id)); - if (orders.Count == 0) return; //不存在符合条件的订单，直接提交信号丢弃 + if (orders.Count == 0) return; var orderIds = orders.Select(x => x.Id).ToHashSet(); - // 2.查询订单每个轨迹状态的最早轨迹 - var orderTracksDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds); - // 3.以单的维度处理 - // todo:优化流程减少N+1 + // 2.查询订单每个轨迹状态的最早轨迹 + var orderTrackStatusEarliestDictDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds); + + // 3.查规则主体 + var conditionDict = orders.GroupBy(x => (x.SubSystemId, x.CustomId)).ToDictionary(x => x.Key, y => y.ToList()); + var conditions = conditionDict.Keys.ToHashSet(); + var ruleDict = await _warningRuleRepository.GetWarningRuleForEnabledAsync(conditions, WarningRuleType.完整预警); + if (ruleDict.Count == 0) return; + + // 4.查规则项目 + var ruleIds = ruleDict.Keys.ToHashSet(); + var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(x => ruleIds.Contains(x.WarningId)); + if (ruleItems.Count == 0) return; + var ruleItemDict = ruleItems.ToDictionary(x => x.Id); + + // 5.构建订单id 命中规则，完成规则 字典 + var orderIdHitRuleItemDict = new Dictionary>(); + var orderIdDoneRuleItemIdsDict = new Dictionary>(); + var orderIdRuleItemsDict = new Dictionary>(); + foreach (var group in ruleItems.GroupBy(x => x.WarningId)) + { + var rule = ruleDict[group.Key]; + var hitRuleOrders = conditionDict[(rule.SubSystemId, rule.CustomId)]; //tips:一定存在 + foreach (var order in hitRuleOrders) + { + var trackStatusIdTrackDict = orderTrackStatusEarliestDictDict.GetValueOrDefault(order.Id); + if (trackStatusIdTrackDict == null) continue; + var trackStatusIds = trackStatusIdTrackDict.Keys.ToHashSet(); + foreach (var ruleItem in group) + { + if (ruleItem.CanHitRule(order.OrderType, trackStatusIds)) + { + orderIdRuleItemsDict.GetOrAdd(order.Id, _ => new HashSet()).Add(ruleItem.Id); + orderIdHitRuleItemDict.GetOrAdd(order.Id, _ => new List()).Add(ruleItem); + continue; + } + + if (ruleItem.IsDone(order.OrderType, trackStatusIds)) + { + orderIdRuleItemsDict.GetOrAdd(order.Id, _ => new HashSet()).Add(ruleItem.Id); + orderIdDoneRuleItemIdsDict.GetOrAdd(order.Id, _ => new HashSet()).Add(ruleItem.Id); + } + } + } + } + + // 6.查预警信息 + var allWarningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO + { + OrderIdRuleItemIdsDict = orderIdRuleItemsDict, + }); + var orderIdWarningDetailDict = allWarningDetails + .GroupBy(x => x.OrderId) + .ToDictionary(x => x.Key, y => y.ToList()); + + // 7.以单的维度处理 var addDetails = new List(); var updateDetails = new List(); - var closeDetailsIds = new HashSet(); foreach (var order in orders) { - var trackStatusIdTrackDict = orderTracksDict.GetValueOrDefault(order.Id); - if (trackStatusIdTrackDict == null) continue; + // 订单轨迹状态 + var orderTrackEarliestTracksDict = orderTrackStatusEarliestDictDict.GetValueOrDefault(order.Id); + if (orderTrackEarliestTracksDict == null) continue; + var orderTrackStatusIds = orderTrackEarliestTracksDict.Keys.ToHashSet(); - // 查询规则主体 - var rule = await _warningRuleRepository.GetWarningRuleForEnabledAsync(order.SubSystemId, order.CustomId, WarningRuleType.完整预警); - if (rule == null) continue; + // 订单命中的预警规则项目 + var hitRuleItems = orderIdHitRuleItemDict.GetValueOrDefault(order.Id, Empty.List); + var hitRuleItemIds = hitRuleItems.Select(x => x.Id).ToHashSet(); - // 查询规则项目 - var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(rule.Id); - if (ruleItems.Count == 0) continue; - - // 匹配规则项目（要生成预警的，已完成的） - var trackStatusIds = trackStatusIdTrackDict.Keys.ToHashSet(); - var hitRuleItems = new List(); - var doneRuleItemIds = new HashSet(); - foreach (var ruleItem in ruleItems) + // 处理存在的预警明细 + var orderWarningDetails = orderIdWarningDetailDict.GetValueOrDefault(order.Id, Empty.List); + if (orderWarningDetails.Count > 0) { - if (ruleItem.CanHitRule(order.OrderType, trackStatusIds)) - hitRuleItems.Add(ruleItem); - if (ruleItem.IsDone(order.OrderType, trackStatusIds)) - doneRuleItemIds.Add(ruleItem.Id); + var doneRuleItemIds = orderIdDoneRuleItemIdsDict.GetValueOrDefault(order.Id, Empty.Set); + var doneRuleItemDetails = orderWarningDetails.Where(x => doneRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList(); + if (doneRuleItemDetails.Count > 0) + { + doneRuleItemDetails.ForEach(detail => + { + detail.Status = CompleteWarningStatus.已处理.GetHashCode(); + detail.MissingTrackStatusIds = null; // 清空缺失轨迹 + detail.ProcessedTime = DateTime.Now; + }); + updateDetails.AddRange(doneRuleItemDetails); + } + + // 处理已存在的预警明细（命中规则的） + var existsWarningDetails = orderWarningDetails.Where(x => hitRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList(); + var needUpdateDetails = HandleExistsDetails(orderTrackStatusIds, existsWarningDetails, ruleItemDict); + updateDetails.AddRange(needUpdateDetails); + + // 排除已经生成预警信息的规则 + hitRuleItemIds.ExceptWith(existsWarningDetails.Select(x => x.CompleteWarningRuleItemId)); } - if (hitRuleItems.Count == 0 && doneRuleItemIds.Count == 0) return; - var hitRuleItemDict = hitRuleItems.ToDictionary(x => x.Id); - - // 查出规则已经生成预警信息 - var hitRuleItemIds = hitRuleItemDict.Keys.ToHashSet(); - var warningDetails = await _warningDetailsEsRepository.GetCompleteWarningDetailsAsync(new GetCompleteWarningDetailsVO - { - OrderIds = new HashSet(1) - { - order.Id - }, - RuleItemIds = new HashSet() - .Union(hitRuleItemIds) - .Union(doneRuleItemIds) - .ToHashSet(), - }); - - // 已完结的预警规则，稍后修改明细状态为【已处理】 - closeDetailsIds.UnionWith(warningDetails.Where(x => doneRuleItemIds.Contains(x.CompleteWarningRuleItemId)).Select(x => x.Id)); - - // 处理已存在的预警明细（命中规则的） - var existsWarningDetails = warningDetails.Where(x => hitRuleItemIds.Contains(x.CompleteWarningRuleItemId)).ToList(); - var needUpdateDetails = HandleExistsDetails(trackStatusIds, existsWarningDetails, hitRuleItemDict); - updateDetails.AddRange(needUpdateDetails); - - // 排除已经生成预警信息的规则 - hitRuleItemIds.ExceptWith(existsWarningDetails.Select(x => x.CompleteWarningRuleItemId)); + if (hitRuleItemIds.Count == 0) continue; // 生成预警明细 - if (hitRuleItemIds.Count == 0) continue; + var rule = ruleDict[ruleItems.First().WarningId]; //tips：一定存在 var needAddDetails = hitRuleItems .Where(x => hitRuleItemIds.Contains(x.Id)) - .Select(x => BuildCompleteWarningDetails(order, trackStatusIds, x, rule.Version)); - + .Select(x => BuildCompleteWarningDetails(order, orderTrackStatusIds, x, rule.Version)); addDetails.AddRange(needAddDetails); } - // 4.预警由于不缺失轨迹变为失效，修改明细状态为【已处理】，并关闭相关预警通知任务 - if (closeDetailsIds.Count > 0) - await CloseWarningDetailLinksAsync(closeDetailsIds, CompleteWarningStatus.已处理); - // 更新预警明细 + // 8.批量更新预警明细 if (updateDetails.Count > 0) await _warningDetailsEsRepository.BulkUpdateCompleteWarningDetailsAsync(updateDetails); if (addDetails.Count == 0) return; - // 5.批量新增预警明细 + // 9.批量新增预警明细 await _warningDetailsEsRepository.BulkInsertCompleteWarningDetailsAsync(addDetails); - // 6.批量新增预警通知任务 + // 10.批量新增预警通知任务 var notifyTasks = BuildWarningNotifyTask(addDetails); await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(notifyTasks); } diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs index 616beec6..954c77eb 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs @@ -57,7 +57,6 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe { TrackEsIds = records.Select(x => x.TrackEsId).ToHashSet() }); - if (warningDetails.Count == 0) return; var detailIds = warningDetails.Select(x => x.Id).ToHashSet(); @@ -237,32 +236,13 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe if (addNotifyTasks.Count > 0) await _warningTaskRepository.BulkInsertWarningNotifyTaskAsync(addNotifyTasks); - - // 5个订单 每个订单推5个轨迹，失败（整体），5条历史成功推送记录（EsId+版本号）， - // 第一次，生成20预警明细，生成一个回传报文并关联20预警明细，生成重试任务，重试任务重试25条推送 - // 重试阶段的重试结果维护同维护20条预警明细，如果这一次失败超过了最大重试次数，并且命中了kpi考核，则会生成20预警通知任务 - - // 可能最后会生成20条人工介入的明细， - // 如果是人工手动推送则还是有可能会失败，回传失败又会走到这里，不要在进行重试，但是要维护回传报文 - // todo:人工介入直到推送成功才可关闭吗？ - //新增/修改轨迹 -> 系统推送 -> 推送回传 -> 预警计算 - //手动推送 -> 推送回传 -> 预警计算 + //手动推送 -> 推送回传 -> 预警计算 //重试推送 -> 推送回传 -> 预警计算 - - // todo: 预警页面的推送，根据明细id找到明细状态，确保是人工介入才能推送（新接口） - - // 历史没有1成功的记录 - // 1失败，回调，查询1明细包含【已关闭】为0，创建1明细，创建第一次重试任务T1 - // 修改为2，导致1明细关闭 因为是同一个EsId， - // T1第一次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，什么也不做 - // 。。。假设中途没有变更 - // T1第二次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，修改，创建第三次重试任务T1 - // T2第一次重试1，1失败，回调，查询1明细包含【已关闭】数量为1，修改，创建第二次重试任务T2 } // 从推送内容提取轨迹id+版本号 - private (Dictionary> OrderNoTrackIdVersionDict, Dictionary AllTrackIdVersionDict) + private static (Dictionary> OrderNoTrackIdVersionDict, Dictionary AllTrackIdVersionDict) ExtractOrderNoTrackIdVersionDict(List pushInputs, List updateInputs) { var inputs = (pushInputs.Count > 0 @@ -494,7 +474,6 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe { if (addRuleItems.Count > 0) await OnlyAddHandleAsync(rule, addRuleItems); - return; } diff --git a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs index 271f2a56..d69f1d9f 100644 --- a/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs +++ b/src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs @@ -226,7 +226,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen orders = orders.Where(x => earliestTrackStatusTrackDict.ContainsKey(x.Id)).ToList(); if (orders.Count == 0) return; - // 3.查轨迹主体 + // 3.查规则主体 var conditions = orders.Select(x => (x.SubSystemId, x.CustomId)).ToHashSet(); var ruleDict = await _warningRuleRepository.GetWarningRuleForEnabledAsync(conditions, WarningRuleType.运输时效预警); if (ruleDict.Count == 0) return; diff --git a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs index b2650f4f..3d571369 100644 --- a/src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs @@ -100,6 +100,9 @@ public interface IWarningRuleRepository : IBasicRepository // 按规则项目id获取预警规则 Task> GetCompleteRuleItemsAsync(HashSet ruleItemIds); + // 获取完整预警规则项目 + Task> GetCompleteRuleItemsAsync(Expression> whereExpression); + // 批量更新预警规则规则项目 Task BulkUpdateCompleteRuleItemsAsync(List entities); diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.ES.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.ES.cs index 90a530b1..3a9578be 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.ES.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.ES.cs @@ -22,7 +22,7 @@ public partial class GlobalConst /// public const string OrderOperationRecordIndex = "sjzy_track_order_operation_records"; - // todo：以下为新增需要配置索引模板(uat/prod) + // todo：以下为新增需要配置索引模板(prod) 生产环境还需要建立索引！！ /// /// 运输时效预警明细 es索引名称（模板按天） /// diff --git a/src/SJZY.Track.Domain/Dtos/Warning/GetCompleteWarningDetailsVO.cs b/src/SJZY.Track.Domain/Dtos/Warning/GetCompleteWarningDetailsVO.cs index 60a1c0a2..1e704c75 100644 --- a/src/SJZY.Track.Domain/Dtos/Warning/GetCompleteWarningDetailsVO.cs +++ b/src/SJZY.Track.Domain/Dtos/Warning/GetCompleteWarningDetailsVO.cs @@ -20,9 +20,13 @@ public class GetCompleteWarningDetailsVO // 订单id 与 检查点轨迹状态ids 同时满足 public Dictionary>? OrderIdCheckTracksStatusIdsDict { get; set; } + // 订单id 与 规则条目ids 同时满足 + public Dictionary>? OrderIdRuleItemIdsDict { get; set; } + public bool HasDetailsIds => DetailsIds?.Count > 0; public bool HasRuleItemIds => RuleItemIds?.Count > 0; public bool HasWarningIds => WarningIds?.Count > 0; public bool HasOrderIds => OrderIds?.Count > 0; public bool HasOrderIdCheckTracksStatusIdsDict => OrderIdCheckTracksStatusIdsDict?.Count > 0; + public bool HasOrderIdRuleItemIdsDict => OrderIdRuleItemIdsDict?.Count > 0; } diff --git a/src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs b/src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs index e8aa447c..8843d4b4 100644 --- a/src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs @@ -757,6 +757,19 @@ public class WarningDetailsEsRepository : IWarningDetailsEsRepository, ITransien isShouldQuery = true; } + else if (input.HasOrderIdRuleItemIdsDict) + { + foreach (var (orderId, ruleItemIds) in input.OrderIdRuleItemIdsDict!) + { + musts.Add(c => c.Bool(b => b.Must + ( + mu => mu.Term(f => f.OrderId, orderId), + mu => mu.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(ruleItemIds)) + ))); + } + + isShouldQuery = true; + } else { musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.CompleteWarningRuleItemId).Terms(input.RuleItemIds))); diff --git a/src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs b/src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs index 331e2ec8..c79c1c54 100644 --- a/src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs @@ -277,6 +277,13 @@ public class WarningRuleRepository : BasicRepository, IWarnin .ToListAsync(); } + public Task> GetCompleteRuleItemsAsync(Expression> whereExpression) + { + return Context.Queryable() + .Where(whereExpression) + .ToListAsync(); + } + public async Task BulkUpdateCompleteRuleItemsAsync(List entities) { return await Context.Updateable(entities)