From 76019141f138fc615a1a7e19bb484115b04ddebd Mon Sep 17 00:00:00 2001
From: zenghongyao <873884283@qq.com>
Date: Wed, 21 May 2025 10:45:18 +0800
Subject: [PATCH 1/3] =?UTF-8?q?=E4=BC=98=E5=8C=96=E8=BF=81=E7=A7=BBHost?=
 =?UTF-8?q?=EF=BC=8C=E7=A7=BB=E9=99=A4=E4=B8=80=E4=BA=9B=E4=B8=8D=E5=BF=85?=
 =?UTF-8?q?=E8=A6=81=E7=9A=84=E7=BB=84=E4=BB=B6?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 ...CollectBusMigrationHostModule.Configure.cs | 37 ++-----------------
 .../CollectBusMigrationHostModule.cs          |  2 +-
 .../Hangfire/JobRetryLastFilter.cs            | 29 ---------------
 .../JiShe.CollectBus.Migration.Host.csproj    |  2 +
 4 files changed, 6 insertions(+), 64 deletions(-)
 delete mode 100644 web/JiShe.CollectBus.Migration.Host/Hangfire/JobRetryLastFilter.cs

diff --git a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
index 33c2b3a..5816f03 100644
--- a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
+++ b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
@@ -1,7 +1,4 @@
-//using Hangfire;
-//using Hangfire.Redis.StackExchange;
-using JiShe.CollectBus.Migration.Host.Hangfire;
-using JiShe.CollectBus.Migration.Host.HealthChecks;
+using JiShe.CollectBus.Migration.Host.HealthChecks;
 using JiShe.CollectBus.Migration.Host.Swaggers;
 using Microsoft.AspNetCore.Authentication.JwtBearer;
 using Microsoft.AspNetCore.DataProtection;
@@ -12,7 +9,6 @@ using StackExchange.Redis;
 using System.Text;
 using Volo.Abp.AspNetCore.Auditing;
 using Volo.Abp.Auditing;
-using Volo.Abp.BackgroundJobs;
 using Volo.Abp.Caching;
 using Volo.Abp.Modularity;
 
@@ -21,33 +17,6 @@ namespace JiShe.CollectBus.Migration.Host
 {
     public partial class CollectBusMigrationHostModule
     {
-        /// <summary>
-        /// Configures the hangfire.
-        /// </summary>
-        /// <param name="context">The context.</param>
-        //private void ConfigureHangfire(ServiceConfigurationContext context)
-        //{
-        //    var redisStorageOptions = new RedisStorageOptions()
-        //    {
-        //        Db = context.Services.GetConfiguration().GetValue<int>("Redis:HangfireDB")
-        //    };
-
-        //    Configure<AbpBackgroundJobOptions>(options => { options.IsJobExecutionEnabled = false; });
-
-        //    context.Services.AddHangfire(config =>
-        //    {
-        //        config.UseRedisStorage(
-        //                context.Services.GetConfiguration().GetValue<string>("Redis:Configuration"), redisStorageOptions)
-        //            .WithJobExpirationTimeout(TimeSpan.FromDays(7));
-        //        var delaysInSeconds = new[] { 10, 60, 60 * 3 }; // 重试时间间隔
-        //        const int Attempts = 3; // 重试次数
-        //        config.UseFilter(new AutomaticRetryAttribute() { Attempts = Attempts, DelaysInSeconds = delaysInSeconds });
-        //        //config.UseFilter(new AutoDeleteAfterSuccessAttribute(TimeSpan.FromDays(7)));
-        //        config.UseFilter(new JobRetryLastFilter(Attempts));
-        //    });
-        //    context.Services.AddHangfireServer();
-        //}
-
         /// <summary>
         /// Configures the JWT authentication.
         /// </summary>
@@ -179,8 +148,8 @@ namespace JiShe.CollectBus.Migration.Host
             (
                 options =>
                 {
-                    options.IsEnabled = true;
-                    options.EntityHistorySelectors.AddAllEntities();
+                    options.IsEnabled = false;
+                    //options.EntityHistorySelectors.AddAllEntities();
                     options.ApplicationName = "JiShe.CollectBus.Migration";
                 }
             );
diff --git a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
index 0f90aa7..49f311a 100644
--- a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
+++ b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
@@ -86,7 +86,7 @@ namespace JiShe.CollectBus.Migration.Host
             }
             //app.UseAuditing();
             app.UseAbpSerilogEnrichers();
-            app.UseUnitOfWork();
+            //app.UseUnitOfWork();
             //app.UseHangfireDashboard("/hangfire", new DashboardOptions
             //{
             //    IgnoreAntiforgeryToken = true
diff --git a/web/JiShe.CollectBus.Migration.Host/Hangfire/JobRetryLastFilter.cs b/web/JiShe.CollectBus.Migration.Host/Hangfire/JobRetryLastFilter.cs
deleted file mode 100644
index 83d68ae..0000000
--- a/web/JiShe.CollectBus.Migration.Host/Hangfire/JobRetryLastFilter.cs
+++ /dev/null
@@ -1,29 +0,0 @@
-using Hangfire.Common;
-using Hangfire.States;
-using Serilog;
-
-namespace JiShe.CollectBus.Migration.Host.Hangfire
-{
-    /// <summary>
-    /// 重试最后一次
-    /// </summary>
-    public class JobRetryLastFilter : JobFilterAttribute, IElectStateFilter
-    {
-        private int RetryCount { get; }
-
-        public JobRetryLastFilter(int retryCount)
-        {
-            RetryCount = retryCount;
-        }
-
-
-        public void OnStateElection(ElectStateContext context)
-        {
-            var retryAttempt = context.GetJobParameter<int>("RetryCount");
-            if (RetryCount == retryAttempt)
-            {
-                Log.Error("最后一次重试");
-            }
-        }
-    }
-}
diff --git a/web/JiShe.CollectBus.Migration.Host/JiShe.CollectBus.Migration.Host.csproj b/web/JiShe.CollectBus.Migration.Host/JiShe.CollectBus.Migration.Host.csproj
index d201582..747a5c8 100644
--- a/web/JiShe.CollectBus.Migration.Host/JiShe.CollectBus.Migration.Host.csproj
+++ b/web/JiShe.CollectBus.Migration.Host/JiShe.CollectBus.Migration.Host.csproj
@@ -46,6 +46,8 @@
     <!--<PackageReference Include="Volo.Abp.BackgroundWorkers.Hangfire" Version="8.3.3" />-->
     
     <PackageReference Include="Volo.Abp.Caching.StackExchangeRedis" Version="8.3.3" />
+    
+    <PackageReference Include="Volo.Abp.Core" Version="8.3.3" />
     <PackageReference Include="Volo.Abp.Swashbuckle" Version="8.3.3" />
     <!--<PackageReference Include="Hangfire.HttpJob" Version="3.8.5" />
     <PackageReference Include="Hangfire.MySqlStorage" Version="2.0.3" />

From 2ead6e8242d5cc4d1dc62bbb12cdba65042aac33 Mon Sep 17 00:00:00 2001
From: zenghongyao <873884283@qq.com>
Date: Wed, 21 May 2025 10:56:49 +0800
Subject: [PATCH 2/3] =?UTF-8?q?=E6=9A=82=E6=97=B6=E5=8E=BB=E6=8E=89?=
 =?UTF-8?q?=E5=AE=A1=E8=AE=A1?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../CollectBusMigrationApplicationModule.cs                   | 2 +-
 .../CollectBusMigrationHostModule.Configure.cs                | 4 ++--
 .../CollectBusMigrationHostModule.cs                          | 3 +--
 3 files changed, 4 insertions(+), 5 deletions(-)

diff --git a/services/JiShe.CollectBus.Migration.Application/CollectBusMigrationApplicationModule.cs b/services/JiShe.CollectBus.Migration.Application/CollectBusMigrationApplicationModule.cs
index 8a82b7e..3a89171 100644
--- a/services/JiShe.CollectBus.Migration.Application/CollectBusMigrationApplicationModule.cs
+++ b/services/JiShe.CollectBus.Migration.Application/CollectBusMigrationApplicationModule.cs
@@ -14,7 +14,7 @@ using Volo.Abp.Modularity;
 namespace JiShe.CollectBus.Migration;
 
 [DependsOn(
-   typeof(CollectBusDomainModule),
+   //typeof(CollectBusDomainModule),
     typeof(CollectBusMigrationApplicationContractsModule),
     typeof(AbpDddApplicationModule),
     typeof(AbpAutoMapperModule),
diff --git a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
index 5816f03..ebe0541 100644
--- a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
+++ b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.Configure.cs
@@ -148,8 +148,8 @@ namespace JiShe.CollectBus.Migration.Host
             (
                 options =>
                 {
-                    options.IsEnabled = false;
-                    //options.EntityHistorySelectors.AddAllEntities();
+                    options.IsEnabled = true;
+                    options.EntityHistorySelectors.AddAllEntities();
                     options.ApplicationName = "JiShe.CollectBus.Migration";
                 }
             );
diff --git a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
index 49f311a..c181f60 100644
--- a/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
+++ b/web/JiShe.CollectBus.Migration.Host/CollectBusMigrationHostModule.cs
@@ -42,7 +42,6 @@ namespace JiShe.CollectBus.Migration.Host
             ConfigureSwaggerServices(context, configuration);
             //ConfigureNetwork(context, configuration);
             ConfigureJwtAuthentication(context, configuration);
-            //ConfigureHangfire(context);
             ConfigureAuditLog(context);
             ConfigureCustom(context, configuration);
             ConfigureHealthChecks(context, configuration);
@@ -84,7 +83,7 @@ namespace JiShe.CollectBus.Migration.Host
                     options.DefaultModelsExpandDepth(-1);
                 });
             }
-            //app.UseAuditing();
+            app.UseAuditing();
             app.UseAbpSerilogEnrichers();
             //app.UseUnitOfWork();
             //app.UseHangfireDashboard("/hangfire", new DashboardOptions

From 2d2bb0dcc03bcb88a7ec6e8caef13fd5b2e35f45 Mon Sep 17 00:00:00 2001
From: zenghongyao <873884283@qq.com>
Date: Wed, 21 May 2025 14:08:26 +0800
Subject: [PATCH 3/3] =?UTF-8?q?=E5=A2=9E=E5=8A=A0=E6=98=AF=E5=90=A6?=
 =?UTF-8?q?=E8=BF=81=E7=A7=BB=E6=A0=87=E8=AF=86=E5=92=8C=E5=88=86=E5=BA=93?=
 =?UTF-8?q?=E6=A0=87=E8=AF=86?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 .../AnalysisData/DataStorage.cs               | 74 +++++++++++++++++--
 .../Consts/T37612012PacketItemCodeConst.cs    | 14 +++-
 2 files changed, 81 insertions(+), 7 deletions(-)

diff --git a/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs b/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs
index a997274..ea11af8 100644
--- a/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs
+++ b/protocols/JiShe.CollectBus.Protocol.T37612012/AnalysisData/DataStorage.cs
@@ -1,4 +1,5 @@
-using FreeRedis;
+using Confluent.Kafka;
+using FreeRedis;
 using JiShe.CollectBus.Common;
 using JiShe.CollectBus.Common.Consts;
 using JiShe.CollectBus.Common.Encrypt;
@@ -29,6 +30,7 @@ using Volo.Abp.DependencyInjection;
 using Volo.Abp.Guids;
 using YamlDotNet.Core.Tokens;
 using static FreeSql.Internal.GlobalFilter;
+using static IClientRPCService;
 using static JiShe.CollectBus.Common.Consts.T37612012PacketItemCodeConst;
 
 namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
@@ -211,6 +213,31 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
             if (!string.IsNullOrWhiteSpace(data.FiledName))
             { 
                 await _dbProvider.GetSessionPool(false).InsertAsync(meter);
+                // 增加标识字段
+                var meterIsSync = new TreeModelSingleMeasuringEntity<bool>()
+                {
+                    SystemName = _applicationOptions.SystemType,
+                    DeviceId = $"{data.DeviceId}",
+                    DeviceType = $"{data.DeviceType}",
+                    ProjectId = $"{data.ProjectId}",
+                    DataType = analysisBaseDto.DataType,
+                    Timestamps = meter.Timestamps, 
+                    SingleMeasuring = (IotDbFieldConst.IsSync, false)
+                };
+                await _dbProvider.GetSessionPool(false).InsertAsync(meterIsSync);
+
+                // 数据库业务ID
+                var meterIsDatabaseBusiID = new TreeModelSingleMeasuringEntity<int>()
+                {
+                    SystemName = _applicationOptions.SystemType,
+                    DeviceId = $"{data.DeviceId}",
+                    DeviceType = $"{data.DeviceType}",
+                    ProjectId = $"{data.ProjectId}",
+                    DataType = analysisBaseDto.DataType,
+                    Timestamps = meter.Timestamps,
+                    SingleMeasuring = (IotDbFieldConst.DatabaseBusiID, data.DatabaseBusiID)
+                };
+                await _dbProvider.GetSessionPool(false).InsertAsync(meterIsDatabaseBusiID);
             }
             return await Task.FromResult(true);
         }
@@ -228,6 +255,9 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
             var data = analysisBaseDto.Data!;
             List<MeterReadingTelemetryPacketInfo> meterReadingTelemetryPacketInfos = new List<MeterReadingTelemetryPacketInfo>();
             List<TreeModelSingleMeasuringEntity<T>> treeModelSingleMeasuringEntities = new List<TreeModelSingleMeasuringEntity<T>>();
+            List<TreeModelSingleMeasuringEntity<bool>> meterIsSyncs = new List<TreeModelSingleMeasuringEntity<bool>>();
+            List<TreeModelSingleMeasuringEntity<int>> meterDataBaseIDs = new List<TreeModelSingleMeasuringEntity<int>>();
+
             foreach (var item in data)
             {
                 if(!item.TimeSpan.HasValue)
@@ -320,6 +350,35 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                 if (!string.IsNullOrWhiteSpace(item.FiledName) && item.ProjectId>0)
                 {
                     treeModelSingleMeasuringEntities.Add(meter);
+
+                    // 增加标识字段
+                    var meterIsSync = new TreeModelSingleMeasuringEntity<bool>()
+                    {
+                        SystemName = _applicationOptions.SystemType,
+                        DeviceId = $"{item.DeviceId}",
+                        DeviceType = $"{item.DeviceType}",
+                        ProjectId = $"{item.ProjectId}",
+                        DataType = analysisBaseDto.DataType,
+                        Timestamps = meter.Timestamps,
+                        SingleMeasuring = (IotDbFieldConst.IsSync, false)
+                    };
+                    if(!meterIsSyncs.Any(a=> a.DataType == meterIsSync.DataType && a.ProjectId == meterIsSync.ProjectId && a.SystemName== meterIsSync.SystemName && a.DeviceId== meterIsSync.DeviceId && a.Timestamps== meterIsSync.Timestamps))
+                        meterIsSyncs.Add(meterIsSync);
+
+                    // 数据库业务ID
+                    var meterIsDatabaseBusiID = new TreeModelSingleMeasuringEntity<int>()
+                    {
+                        SystemName = _applicationOptions.SystemType,
+                        DeviceId = $"{item.DeviceId}",
+                        DeviceType = $"{item.DeviceType}",
+                        ProjectId = $"{item.ProjectId}",
+                        DataType = analysisBaseDto.DataType,
+                        Timestamps = meter.Timestamps,
+                        SingleMeasuring = (IotDbFieldConst.DatabaseBusiID, item.DatabaseBusiID)
+                    };
+                    if (!meterDataBaseIDs.Any(a => a.DataType == meterIsSync.DataType && a.ProjectId == meterIsSync.ProjectId && a.SystemName == meterIsSync.SystemName && a.DeviceId == meterIsSync.DeviceId && a.Timestamps == meterIsSync.Timestamps))
+                        meterDataBaseIDs.Add(meterIsDatabaseBusiID);
+                    
                 }
             }
             // 批量保存数据 
@@ -327,6 +386,11 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
             if (treeModelSingleMeasuringEntities.Count > 0)
             { 
                 await _dbProvider.GetSessionPool(false).BatchInsertAsync(treeModelSingleMeasuringEntities);
+
+                // 报存标识字段
+                await _dbProvider.GetSessionPool(false).BatchInsertAsync(meterIsSyncs);
+                await _dbProvider.GetSessionPool(false).BatchInsertAsync(meterDataBaseIDs);
+
             }
             return await Task.FromResult(true);
         }
@@ -371,7 +435,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                     ProjectId = $"{data.ProjectId}",
                     DataType = IOTDBDataTypeConst.Status,
                     Timestamps = timestamps,
-                    SingleMeasuring = (ConcentratorStatusFieldConst.FrameData, analysisBaseDto.ReceivedHexMessage ?? string.Empty)
+                    SingleMeasuring = (IotDbFieldConst.FrameData, analysisBaseDto.ReceivedHexMessage ?? string.Empty)
                 };
                 await _dbProvider.GetSessionPool(false).InsertAsync(treeFrameData);
 
@@ -384,7 +448,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                     ProjectId = $"{data.ProjectId}",
                     Timestamps = timestamps,
                     DataType = IOTDBDataTypeConst.Status,
-                    SingleMeasuring = (ConcentratorStatusFieldConst.RecordingTime, data.TimeSpan.HasValue ? data.TimeSpan.Value : DateTime.Now)
+                    SingleMeasuring = (IotDbFieldConst.RecordingTime, data.TimeSpan.HasValue ? data.TimeSpan.Value : DateTime.Now)
                 };
                 await _dbProvider.GetSessionPool(false).InsertAsync(treeRecordingTimeData);
             }
@@ -472,7 +536,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                         ProjectId = $"{item.ProjectId}",
                         DataType = IOTDBDataTypeConst.Status,
                         Timestamps = timestamps,
-                        SingleMeasuring = (ConcentratorStatusFieldConst.FrameData, analysisBaseDto.ReceivedHexMessage ?? string.Empty)
+                        SingleMeasuring = (IotDbFieldConst.FrameData, analysisBaseDto.ReceivedHexMessage ?? string.Empty)
                     };
 
                     await _dbProvider.GetSessionPool(false).InsertAsync(treeFrameData);
@@ -485,7 +549,7 @@ namespace JiShe.CollectBus.Protocol.T37612012.AnalysisData
                         ProjectId = $"{item.ProjectId}",
                         DataType = IOTDBDataTypeConst.Status,
                         Timestamps = timestamps,
-                        SingleMeasuring = (ConcentratorStatusFieldConst.RecordingTime, item.TimeSpan.HasValue ? item.TimeSpan.Value : DateTime.Now)
+                        SingleMeasuring = (IotDbFieldConst.RecordingTime, item.TimeSpan.HasValue ? item.TimeSpan.Value : DateTime.Now)
                     };
                     await _dbProvider.GetSessionPool(false).InsertAsync(treeRecordingTimeData);
 
diff --git a/shared/JiShe.CollectBus.Common/Consts/T37612012PacketItemCodeConst.cs b/shared/JiShe.CollectBus.Common/Consts/T37612012PacketItemCodeConst.cs
index 69e33e5..b556524 100644
--- a/shared/JiShe.CollectBus.Common/Consts/T37612012PacketItemCodeConst.cs
+++ b/shared/JiShe.CollectBus.Common/Consts/T37612012PacketItemCodeConst.cs
@@ -170,9 +170,9 @@ namespace JiShe.CollectBus.Common.Consts
 
 
         /// <summary>
-        /// 集中器状态字段
+        /// IotDB存储字段字段
         /// </summary>
-        public class ConcentratorStatusFieldConst
+        public class IotDbFieldConst
         {
 
             /// <summary>
@@ -185,6 +185,16 @@ namespace JiShe.CollectBus.Common.Consts
             /// </summary>
             public const string FrameData = "FrameData";
 
+            /// <summary>
+            /// 是否同步
+            /// </summary>
+            public const string IsSync = "IsSync";
+
+            /// <summary>
+            /// 数据库业务ID
+            /// </summary>
+            public const string DatabaseBusiID= "DatabaseBusiID";
+
         }
         #endregion