399 lines
18 KiB
C#
399 lines
18 KiB
C#
using Microsoft.Extensions.Options;
|
||
using SJZY.Track.Application.Contract.Dtos.Warning;
|
||
using SJZY.Track.Application.Helpers;
|
||
using SJZY.Track.Application.Options;
|
||
using SJZY.Track.Application.Services.Warning.Model;
|
||
using SJZY.Track.Domain.Aggregates.KPIAreaAggregate;
|
||
using SJZY.Track.Domain.Aggregates.OrderAggregate;
|
||
using SJZY.Track.Domain.Aggregates.TrackDetailsAggregate;
|
||
using SJZY.Track.Domain.Aggregates.WarningAggregate;
|
||
using SJZY.Track.Domain.Dtos.Forewarning;
|
||
using SJZY.Track.Domain.Dtos.Warning;
|
||
using SJZY.Track.Domain.Enums;
|
||
using SJZY.Track.Domain.Extensions;
|
||
using Volo.Abp.DependencyInjection;
|
||
|
||
namespace SJZY.Track.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<NotifyConfig> 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), y => y);
|
||
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<NotifyConfig> BuildNotifyConfigs(List<WarningNotifyTaskEntity> needNotifyTasks)
|
||
{
|
||
var result = new List<NotifyConfig>();
|
||
if (needNotifyTasks.Count == 0) return result;
|
||
|
||
// 1.以人的维度对不同类别的预警通知分组聚合
|
||
// key: userId value: {key:type,value:notify }
|
||
var userIdNotifyTaskDict = new Dictionary<int, Dictionary<int, List<WarningNotifyTaskEntity>>>();
|
||
foreach (var task in needNotifyTasks)
|
||
{
|
||
var notifyUsers = TextJsonHelper.Deserialize<List<NotifyUserDto>>(task.NotifyUsers)!;
|
||
var notifyUserIds = notifyUsers.Select(x => x.UserId).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)
|
||
{
|
||
var notifyMessage = BuildNotifyMessage(typeNotifyTaskDict);
|
||
//聚合时出现 用户a
|
||
//接收回传预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书)
|
||
//接收运输预警，命中规则1，通知方式(飞书,站内信) ，命中规则2，通知方式(飞书)
|
||
//接收完整预警，命中规则1，通知方式(飞书) ，命中规则2，通知方式(飞书，站内信)
|
||
//此时聚合后，消息为：总共6条预警消息,其中回传xx条，运输xx条，完整xx条
|
||
//同时使用飞书和站内信两种方式通知
|
||
result.Add(NotifyConfig.CreateConfig(userId, notifyMessage, new List<int>
|
||
{
|
||
1,
|
||
2
|
||
}));
|
||
}
|
||
|
||
return result;
|
||
}
|
||
|
||
private string BuildNotifyMessage(Dictionary<int, List<WarningNotifyTaskEntity>> typeNotifyTaskDict)
|
||
{
|
||
var options = _optionsMonitor.CurrentValue;
|
||
var hours = options.ScanWarningNotifyTaskHourInterval;
|
||
var transportCount = 0;
|
||
var transportLastNHoursCount = 0;
|
||
var returnCount = 0;
|
||
var returnLastNHoursCount = 0;
|
||
var completeCount = 0;
|
||
var completeLastNHoursCount = 0;
|
||
Func<WarningNotifyTaskEntity, bool> hoursNBefore = x => DateTime.Now - x.CreateTime <= TimeSpan.FromHours(hours);
|
||
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.运输时效预警.GetHashCode(), out var transportTimelinessNotifyTasks))
|
||
{
|
||
transportLastNHoursCount = transportTimelinessNotifyTasks.Count(hoursNBefore);
|
||
transportCount = transportTimelinessNotifyTasks.Count;
|
||
}
|
||
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.回传时效预警.GetHashCode(), out var returnTimelinessNotifyTasks))
|
||
{
|
||
returnLastNHoursCount = returnTimelinessNotifyTasks.Count(hoursNBefore);
|
||
returnCount = returnTimelinessNotifyTasks.Count;
|
||
}
|
||
|
||
if (typeNotifyTaskDict.TryGetValue(WarningRuleType.完整预警.GetHashCode(), out var completeWarningNotifyTasks))
|
||
{
|
||
completeLastNHoursCount = completeWarningNotifyTasks.Count(hoursNBefore);
|
||
completeCount = completeWarningNotifyTasks.Count;
|
||
}
|
||
|
||
return NotifyMessageBuilder.Build(
|
||
hours,
|
||
transportCount,
|
||
transportLastNHoursCount,
|
||
returnCount,
|
||
returnLastNHoursCount,
|
||
completeCount,
|
||
completeLastNHoursCount
|
||
);
|
||
}
|
||
}
|
||
|
||
// 标题：预警通知
|
||
// 内容：尊敬的用户，截止到2024-12-13 10:00:01(UTC +8:00)，轨迹中心监测到与您相关的预警风险信息共计299 条，请及时处理，详情如下：
|
||
// 运输时效预警共计100条，其中过去4小时内新增的预警10条，查看详情
|
||
// 回传时效预警共计99条，其中过去4小时内新增的预警10条，查看详情
|
||
// 轨迹完整预警共计100条，其中过去4小时内新增的预警10条，查看详情
|
||
public static class NotifyMessageBuilder
|
||
{
|
||
public const string Title = "预警通知";
|
||
|
||
public static string Build(int hours,
|
||
int transportCount, int transportLastNHoursCount,
|
||
int returnCount, int returnLastNHoursCount,
|
||
int completeCount, int completeLastNHoursCount)
|
||
{
|
||
var dateTime = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss");
|
||
var totalCount = transportCount + returnCount + completeCount;
|
||
// todo:无法使用a标签填充（可能是为了安全考虑），需要具体给定一个样式
|
||
var url1 = "";
|
||
var url2 = "";
|
||
var url3 = "";
|
||
|
||
var message = $"{Title}\n尊敬的用户，截止到{dateTime}(UTC {BuildLocalTimezoneStr()})，轨迹中心监测到与您相关的预警风险信息共计{totalCount} 条，请及时处理，详情如下：";
|
||
|
||
if (transportCount > 0)
|
||
message += $"\n运输时效预警共计{transportCount}条，其中过去{hours}小时内新增的预警{transportLastNHoursCount}条";
|
||
if (returnCount > 0)
|
||
message += $"\n回传时效预警共计{returnCount}条，其中过去{hours}小时内新增的预警{returnLastNHoursCount}条";
|
||
if (completeCount > 0)
|
||
message += $"\n轨迹完整预警共计{completeCount}条，其中过去{hours}小时内新增的预警{completeLastNHoursCount}条";
|
||
|
||
return message;
|
||
}
|
||
|
||
// 构建本地时区字符串标识 e.g UTC +8:00
|
||
private static string BuildLocalTimezoneStr()
|
||
{
|
||
var offset = TimeZoneInfo.Local.BaseUtcOffset;
|
||
var flag = offset.Hours > 0 ? "+" : "";
|
||
var offsetStr = offset.ToString().RemovePostFix(":00");
|
||
return $"UTC {flag}{offsetStr}";
|
||
}
|
||
}
|