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;
///
/// 预警通知聚合器
///
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 _optionsMonitor;
public WarningNotifyAggregator(IWarningTaskRepository warningTaskRepository, IWarningRuleRepository warningRuleRepository, IOrderRepository orderRepository, ITrackDetailsEsRepository trackDetailsEsRepository, IWarningDetailsEsRepository warningDetailsEsRepository, IKpiAreaRepository kpiAreaRepository, IOptionsMonitor 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 NotifyConfigs, HashSet 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> FilterAsync(List 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.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.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(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> TransportFilterAsync(
List tasks, Dictionary orderDict, Dictionary> 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.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>(0);
// 2.2场景验证
var validTasks = new List(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.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.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();
var statusIds = statusTrack.Keys.ToHashSet();
return group.Where(task =>
ruleItemDict.TryGetValue(task.WarningRuleItemId, out var ruleItem)
&& ruleItem.CanReHit(statusIds));
})
.ToList();
}
private async Task> ReturnFilterAsync(
List tasks, Dictionary orderDict, Dictionary> orderTrackStatusTrackDict)
{
// 1.规则条目存在
var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet();
var ruleItems = await _warningRuleRepository.GetReturnRuleItemsAsync(ruleItemIds);
if (ruleItems.Count == 0) return Empty.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(1)
{
ReturnWarningStatus.人工介入.GetHashCode()
},
});
if (warningDetails.Count == 0) return Empty.List;
// 过滤掉不为人工介入的预警通知
var detailsIds = warningDetails.Select(x => x.Id).ToHashSet();
return filtered.Where(x => detailsIds.Contains(x.WarningDetailsId)).ToList();
}
private async Task> CompleteFilterAsync(
List tasks, Dictionary orderDict, Dictionary> orderTrackStatusTrackDict)
{
// 1.规则条目存在
var ruleItemIds = tasks.Select(x => x.WarningRuleItemId).ToHashSet();
var ruleItems = await _warningRuleRepository.GetCompleteRuleItemsAsync(ruleItemIds);
if (ruleItems.Count == 0) return Empty.List;
var ruleItemDict = ruleItems.ToDictionary(x => x.Id);
// 过滤掉不存在规则的通知
var filtered = tasks.Where(x => ruleItemDict.ContainsKey(x.WarningRuleItemId));
// 2.验证规则条目
var result = new List();
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 BuildNotifyConfigs(List needNotifyTasks)
{
var result = new List();
if (needNotifyTasks.Count == 0) return result;
// 1.以人的维度对不同类别的预警通知分组聚合
// key: userId value: {key:type,value:notify }
var userIdNotifyTaskDict = new Dictionary>>();
foreach (var task in needNotifyTasks)
{
var notifyUsers = TextJsonHelper.Deserialize>(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>();
userIdNotifyTaskDict[userId] = notifyTask;
}
if (!notifyTask.TryGetValue(task.WarningType, out var notify))
{
notify = new List();
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
{
1,
2
}));
}
return result;
}
private string BuildNotifyMessage(Dictionary> 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 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}({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}";
}
}