HealthMonitor/HealthMonitor.WebApi/Worker.cs

using dotnet_etcd;
using Etcdserverpb;
using Google.Protobuf.WellKnownTypes;
using HealthMonitor.Common;
using HealthMonitor.Common.helper;
using HealthMonitor.Core.Common.Extensions;
using HealthMonitor.Model.Service;
using HealthMonitor.Service.Biz;
using HealthMonitor.Service.Biz.db;
using HealthMonitor.Service.Etcd;
using HealthMonitor.Service.Sub;
using Microsoft.AspNetCore.Mvc.RazorPages;
using Microsoft.EntityFrameworkCore.Metadata.Internal;
using Newtonsoft.Json;
using Newtonsoft.Json.Linq;
using System.Reflection;
using System.Threading.Channels;
using TDengineDriver;
using TDengineTMQ;

namespace HealthMonitor.WebApi
{
    public class Worker : BackgroundService
    {
        private readonly ILogger<Worker> _logger;
        private readonly IServiceProvider _services;
        private readonly TDengineDataSubcribe _tdEngineDataSubcribe;
        private readonly PackageProcess _processor;
        private readonly TDengineService _serviceTDengine;
        private readonly EtcdService _serviceEtcd;
        private readonly HttpHelper _httpHelper = default!;
        private readonly IotWebApiService _serviceIotWebApi;

        private CancellationTokenSource _tokenSource=default!;

        public Worker(ILogger<Worker> logger, IServiceProvider services, IotWebApiService iotWebApiService, PackageProcess processor,TDengineDataSubcribe tdEngineDataSubcribe, TDengineService serviceDengine, HttpHelper httpHelper, EtcdService serviceEtcd) 
        {
            _logger = logger;
            _tdEngineDataSubcribe = tdEngineDataSubcribe;
            _services = services;
            _serviceIotWebApi = iotWebApiService;
            _processor = processor;
            _serviceEtcd = serviceEtcd; 
            _serviceTDengine = serviceDengine;
            _httpHelper = httpHelper;
        }

        public override Task StartAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("------StartAsync");
            _tokenSource = CancellationTokenSource.CreateLinkedTokenSource(cancellationToken);
            // 创建消费者
          //  _tdEngineDataSubcribe.CreateConsumer();
            return base.StartAsync(cancellationToken);
        }

        public override Task StopAsync(CancellationToken cancellationToken)
        {
            _logger.LogInformation("------StopAsync");
            _tokenSource.Cancel();  //停止工作线程
            // 关闭消费者
          //  _tdEngineDataSubcribe.CloseConsumer();
            return base.StopAsync(cancellationToken);
        }

        protected override Task ExecuteAsync(CancellationToken stoppingToken)
        {
            // var processor = _services.GetService<PackageProcess>();
            TaskFactory factory = new(_tokenSource.Token);
            factory.StartNew(async () =>
            {
                if (_tokenSource.IsCancellationRequested)
                    _logger.LogWarning("Worker exit");

                _logger.LogInformation("------ResolveAsync");
                while (!_tokenSource.IsCancellationRequested)
                {
                    // 
                    await _processor.ResolveAsync().ConfigureAwait(false);
                    //  await  _tdEngineDataSubcribe.ProcessMsg();
                }

            }, TaskCreationOptions.LongRunning);

            factory.StartNew(() =>
            {
                _logger.LogInformation("------_tdEngineDataSubcribe");
                while (!_tokenSource.IsCancellationRequested)
                {
                   
                    _tdEngineDataSubcribe.BeginListen(_tokenSource.Token);
                }
            }, TaskCreationOptions.LongRunning);

            Task.Run(() =>
                _serviceEtcd.WacthKeysWithPrefixResponseAsync($"health_moniter/schedule_push", WatchEvents)
            , stoppingToken);

            // watch
            //factory.StartNew(() =>
            //{
            //    while (!_tokenSource.IsCancellationRequested)
            //    {
            //        //_serviceEtcd.WacthKeysWithPrefixAsync($"health_moniter/schedule_push", watchEvents => WatchEvents(watchEvents));
            //        _serviceEtcd.WacthKeysWithPrefixResponseAsync($"health_moniter/schedule_push", WatchEvents);
            //    }
            //}, TaskCreationOptions.LongRunning);

            // _serviceEtcd.WacthKeysWithPrefixResponseAsync($"health_moniter/schedule_push", WatchEvents);
            return Task.Delay(1000, _tokenSource.Token);

        }

        private void WatchEvents(WatchEvent[] response)
        {
            foreach (WatchEvent e1 in response)
            {
                // Console.WriteLine($"{nameof(WatchEventsAsync)} --- {e1.Key}:{e1.Value}:{e1.Type}");

                switch (e1.Type.ToString())
                {
                    //case "Put":
                    //    // 获取时间点计算TTL
                    //    break;

                    case "Delete":
                        // TTL到了重新计算TTL，下发
                        Console.WriteLine($"--- {e1.Key}:{e1.Value}:{e1.Type}");
                        break;
                }
            }
        }


        private  void WatchEvents(WatchResponse response)
        {

            response.Events.ToList().ForEach(async e =>
            {

                switch (e.Type.ToString())
                {
                    case "Put":
                        // 获取时间点计算TTL
                        Console.BackgroundColor = ConsoleColor.Blue;
                        Console.WriteLine($"--- {e.Type}--{e.Kv.Key.ToStringUtf8()}--{e.Kv.Value.ToStringUtf8()}---{DateTime.Now}");
                        Console.BackgroundColor = ConsoleColor.Black;
                        break;

                    case "Delete":
                        // TTL到了重新计算TTL，下发
                        Console.BackgroundColor = ConsoleColor.Green;
                        Console.WriteLine($"--- {e.Type}--{e.Kv.Key.ToStringUtf8()}--{e.Kv.Value.ToStringUtf8()}---{DateTime.Now}");

                        // var key = $"health_moniter/schedule_push/imei/{bp.Serialno}";
                        var key = e.Kv.Key.ToStringUtf8();
                        var imeiDel = key.Split('/')[3];
                        var schedule_push = await _serviceEtcd.GetValAsync(key).ConfigureAwait(false);
                        if (string.IsNullOrWhiteSpace(schedule_push))
                        {
                           

                            var startTime =DateTime.Now; 
                            // 下发增量值
                            #region 定时下发增量值
                            var last= await _serviceTDengine.GetLastAsync("stb_hm_bloodpress_stats_inc", $"serialno='{imeiDel}' order by last_update desc");
                            var ts = last?[0];
                            if (DateTime.TryParse(ts?.ToString(),out DateTime newTs))
                            {
                                // 7 天有效数据
                                if (newTs.AddDays(7) > DateTime.Now)
                                {
                                    Console.WriteLine(ts);
                                    var systolic_ref_value = last?[5];
                                    var diastolic_ref_value = last?[12];
                                    var systolic_inc_value = last?[10];
                                    var diastolic_inc_value = last?[17];

                                    #region 判断是否初始化remark


                                    // 判断是否初始化remark
                                    /**
                                     *  
                                    var imei = imeiDel;
                                    var last_push = await _serviceTDengine.GetLastAsync("stb_hm_bp_push_ref_inc_value", $"serialno='{imei}' order by ts desc");
                                    if (last_push?.Count == 0)
                                    {
                                        var dataServiceBaseUrl = $"https://id.ssjlai.com/data";
                                        var DataServicePersionGet = $"{dataServiceBaseUrl}/api/GpsCard/GpsPerson/GetFirst";
                                        List<KeyValuePair<string, string>> Dataheaders = new()
                                            {
                                                new KeyValuePair<string, string>("requestId", $"{imei}")
                                            };
                                        var dataPersion = new
                                        {
                                            filters = new List<object>() { new { key = "serialno", value = $"{imei}", valueType = "string", @operator = "Equal" } },
                                            orderBys = new List<object>() { new { key = "serialno", isDesc = true } }
                                        };
                                        var resRef = await _httpHelper.HttpToPostAsync(DataServicePersionGet, dataPersion, Dataheaders).ConfigureAwait(false);
                                        var resObj = JsonConvert.DeserializeObject(resRef!) as JToken;
                                        var remark = resObj?["remarks"]?.ToString();

                                        // 修改数据库
                                        if (string.IsNullOrWhiteSpace(remark))
                                        {
                                            var newRemarkData = new
                                            {
                                                imei,
                                                time = DateTime.Now.ToString("yyyy-MM-dd HH:mm:ss"),
                                                commandValue = new
                                                {
                                                    systolicCalibrationValue = systolic_ref_value,  //收缩压标定值，值为0 表示不生效
                                                    diastolicCalibrationValue = diastolic_ref_value, //舒张压标定值，值为0表示不生效
                                                    systolicIncValue = 0, //收缩压显示增量，值为0 表示不生效
                                                    diastolicIncValue = 0   //舒张压显示增量，值为0 表示不生效
                                                }
                                            };
                                            resObj!["remarks"] = $"is_blood_press:{JsonConvert.SerializeObject(newRemarkData)}|";
                                            var DataServicePersionUpdate = $"{dataServiceBaseUrl}/api/GpsCard/GpsPerson/Update";
                                            var resUpdate = await _httpHelper.HttpToPostAsync(DataServicePersionUpdate, resObj, Dataheaders).ConfigureAwait(false);
                                            _logger.LogInformation($"更新Person数据库|{resUpdate}");

                                            //更新缓存
                                            List<KeyValuePair<string, string>> headersCache = new()
                                    {
                                        new KeyValuePair<string, string>("AuthKey", "key1")
                                    };

                                            var cacheUrl = $"http://id.ssjlai.com/webapi/api/Device/UpdatePersonInfoCache?imei={imei}";
                                            var updateCache = await _httpHelper.HttpToGetAsync(cacheUrl, headersCache);
                                            _logger.LogInformation($"更新Person缓存|{updateCache}");
                                        }
                                    }
                                    */
                                    #endregion

                                    //var bpData = new
                                    //{
                                    //    imei = imeiDel,
                                    //    //systolicCalibrationValue = last?[5],  //收缩压标定值，值为0 表示不生效
                                    //    //diastolicCalibrationValue = last?[12], //舒张压标定值，值为0表示不生效
                                    //    //systolicIncValue = last?[10], //收缩压显示增量，值为0 表示不生效
                                    //    //diastolicIncValue = last?[17]   //舒张压显示增量，值为0 表示不生效

                                    //    systolicCalibrationValue = systolic_ref_value,  //收缩压标定值，值为0 表示不生效
                                    //    diastolicCalibrationValue = diastolic_ref_value, //舒张压标定值，值为0表示不生效
                                    //    systolicIncValue = systolic_inc_value, //收缩压显示增量，值为0 表示不生效
                                    //    diastolicIncValue = diastolic_inc_value   //舒张压显示增量，值为0 表示不生效
                                    //};
                                    //var str = JsonConvert.SerializeObject(bpData);
                                    //var url = $"http://id.ssjlai.com/webapi/api/Command/SetBloodPressCalibrationConfig";
                                    //List<KeyValuePair<string, string>> headers = new()
                                    //{
                                    //    new KeyValuePair<string, string>("AuthKey", "key1")
                                    //};
                                    //var res = await _httpHelper.HttpToPostAsync(url, bpData, headers).ConfigureAwait(false);
                                    //_logger.LogInformation($"向{imeiDel}下发增量值数据:{str}，响应：{res}");
                                    //var resJToken = JsonConvert.DeserializeObject(res!) as JToken;


                                    //if (resJToken!["message"]!.ToString().Equals("ok"))

                                    BloodPressCalibrationConfigModel bpIncData = new()
                                    {

                                        Imei = imeiDel,
                                        SystolicRefValue = SafeType.SafeInt(((int)systolic_ref_value!)),  //收缩压标定值，值为0 表示不生效
                                        DiastolicRefValue = SafeType.SafeInt(((int)diastolic_ref_value!)), //舒张压标定值，值为0表示不生效
                                        SystolicIncValue = SafeType.SafeInt(((int)systolic_inc_value!)), //收缩压显示增量，值为0 表示不生效
                                        DiastolicIncValue = SafeType.SafeInt(((int)diastolic_inc_value!))   //舒张压显示增量，值为0 表示不生效
                                    };
                                    var pushedBP = await _serviceIotWebApi.SetBloodPressCalibrationConfigAsync(bpIncData).ConfigureAwait(false);
                                    if (pushedBP)
                                    {

                                        #region 保存下推记录 stb_hm_bp_push_ref_inc_value
                                        var sql = $"INSERT INTO health_monitor.hm_bp_push_ref_inc_value_{imeiDel.Substring(imeiDel.Length - 2)} " +
                                             $"USING health_monitor.stb_hm_bp_push_ref_inc_value " +
                                           $"TAGS ('{imeiDel.Substring(imeiDel.Length - 2)}')  " +
                                           $"VALUES(" +
                                               $"'{DateTime.Now:yyyy-MM-dd HH:mm:ss.fff}'," +
                                               $"'{imeiDel}'," +
                                               $"{systolic_ref_value}," +
                                               $"{diastolic_ref_value}," +
                                               $"{systolic_inc_value}," +
                                               $"{diastolic_inc_value}," +
                                               $"{false})";                                    

                                        _serviceTDengine.ExecuteInsertSQL(sql);
                                        #endregion

                                        // 注册下次下推
                                        var endTime = DateTime.Now;

                                       


#if DEBUG


                                        //long ttl = (long)((60 * 1000-(endTime-startTime).TotalMilliseconds)/1000);
                                        //await _serviceEtcd.PutValAsync(key, imeiDel,ttl, false).ConfigureAwait(false);

                                        var interval = 0;
                                        // 获取当前时间
                                        DateTime now = DateTime.Now;

                                        // 计算距离下一个$interval天后的8点的时间间隔
                                        DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, now.Hour, now.Minute + 2, 0).AddDays(interval);
                                        TimeSpan timeUntilNextRun = nextRunTime - now;

                                        // 如果当前时间已经超过了8点，将等待到明天后的8点
                                        if (timeUntilNextRun < TimeSpan.Zero)
                                        {
                                            timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromMinutes(1));
                                            nextRunTime += timeUntilNextRun;
                                        }

                                        // var ttl = timeUntilNextRun.TotalMilliseconds;
                                        long ttl = (long)((timeUntilNextRun.TotalMilliseconds - (endTime - startTime).TotalMilliseconds) / 1000);
                                        var data = new
                                        {
                                            imei = imeiDel,
                                            create_time = now.ToString("yyyy-MM-dd HH:mm:ss"),
                                            ttl,
                                            next_run_time = nextRunTime.ToString("yyyy-MM-dd HH:mm:ss")
                                        };
                                        var result = JsonConvert.SerializeObject(data);

                                        await _serviceEtcd.PutValAsync(key, result, ttl, false).ConfigureAwait(false);


#else
                                        // 每$interval天,晚上8点
                                        var interval = 1;
                                        // 获取当前时间
                                        DateTime now = DateTime.Now;

                                        // 计算距离下一个$interval天后的8点的时间间隔
                                        DateTime nextRunTime = new DateTime(now.Year, now.Month, now.Day, 20, 0, 0).AddDays(interval);
                                        TimeSpan timeUntilNextRun = nextRunTime - now;

                                        // 如果当前时间已经超过了8点，将等待到明天后的8点
                                        if (timeUntilNextRun < TimeSpan.Zero)
                                        {
                                            timeUntilNextRun = timeUntilNextRun.Add(TimeSpan.FromDays(1));
                                            nextRunTime += timeUntilNextRun;
                                        }

                                       // var ttl = timeUntilNextRun.TotalMilliseconds;
                                        long ttl = (long)((timeUntilNextRun.TotalMilliseconds-(endTime-startTime).TotalMilliseconds)/1000);
                                        var data = new
                                        {
                                            imei = imeiDel,
                                            create_time = now.ToString("yyyy-MM-dd HH:mm:ss"),
                                            ttl,
                                            next_run_time = nextRunTime.ToString("yyyy-MM-dd HH:mm:ss")
                                        };
                                        var result = JsonConvert.SerializeObject(data);
                                        await _serviceEtcd.PutValAsync(key, result, ttl, false).ConfigureAwait(false);
#endif


                                    }
                                    else
                                    {
                                        _logger.LogInformation($"错误响应，没有下推数据");
                                    }

                                }
                            }
                            else
                            {
                                _logger.LogInformation($"向{imeiDel}准备下推的数据已经失效ts,{ts}");
                            }


                            #endregion
                            //if (imeiDel.Equals("861281060086216"))
                            //{
                            //    var result = await _httpHelper.HttpToPostAsync(url, data, headers).ConfigureAwait(false);
                            //    _logger.LogInformation($"向{imeiDel}下发增量值数据:{str}，响应：{result}");
                            //    Console.WriteLine(str);

                            //}
                            //Console.BackgroundColor = ConsoleColor.Black;

                        }

                        break;
                }
             
            });
            //if (response.Events.Count == 0)
            //{
            //    Console.WriteLine(response);
            //}
            //else
            //{
            //    Console.WriteLine($"{response.Events[0].Kv.Key.ToStringUtf8()}:{response.Events.Kv.Value.ToStringUtf8()}");
            //}
        }

    }
}