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.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 { /// /// redis key /// public static class RedisExpandKey { /// /// 基础key /// public const string BaseKey = "VideoAnalysis:"; /// /// 基础Channel key /// public const string ChannelKey = BaseKey + "TaskChannel"; /// /// 下载文件 /// public const string DownloadFile = ChannelKey + "DownloadFile"; /// /// 分离音频 /// public const string SeparateAudio = ChannelKey + "SeparateAudio"; /// /// 解析字幕 /// public const string ParsingCaptions = ChannelKey + "ParsingCaptions"; /// /// 解析说话人 /// public const string ParsingSpeaker = ChannelKey + "ParsingSpeaker"; /// /// Chat模型分析 /// public const string ChatModelAnalysis = ChannelKey + "ChatModelAnalysis"; /// /// 任务数组 /// public const string TaskArr = BaseKey + "TaskArr"; /// /// 任务日志缓存 /// public static string TaskLog => BaseKey + "TaskLog:" + AppCommon.Config.ID; /// /// 任务对象地址 /// public static string Task(object taskId) => BaseKey + "Info:" + taskId; public static string IDTask => BaseKey + "Services:" + AppCommon.Config.ID; public static string TaskGPT(object taskId) => BaseKey + "GPTCached:" + taskId; /// /// 初始化 redis /// 需要在初始化配置文件时候调用 /// public static void AddTaskSubscribe(this IServiceCollection service) { Console.WriteLine($"{DateTime.Now}=>初始化 Redis任务队列"); service.AddSingleton(); } public static void AddRedisExpand(this IServiceCollection service) { 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(); } } public class RedisInit { public FFMPGEHandle FFMPGE { get; set; } public SenseVoice senseVoice { get; set; } public RedisManager redisManager { get; set; } public RedisInit(FFMPGEHandle fFMPGE, SenseVoice senseVoice, RedisManager redisManager) { FFMPGE = fFMPGE; this.senseVoice = senseVoice; this.redisManager = redisManager; Init(); redisManager.InitChannel(); } public void Init() { var SubscribeList = RedisManager.SubscribeList; SubscribeList.Add(RedisChannelEnum.排队中, async (task) => { await Task.CompletedTask; }); SubscribeList.Add(RedisChannelEnum.下载文件, async (task) => { using var scope = AppCommon.Services?.CreateScope(); if (scope is null || scope.ServiceProvider.GetService() is null) throw new Exception("DownloadFile 未注入"); else await scope.ServiceProvider.GetService()?.RunTask(task); }); SubscribeList.Add(RedisChannelEnum.分离音频, FFMPGE.RunAsync); SubscribeList.Add(RedisChannelEnum.解析字幕, senseVoice.RunTask); //SubscribeList.Add(RedisChannelEnum.解析说话人,Speaker.Run); SubscribeList.Add(RedisChannelEnum.AI课程类型, async (task) => { using var scope = AppCommon.Services?.CreateScope(); if (scope is null || scope.ServiceProvider.GetService() is null) throw new Exception("IBserGPT 未注入"); else await scope.ServiceProvider.GetService()?.GetVideoType(task); }); SubscribeList.Add(RedisChannelEnum.AI模型分析, async (task) => { using var scope = AppCommon.Services?.CreateScope(); if (scope is null || scope.ServiceProvider.GetService() is null) throw new Exception("IBserGPT 未注入"); else await scope.ServiceProvider?.GetService()?.GetKnow(task); }); SubscribeList.Add(RedisChannelEnum.AI分析试题, async (task) => { using var scope = AppCommon.Services?.CreateScope(); if (scope is null || scope.ServiceProvider.GetService() is null) throw new Exception("IBserGPT 未注入"); else await scope.ServiceProvider?.GetService()?.GetVideoQuestion(task); }); SubscribeList.Add(RedisChannelEnum.结束任务, redisManager.TaskEnd); } } /// /// redis拓展 /// public class RedisManager { public static bool StopTask { get; set; } = false; public static Dictionary> SubscribeList = new Dictionary>(); /// /// 队列池 /// static SubscribeListObject? Subscribe; public RedisClient Redis { get; set; } public Repository videoTaskDB { get; set; } public Repository taskLogDB { get; set; } public RedisManager(RedisClient redis, Repository videoTaskDB, Repository taskLogDB) { Redis = redis; this.videoTaskDB = videoTaskDB; this.taskLogDB = taskLogDB; } /// /// 缓存GPT任务缓存 /// /// public void SetTaskGPTCached(object taskId, string time, object? data) { Redis.Set(RedisExpandKey.TaskGPT(taskId) + ":" + time, data, timeoutSeconds: 3600 * 24); } /// /// 加入到消费队列 /// /// public void JoinQueue(params long[] taskIds) { //事务 if (taskIds is null || taskIds.Length == 0) return; using (var tran = Redis.Multi()) { foreach (var item in taskIds) tran.LPush(RedisExpandKey.ChannelKey, item); tran.Exec(); } } /// /// 添加日志 /// /// 任务id /// 内容 public async Task AddTaskLog(object taskId, string msg) { #if DEBUG Console.WriteLine($"{DateTime.Now.ToString("MM-dd HH:mm:ss")} => {taskId} \r\n{msg}\r\n"); #endif await Redis.RPushAsync(RedisExpandKey.TaskLog, new TaskLog() { VideoTaskId = long.Parse(taskId.ToString()), CreateTime = DateTime.Now, Message = msg }); var count = 50; lock (RedisExpandKey.TaskLog) { var oldTaskCount = Redis.LLen(RedisExpandKey.TaskLog); if (oldTaskCount > count) { try { var insertData = Redis.LRange(RedisExpandKey.TaskLog, 0, count -1); taskLogDB.AsInsertable(insertData).ExecuteCommand(); //同步删除redis Redis.LTrim(RedisExpandKey.TaskLog, count, 1000); } catch (Exception ex) { Console.WriteLine("写入任务日志出错" + "\r\n" + ex.Message + "\r\n" + ex.StackTrace); } } } } /// /// 获取任务进度 /// /// public float GetTaskProgress(object taskId) { return Redis.HMGet(RedisExpandKey.Task(taskId), "Progress")[0]; } /// /// 设置任务进度 /// /// 进度百分比 /// public void SetTaskProgress(object taskId, object p) { Redis.HMSet(RedisExpandKey.Task(taskId), "Progress", p.ToString()); } /// /// 将任务 插入 队列 /// /// 枚举 /// 任务id public async Task InsertChannel(RedisChannelEnum @enum, object taskId) { if (taskId is null) throw new Exception("taskId为空"); if (Redis is null) throw new Exception("redis未初始化"); //设置任务Redis缓存过期时间 var startTime = Redis.HMGet>(RedisExpandKey.Task(taskId), "StartTime").FirstOrDefault(); if (startTime is null) startTime = new Dictionary(); if (!SubscribeList.ContainsKey(@enum)) throw new Exception(@enum + " 未实现"); var tId = taskId.ToString(); await AddTaskLog(tId, "-------------> 开始执行任务 "); try { while (true) { if (!startTime.ContainsKey(@enum)) startTime.Add(@enum, DateTime.Now); else startTime[@enum] = DateTime.Now; Redis.HMSet(RedisExpandKey.Task(taskId), "StartTime", startTime); await TouchChannel(@enum, tId, SubscribeList[@enum]); //await SubscribeList[@enum](tId); var e = @enum.NextEnum(); if (e is null) break; @enum = e.Value; } } catch (Exception ex) { await SetTaskErrorMessage(long.Parse(tId), ex); } Redis.Expire(RedisExpandKey.Task(taskId), 60 * 60 * 24 * 15); } public async Task TaskEnd(string task) { var tId = long.Parse(task); //var gptRes = (await Redis // .HMGetAsync(RedisExpandKey.Task(task), "ChatAnalysis")).FirstOrDefault(); //if (gptRes is null) // throw new Exception("未能读取到GPT处理结果"); //删除任务执行状态 await Redis.LRemAsync(RedisExpandKey.IDTask, 1, task); var taskData = await videoTaskDB .GetFirstAsync(s => s.Id == tId); if (taskData.Captions == "[]") taskData.Captions = (await Redis.HMGetAsync(RedisExpandKey.Task(task), "Captions")).First(); //if (taskData.Speaker == "[]") // taskData.Speaker = (await Redis.HMGetAsync(RedisExpandKey.Task(task), "Speaker"))?.FirstOrDefault()??"[]"; //未使用结果暂时屏蔽 //taskData.ChatAnalysis = JsonSerializer.Serialize(gptRes); taskData.ChatAnalysisScore = 0; taskData.ErrorMessage = string.Empty; taskData.LastEnum = RedisChannelEnum.结束任务; taskData.EndTime = DateTime.Now; await videoTaskDB.AsUpdateable(taskData) .UpdateColumns(it => new { //it.ChatAnalysis, it.Captions, it.Speaker, it.ChatAnalysisScore, it.ErrorMessage, it.TotalTokens, it.LastEnum, it.EndTime }).ExecuteCommandAsync(); try { await ExpandFunction.DeleteTaskFileAsync(tId, this); } catch (Exception) { throw; } //NewTask(); } /// /// 初始化 队列 任务 /// public async Task InitChannel() { if (Redis is null) throw new Exception("redis未初始化"); //处理之前程序结束前未能执行完的情况 var oldTaskCount = Redis.LLen(RedisExpandKey.IDTask); //重试任务并发过多可能会导致程序崩溃 // 未能重新分析的中断任务 则单独开一个网页来处理 if (oldTaskCount > 0 ) { var oldTaskArr = Redis.LRange(RedisExpandKey.IDTask, 0, oldTaskCount); //不自动清理未完成任务 等待执行完毕/失败后自动清理 //Redis.LTrim(RedisExpandKey.IDTask, 1, 0);//删除 redis 列表 //最多执行5个上次中断/或者未完成的任务 foreach (var oldTask in oldTaskArr.Take(5)) { _ = Task.Run(async () => { try { await AddTaskLog(oldTask, DateTime.Now.ToString("HH:mm:ss") + "-------------> 接收上次未完成任务 " + oldTask); await ClearTaskError(long.Parse(oldTask)); var lastEnum = (await Redis.HMGetAsync(RedisExpandKey.Task(oldTask), "LastEnum")).FirstOrDefault(); await InsertChannel(lastEnum, oldTask); } catch (Exception ex) { await SetTaskErrorMessage(long.Parse(oldTask), ex); throw; } }); } } else { Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-------------> 接收新任务!"); ReceivingTaskAsync(); } } /// /// 停止接收新任务 /// public void StopTaskAsync() { StopTask = true; try { //取消接收任务监听 Subscribe?.Dispose(); } catch (Exception) { throw; } } /// /// 开始接收新任务 /// public void RestartTask() { StopTask = false; NewTask(); } /// /// 重新执行新任务 /// /// public void NewTask() { // 取消 消费机的任务订阅 if (StopTask) { Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-------------> 接收任务已经暂停 "); return; } ReceivingTaskAsync(); } /// /// 重新接收新任务 /// public void ReceivingTaskAsync() { if (AppCommon.Config.TaskSetting.IS_Server) { Console.WriteLine($"{DateTime.Now} =>服务端不接收任务"); return; } lock (Redis) { if (Subscribe?.IsUnsubscribed == false)//排除重试机制后 多次接收任务导致内存泄露 return; Subscribe = Redis.SubscribeList(RedisExpandKey.ChannelKey, async (taskId) => { if (taskId is null) return; Subscribe?.Dispose();//取消接收任务监听 Redis.LPush(RedisExpandKey.IDTask, taskId); await AddTaskLog(taskId, "-------------> 接收到任务 "); await InsertChannel(RedisChannelEnum.下载文件, taskId); }); } } /// /// 写入任务异常 /// /// /// /// public async Task SetTaskErrorMessage(long taskID, Exception? ex) { var error = string.Empty; if (ex != null) { await Redis.LRemAsync(RedisExpandKey.IDTask, 1, taskID.ToString()); //执行任务时出现异常 error = ex.Message + ex.StackTrace; await AddTaskLog(taskID, $""" 出现异常 {ex.Message} {ex.StackTrace} """); //清除失败任务 重新接收任务 NewTask(); } return await SetTaskError(taskID, error); } /// /// 清除 任务的错误信息 /// /// /// public async Task ClearTaskError(long taskID) => await SetTaskError(taskID, string.Empty); /// /// 修改任务的错误信息 /// /// /// /// public async Task SetTaskError(long taskID, string? error) { var vDB = AppCommon.Services.GetService>(); Redis.HMSet(RedisExpandKey.Task(taskID), "ErrorMessage", error); return await vDB.CopyNew().AsUpdateable() .SetColumns(it => it.ErrorMessage == error)//SetColumns是可以叠加的 写2个就2个字段赋值 .Where(it => it.Id == taskID) .ExecuteCommandAsync() == 1; } /// /// 触发 /// /// /// /// public async Task TouchChannel(RedisChannelEnum key, string taskId, Func action = null) { if (taskId is null) return; var tID = long.Parse(taskId); if (action is not null) { var tryCount = 1; for (int i = 0; i < tryCount; i++) { await AddTaskLog(taskId, " 开始执行 " + key + " " + taskId); try { Redis.HMSet(RedisExpandKey.Task(taskId), "LastEnum", key); Redis.HMSet(RedisExpandKey.Task(taskId), "Progress", 0); var vDB = AppCommon.Services.GetService>(); await vDB.CopyNew().AsUpdateable() .SetColumns(it => it.LastEnum == key) .Where(it => it.Id == tID) .ExecuteCommandAsync(); await action(taskId); return; } catch (Exception ex) { await AddTaskLog(taskId, $""" 出现异常 {ex.Message} {ex.StackTrace} """); Thread.Sleep(1000); await AddTaskLog(taskId, "稍后后重试." + key + " " + taskId); if (i + 1 == tryCount) throw; } } } else { await AddTaskLog(taskId, "任务函数 未实现." + key); } } } }