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)
{
await Redis.RPushAsync(RedisExpandKey.TaskLog,
new TaskLog()
{
VideoTaskId = long.Parse(taskId.ToString()),
CreateTime = DateTime.Now,
Message = msg
});
lock (RedisExpandKey.TaskLog)
{
var oldTaskCount = Redis.LLen(RedisExpandKey.TaskLog);
if (oldTaskCount > 100)
{
try
{
var insertData = Redis.LRange(RedisExpandKey.TaskLog, 0, 99);
taskLogDB.AsInsertable(insertData).ExecuteCommand();
//同步删除redis
Redis.LTrim(RedisExpandKey.TaskLog, 100, 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);
}
}
}
}