Merge remote-tracking branch 'origin/main'

2. mqtt消息处理去重

OpenAuth.App/ServiceApp/ManageApp.cs

@@ -13,6 +13,7 @@ using MetadataExtractor;
 using MetadataExtractor.Formats.Exif;
 using MetadataExtractor.Formats.Xmp;
 using Microsoft.AspNetCore.Http;
+using Microsoft.Extensions.Logging;
 using Microsoft.IdentityModel.Tokens;
 using Microsoft.Net.Http.Headers;
 using Newtonsoft.Json;
@@ -41,17 +42,19 @@ namespace OpenAuth.App.ServiceApp
         private readonly MqttClientManager _mqttClientManager;
         private readonly MinioService _minioService;
         private readonly OpenJobApp _openJobApp;
+        private readonly ILogger<LasaDronePort> _logger;
         CommonDataManager _commonDataManager;
 
         public ManageApp(ISugarUnitOfWork<SugarDbContext> unitWork, ISimpleClient<LasaDronePort> repository, IAuth auth,
             MqttClientManager mqttClientManager, CommonDataManager commonDataManager, MinioService minioService,
-            OpenJobApp openJobApp)
+            OpenJobApp openJobApp, ILogger<LasaDronePort> logger)
             : base(unitWork, repository, auth)
         {
             _mqttClientManager = mqttClientManager;
             _minioService = minioService;
             _commonDataManager = commonDataManager;
             _openJobApp = openJobApp;
+            _logger = logger;
         }
 
         #region 机场管理
@@ -2381,6 +2384,7 @@ namespace OpenAuth.App.ServiceApp
 
         public async Task<Response<bool>> CallAiModel(CallAiModel req)
         {
+            _logger.LogDebug("成功调用CallAIModel");
             var task = await Repository
                 .ChangeRepository<SugarRepositiry<LasaTask>>()
                 .GetByIdAsync(req.TaskId);

OpenAuth.App/ServiceApp/Subscribe/ConfigSubscribe.cs

@@ -1,4 +1,5 @@
-using System.Dynamic;
+using System.Collections.Concurrent;
+using System.Dynamic;
 using System.Net;
 using System.Net.Http.Headers;
 using System.Text;
@@ -7,13 +8,13 @@ using Infrastructure.CloudSdk;
 using Infrastructure.CloudSdk.minio;
 using Infrastructure.CloudSdk.wayline;
 using Infrastructure.Extensions;
+using Infrastructure.Helpers;
 using MetadataExtractor;
 using MetadataExtractor.Formats.Exif;
 using MetadataExtractor.Formats.Xmp;
 using Microsoft.Extensions.Logging;
 using MQTTnet.Client;
 using Newtonsoft.Json;
-using NPOI.SS.Formula.Functions;
 using OpenAuth.App.ServiceApp;
 using OpenAuth.App.ServiceApp.FlyTask.Request;
 using OpenAuth.Repository.Domain;
@@ -33,6 +34,8 @@ public class ConfigSubscribe : IJob
     private object _locker = new();
     private object _dockUploadFileLocker = new();
     private readonly ILogger<ConfigSubscribe> _logger;
+    private readonly ConcurrentDictionary<string, DateTime> _processedMessages = new();
+    private readonly TimeSpan _deduplicationWindow = TimeSpan.FromMinutes(1);
 
     public ConfigSubscribe(MqttClientManager mqttClientManager, ISqlSugarClient sqlSugarClient,
         ICacheContext redisCacheContext, ManageApp manageApp, MinioService minioService,
@@ -570,7 +573,14 @@ public class ConfigSubscribe : IJob
                         // "28":"降落以后的关盖"
                         // "29":"机场退出工作模式"
                         // "30":"机场异常恢复","31":"机场上传飞行系统日志","32":"相机录像状态检查","33":"获取媒体文件数量","34":"机场起飞开盖的异常恢复","35":"通知任务结果","36":"日志列表拉取 - 飞行器列表","37":"日志列表拉取 - 拉取机场列表","38":"日志列表拉取 - 上传日志列表结果","39":"日志拉取-拉取飞行器日志","40":"日志拉取-拉取机场日志","41":"日志拉取-压缩飞行器日志","42":"日志拉取-压缩机场日志","43":"日志拉取-上传飞行器日志","44":"日志拉取-上传机场日志","45":"日志拉取-通知结果","46":"自定义飞行区文件更新准备中","47":"自定义飞行区更新中","48":"离线地图更新准备中","49":"离线地图更新中","65533":"结束后等待服务回包","65534":"无具体状态","65535":"UNKNOWN"}
+                        _logger.LogDebug($"md5: {Md5.Encrypt(message)} 重复否：{IsDuplicate(Md5.Encrypt(message))} 信息：{message} ");
+                        if (IsDuplicate(Md5.Encrypt(message)))
+                        {
+                            _logger.LogInformation("跳过处理");
+                            break;
+                        }
                         code = data.result; // result 
+
                         var waylineMissionState = (int)data.output.ext.wayline_mission_state;
                         string flightId1 = (string)data.output.ext.flight_id;
                         var taskAssign1 = _manageApp.GetTaskAssignByFlightId(flightId1);
@@ -613,6 +623,7 @@ public class ConfigSubscribe : IJob
                         else
                         {
                             var step = (int)data.output.progress.current_step;
+                            _logger.LogDebug($"航线进度：{waylineMissionState} {step} {message}");
                             if (step.Equals(25)) // 航线执行
                             {
                                 var task = await _sqlSugarClient.Queryable<LasaTask>()
@@ -620,18 +631,20 @@ public class ConfigSubscribe : IJob
                                 if (task != null && !string.IsNullOrEmpty(task.AIInspection) &&
                                     task.AIInspection.Equals("true") && string.IsNullOrEmpty(task.PushUrl))
                                 {
+                                    _logger.LogDebug("执行AI 智能巡检。。。。。");
                                     var param = @$"{{
-    ""bid"": {Guid.NewGuid().ToString()},
-    ""method"": ""live_start_push"",
-    ""tid"": {Guid.NewGuid().ToString()},
-    ""timestamp"": {DateTimeOffset.Now.ToUnixTimeMilliseconds()},
-    ""data"": {{
-        ""url_type"": 1,
-        ""url"": ""rtmp://box.wisestcity.com:1935/live/7"",
-        ""video_id"": ""1581F8HGX254V00A0BUY/99-0-0/normal-0"",
-        ""video_quality"": 3
-    }}
-}}";
+                                            ""bid"": {Guid.NewGuid().ToString()},
+                                            ""method"": ""live_start_push"",
+                                            ""tid"": {Guid.NewGuid().ToString()},
+                                            ""timestamp"": {DateTimeOffset.Now.ToUnixTimeMilliseconds()},
+                                            ""data"": {{
+                                                ""url_type"": 1,
+                                                ""url"": ""rtmp://box.wisestcity.com:1935/live/7"",
+                                                ""video_id"": ""1581F8HGX254V00A0BUY/99-0-0/normal-0"",
+                                                ""video_quality"": 3
+                                                }}
+                                        }}";
+                                    _logger.LogDebug($"直播参数：{param}");
                                     var topicRequest = $"thing/product/{sn}/services";
                                     await _mqttClientManager.PublishAsync(topicRequest, param);
                                     var req = new CallAiModel { TaskId = taskAssign1.TaskId };
@@ -712,6 +725,9 @@ public class ConfigSubscribe : IJob
             case "thing/product/*/services_reply":
                 switch (method)
                 {
+                    case "live_start_push":
+                        _logger.LogDebug($"开启直播成功 {message}");
+                        break;
                     case "flighttask_prepare": // 下发任务响应
                         // 顺序处理，多余的不再处理
                         lock (_locker)
@@ -825,6 +841,29 @@ public class ConfigSubscribe : IJob
         return DateTimeOffset.FromUnixTimeSeconds(seconds).DateTime;
     }
 
+
+    public bool IsDuplicate(string messageId)
+    {
+        var now = DateTime.UtcNow;
+        // 清理过期消息
+        foreach (var kvp in _processedMessages)
+        {
+            if (now - kvp.Value > _deduplicationWindow)
+            {
+                _processedMessages.TryRemove(kvp.Key, out _);
+            }
+        }
+
+        // 检查是否已存在
+        if (_processedMessages.ContainsKey(messageId))
+        {
+            return true;
+        }
+
+        _processedMessages[messageId] = now;
+        return false;
+    }
+
     public static Dictionary<long, string> ErrorMap = new()
     {
         { 312014, "设备升级中，请勿重复操作" },