Learn.VideoAnalysis/VideoAnalysisCore/Common/RedisExpand.cs

using FreeRedis;
using FreeRedis.Internal;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.IdentityModel.Tokens;
using NetTaste;
using Newtonsoft.Json.Schema;
using SqlSugar.IOC;
using System;
using System.Security.Cryptography;
using System.Text.Json;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Collections.Concurrent;
using System.Xml.Linq;
using UserCenter.Model.Enum;
using VideoAnalysisCore.AICore.FFMPGE;
using VideoAnalysisCore.AICore.GPT;
using VideoAnalysisCore.AICore.GPT.Dto;


//using VideoAnalysisCore.AICore.FFMPGE;
using VideoAnalysisCore.AICore.SherpaOnnx;
using VideoAnalysisCore.AICore.Whisper;
using VideoAnalysisCore.Model;
using VideoAnalysisCore.Model.Dto;
using VideoAnalysisCore.Model.Enum;

namespace VideoAnalysisCore.Common
{
    /// <summary>
    /// redis key
    /// </summary>
    public static class RedisExpandKey
    {

        /// <summary>
        /// 基础key
        /// </summary>
        public const string BaseKey = "VideoAnalysis:";
        /// <summary>
        /// 基础Channel key
        /// </summary>
        public const string ChannelKey = BaseKey + "TaskChannel";
        /// <summary>
        /// TidySlide 工作流 Channel key
        /// </summary>
        public const string TidySlideChannelKey = BaseKey + "TidySlideTaskChannel";
        /// <summary>
        /// 下载文件
        /// </summary>
        public const string DownloadFile = ChannelKey + "DownloadFile";
        /// <summary>
        /// 分离音频
        /// </summary>
        public const string SeparateAudio = ChannelKey + "SeparateAudio";
        /// <summary>
        /// 解析字幕
        /// </summary>
        public const string ParsingCaptions = ChannelKey + "ParsingCaptions";
        /// <summary>
        /// 解析说话人
        /// </summary>
        public const string ParsingSpeaker = ChannelKey + "ParsingSpeaker";
        /// <summary>
        /// Chat模型分析
        /// </summary>
        public const string ChatModelAnalysis = ChannelKey + "ChatModelAnalysis";

        /// <summary>
        /// 任务数组
        /// </summary>
        public const string TaskArr = BaseKey + "TaskArr";
        /// <summary>
        /// 任务日志缓存
        /// </summary>
        public static string TaskLog => BaseKey + "TaskLog:" + AppCommon.Config.ID;


        /// <summary>
        /// 任务对象地址
        /// </summary>
        public static string Task(object taskId) => BaseKey + "Info:" + taskId;
        public static string IDTask => BaseKey + "Services:" + AppCommon.Config.ID;
        public static string DeviceTaskLog(string deviceId) => BaseKey + "Services:" + deviceId;
        /// <summary>
        /// 在线设备Key集合 (已弃用，直接扫描 Heartbeat)
        /// </summary>
        // public static string OnlineDevices => BaseKey + "OnlineDevices"; 
        /// <summary>
        /// 设备心跳Key前缀 (VideoAnalysis:Heartbeat:{DeviceId})
        /// </summary>
        public static string DeviceHeartbeat(string deviceId) => BaseKey + "Heartbeat:" + deviceId;
        public static string TaskGPT(object taskId) => BaseKey + "GPTCached:" + taskId;
        /// <summary>
        /// 初始化 redis
        /// <para>需要在初始化配置文件时候调用</para>
        /// </summary>
        public static void AddTaskSubscribe(this IServiceCollection service)
        {

            Console.WriteLine($"{DateTime.Now}=>初始化 Redis任务队列");
            service.AddSingleton<RedisInit>();
        }
        /// <summary>
        /// redis连接拓展(包含消息队列任务)
        /// </summary>
        /// <param name="service"></param>
        public static void AddRedisExpand(this IServiceCollection service,bool needWorkflow=true)
        {
            Console.WriteLine($"{DateTime.Now}=>初始化 Redis");
            var redis = new RedisClient(AppCommon.Config.Redis.ConnectionString);
            redis.Serialize = obj =>
                JsonSerializer.Serialize(obj);
            redis.Deserialize = (json, type) =>
                JsonSerializer.Deserialize(json, type);
            service.AddSingleton(redis);
            service.AddSingleton<RedisManager>();
            if (needWorkflow)
            {
                service.AddVideoSliceWorkflow();
                service.AddTidySlideWorkflow();
            }
            
            // 注册心跳 Job
            // service.AddTransient<DeviceHeartbeatJob>(); // 迁移到 CoravelExpand 中统一管理
            // service.AddTransient<TaskFileClearJob>();
            // service.AddTransient<ClearAllCacheJob>();
        }
    }

    public class RedisInit
    {

        public RedisInit(IServiceProvider serviceProvider)
        {
            serviceProvider.GetService<VideoSliceWorkflowInit>();
            serviceProvider.GetService<TidySlideWorkflowInit>();
            // serviceProvider.GetService<RedisManager>().InitChannel(); // 已废弃，由各工作流自行初始化
        }
    }
    /// <summary>
    /// redis拓展
    /// </summary>
    public class RedisManager
    {
        public static bool StopTask { get; set; } = false;
        private static Task? _workerTask;

        public RedisClient Redis { get; set; }

        public RedisManager(RedisClient redis)
        {
            Redis = redis;
        }

        public void JoinQueue(string ChannelKey,params long[] taskIds)
        {
            if (taskIds is null || taskIds.Length == 0)
                return;
            // 直接批量推入，避免循环和事务开销
            var d = taskIds.Select(s => (object)s).ToArray();
            Redis.LPush(ChannelKey, d);
        }
        /// <summary>
        /// 缓存GPT任务缓存
        /// </summary>
        /// <param name="taskId"></param>
        public void SetTaskGPTCached(object taskId, string time, object? data)
        {
            Redis.Set(RedisExpandKey.TaskGPT(taskId) + ":" + time, data, timeoutSeconds: 3600 * 24);
        }
    }
}