定时任务轨迹运输时效预警推送

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
     /// </summary>
     public int OrderId { get; set; }
 
+    /// <summary>
+    /// 预警次数
+    /// </summary>
+    public int Counter { get; set; }
+
     public override string ToString()
     {
-        return $"{RuleId}_{OrderId}";
+        return $"{RuleId}_{OrderId}_{Counter}";
     }
 }

src/SJZY.Track.Application.Contract/Events/TrackInsertForNotMappedEvent.cs

@@ -1,49 +0,0 @@
-namespace SJZY.Track.Application.Contract.Events;
-
-public class TrackInsertForNotMappedEvent
-{
-    /// <summary>
-    /// 订单id
-    /// </summary>
-    public int OrderId { get; set; }
-
-    /// <summary>
-    /// 订单类型
-    /// </summary>
-    public int OrderType { get; set; }
-
-    /// <summary>
-    /// 订单号
-    /// </summary>
-    public string OrderNo { get; set; } = null!;
-
-    /// <summary>
-    /// 服务商
-    /// </summary>
-    public string ServiceProviderName { get; set; } = null!;
-
-    /// <summary>
-    /// 原轨迹内容
-    /// </summary>
-    public string? OriginTrackContent { get; set; }
-
-    /// <summary>
-    /// 原轨迹报文
-    /// </summary>
-    public string? OriginTrackData { get; set; }
-
-    /// <summary>
-    /// 轨迹发生地
-    /// </summary>
-    public string TrackPlace { get; set; } = null!;
-
-    /// <summary>
-    /// 轨迹发生时间
-    /// </summary>
-    public DateTime TrackTime { get; set; }
-
-    /// <summary>
-    /// 轨迹发生时区
-    /// </summary>
-    public float TrackTimeZone { get; set; }
-}

src/SJZY.Track.Application.Contract/Events/TrackUpsertForRefreshEvent.cs

@@ -1,94 +0,0 @@
-namespace SJZY.Track.Application.Contract.Events;
-
-public class TrackUpsertForRefreshEvent
-{
-    /// <summary>
-    /// 子系统id（操作的订单都是相同子系统下的）
-    /// </summary>
-    public int SubSystemId { get; set; }
-
-    /// <summary>
-    /// 订单
-    /// </summary>
-    public OrderIdentifier OrderIdentifiers { get; set; }
-
-    /// <summary>
-    /// 服务商
-    /// </summary>
-    public string? ServiceProviderName { get; set; }
-
-    /// <summary>
-    /// 原轨迹内容
-    /// </summary>
-    public string? OriginTrackContent { get; set; }
-
-    /// <summary>
-    /// 原轨迹报文
-    /// </summary>
-    public string? OriginTrackData { get; set; }
-
-    /// <summary>
-    /// 表示订单是否完成（只由轨迹刷新配置决定）
-    /// tips：能够使得IsCompleted为true的点，全局只有一个：轨迹刷新
-    /// </summary>
-    public bool IsCompleted { get; set; }
-
-    /// <summary>
-    /// 轨迹内容代码
-    /// </summary>
-    public string TrackCode { get; set; } = null!;
-
-    /// <summary>
-    /// 轨迹发生地
-    /// </summary>
-    public string TrackPlace { get; set; } = null!;
-
-    /// <summary>
-    /// 轨迹发生时间
-    /// </summary>
-    public DateTime TrackTime { get; set; }
-
-    /// <summary>
-    /// 轨迹发生时区
-    /// </summary>
-    public float TrackTimeZone { get; set; }
-}
-
-public class TrackHandledDto
-{
-    /// <summary>
-    /// 原轨迹内容
-    /// </summary>
-    public string? OriginTrackContent { get; set; }
-
-    /// <summary>
-    /// 原轨迹报文
-    /// </summary>
-    public string? OriginTrackData { get; set; }
-
-    /// <summary>
-    /// 表示订单是否完成（只由轨迹刷新配置决定）
-    /// tips：能够使得IsCompleted为true的点，全局只有一个：轨迹刷新
-    /// </summary>
-    public bool? IsCompleted { get; set; }
-
-    /// <summary>
-    /// 轨迹内容代码
-    /// </summary>
-    public string? TrackCode { get; set; }
-
-    /// <summary>
-    /// 轨迹发生地
-    /// </summary>
-    public string? TrackPlace { get; set; }
-
-    /// <summary>
-    /// 轨迹发生时间
-    /// </summary>
-    public DateTime? TrackTime { get; set; }
-
-    /// <summary>
-    /// 轨迹发生时区
-    /// </summary>
-    public float? TrackTimeZone { get; set; }
-}

src/SJZY.Track.Application.Contract/Services/IForewarningNotifyService.cs

@@ -18,5 +18,5 @@ public interface IForewarningNotifyService
     /// <summary>
     /// 发送通知。
     /// </summary>
-    Task SendNotificationAsync();
+    Task SendNotificationAsync(int subSystemId);
 }

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 adds = new List<(int ruleId, int orderId, int counter, long ts)>();
-        var dels = new List<(int ruleId, int orderId)>();
+        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 tran.ZRemAsync(redisKey, right.Select(x => x.ToString()).ToArray());
-            await _warningRecordEsRepository.BulkUpdateAsync(dels); // 标记处理完成
         }
 
         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<WarningRecordIndex>();
         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<GetWarningNotifiesVo> rules, List<WarningNotifyCacheEntry> entries,
         NotifyMode mode)
     {

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;
     }
 
     /// <summary>
     /// 获取订单运输时效预警分页列表
     /// </summary>
     /// <param name="inputVo"></param>
-    /// <param name="warningRecordEsRepository"></param>
     /// <returns></returns>
     [HttpPost("GetOrderWarningPageList")]
     public async Task<UnifyPageResultDto<GetOrderWarningPageOutput>> 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<GetOrderWarningPageOutput>(list, inputVo.PageIndex, inputVo.PageSize,
             (int) result.total);
     }
+
+
+    /// <summary>
+    /// 定时任务轨迹运输时效预警推送
+    /// </summary>
+    /// <returns></returns>
+    [AllowAnonymous]
+    [HttpPost("Track/SendWarningForTask")]
+    public async Task<bool> SendWarningForTaskAsync([FromBody] IConfiguration configuration)
+    {
+        var configs = configuration.GetSection("SubSystemWarning").Get<Dictionary<string, int>>();
+        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;
+    }
 }

src/SJZY.Track.Domain/Aggregates/ForewarningAggregate/IWarningRecordEsRepository.cs

@@ -23,7 +23,7 @@ public interface IWarningRecordEsRepository
     /// </summary>
     /// <param name="warningRecords"></param>
     /// <returns></returns>
-    Task BulkUpdateAsync(List<(int ruleId, int orderId)> warningRecords);
+    Task BulkUpdateAsync(List<(int ruleId, int orderId, int counter)> warningRecords);
 
     /// <summary>
     /// 获取订单运输时效预警分页列表

src/SJZY.Track.Domain/Aggregates/OrderAggregate/IOrderRepository.cs

@@ -46,6 +46,13 @@ public interface IOrderRepository : IBasicRepository<OrderEntity>
     /// <returns></returns>
     Task<int> UpdateBatchIdsAsync(List<int> ids, int newValue);
 
+    /// <summary>
+    /// 根据子系统code获取子系统id
+    /// </summary>
+    /// <param name="codes"></param>
+    /// <returns></returns>
+    Task<List<(string code, int id)>> GetSubSystemListByCodesAsync(List<string> codes);
+
     #region 订单关联
 
     /// <summary>

src/SJZY.Track.Domain/Consts/GlobalConst.MQ.cs

@@ -51,4 +51,9 @@ public partial class GlobalConst
     /// 推送预警任务
     /// </summary>
     public const string TrackPushWarning = "SJZY.Track.TrackPushWarning";
+
+    /// <summary>
+    /// 预警计算任务
+    /// </summary>
+    public const string TrackWarningCalc = "SJZY.Track.TrackWarningCalc";
 }

src/SJZY.Track.Domain/Consts/GlobalConst.Redis.cs

@@ -23,14 +23,14 @@ public partial class GlobalConst
     public const string TrackPullContentKeyPrefix = "TrackPullContent";
 
     /// <summary>
-    /// 预警通知redis key
+    /// 预警通知set redis key
     /// </summary>
-    public const string ForewarningNotifyKey = "ForewarningNotify";
+    public const string ForewarningNotifySetKey = "ForewarningNotifySet";
 
     /// <summary>
-    /// 预警通知剩余次数redis key
+    /// 预警通知定时任务罚时 redis key
     /// </summary>
-    public const string ForewarningNotifyCntKey = "ForewarningNotifyCnt";
+    public const string ForewarningNotifyCheckKey = "ForewarningNotifyCheck";
 
     /// <summary>
     /// kpi区域redis key前缀

src/SJZY.Track.Repository/Repositories/OrderRepository.cs

@@ -148,6 +148,20 @@ public class OrderRepository : BasicRepository<OrderEntity>, IOrderRepository, I
         return ordersToUpdate.Count;
     }
 
+    public async Task<List<(string code, int id)>> GetSubSystemListByCodesAsync(List<string> codes)
+    {
+        var res = await Context.Queryable<VSystemEntity>()
+            .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<int> ids)
     {
         var idsString = string.Join(",", ids);

src/SJZY.Track.Repository/Repositories/WarningRecordEsRepository.cs

@@ -44,7 +44,7 @@ public class WarningRecordEsRepository : ElasticSearchContextAbstract<WarningRec
         }
     }
 
-    public async Task BulkUpdateAsync(List<(int ruleId, int orderId)> rules)
+    public async Task BulkUpdateAsync(List<(int ruleId, int orderId, int counter)> rules)
     {
         foreach (var rule in rules)
         {