Learn.VideoAnalysis/VideoAnalysisCore/Common/RedisExpand.cs

using FreeRedis;
using Microsoft.Extensions.DependencyInjection;
using System;
using System.Threading.Channels;
using System.Threading.Tasks;
using System.Xml.Linq;
using VideoAnalysisCore.AICore.ChatGPT;
using VideoAnalysisCore.AICore.FFMPGE;

//using VideoAnalysisCore.AICore.FFMPGE;
using VideoAnalysisCore.AICore.SherpaOnnx;
using VideoAnalysisCore.AICore.Whisper;
using VideoAnalysisCore.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 + "Channel:";
        /// <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>
        /// 获取枚举RedisKey
        /// </summary>
        /// <param name="e"></param>
        /// <returns></returns>
        public static string EnumKey(RedisChannelEnum e)
        {
            return ChannelKey + e.ToString();
        }

        /// <summary>
        /// 任务对象地址
        /// </summary>
        public static string Task(object taskId) => BaseKey + "Task:" + taskId;

    }
    /// <summary>
    /// redis拓展
    /// </summary>
    public class RedisExpand
    {
        /// <summary>
        /// redis 连接
        /// </summary>
        public static RedisClient Redis = new RedisClient(AppCommon.Config.Redis.ConnectionString);
        /// <summary>
        /// 初始化 redis
        /// <para>需要在初始化配置文件时候调用</para>
        /// </summary>
        public static void Init()
        {
            Console.WriteLine("初始化 redis");
            Redis.Serialize = obj => System.Text.Json.JsonSerializer.Serialize(obj);
            Redis.Deserialize = (json, type) => System.Text.Json.JsonSerializer.Deserialize(json, type);
            InitChannel();
        }
        /// <summary>
        /// 将任务 插入 队列
        /// </summary>
        /// <param name="enum">枚举</param>
        /// <param name="taskId">任务id</param>
        public static void InsertChannel(RedisChannelEnum @enum, object taskId)
        {
            if (Redis is null) throw new Exception("redis未初始化");

            var startTime = Redis.HMGet<Dictionary<RedisChannelEnum, DateTime>>(RedisExpandKey.Task(taskId), "StartTime").FirstOrDefault();
            if (startTime is null)
                startTime = new Dictionary<RedisChannelEnum, DateTime>();
            if (!startTime.ContainsKey(@enum))
                startTime.Add(@enum, DateTime.Now);
            else
                startTime[@enum] = DateTime.Now;


            Redis.HMSet(RedisExpandKey.Task(taskId), "StartTime", startTime);
            Redis.LPush(RedisExpandKey.EnumKey(@enum), taskId);

        }

        /// <summary>
        /// 初始化 队列 任务
        /// </summary>
        public static void InitChannel()
        {
            if (Redis is null) throw new Exception("redis未初始化");

            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.DownloadFile),
                (msg) => { TouchChannel(RedisChannelEnum.DownloadFile, msg, DownloadFile.RunTask); });

            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.SeparateAudio),
                (msg) => { TouchChannel(RedisChannelEnum.SeparateAudio, msg, FFMPGEHandle.Audio2WAV16KAsync); });

            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.ParsingCaptions),
                (msg) => { TouchChannel(RedisChannelEnum.ParsingCaptions, msg, SenseVoice.RunTask); });
            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.ParsingSpeaker),
                (msg) => { TouchChannel(RedisChannelEnum.ParsingSpeaker, msg, Speaker.Run); });
            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.ChatModelAnalysis),
                (msg) =>
                {
                    TouchChannel(RedisChannelEnum.ChatModelAnalysis, msg,
                    (task) =>
                    {
                        using var scope = AppCommon.Services?.CreateScope();
                        if (scope is null || scope.ServiceProvider.GetService<IBserGPT>() is null)
                            throw new Exception("IBserGPT 未注入");
                        else
                            return scope.ServiceProvider.GetService<IBserGPT>()?.CallGPT(task)??Task.CompletedTask;
                    });
                });
            Redis.SubscribeList(RedisExpandKey.EnumKey(RedisChannelEnum.CallBackSystem),
               (msg) => { TouchChannel(RedisChannelEnum.ParsingSpeaker, msg); });


        }
        /// <summary>
        /// 
        /// </summary>
        /// <param name="key"></param>
        /// <param name="msg"></param>
        /// <param name="action"></param>
        public static async void TouchChannel(RedisChannelEnum key, string msg, Func<string, Task> action = null)
        {
            if (msg is null) return;
            Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-> " + key + " " + msg);
            if (action is not null)
            {
                Redis.HMSet(RedisExpandKey.Task(msg), "LastEnum", key);
                await action(msg);
            }
            else
            {
                Console.WriteLine(key + "  任务函数 未实现");
            }
        }

    }
}