ideship/track
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge branch 'S07' into test

Browse Source
This commit is contained in:
hardstylewyl 2025-04-10 09:17:29 +08:00
commit 0c1cca5a8d
8 changed files with 236 additions and 199 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
src/SJZY.Track.Application/Hangfire/WarningNotifyAggregator.cs
View File

@ -166,7 +166,7 @@ public sealed class WarningNotifyAggregator : ITransientDependency
// 3.场景规则存在验证
var ruleItemIds = validTasks.Select(x => x.WarningRuleItemId).ToHashSet();
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id), y => y);
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id));
if (ruleItems.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
// 过滤掉不存在的场景规则

12
src/SJZY.Track.Application/Services/Warning/Handlers/ReturnWarningHandler.cs
View File

@ -66,7 +66,8 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
!x.IsClosed
&& detailIds.Contains(x.WarningDetailsId)
&& x.WarningType == WarningRuleType.回传时效预警.GetHashCode());
&& x.WarningType == WarningRuleType.回传时效预警.GetHashCode()
);
// 3.修改预警明细状态为【推送完成】，补充回传时间
var trackReturnTime = DateTime.Now;
@ -427,11 +428,10 @@ public sealed class ReturnWarningHandler : IReturnWarningHandler, ITransientDepe
if (orderIdTracksDict.Count == 0) return; //无需进行处理，直接提交信号丢弃
// 1.2 构建订单和轨迹内容id字典
var orderIdTrackContentIdsDict = orderIdTracksDict
.ToDictionary(
x => x.Key,
y => y.Value.Select(x => x.TrackContentId).ToHashSet()
);
var orderIdTrackContentIdsDict = orderIdTracksDict.ToDictionary(
x => x.Key,
y => y.Value.Select(x => x.TrackContentId).ToHashSet()
);
// 2.获取预警明细，订单id+轨迹内容ids 默认关注（重试中，人工介入）
var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO

365
src/SJZY.Track.Application/Services/Warning/Handlers/TransportWarningHandler.cs
View File

@ -1,3 +1,4 @@
using System.Linq.Expressions;
using SJZY.Track.Application.Services.Warning.Model;
using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
using SJZY.Track.Domain.Aggregates.OrderAggregate;
@ -11,6 +12,7 @@ using SJZY.Track.Domain.Dtos.TrackDetails;
using SJZY.Track.Domain.Dtos.Warning;
using SJZY.Track.Domain.Enums;
using SJZY.Track.Domain.Extensions;
using SqlSugar;
using Volo.Abp.DependencyInjection;
namespace SJZY.Track.Application.Services.Warning.Handlers;
@ -36,9 +38,10 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
_trackStatusRepository = trackStatusRepository;
}
#region todo-test: 预警信号批处理-删除轨迹
#region 预警信号批处理-删除轨迹
// tips:删除轨迹信号在运输预警处理只关注被删除的轨迹的状态
// 可能影响【运输预警明细】【运输预警明细生成任务】
public async Task TrackDeletedHandleAsync(List<TrackDeletedSignal> deletedSignals)
{
// 1.查询订单
@ -50,42 +53,35 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
)).ToHashSet();
if (orderIds.Count == 0) return;
// 1.1 构建订单和轨迹状态ids字典
// 利用信号构建订单和轨迹状态ids字典
var orderIdTrackStatusIdsDict = deletedSignals
.Where(s => orderIds.Contains(s.OrderId)) // tips:信号过滤
.Where(s => orderIds.Contains(s.OrderId)) // tips:过滤掉无需处理的信号
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.Select(z => z.DeletedTrack.TrackStatusId).ToHashSet());
if (orderIdTrackStatusIdsDict.Count == 0) return;
// 2.查询订单每个状态最早轨迹的字点(tips：由于已经是删除，查询出的轨迹不会包含当前已经删除的轨迹)
var trackStatusEarliestDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
var orderIdTrackStatusIdEarliestDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
// 3.查订单的运输时效预警明细，命中起始轨迹状态的(即将超时/已超时)
var warningDetails = await _warningDetailsEsRepository.GetTransportWarningDetailsAsync(new GetTransportWarningDetailsVO
{
// tips:要求订单id+轨迹状态id同时满足的预警明细
OrderIdStartTracksStatusIdsDict = orderIdTrackStatusIdsDict
OrderIdStartTracksStatusIdsDict = orderIdTrackStatusIdsDict // tips:要求订单id+轨迹状态id同时满足的预警明细
});
// 订单和它的预警明细字典
var orderDetailsDict = warningDetails
var orderIdWarningDetailsDict = warningDetails
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.ToList());
// 4.查询预警明细生成任务，命中起始轨迹状态的(即将超时/已超时)
// 4.查询预警明细生成任务
var generateTasks = await _warningTaskRepository.GetTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& orderIds.Contains(x.OrderId)
// && orderIdTrackStatusIdsDict[x.OrderId].Contains(x.StartTrackStatusId)
!x.IsClosed
&& orderIds.Contains(x.OrderId)
);
generateTasks = generateTasks.Where(x =>
{
var statusIds = orderIdTrackStatusIdsDict[x.OrderId];
return statusIds.Contains(x.StartTrackStatusId);
}).ToList();
// 订单和它的预警生成任务字典
var oderGenerateTaskDict = generateTasks
// 订单和它的预警生成任务字典, tips:只选择命中起始轨迹状态的任务
var orderIdGenerateTaskDict = generateTasks
.Where(x => orderIdTrackStatusIdsDict[x.OrderId].Contains(x.StartTrackStatusId)) // tips:字典一定存在该订单
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.ToList());
@ -95,24 +91,23 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 以订单的维度处理预警明细
foreach (var orderId in orderIds)
{
var orderDetails = orderDetailsDict.GetValueOrDefault(orderId, Empty<TransportWarningDetailsIndex>.List);
var orderTrackStatusEarliestDict = trackStatusEarliestDict.GetValueOrDefault(orderId, new Dictionary<int, TrackDetailsIndex>(0));
var orderTasks = oderGenerateTaskDict.GetValueOrDefault(orderId, Empty<TransportWarningDetailsGenerateTaskEntity>.List);
var orderWarningDetails = orderIdWarningDetailsDict.GetValueOrDefault(orderId, Empty<TransportWarningDetailsIndex>.List);
var orderTrackStatusEarliestDict = orderIdTrackStatusIdEarliestDict.GetValueOrDefault(orderId, Empty<int, TrackDetailsIndex>.Dict);
var orderTasks = orderIdGenerateTaskDict.GetValueOrDefault(orderId, Empty<TransportWarningDetailsGenerateTaskEntity>.List);
// 明细和生成任务都不存在跳过
if (orderDetails.Count == 0 && orderTasks.Count == 0)
continue;
// 预警明细和生成任务都不存在跳过
if (orderWarningDetails.Count == 0 && orderTasks.Count == 0) continue;
// 轨迹全部被删除直接关闭即可
// 轨迹全部被删除 关闭（预警生成任务，预警明细）
if (orderTrackStatusEarliestDict.Count == 0)
{
orderDetails.ForEach(x =>
orderWarningDetails.ForEach(x =>
{
x.Status = TransportWarningStatus.已关闭.GetHashCode();
x.CumulativeTimeSpan = x.CalcCumulativeTimeSpan(); //tips:完成态缓存累计运输时间
});
orderTasks.ForEach(x => x.IsClosed = true);
needUpdateDetails.AddRange(orderDetails);
needUpdateDetails.AddRange(orderWarningDetails);
needUpdateGenerateTask.AddRange(orderTasks);
continue;
}
@ -125,8 +120,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 尝试寻找该轨迹状态的最早轨迹
var earliestTrack = orderTrackStatusEarliestDict.GetValueOrDefault(group.Key);
// 不存在起始轨迹状态轨迹？
// 关闭预警生成任务
// 不存在起始轨迹状态轨迹？关闭预警生成任务
if (earliestTrack == null)
{
tasks.ForEach(x => x.IsClosed = true);
@ -136,18 +130,18 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 存在起始状态最早轨迹？
// 如果任务上的起始轨迹发生时间有变动则修改为最新的
// todo:需要补充转化未utc0时区
foreach (var task in tasks.Where(x => x.StartTrackTime != earliestTrack.TrackTime))
foreach (var task in tasks.Where(x => x.GetUtc0StartTrackTime() != earliestTrack.GetUtc0TrackTime()))
{
task.StartTrackTime = earliestTrack.TrackTime;
task.StartTrackTimeZone = earliestTrack.TrackTimeZone;
task.PlanExecutionTime = earliestTrack.TrackTime.AddHours(task.StartTrackWarningTimeSpan);
// tips:任务执行的时区为+8，所以需要将起始状态最早轨迹转化为+8时区来计算计划执行时间
task.PlanExecutionTime = earliestTrack.GetUtc0TrackTime().AddHours(task.StartTrackWarningTimeSpan + 8f);
needUpdateGenerateTask.Add(task);
}
}
// 以起始轨迹状态id分组处理预警明细
foreach (var group in orderDetails.GroupBy(x => x.StartTrackStatusId))
foreach (var group in orderWarningDetails.GroupBy(x => x.StartTrackStatusId))
{
var details = group.ToList();
@ -173,18 +167,17 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
}
// 存在起始状态最早轨迹？
// 最早轨迹如果早于预警明细中的起始轨迹时间，则修改预警明细
// todo:时区信息如果未提供则比较时间无意义!!,出现一个有时区一个无时区（后续可能强约束时区必填）？
// 最早轨迹如果有变化，则同步修改预警明细
foreach (var detail in details)
{
//保持预警明细中始终为起始状态的最早轨迹
if (detail.GetUtc0StartTrackTime() == earliestTrack.GetUtc0TrackTime()) continue;
detail.StartTrackContentId = earliestTrack.TrackContentId;
detail.StartTrackContent = earliestTrack.TrackContent;
detail.StartTrackTime = earliestTrack.TrackTime;
detail.StartTrackTimeZone = earliestTrack.TrackTimeZone;
needUpdateDetails.Add(detail);
}
needUpdateDetails.AddRange(details);
}
}
@ -203,81 +196,116 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
#endregion
#region todo-test: 预警信号批处理-新增修改轨迹
#region 预警信号批处理-新增修改轨迹
// tips:新增修改轨迹信号，在运输处理关注最新的轨迹
public async Task TrackAddedOrUpdatedHandleAsync(List<IWarningSignal> addOrUpdateSignals)
{
// 1.查询订单
var orderIdSet = addOrUpdateSignals.Select(x => x.OrderId).ToHashSet();
// 只处理未完结并且未删除的订单
// tips:只处理未完结并且未删除的订单
var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id) && x.IsCompleted == false);
if (orders.Count == 0) return; //无任何订单需要处理，直接提交信号丢弃
if (orders.Count == 0) return;
var orderIds = orders.Select(x => x.Id).ToHashSet();
// 1.1 构建订单和轨迹状态ids字典
var orderIdTrackStatusIdsDict = addOrUpdateSignals
.Where(x => orderIds.Contains(x.OrderId)) // tips:信号过滤
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y =>
y.Select(x => x.Type == WarningSignalType.轨迹新增
? ((TrackAddedSignal) x).TrackStatusId
: ((TrackUpdatedSignal) x).BeforeChangeTrack.TrackStatusId
).ToHashSet()
);
if (orderIdTrackStatusIdsDict.Count == 0) return;
// 构建订单和轨迹状态ids字典 tips:可以利用此优化查询轨迹
// var orderIdTrackStatusIdsDict = addOrUpdateSignals
// .Where(x => orderIds.Contains(x.OrderId)) // tips:信号过滤
// .GroupBy(x => x.OrderId)
// .ToDictionary(x => x.Key, y =>
// y.Select(x => x.Type == WarningSignalType.轨迹新增
// ? ((TrackAddedSignal) x).TrackStatusId
// : ((TrackUpdatedSignal) x).BeforeChangeTrack.TrackStatusId
// ).ToHashSet()
// );
// 2.查询订单所有轨迹(tips：查最新轨迹) k:订单id v:[k:轨迹状态id v:轨迹状态下最早轨迹]
var earliestTrackStatusTrackDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
// 3.以订单维度处理
// 过滤没有轨迹的订单
orders = orders.Where(x => earliestTrackStatusTrackDict.ContainsKey(x.Id)).ToList();
if (orders.Count == 0) return;
// 3.查轨迹主体
var conditions = orders.Select(x => (x.SubSystemId, x.CustomId)).ToHashSet();
var ruleDict = await _warningRuleRepository.GetWarningRuleForEnabledAsync(conditions, WarningRuleType.运输时效预警);
if (ruleDict.Count == 0) return;
// 4.查规则场景
var ruleIds = ruleDict.Keys.ToHashSet();
var scenes = await _warningRuleRepository.GetTransportSceneAsync(x => ruleIds.Contains(x.WarningId));
if (scenes.Count == 0) return;
// 5.查区域信息
var areaIds = scenes.Where(x => x.AreaId > 0).Select(x => x.AreaId!.Value).ToHashSet();
var ruleAreaInfosDict = await _kpiAreaRepository.GetAreaInfosDictByAreaId(areaIds);
// 选出订单命中的场景
var hitSceneIds = new HashSet<int>();
var orderIdHitSceneIdsDict = new Dictionary<int, HashSet<int>>();
var sceneBos = TransportWarningSceneBO.From(scenes);
foreach (var order in orders)
{
var sceneSet = new HashSet<int>();
foreach (var scene in sceneBos)
{
if (!scene.CanHitScene(order, ruleAreaInfosDict.GetValueOrDefault(scene.AreaId ?? 0))) continue;
sceneSet.Add(scene.Id);
hitSceneIds.Add(scene.Id);
}
if (sceneSet.Count > 0)
orderIdHitSceneIdsDict.TryAdd(order.Id, sceneSet);
}
if (orderIdHitSceneIdsDict.Count == 0) return;
// 6.查预警规则项目，所有被命中的场景下
var allRuleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => hitSceneIds.Contains(x.TransportWarningSceneId));
if (allRuleItems.Count == 0) return;
var allRuleItemBos = TransportWarningSceneRuleItemBO.From(allRuleItems);
var ruleItemBoDict = allRuleItemBos.ToDictionary(x => x.Id);
// 建立订单和预警规则ids的关联字典
var orderIdRuleItemIdsDict = new Dictionary<int, HashSet<int>>();
foreach (var order in orders)
{
var sceneIds = orderIdHitSceneIdsDict.GetValueOrDefault(order.Id);
if (sceneIds == null) continue;
var ruleItems = allRuleItemBos.Where(x => sceneIds.Contains(x.SceneId)).ToList();
if (ruleItems.Count == 0) continue;
orderIdRuleItemIdsDict.TryAdd(order.Id, ruleItems.Select(x => x.Id).ToHashSet());
}
// 7.查预警明细 订单所有跟规则相关的明细，要求（即将超时/已超时）
var allWarningDetails = await _warningDetailsEsRepository.GetTransportWarningDetailsAsync(new GetTransportWarningDetailsVO
{
OrderIdRuleItemIdsDict = orderIdRuleItemIdsDict,
});
var orderIdWarningDetailDict = allWarningDetails
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.ToList());
// 8.查预警生成任务
Expression<Func<TransportWarningDetailsGenerateTaskEntity, bool>> baseExp = x => x.IsClosed == false;
var exp = Expressionable.Create<TransportWarningDetailsGenerateTaskEntity>();
foreach (var (orderId, ruleItemIds) in orderIdRuleItemIdsDict)
exp.Or(x => x.OrderId == orderId && ruleItemIds.Contains(x.WarningSceneRuleItemId));
var allWarningTasks = await _warningTaskRepository.GetTransportWarningDetailsGenerateTaskAsync(baseExp.And(exp.ToExpression()));
var orderIdWarningTaskDict = allWarningTasks
.GroupBy(x => x.OrderId)
.ToDictionary(x => x.Key, y => y.ToList());
// 9.以订单维度处理
var updateDetails = new List<TransportWarningDetailsIndex>();
var addDetailGenerateTasks = new List<TransportWarningDetailsGenerateTaskEntity>();
var updateDetailGenerateTasks = new List<TransportWarningDetailsGenerateTaskEntity>();
var closeNotifyDetailsIds = new HashSet<string>();
// todo:优化流程
foreach (var order in orders)
{
var trackStatusTrackDict = earliestTrackStatusTrackDict.GetValueOrDefault(order.Id);
if (trackStatusTrackDict == null) continue;
// 匹配规则主体，要求已经启用
var rule = await _warningRuleRepository.GetWarningRuleForEnabledAsync(order.SubSystemId, order.CustomId, WarningRuleType.运输时效预警);
if (rule == null) continue;
// 查找主体下所有场景
var scenes = await _warningRuleRepository.GetTransportSceneAsync(rule.Id);
if (scenes.Count == 0) continue;
// 查找场景关联的区域信息
var areaIds = scenes.Select(x => x.AreaId ?? 0).ToHashSet();
var ruleAreaInfosDict = await _kpiAreaRepository.GetAreaInfosDictByAreaId(areaIds);
// 选出订单命中的场景ids
var sceneBos = TransportWarningSceneBO.From(scenes);
var hitSceneIds = sceneBos
.Where(x => x.CanHitScene(order, ruleAreaInfosDict.GetValueOrDefault(x.AreaId ?? 0)))
.Select(x => x.Id)
.ToHashSet();
// 查询命中场景下的所有规则项目
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => hitSceneIds.Contains(x.TransportWarningSceneId), y => y);
if (ruleItems.Count == 0) continue;
var ruleItemIds = ruleItems.Select(x => x.Id).ToHashSet();
var ruleItemBoDict = TransportWarningSceneRuleItemBO
.From(ruleItems)
.ToDictionary(x => x.Id);
// 查订单所有跟规则相关的明细，要求（即将超时/已超时）
var warningDetails = await _warningDetailsEsRepository.GetTransportWarningDetailsAsync(new GetTransportWarningDetailsVO
{
RuleItemIds = ruleItemIds,
OrderIds = new HashSet<int>(1)
{
order.Id
},
});
var trackStatusTrackDict = earliestTrackStatusTrackDict[order.Id]; // 一定存在，已经存在前置过滤
var warningDetails = orderIdWarningDetailDict.GetValueOrDefault(order.Id, Empty<TransportWarningDetailsIndex>.List);
// 处理【通知阶段】的预警明细
if (warningDetails.Count > 0)
@ -287,12 +315,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
closeNotifyDetailsIds.UnionWith(needCloseNotifyDetailsIds);
}
// 查订单所有跟规则相关的明细生成任务(未来可能会执行)，要求（未关闭）
var orderTasks = await _warningTaskRepository.GetTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& x.OrderId == order.Id
&& ruleItemIds.Contains(x.WarningSceneRuleItemId)
);
var orderTasks = orderIdWarningTaskDict.GetValueOrDefault(order.Id, Empty<TransportWarningDetailsGenerateTaskEntity>.List);
// 处理【预警明细生成就绪阶段】预警明细生成任务
var (updateTasks, addTasks) = HandleDetailGenerateTask(order, orderTasks, warningDetails, ruleItemBoDict, trackStatusTrackDict);
@ -300,7 +323,6 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
updateDetailGenerateTasks.AddRange(updateTasks);
}
// 4.批量操作
if (closeNotifyDetailsIds.Count > 0)
{
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
@ -321,7 +343,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 处理【通知阶段】的预警明细
// 起始轨迹状态轨迹变更（存在起始，不存在截止）
// 收到了截止轨迹（存在起始，存在截止）
private (List<TransportWarningDetailsIndex> UpdateDeatils, HashSet<string> CloseNotifyDetailIds) HandleExistsDetails(
private static (List<TransportWarningDetailsIndex> UpdateDeatils, HashSet<string> CloseNotifyDetailIds) HandleExistsDetails(
List<TransportWarningDetailsIndex> warningDetails,
Dictionary<int, TransportWarningSceneRuleItemBO> ruleItemBoDict,
Dictionary<int, TrackDetailsIndex> trackStatusIdEarliestTrackDict
@ -400,7 +422,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 命中：存在任务，起始轨迹状态轨迹变更则更新任务信息，
// 命中：不存在任务，添加生成任务
// 未命中，已有的任务关闭，没有的忽略生成
private (List<TransportWarningDetailsGenerateTaskEntity> UpdateTasks, List<TransportWarningDetailsGenerateTaskEntity> AddTasks) HandleDetailGenerateTask(OrderEntity order,
private static (List<TransportWarningDetailsGenerateTaskEntity> UpdateTasks, List<TransportWarningDetailsGenerateTaskEntity> AddTasks) HandleDetailGenerateTask(OrderEntity order,
List<TransportWarningDetailsGenerateTaskEntity> tasks,
List<TransportWarningDetailsIndex> warningDetails,
Dictionary<int, TransportWarningSceneRuleItemBO> ruleItemBoDict,
@ -460,7 +482,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
return (updateTasks, insertTasks);
}
private TransportWarningDetailsGenerateTaskEntity BuildDetailsGenerateTask(
private static TransportWarningDetailsGenerateTaskEntity BuildDetailsGenerateTask(
OrderEntity order,
TransportWarningSceneRuleItemBO ruleItem,
TrackDetailsIndex earliestTrack
@ -485,7 +507,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
#endregion 预警信号处理
#region todo-test: 运输预警主体关闭
#region 运输预警主体关闭
public async Task WarningRuleToDisabledHandleAsync(WarningRuleEntity rule)
{
@ -517,7 +539,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
#endregion
#region todo-test: 规则场景变更
#region 规则场景变更
// tips:由于只是改动了场景，只要订单仍然命中场景，那么订单相关的预警生成任务/预警通知任务无需处理，预警明细更新部分场景信息。未命中则相关联条目全部删除
// 更新:删除不符合【场景】条件的【预警生成任务】【预警明细】【预警通知任务】，符合场景则修改预警明细的【预警规则描述】
@ -526,17 +548,17 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
if (!CanSceneChange(oldScene, postScene))
return;
// 1.1 找到该场景下所有的预警生成任务(未关闭的)
// 1.找到该场景下所有的预警生成任务(未关闭的)
var generateTasks = await _warningTaskRepository.GetTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& x.WarningId == postScene.Id
&& x.WarningId == postScene.WarningId
&& x.WarningSceneId == postScene.Id
);
// 1.1 预警生成任务反找订单id
// 预警生成任务反找订单id
var orderIds = generateTasks.Select(x => x.OrderId).ToHashSet();
// 2.1 找到该场景下所有的预警明细（即将超时/已超时）
// 2.找到该场景下所有的预警明细（即将超时/已超时）
var warningDetails = await _warningDetailsEsRepository.GetTransportWarningDetailsAsync(new GetTransportWarningDetailsVO()
{
SceneIds = new HashSet<int>(1)
@ -545,7 +567,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
},
});
// 2.2 预警明细反找订单id并合并
// 预警明细反找订单id并合并
orderIds.UnionWith(warningDetails.Select(x => x.OrderId));
// tips:无需在通过预警通知任务反找订单，因为产生预警通知任务时预警明细已经存在！
@ -574,9 +596,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 明细->规则
var updateDetails = warningDetails.Where(x => reHitSceneOrderIds.Contains(x.OrderId)).ToList();
var updateRuleItemIds = updateDetails.Select(x => x.WarningSceneRuleItemId).ToHashSet();
var updateRuleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(
x => updateRuleItemIds.Contains(x.Id),
y => y);
var updateRuleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => updateRuleItemIds.Contains(x.Id));
var ruleItemDict = updateRuleItems.ToDictionary(x => x.Id);
// 场景->区域
@ -596,8 +616,8 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
if (needUpdateDetails.Count > 0)
await _warningDetailsEsRepository.BulkUpdateTransportWarningDetailsAsync(needUpdateDetails);
if (noHitSceneOrderIds.Count == 0)
return;
// 没有任何未命中场景的订单
if (noHitSceneOrderIds.Count == 0) return;
// 5.关闭不符合场景的【预警明细】【预警生成任务】【预警通知任务】
// 预警明细状态修改为已关闭（即将超时/已超时）
@ -654,7 +674,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
);
}
private bool CanSceneChange(TransportWarningSceneBO oldScene, TransportWarningSceneBO postScene)
private static bool CanSceneChange(TransportWarningSceneBO oldScene, TransportWarningSceneBO postScene)
{
return oldScene.CargoType != postScene.CargoType || //多个 普货,危险品
oldScene.Months != postScene.Months || //多个 约束提交的月份组从小到大 1/3/4，不能出现 4/1/3这种 (如果出现则SequenceEqual)
@ -666,7 +686,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
#endregion todo: 规则场景变更
#region todo-test: 规则项目变更
#region 规则项目变更
public async Task HandleEditAsync(WarningRuleEntity rule, TransportWarningSceneBO scene, List<TransportWarningSceneRuleItemBO> postRuleItems)
{
@ -746,7 +766,6 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
await HandleDeleteAsync(ruleItemIds);
}
// todo:需要优化
private async Task HandleRuleItemsChangeAsync(
WarningRuleEntity rule,
TransportWarningSceneBO scene,
@ -771,31 +790,15 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
var deleteRuleItemIds = deleteRuleItems.Select(x => x.Id).ToHashSet();
if (deleteRuleItemIds.Count > 0)
await _warningRuleRepository.DeleteTransportSceneRuleItemsAsync(deleteRuleItemIds);
// 不存在更新需要关注的，也不存在删除的
if (updateFollowRuleItems.Count == 0 && deleteRuleItemIds.Count == 0) return;
// 只存在删除的规则项目,直接关闭处理返回即可
if (updateFollowRuleItems.Count == 0)
{
// 关闭相关的预警生成任务（未关闭的）
await _warningTaskRepository.CloseTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& x.WarningId == rule.Id
&& deleteRuleItemIds.Contains(x.WarningSceneRuleItemId)
);
// 关闭关联的预警通知任务（未关闭的）
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
!x.IsClosed
&& x.WarningId == rule.Id
&& x.WarningType == WarningRuleType.运输时效预警.GetHashCode()
&& deleteRuleItemIds.Contains(x.WarningRuleItemId)
);
return;
await _warningRuleRepository.DeleteTransportSceneRuleItemsAsync(deleteRuleItemIds);
// 处理删除规则后的【预警明细】和【预警生成任务】和【预警通知任务】
await HandleDeleteAsync(deleteRuleItemIds);
}
// 不存在更新需要关注的
if (updateFollowRuleItems.Count == 0) return;
// 1.查询变更配置相关的预警生成任务(未关闭的)
var updateFollowRuleItemIds = updateFollowRuleItems.Select(x => x.Id).ToHashSet();
var generateTasks = await _warningTaskRepository.GetTransportWarningDetailsGenerateTaskAsync(x =>
@ -915,25 +918,17 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
await _warningTaskRepository.BulkUpdateTransportWarningDetailsGenerateTaskAsync(updateGenerateTasks);
// 3.3 处理需要关闭的【预警明细】【预警生成任务】【预警通知任务】(未命中规则)
var typeIdsDict = handleBlocks
.Except(reHitRuleBlocks)
var notHitHandleBlocksDict = handleBlocks
.Except(reHitRuleBlocks) // 排除掉已命中
.GroupBy(x => x.Type)
.ToDictionary(x => x.Key, y =>
{
return y.Key switch
{
HandleType.Detail => y.Select(x => x.Detail.Id).ToHashSet(),
HandleType.GenerateTask => y.Select(x => x.GenerateTask.Id.ToString()).ToHashSet(),
HandleType.NotifyTask => y.Select(x => x.NotifyTask.Id.ToString()).ToHashSet(),
_ => throw new ArgumentOutOfRangeException()
};
});
.ToDictionary(x => x.Key, y => y.ToList());
// 4.处理删除规则的相关联的【预警明细】【预警生成任务】【预警通知任务】
// 预警信息状态修改为已关闭（即将超时/已超时）
var closeDetailIds = typeIdsDict.GetValueOrDefault(HandleType.Detail, new HashSet<string>());
if (closeDetailIds.Count > 0)
var closeDetailHandleBlocks = notHitHandleBlocksDict.GetValueOrDefault(HandleType.Detail, Empty<HandleBlock>.List);
if (closeDetailHandleBlocks.Count > 0)
{
var closeDetailIds = closeDetailHandleBlocks.Select(x => x.Detail.Id).ToHashSet();
await _warningDetailsEsRepository.BulkCloseTransportWarningDetailsAsync(new BulkCloseTransportWarningDetailsVO
{
DetailsIds = closeDetailIds,
@ -941,25 +936,26 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
}
// 关闭相关的预警生成任务（未关闭的）
var closeGenerateTaskIds = typeIdsDict.GetValueOrDefault(HandleType.GenerateTask, new HashSet<string>());
await _warningTaskRepository.CloseTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& (
closeGenerateTaskIds.Contains(x.Id.ToString())
|| x.WarningId == rule.Id && deleteRuleItemIds.Contains(x.WarningSceneRuleItemId)
)
);
var closeGenerateTaskHandleBlocks = notHitHandleBlocksDict.GetValueOrDefault(HandleType.GenerateTask, Empty<HandleBlock>.List);
if (closeGenerateTaskHandleBlocks.Count > 0)
{
var closeGenerateTaskIds = closeGenerateTaskHandleBlocks.Select(x => x.GenerateTask.Id).ToHashSet();
await _warningTaskRepository.CloseTransportWarningDetailsGenerateTaskAsync(x =>
!x.IsClosed
&& closeGenerateTaskIds.Contains(x.Id)
);
}
// 关闭关联的预警通知任务（未关闭的）
var closeNotifyTaskIds = typeIdsDict.GetValueOrDefault(HandleType.NotifyTask, new HashSet<string>());
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
!x.IsClosed
&& (
closeNotifyTaskIds.Contains(x.Id.ToString())
|| x.WarningId == rule.Id
&& x.WarningType == WarningRuleType.运输时效预警.GetHashCode()
&& deleteRuleItemIds.Contains(x.WarningRuleItemId)
));
var closeNotifyTaskHandleBlocks = notHitHandleBlocksDict.GetValueOrDefault(HandleType.NotifyTask, Empty<HandleBlock>.List);
if (closeNotifyTaskHandleBlocks.Count > 0)
{
var closeNotifyTaskIds = closeNotifyTaskHandleBlocks.Select(x => x.NotifyTask.Id).ToHashSet();
await _warningTaskRepository.CloseWarningNotifyTaskAsync(x =>
!x.IsClosed
&& closeNotifyTaskIds.Contains(x.Id)
);
}
}
private async Task HandleDeleteAsync(HashSet<int> deleteRuleItemIds)
@ -1060,17 +1056,14 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
#endregion todo: 规则项目变更
#region todo-test: 批量处理预警明细生成任务
#region 批量处理预警明细生成任务
public async Task HandleWarningDetailsGenerateTask(List<TransportWarningDetailsGenerateTaskEntity> tasks)
{
// 1.查询订单,tips:只处理未完结并且未删除的订单
var orderIdSet = tasks.Select(x => x.OrderId).ToHashSet();
var orders = await _orderRepository.GetListAsync(x =>
orderIdSet.Contains(x.Id)
&& x.IsCompleted == false
);
if (orders.Count == 0) return; //不存在符合条件的订单，直接全部关闭
var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id) && x.IsCompleted == false);
if (orders.Count == 0) return;
// 构建订单id和订单字典
var orderDict = orders.ToDictionary(x => x.Id);
@ -1095,13 +1088,13 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
{
OrderIdTracksStatusIdsDict = orderIdTrackStatusIdsDict
});
if (orderTracksDict.Count == 0) return; //不存在轨迹，直接全部关闭
if (orderTracksDict.Count == 0) return;
// 3.查出任务相关，运输时效规则配置
// 查出启用的规则主体
var warningIds = tasks.Select(x => x.WarningId).ToHashSet();
var waningRules = await _warningRuleRepository.GetListAsync(x => x.IsEnabled && warningIds.Contains(x.Id));
if (waningRules.Count == 0) return; //不存在启用的规则主体，直接全部关闭
if (waningRules.Count == 0) return;
// 构建规则主体id和版本号字典,稍后用于填充明细的版本号
var ruleDict = waningRules.ToDictionary(x => x.Id);
@ -1114,8 +1107,8 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
if (ruleItemIds.Count == 0) return;
// 查出规则配置
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id), y => y);
if (ruleItems.Count == 0) return; //不存在规则配置，直接全部关闭
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id));
if (ruleItems.Count == 0) return;
// 构建规则id和规则条目字典
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
@ -1123,7 +1116,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
// 4.选出可以执行的任务和需要关闭的任务
var needCloseTaskIds = new HashSet<int>();
var addDetails = new List<TransportWarningDetailsIndex>();
foreach ((int orderId, List<TransportWarningDetailsGenerateTaskEntity> orderTasks) in orderIdTasksDict)
foreach (var (orderId, orderTasks) in orderIdTasksDict)
{
var order = orderDict.GetValueOrDefault(orderId);
// 订单不存在/已完结 直接关闭任务即可
@ -1216,7 +1209,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
}
private TransportWarningDetailsIndex BuildTransportWarningDetail(
private static TransportWarningDetailsIndex BuildTransportWarningDetail(
OrderEntity order,
TransportWarningSceneRuleItemEntity ruleItem,
List<TrackDetailsIndex> tracks
@ -1269,7 +1262,7 @@ public class TransportWarningHandler : ITransportWarningHandler, ITransientDepen
return detail;
}
private List<WarningNotifyTaskEntity> BuildWarningNotifyTask(List<TransportWarningDetailsIndex> details)
private static List<WarningNotifyTaskEntity> BuildWarningNotifyTask(List<TransportWarningDetailsIndex> details)
{
return details.Select(x => new WarningNotifyTaskEntity
{

8
src/SJZY.Track.Domain/Aggregates/WarningAggregate/IWarningRuleRepository.cs
View File

@ -30,6 +30,9 @@ public interface IWarningRuleRepository : IBasicRepository<WarningRuleEntity>
// 获取启用的预警规则主体
Task<WarningRuleEntity?> GetWarningRuleForEnabledAsync(int subSystemId, int customId, WarningRuleType warningType);
// 获取启用的预警规则主体
Task<IReadOnlyDictionary<int, WarningRuleEntity>> GetWarningRuleForEnabledAsync(HashSet<(int SubSystemId, int CustomId)> conditions, WarningRuleType warningType);
// 获取预警规则主体根据id
Task<WarningRuleEntity?> GetWarningRuleByIdAsync(int id, WarningRuleType? warningType = null);
@ -50,9 +53,8 @@ public interface IWarningRuleRepository : IBasicRepository<WarningRuleEntity>
Task<List<TransportWarningSceneRuleItemEntity>> GetTransportSceneRuleItemsAsync(int warningId, int sceneId);
// 获取运输预警场景规则项目
Task<List<TResult>> GetTransportSceneRuleItemsAsync<TResult>(
Expression<Func<TransportWarningSceneRuleItemEntity, bool>> whereExpression,
Expression<Func<TransportWarningSceneRuleItemEntity, TResult>> resultExpression);
Task<List<TransportWarningSceneRuleItemEntity>> GetTransportSceneRuleItemsAsync(
Expression<Func<TransportWarningSceneRuleItemEntity, bool>> whereExpression);
// 新增/更新 运输预警场景
Task<int> UpsertTransportSceneReturnIdAsync(TransportWarningSceneEntity entity);

10
src/SJZY.Track.Domain/Aggregates/WarningAggregate/Transport/TransportWarningDetailsGenerateTaskEntity.cs
View File

@ -1,3 +1,5 @@
using SJZY.Track.Domain.Consts;
namespace SJZY.Track.Domain.Aggregates.WarningAggregate.Transport;
[SugarTable("t_warning_transport_generate_task", TableDescription = "运输时效预警明细生成任务")]
@ -39,4 +41,12 @@ public sealed class TransportWarningDetailsGenerateTaskEntity : IFullAuditedBasi
[SugarColumn(ColumnName = "IsClosed", ColumnDescription = "是否关闭")]
public bool IsClosed { get; set; } //是否关闭，执行完成被关闭/手动关闭
// 如果时区不存在，默认认为为+8时区
public DateTime GetUtc0StartTrackTime()
{
return Math.Abs(StartTrackTimeZone - GlobalConst.NoneTrackTimeZone) < 0.01f
? StartTrackTime.AddHours(-8.0f)
: StartTrackTime.AddHours(-StartTrackTimeZone);
}
}

4
src/SJZY.Track.Domain/Dtos/Warning/GetTransportWarningDetailsVO.cs
View File

@ -21,10 +21,14 @@ public class GetTransportWarningDetailsVO
// 订单id 与 起始轨迹状态ids 同时满足
public Dictionary<int, HashSet<int>>? OrderIdStartTracksStatusIdsDict { get; set; }
// 订单id 与 预警规则项目ids 同时满足
public Dictionary<int, HashSet<int>>? OrderIdRuleItemIdsDict { get; set; }
public bool HasDetailsIds => DetailsIds?.Count > 0;
public bool HasRuleItemIds => RuleItemIds?.Count > 0;
public bool HasSceneIds => SceneIds?.Count > 0;
public bool HasWarningIds => WarningIds?.Count > 0;
public bool HasOrderIds => OrderIds?.Count > 0;
public bool HasOrderIdStartTracksStatusIdDict => OrderIdStartTracksStatusIdsDict?.Count > 0;
public bool HasOrderIdRuleItemIdsDict => OrderIdRuleItemIdsDict?.Count > 0;
}

13
src/SJZY.Track.Repository/Repositories/WarningDetailsEsRepository.cs
View File

@ -80,6 +80,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.WarningSceneRuleItemId).Terms(ruleItemIds))
)));
}
isShouldQuery = true;
}
else
{
musts.AddIf(x => x.HasRuleItemIds, c => c.Terms(t => t.Field(f => f.WarningSceneRuleItemId).Terms(input.RuleItemIds)));

21
src/SJZY.Track.Repository/Repositories/WarningRuleRepository.cs
View File

@ -100,6 +100,23 @@ public class WarningRuleRepository : BasicRepository<WarningRuleEntity>, IWarnin
return warningRules.FirstOrDefault();
}
public async Task<IReadOnlyDictionary<int, WarningRuleEntity>> GetWarningRuleForEnabledAsync(HashSet<(int SubSystemId, int CustomId)> conditions, WarningRuleType warningType)
{
Expression<Func<WarningRuleEntity, bool>> baseExp = x => x.WarningType == (int) warningType && x.IsEnabled;
var exp = Expressionable.Create<WarningRuleEntity>();
foreach (var condition in conditions)
exp.Or(x => x.CustomId == condition.CustomId && x.SubSystemId == condition.SubSystemId);
var list = await Context.Queryable<WarningRuleEntity>()
.Where(baseExp.And(exp.ToExpression()))
.ToListAsync();
return list.Count == 0
? Empty<int, WarningRuleEntity>.Dict
: list.ToDictionary(x => x.Id);
}
public async Task<WarningRuleEntity?> GetWarningRuleByIdAsync(int id, WarningRuleType? warningType = null)
{
var result = await Context.Queryable<WarningRuleEntity>()
@ -142,12 +159,10 @@ public class WarningRuleRepository : BasicRepository<WarningRuleEntity>, IWarnin
.ToListAsync();
}
public Task<List<TResult>> GetTransportSceneRuleItemsAsync<TResult>(Expression<Func<TransportWarningSceneRuleItemEntity, bool>> whereExpression,
Expression<Func<TransportWarningSceneRuleItemEntity, TResult>> resultExpression)
public Task<List<TransportWarningSceneRuleItemEntity>> GetTransportSceneRuleItemsAsync(Expression<Func<TransportWarningSceneRuleItemEntity, bool>> whereExpression)
{
return Context.Queryable<TransportWarningSceneRuleItemEntity>()
.Where(whereExpression)
.Select(resultExpression)
.ToListAsync();
}