diff --git a/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningNotifyCacheEntry.cs b/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningNotifyCacheEntry.cs index 9d07c54e..591be62e 100644 --- a/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningNotifyCacheEntry.cs +++ b/src/SJZY.Track.Application.Contract/Dtos/Forewarning/WarningNotifyCacheEntry.cs @@ -2,10 +2,11 @@ namespace SJZY.Track.Application.Contract.Dtos.Forewarning; public class WarningNotifyCacheEntry { - public WarningNotifyCacheEntry(int ruleId, int orderId) + public WarningNotifyCacheEntry(int ruleId, int orderId, int counter) { RuleId = ruleId; OrderId = orderId; + Counter = counter; } public WarningNotifyCacheEntry(string notifyValue) @@ -25,8 +26,13 @@ public class WarningNotifyCacheEntry /// public int OrderId { get; set; } + /// + /// 预警次数 + /// + public int Counter { get; set; } + public override string ToString() { - return $"{RuleId}_{OrderId}"; + return $"{RuleId}_{OrderId}_{Counter}"; } } diff --git a/src/SJZY.Track.Application.Contract/Events/TrackInsertForNotMappedEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackInsertForNotMappedEvent.cs deleted file mode 100644 index f67be132..00000000 --- a/src/SJZY.Track.Application.Contract/Events/TrackInsertForNotMappedEvent.cs +++ /dev/null @@ -1,49 +0,0 @@ -namespace SJZY.Track.Application.Contract.Events; - -public class TrackInsertForNotMappedEvent -{ - /// - /// 订单id - /// - public int OrderId { get; set; } - - /// - /// 订单类型 - /// - public int OrderType { get; set; } - - /// - /// 订单号 - /// - public string OrderNo { get; set; } = null!; - - /// - /// 服务商 - /// - public string ServiceProviderName { get; set; } = null!; - - /// - /// 原轨迹内容 - /// - public string? OriginTrackContent { get; set; } - - /// - /// 原轨迹报文 - /// - public string? OriginTrackData { get; set; } - - /// - /// 轨迹发生地 - /// - public string TrackPlace { get; set; } = null!; - - /// - /// 轨迹发生时间 - /// - public DateTime TrackTime { get; set; } - - /// - /// 轨迹发生时区 - /// - public float TrackTimeZone { get; set; } -} diff --git a/src/SJZY.Track.Application.Contract/Events/TrackUpsertForRefreshEvent.cs b/src/SJZY.Track.Application.Contract/Events/TrackUpsertForRefreshEvent.cs deleted file mode 100644 index 4ae99f4f..00000000 --- a/src/SJZY.Track.Application.Contract/Events/TrackUpsertForRefreshEvent.cs +++ /dev/null @@ -1,94 +0,0 @@ -namespace SJZY.Track.Application.Contract.Events; - -public class TrackUpsertForRefreshEvent -{ - /// - /// 子系统id（操作的订单都是相同子系统下的） - /// - public int SubSystemId { get; set; } - - /// - /// 订单 - /// - public OrderIdentifier OrderIdentifiers { get; set; } - - /// - /// 服务商 - /// - public string? ServiceProviderName { get; set; } - - /// - /// 原轨迹内容 - /// - public string? OriginTrackContent { get; set; } - - /// - /// 原轨迹报文 - /// - public string? OriginTrackData { get; set; } - - /// - /// 表示订单是否完成（只由轨迹刷新配置决定） - /// tips：能够使得IsCompleted为true的点，全局只有一个：轨迹刷新 - /// - public bool IsCompleted { get; set; } - - /// - /// 轨迹内容代码 - /// - public string TrackCode { get; set; } = null!; - - /// - /// 轨迹发生地 - /// - public string TrackPlace { get; set; } = null!; - - /// - /// 轨迹发生时间 - /// - public DateTime TrackTime { get; set; } - - /// - /// 轨迹发生时区 - /// - public float TrackTimeZone { get; set; } -} - -public class TrackHandledDto -{ - /// - /// 原轨迹内容 - /// - public string? OriginTrackContent { get; set; } - - /// - /// 原轨迹报文 - /// - public string? OriginTrackData { get; set; } - - /// - /// 表示订单是否完成（只由轨迹刷新配置决定） - /// tips：能够使得IsCompleted为true的点，全局只有一个：轨迹刷新 - /// - public bool? IsCompleted { get; set; } - - /// - /// 轨迹内容代码 - /// - public string? TrackCode { get; set; } - - /// - /// 轨迹发生地 - /// - public string? TrackPlace { get; set; } - - /// - /// 轨迹发生时间 - /// - public DateTime? TrackTime { get; set; } - - /// - /// 轨迹发生时区 - /// - public float? TrackTimeZone { get; set; } -} diff --git a/src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs b/src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs index df0272c6..14647fcc 100644 --- a/src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs +++ b/src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs @@ -18,5 +18,5 @@ public interface IForewarningNotifyService /// /// 发送通知。 /// - Task SendNotificationAsync(); + Task SendNotificationAsync(int subSystemId); } diff --git a/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs b/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs index 59e99f1f..4f104e7d 100644 --- a/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs +++ b/src/SJZY.Track.Application/Services/ForewarningServices/ForewarningNotifyService.cs @@ -40,8 +40,8 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep var rules = await _cacheService.GetRulesAsync(dto.SubSystemId, dto.CustomerId); if (rules == null || rules.Count == 0) return; - var adds = new List<(int ruleId, int orderId, long ts)>(); - var dels = new List<(int ruleId, int orderId)>(); + var adds = new List<(int ruleId, int orderId, int counter, long ts)>(); + var dels = new List<(int ruleId, int orderId, int counter)>(); foreach (var order in dto.Orders) { // 加左区间 @@ -59,7 +59,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep startTime = startTime.AddMinutes(interval); // 计算预警时间戳，作为排序依据 - adds.Add((rule.Id, order.Id, timestamp)); + adds.Add((rule.Id, order.Id, i, timestamp)); } }); @@ -67,33 +67,43 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep MatchRule( rules.Where(rule => rule.EndTrackStatusId == dto.TrackStatusId).ToList(), order - ).ForEach(rule => { dels.Add((rule.Id, order.Id)); }); + ).ForEach(rule => + { + for (var i = 0; i < rule.WarnCount; i++) + { + dels.Add((rule.Id, order.Id, i)); + } + }); } + var redisKey = GetForewarningNotifySetKeyRedisKey(dto.SubSystemId); + var tran = RedisHelper.Client.Multi(); // 左区间加入预警队列 - await tran.ZAddAsync(GlobalConst.ForewarningNotifyKey, adds.Select(x => + await tran.ZAddAsync(redisKey, adds.Select(x => { - var entry = new WarningNotifyCacheEntry(x.ruleId, x.orderId); + var entry = new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter); return new ZMember(entry.ToString(), x.ts); }).ToArray()); // 右区间移出预警队列 - var right = dels.Select(x => new WarningNotifyCacheEntry(x.ruleId, x.orderId)).ToList(); + var right = dels.Select(x => new WarningNotifyCacheEntry(x.ruleId, x.orderId, x.counter)).ToList(); if (!right.IsNullOrEmpty()) { - await tran.ZRemAsync(GlobalConst.ForewarningNotifyKey, right.Select(x => x.ToString()).ToArray()); - await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成 + await tran.ZRemAsync(redisKey, right.Select(x => x.ToString()).ToArray()); } tran.Exec(); + + await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成 } - public async Task SendNotificationAsync() + public async Task SendNotificationAsync(int subSystemId) { + var redisKey = GetForewarningNotifySetKeyRedisKey(subSystemId); var timestamp = GetTimestamp(DateTime.Now); - var set = await RedisHelper.Client.ZRangeByScoreAsync(GlobalConst.ForewarningNotifyKey, 0, timestamp); + var set = await RedisHelper.Client.ZRangeByScoreAsync(redisKey, 0, timestamp); if (set.IsNullOrEmpty()) { @@ -119,6 +129,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep } } + // 写入es预警记录 var warningRecords = new List(); foreach (var entry in entries) { @@ -141,6 +152,7 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep await _warningRecordEsRepository.BulkInsertAsync(warningRecords); + // 发通知 foreach (var userRule in userRules) { var siteInternalContent = BuildNotifyContent(userRule.Value, entries, NotifyMode.SiteInternal); @@ -156,8 +168,13 @@ public class ForewarningNotifyService : IForewarningNotifyService, ITransientDep // todo：发送飞书通知 } } + + await RedisHelper.Client.ZRemRangeByScoreAsync(redisKey, 0, timestamp); } + private string GetForewarningNotifySetKeyRedisKey(int subSystemId) => + $"{GlobalConst.ForewarningNotifySetKey}:{subSystemId}"; + private string BuildNotifyContent(List rules, List entries, NotifyMode mode) { diff --git a/src/SJZY.Track.Application/Services/TrackOrders/OrderWarningService.cs b/src/SJZY.Track.Application/Services/TrackOrders/OrderWarningService.cs index cd2f862d..86bf42b9 100644 --- a/src/SJZY.Track.Application/Services/TrackOrders/OrderWarningService.cs +++ b/src/SJZY.Track.Application/Services/TrackOrders/OrderWarningService.cs @@ -1,3 +1,8 @@ +using Microsoft.AspNetCore.Authorization; +using Microsoft.Extensions.Configuration; +using SJZY.Expand.ABP.Core; +using SJZY.Expand.ABP.Core.Logger; +using SJZY.Track.Application.Contract.Services; using SJZY.Track.Domain.Aggregates.ForewarningAggregate; using SJZY.Track.Domain.Aggregates.OrderAggregate; using SJZY.Track.Domain.Dtos.Forewarning; @@ -6,25 +11,28 @@ namespace SJZY.Track.Application.Services.TrackOrders; public class OrderWarningService { + private readonly IForewarningNotifyService _forewarningNotifyService; private readonly IOrderRepository _orderRepository; private readonly IWarningRecordEsRepository _warningRecordEsRepository; - public OrderWarningService(IWarningRecordEsRepository warningRecordEsRepository, IOrderRepository orderRepository) + public OrderWarningService(IWarningRecordEsRepository warningRecordEsRepository, IOrderRepository orderRepository, + IForewarningNotifyService forewarningNotifyService) { _warningRecordEsRepository = warningRecordEsRepository; _orderRepository = orderRepository; + _forewarningNotifyService = forewarningNotifyService; } /// /// 获取订单运输时效预警分页列表 /// /// - /// /// [HttpPost("GetOrderWarningPageList")] public async Task> GetOrderWarningPageListAsync( GetOrderWarningPageListVo inputVo) { + // todo：产品要的，排序字段，查询方式需要讨论 var result = await _warningRecordEsRepository.GetOrderWarningPageListAsync(inputVo); var orderIds = result.list.Select(x => x.OrderId).ToList(); var orders = await _orderRepository.GetListAsync(x => orderIds.Contains(x.Id), x => new @@ -61,4 +69,49 @@ public class OrderWarningService return new UnifyPageResultDto(list, inputVo.PageIndex, inputVo.PageSize, (int) result.total); } + + + /// + /// 定时任务轨迹运输时效预警推送 + /// + /// + [AllowAnonymous] + [HttpPost("Track/SendWarningForTask")] + public async Task SendWarningForTaskAsync([FromBody] IConfiguration configuration) + { + var configs = configuration.GetSection("SubSystemWarning").Get>(); + if (configs == null || configs.Keys.Count == 0) + { + throw new BusinessException("未配置预警系统"); + } + + var subSystemIds = await _orderRepository.GetSubSystemListByCodesAsync(configs.Keys.ToList()); + foreach (var item in subSystemIds) + { + // 如果已经发送过通知，不再发送 + var redisKey = $"{GlobalConst.ForewarningNotifyCheckKey}:{subSystemIds}"; + if (await RedisHelper.Client.ExistsAsync(redisKey)) + { + continue; + } + + // 异步发送通知 + _ = Task.Run(async () => + { + try + { + await _forewarningNotifyService.SendNotificationAsync(item.id); + } + catch (Exception e) + { + LogHelper.Error(e.ToString()); + await RedisHelper.Client.DelAsync(redisKey); + } + }); + + await RedisHelper.Client.SetExAsync(redisKey, configs[item.code] * 60, 1); + } + + return true; + } } diff --git a/src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRecordEsRepository.cs b/src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRecordEsRepository.cs index eb52ca54..5f7d1ad3 100644 --- a/src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRecordEsRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRecordEsRepository.cs @@ -23,7 +23,7 @@ public interface IWarningRecordEsRepository /// /// /// - Task BulkUpdateAsync(List<(int ruleId, int orderId)> warningRecords); + Task BulkUpdateAsync(List<(int ruleId, int orderId, int counter)> warningRecords); /// /// 获取订单运输时效预警分页列表 diff --git a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs index d71e24b6..4839a628 100644 --- a/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs +++ b/src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs @@ -46,6 +46,13 @@ public interface IOrderRepository : IBasicRepository /// Task UpdateBatchIdsAsync(List ids, int newValue); + /// + /// 根据子系统code获取子系统id + /// + /// + /// + Task> GetSubSystemListByCodesAsync(List codes); + #region 订单关联 /// diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs index e75305e1..488eced1 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs @@ -51,4 +51,9 @@ public partial class GlobalConst /// 推送预警任务 /// public const string TrackPushWarning = "SJZY.Track.TrackPushWarning"; + + /// + /// 预警计算任务 + /// + public const string TrackWarningCalc = "SJZY.Track.TrackWarningCalc"; } diff --git a/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs b/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs index 37e054a5..6e2e6791 100644 --- a/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs +++ b/src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs @@ -23,14 +23,14 @@ public partial class GlobalConst public const string TrackPullContentKeyPrefix = "TrackPullContent"; /// - /// 预警通知redis key + /// 预警通知set redis key /// - public const string ForewarningNotifyKey = "ForewarningNotify"; + public const string ForewarningNotifySetKey = "ForewarningNotifySet"; /// - /// 预警通知剩余次数redis key + /// 预警通知定时任务罚时 redis key /// - public const string ForewarningNotifyCntKey = "ForewarningNotifyCnt"; + public const string ForewarningNotifyCheckKey = "ForewarningNotifyCheck"; /// /// kpi区域redis key前缀 diff --git a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs index c6a68df9..e7517f86 100644 --- a/src/SJZY.Track.Repository/Repositories/OrderRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/OrderRepository.cs @@ -148,6 +148,20 @@ public class OrderRepository : BasicRepository, IOrderRepository, I return ordersToUpdate.Count; } + public async Task> GetSubSystemListByCodesAsync(List codes) + { + var res = await Context.Queryable() + .Where(x => codes.Contains(x.Key)) + .Select(x => new + { + x.Key, + x.Id + }) + .ToListAsync(); + + return res.Select(x => (x.Key, x.Id)).ToList(); + } + private int UpdateOrderQuestionType(int questionType, IEnumerable ids) { var idsString = string.Join(",", ids); diff --git a/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs b/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs index 490570f0..a3d73b5e 100644 --- a/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs +++ b/src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs @@ -44,7 +44,7 @@ public class WarningRecordEsRepository : ElasticSearchContextAbstract rules) + public async Task BulkUpdateAsync(List<(int ruleId, int orderId, int counter)> rules) { foreach (var rule in rules) {