xiaolipro
f519405c63
Added a new JSON converter to handle flexible string conversions. Updated NotifyUserDto to change UserId from int to string and removed the [IdRequired] attribute. This aligns with the new converter's functionality to handle different ID formats. I need to condense this. The core
342 lines
16 KiB
C#
342 lines
16 KiB
C#
using Microsoft.Extensions.Options;
|
||
using TrackCenter.Application.Dtos.Warning;
|
||
using TrackCenter.Application.Options;
|
||
using TrackCenter.Application.Services.Warning.Model;
|
||
using TrackCenter.Domain.Aggregates.KPIAreaAggregate;
|
||
using TrackCenter.Domain.Aggregates.OrderAggregate;
|
||
using TrackCenter.Domain.Aggregates.TrackDetailsAggregate;
|
||
using TrackCenter.Domain.Aggregates.WarningAggregate;
|
||
using TrackCenter.Domain.Dtos.Forewarning;
|
||
using TrackCenter.Domain.Dtos.Warning;
|
||
|
||
namespace TrackCenter.Application.Hangfire;
|
||
|
||
/// <summary>
|
||
/// 预警通知聚合器
|
||
/// </summary>
|
||
public sealed class WarningNotifyAggregator : ITransientDependency
|
||
{
|
||
private readonly IOrderRepository _orderRepository;
|
||
private readonly IKpiAreaRepository _kpiAreaRepository;
|
||
private readonly ITrackDetailsEsRepository _trackDetailsEsRepository;
|
||
private readonly IWarningTaskRepository _warningTaskRepository;
|
||
private readonly IWarningRuleRepository _warningRuleRepository;
|
||
private readonly IWarningDetailsEsRepository _warningDetailsEsRepository;
|
||
private readonly IOptionsMonitor<TrackWarningTaskOptions> _optionsMonitor;
|
||
|
||
public WarningNotifyAggregator(IWarningTaskRepository warningTaskRepository, IWarningRuleRepository warningRuleRepository, IOrderRepository orderRepository, ITrackDetailsEsRepository trackDetailsEsRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IKpiAreaRepository kpiAreaRepository, IOptionsMonitor<TrackWarningTaskOptions> optionsMonitor)
|
||
{
|
||
_warningTaskRepository = warningTaskRepository;
|
||
_warningRuleRepository = warningRuleRepository;
|
||
_orderRepository = orderRepository;
|
||
_trackDetailsEsRepository = trackDetailsEsRepository;
|
||
_warningDetailsEsRepository = warningDetailsEsRepository;
|
||
_kpiAreaRepository = kpiAreaRepository;
|
||
_optionsMonitor = optionsMonitor;
|
||
}
|
||
|
||
// 1.可行性过滤（不符合的统一关闭）
|
||
// 2.以人的维度聚合通知信息，并生成配置参数NotifyConfig
|
||
// 3.构建并调度一个调用通知接口的立即执行任务
|
||
// 4.绑定这个任务id(可能是多个)和jobId用户跟踪任务状态
|
||
public async Task<(List<WarningNotifyParam> NotifyConfigs, HashSet<int> taskIds)> ScanNotifyAndAggregateAsync()
|
||
{
|
||
// 构建查询游标 todo:全量查询需要优化，直接存缓存？
|
||
// var options = _optionsMonitor.CurrentValue;
|
||
// var cursor = DateTime.Now.Subtract(TimeSpan.FromHours(options.ScanWarningNotifyTaskHourInterval));
|
||
|
||
// 1.扫描出可能进行的任务
|
||
var tasks = await _warningTaskRepository.ScanNotifyTaskAsync();
|
||
|
||
// 2.可行性过滤
|
||
var filtered = await FilterAsync(tasks);
|
||
|
||
// 3.以人的维度聚合通知构建NotifyConfig
|
||
return (BuildNotifyConfigs(filtered), tasks.Select(x => x.Id).ToHashSet());
|
||
}
|
||
|
||
|
||
// 要求订单存在，预警主体存在过滤
|
||
private async Task<List<WarningNotifyTaskEntity>> FilterAsync(List<WarningNotifyTaskEntity> notifyTasks)
|
||
{
|
||
// 1.只通知存在的订单
|
||
var orderIdSet = notifyTasks.Select(x => x.OrderId).ToHashSet();
|
||
var orders = await _orderRepository.GetListAsync(x => orderIdSet.Contains(x.Id));
|
||
if (orders.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
var orderDict = orders.ToDictionary(x => x.Id);
|
||
// 2.过滤掉不存在的订单
|
||
var filtered = notifyTasks.Where(x => orderDict.ContainsKey(x.OrderId)).ToList();
|
||
|
||
// 3.查询规则主体（存在&启用）
|
||
var warningSet = filtered.Select(x => x.WarningId).ToHashSet();
|
||
var warningIds = (await _warningRuleRepository.GetListAsync(
|
||
x => warningSet.Contains(x.Id) && x.IsEnabled == true,
|
||
y => y.Id //不回表
|
||
)).ToHashSet();
|
||
if (warningIds.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
// 4.过滤掉规则主体禁用
|
||
filtered = filtered.Where(x => warningIds.Contains(x.WarningId)).ToList();
|
||
|
||
// 5.查询订单每个轨迹状态的最早轨迹，后续过滤规则使用
|
||
var orderIds = filtered.Select(x => x.OrderId).ToHashSet();
|
||
var orderTrackStatusTrackDict = await _trackDetailsEsRepository.GetTrackStatusEarliestDictAsync(orderIds);
|
||
|
||
var result = new List<WarningNotifyTaskEntity>(filtered.Count);
|
||
// 6.按预警类型分组，重新验证规则
|
||
foreach (var group in filtered.GroupBy(x => x.WarningType))
|
||
{
|
||
switch (group.Key)
|
||
{
|
||
case (int) WarningRuleType.运输时效预警:
|
||
{
|
||
var filterTasks = await TransportFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict);
|
||
if (filterTasks.Count > 0)
|
||
result.AddRange(filterTasks);
|
||
}
|
||
break;
|
||
case (int) WarningRuleType.回传时效预警:
|
||
{
|
||
var filterTasks = await ReturnFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict);
|
||
if (filterTasks.Count > 0)
|
||
result.AddRange(filterTasks);
|
||
break;
|
||
}
|
||
case (int) WarningRuleType.完整预警:
|
||
{
|
||
var filterTasks = await CompleteFilterAsync(group.ToList(), orderDict, orderTrackStatusTrackDict);
|
||
if (filterTasks.Count > 0)
|
||
result.AddRange(filterTasks);
|
||
break;
|
||
}
|
||
}
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
private async Task<List<WarningNotifyTaskEntity>> TransportFilterAsync(
|
||
List<WarningNotifyTaskEntity> tasks, Dictionary<int, OrderEntity> orderDict, Dictionary<int, Dictionary<int, TrackDetailsIndex>> orderTrackStatusTrackDict)
|
||
{
|
||
// 1.规则场景存在验证
|
||
// 场景id附在任务的辅助id上,只有运输时效预警通知才有这个值！！
|
||
var sceneIds = tasks
|
||
.Select(x => (Success: int.TryParse(x.AuxiliaryId, out var id), Id: id))
|
||
.Where(x => x.Success)
|
||
.Select(x => x.Id)
|
||
.ToHashSet();
|
||
var scenes = await _warningRuleRepository.GetTransportSceneAsync(x => sceneIds.Contains(x.Id));
|
||
if (scenes.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
// 1.1过滤掉不存在场景的通知任务
|
||
var validSceneIds = scenes.Select(x => x.Id).ToHashSet();
|
||
var filtered = tasks
|
||
.Where(x => x.AuxiliaryId != null
|
||
&& int.TryParse(x.AuxiliaryId, out var id)
|
||
&& validSceneIds.Contains(id))
|
||
.ToList();
|
||
|
||
// 2.场景验证
|
||
var sceneDict = scenes.ToDictionary(x => x.Id.ToString(), TransportWarningSceneBO.From);
|
||
|
||
// 2.1查找场景关联的区域信息
|
||
var areaIds = scenes.Select(x => x.AreaId ?? 0).Where(id => id != 0).ToHashSet();
|
||
var ruleAreaInfosDict = areaIds.Count > 0
|
||
? await _kpiAreaRepository.GetAreaInfosDictByAreaId(areaIds)
|
||
: new Dictionary<int, List<KpiAreaInfo>>(0);
|
||
|
||
// 2.2场景验证
|
||
var validTasks = new List<WarningNotifyTaskEntity>(filtered.Count);
|
||
foreach (var task in filtered)
|
||
{
|
||
if (sceneDict.TryGetValue(task.AuxiliaryId!, out var scene)
|
||
&& orderDict.TryGetValue(task.OrderId, out var order)
|
||
&& scene.CanHitScene(order, ruleAreaInfosDict.GetValueOrDefault(scene.AreaId ?? 0)))
|
||
{
|
||
validTasks.Add(task);
|
||
}
|
||
}
|
||
|
||
if (validTasks.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
// 3.场景规则存在验证
|
||
var ruleItemIds = validTasks.Select(x => x.WarningRuleItemId).ToHashSet();
|
||
var ruleItems = await _warningRuleRepository.GetTransportSceneRuleItemsAsync(x => ruleItemIds.Contains(x.Id));
|
||
if (ruleItems.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
// 过滤掉不存在的场景规则
|
||
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
|
||
// 使用 Join 替代 Where + Contains 提升性能
|
||
filtered = validTasks
|
||
.Join(ruleItemDict.Keys,
|
||
task => task.WarningRuleItemId,
|
||
id => id,
|
||
(task, _) => task)
|
||
.ToList();
|
||
|
||
// 4.场景规则验证
|
||
return filtered
|
||
.GroupBy(x => x.OrderId)
|
||
.SelectMany(group =>
|
||
{
|
||
var orderId = group.Key;
|
||
if (!orderTrackStatusTrackDict.TryGetValue(orderId, out var statusTrack))
|
||
return Enumerable.Empty<WarningNotifyTaskEntity>();
|
||
|
||
var statusIds = statusTrack.Keys.ToHashSet();
|
||
return group.Where(task =>
|
||
ruleItemDict.TryGetValue(task.WarningRuleItemId, out var ruleItem)
|
||
&& ruleItem.CanReHit(statusIds));
|
||
})
|
||
.ToList();
|
||
}
|
||
|
||
private async Task<List<WarningNotifyTaskEntity>> ReturnFilterAsync(
|
||
List<WarningNotifyTaskEntity> tasks, Dictionary<int, OrderEntity> orderDict, Dictionary<int, Dictionary<int, TrackDetailsIndex>> orderTrackStatusTrackDict)
|
||
{
|
||
// 1.规则条目存在
|
||
var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet();
|
||
var ruleItems = await _warningRuleRepository.GetReturnRuleItemsAsync(ruleItemIds);
|
||
if (ruleItems.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
|
||
// 过滤掉不存在规则的通知
|
||
var filtered = tasks.Where(x => ruleItemDict.ContainsKey(x.WarningRuleItemId)).ToList();
|
||
|
||
// 2.检查明细中为人工介入的状态
|
||
var detailsIdSet = filtered.Select(x => x.WarningDetailsId).ToHashSet();
|
||
var warningDetails = await _warningDetailsEsRepository.GetReturnWarningDetailsAsync(new GetReturnWarningDetailsVO
|
||
{
|
||
DetailsIds = detailsIdSet,
|
||
Status = new HashSet<int>(1)
|
||
{
|
||
ReturnWarningStatus.人工介入.GetHashCode()
|
||
},
|
||
});
|
||
if (warningDetails.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
// 过滤掉不为人工介入的预警通知
|
||
var detailsIds = warningDetails.Select(x => x.Id).ToHashSet();
|
||
return filtered.Where(x => detailsIds.Contains(x.WarningDetailsId)).ToList();
|
||
}
|
||
|
||
private async Task<List<WarningNotifyTaskEntity>> CompleteFilterAsync(
|
||
List<WarningNotifyTaskEntity> tasks, Dictionary<int, OrderEntity> orderDict, Dictionary<int, Dictionary<int, TrackDetailsIndex>> orderTrackStatusTrackDict)
|
||
{
|
||
// 1.规则条目存在
|
||
var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet();
|
||
var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds);
|
||
if (ruleItems.Count == 0) return Empty<WarningNotifyTaskEntity>.List;
|
||
|
||
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
|
||
// 过滤掉不存在规则的通知
|
||
var filtered = tasks.Where(x => ruleItemDict.ContainsKey(x.WarningRuleItemId));
|
||
|
||
// 2.验证规则条目
|
||
var result = new List<WarningNotifyTaskEntity>();
|
||
foreach (var group in filtered.GroupBy(x => x.OrderId))
|
||
{
|
||
var orderId = group.Key;
|
||
var orderTrackStatusTrack = orderTrackStatusTrackDict.GetValueOrDefault(orderId);
|
||
|
||
// 一个规则状态也不存在
|
||
if (orderTrackStatusTrack == null)
|
||
continue;
|
||
var order = orderDict[orderId];
|
||
var orderTrackStatusIds = orderTrackStatusTrack.Keys.ToHashSet();
|
||
|
||
foreach (var task in group)
|
||
{
|
||
var ruleItem = ruleItemDict[task.WarningRuleItemId];
|
||
if (ruleItem.CanHitRule(order.OrderType, orderTrackStatusIds))
|
||
result.Add(task);
|
||
}
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
private List<WarningNotifyParam> BuildNotifyConfigs(List<WarningNotifyTaskEntity> needNotifyTasks)
|
||
{
|
||
var result = new List<WarningNotifyParam>();
|
||
if (needNotifyTasks.Count == 0) return result;
|
||
|
||
// 1.以人的维度对不同类别的预警通知分组聚合
|
||
// key: userId value: {key:type,value:notify }
|
||
var userIdNotifyTaskDict = new Dictionary<string, Dictionary<int, List<WarningNotifyTaskEntity>>>();
|
||
foreach (var task in needNotifyTasks)
|
||
{
|
||
var notifyUsers = TextJsonHelper.Deserialize<List<NotifyUserDto>>(task.NotifyUsers)!;
|
||
var notifyUserIds = notifyUsers
|
||
.Select(x => x.UserId.Trim())
|
||
.Where(x => !string.IsNullOrWhiteSpace(x))
|
||
.ToHashSet();
|
||
|
||
foreach (var userId in notifyUserIds)
|
||
{
|
||
if (!userIdNotifyTaskDict.TryGetValue(userId, out var notifyTask))
|
||
{
|
||
notifyTask = new Dictionary<int, List<WarningNotifyTaskEntity>>();
|
||
userIdNotifyTaskDict[userId] = notifyTask;
|
||
}
|
||
|
||
if (!notifyTask.TryGetValue(task.WarningType, out var notify))
|
||
{
|
||
notify = new List<WarningNotifyTaskEntity>();
|
||
notifyTask[task.WarningType] = notify;
|
||
}
|
||
|
||
notify.Add(task);
|
||
}
|
||
}
|
||
|
||
// 2.使用消息模板构建通知参数，稍后排队发送通知
|
||
foreach (var (userId, typeNotifyTaskDict) in userIdNotifyTaskDict)
|
||
{
|
||
//聚合时出现 用户a
|
||
//接收回传预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书)
|
||
//接收运输预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书)
|
||
//接收完整预警，命中规则1，通知方式(飞书) ，命中规则2，通知方式(飞书，站内信)
|
||
//此时聚合后，消息为：总共6条预警消息,其中回传xx条，运输xx条，完整xx条
|
||
//同时使用飞书和站内信两种方式通知
|
||
result.Add(BuildNotifyParam(userId, typeNotifyTaskDict));
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
private WarningNotifyParam BuildNotifyParam(string userId, Dictionary<int, List<WarningNotifyTaskEntity>> typeNotifyTaskDict)
|
||
{
|
||
var options = _optionsMonitor.CurrentValue;
|
||
var hours = options.ScanWarningNotifyTaskHourInterval;
|
||
var param = new WarningNotifyParam
|
||
{
|
||
UserId = userId,
|
||
Hours = hours
|
||
};
|
||
|
||
// tips:获取过去x小时的增量
|
||
Func<WarningNotifyTaskEntity, bool> hoursNBefore = x => DateTime.Now - x.CreateTime <= TimeSpan.FromHours(hours);
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.运输时效预警.GetHashCode(), out var transportTimelinessNotifyTasks))
|
||
{
|
||
param.TransportLastNHoursCount = transportTimelinessNotifyTasks.Count(hoursNBefore);
|
||
param.TransportCount = transportTimelinessNotifyTasks.Count;
|
||
}
|
||
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.回传时效预警.GetHashCode(), out var returnTimelinessNotifyTasks))
|
||
{
|
||
param.ReturnLastNHoursCount = returnTimelinessNotifyTasks.Count(hoursNBefore);
|
||
param.ReturnCount = returnTimelinessNotifyTasks.Count;
|
||
}
|
||
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.完整预警.GetHashCode(), out var completeWarningNotifyTasks))
|
||
{
|
||
param.CompleteLastNHoursCount = completeWarningNotifyTasks.Count(hoursNBefore);
|
||
param.CompleteCount = completeWarningNotifyTasks.Count;
|
||
}
|
||
|
||
return param;
|
||
}
|
||
}
|