420 lines
16 KiB
C#
420 lines
16 KiB
C#
using FreeRedis;
|
|
using FreeRedis.Internal;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
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;
|
|
using static System.Runtime.InteropServices.JavaScript.JSType;
|
|
|
|
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>
|
|
/// 下载文件
|
|
/// </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 Task(object taskId) => BaseKey + "Task:" + taskId;
|
|
public static string IDTask => BaseKey + "Services:" + AppCommon.Config.ID;
|
|
public static string TaskGPT(object taskId) => Task(taskId) + ":GPTCached";
|
|
|
|
}
|
|
/// <summary>
|
|
/// redis拓展
|
|
/// </summary>
|
|
public static class RedisExpand
|
|
{
|
|
|
|
/// <summary>
|
|
/// redis 连接
|
|
/// </summary>
|
|
public static RedisClient Redis = new RedisClient(AppCommon.Config.Redis.ConnectionString);
|
|
public static Dictionary<RedisChannelEnum, Func<string,Task>> SubscribeList = new Dictionary<RedisChannelEnum, Func<string, Task>>();
|
|
/// <summary>
|
|
/// 队列池
|
|
/// </summary>
|
|
static SubscribeListObject? Subscribe;
|
|
|
|
/// <summary>
|
|
/// 初始化 redis
|
|
/// <para>需要在初始化配置文件时候调用</para>
|
|
/// </summary>
|
|
public static void AddRedisExpand(this IServiceCollection service)
|
|
{
|
|
Console.WriteLine($"{DateTime.Now}=>初始化 Redis");
|
|
Redis.Serialize = obj => JsonSerializer.Serialize(obj);
|
|
Redis.Deserialize = (json, type) => JsonSerializer.Deserialize(json, type);
|
|
Task.Run(() =>
|
|
{
|
|
Thread.Sleep(1000 * 10);
|
|
InitChannel();
|
|
});
|
|
}
|
|
|
|
/// <summary>
|
|
/// 缓存GPT任务缓存
|
|
/// </summary>
|
|
/// <param name="taskId"></param>
|
|
public static void SetTaskGPTCached(object taskId,string time, object? data)
|
|
{
|
|
Redis.Set(RedisExpandKey.TaskGPT(taskId) + ":" + time, data, 3600 * 24);
|
|
}
|
|
/// <summary>
|
|
/// 加入到消费队列
|
|
/// </summary>
|
|
/// <param name="taskIds"></param>
|
|
public static 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();
|
|
}
|
|
}
|
|
/// <summary>
|
|
/// 获取任务进度
|
|
/// </summary>
|
|
/// <param name="taskId"></param>
|
|
public static float GetTaskProgress(object taskId)
|
|
{
|
|
return Redis.HMGet<float>(RedisExpandKey.Task(taskId), "Progress")[0];
|
|
}
|
|
/// <summary>
|
|
/// 设置任务进度
|
|
/// </summary>
|
|
/// <param name="p">进度百分比</param>
|
|
/// <param name="taskId"></param>
|
|
public static void SetTaskProgress(object taskId, object p)
|
|
{
|
|
Redis.HMSet(RedisExpandKey.Task(taskId), "Progress", p.ToString());
|
|
|
|
}
|
|
/// <summary>
|
|
/// 将任务 插入 队列
|
|
/// </summary>
|
|
/// <param name="enum">枚举</param>
|
|
/// <param name="taskId">任务id</param>
|
|
public static async Task InsertChannel(RedisChannelEnum @enum, object taskId)
|
|
{
|
|
if (taskId is null) throw new Exception("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 (!SubscribeList.ContainsKey(@enum))
|
|
throw new Exception(@enum + " 未实现");
|
|
var tId = taskId.ToString();
|
|
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)taskId, ex);
|
|
}
|
|
}
|
|
|
|
public static async Task TaskEnd(string task)
|
|
{
|
|
var tId = long.Parse(task);
|
|
//var gptRes = (await Redis
|
|
// .HMGetAsync<TaskRes>(RedisExpandKey.Task(task), "ChatAnalysis")).FirstOrDefault();
|
|
//if (gptRes is null)
|
|
// throw new Exception("未能读取到GPT处理结果");
|
|
//删除任务执行状态
|
|
await Redis.HDelAsync(RedisExpandKey.IDTask,task);
|
|
var taskData = await DbScoped.Sugar.Queryable<VideoTask>()
|
|
.FirstAsync(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 DbScoped.Sugar.Updateable(taskData)
|
|
.UpdateColumns(it => new
|
|
{
|
|
//it.ChatAnalysis,
|
|
it.Captions,
|
|
it.Speaker,
|
|
it.ChatAnalysisScore,
|
|
it.ErrorMessage,
|
|
it.TotalTokens,
|
|
it.LastEnum,
|
|
it.EndTime
|
|
}).ExecuteCommandAsync();
|
|
|
|
//NewTask();
|
|
}
|
|
|
|
/// <summary>
|
|
/// 初始化 队列 任务
|
|
/// </summary>
|
|
public static async void InitChannel()
|
|
{
|
|
if (Redis is null) throw new Exception("redis未初始化");
|
|
|
|
SubscribeList.Add(RedisChannelEnum.下载文件, (task) =>
|
|
{
|
|
using var scope = AppCommon.Services?.CreateScope();
|
|
if (scope is null || scope.ServiceProvider.GetService<DownloadFile>() is null)
|
|
throw new Exception("DownloadFile 未注入");
|
|
else
|
|
return scope.ServiceProvider.GetService<DownloadFile>()?.RunTask(task) ?? Task.CompletedTask;
|
|
});
|
|
SubscribeList.Add(RedisChannelEnum.分离音频, FFMPGEHandle.RunAsync);
|
|
SubscribeList.Add(RedisChannelEnum.解析字幕, SenseVoice.RunTask);
|
|
//SubscribeList.Add(RedisChannelEnum.解析说话人,Speaker.Run);
|
|
SubscribeList.Add(RedisChannelEnum.AI课程类型,
|
|
(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>()?.GetVideoType(task) ?? Task.CompletedTask;
|
|
});
|
|
SubscribeList.Add(RedisChannelEnum.AI模型分析, (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>()?.GetKnow(task) ?? Task.CompletedTask;
|
|
});
|
|
SubscribeList.Add(RedisChannelEnum.AI分析试题, (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>()?.GetVideoQuestion(task) ?? Task.CompletedTask;
|
|
});
|
|
SubscribeList.Add(RedisChannelEnum.结束任务, TaskEnd);
|
|
|
|
ReceivingTaskAsync();
|
|
|
|
}
|
|
|
|
/// <summary>
|
|
/// 重新执行新任务
|
|
/// </summary>
|
|
/// <returns></returns>
|
|
public static async void NewTaskAsync()
|
|
{
|
|
await Redis.DelAsync(RedisExpandKey.IDTask);
|
|
ReceivingTaskAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// 重新接收新任务
|
|
/// </summary>
|
|
public static void ReceivingTaskAsync()
|
|
{
|
|
if (AppCommon.Config.TaskSetting.IS_Server)
|
|
{
|
|
Console.WriteLine($"{DateTime.Now} =>服务端不接收任务");
|
|
return;
|
|
}
|
|
Task.Run(async () =>
|
|
{
|
|
var oldTask = await Redis.GetAsync(RedisExpandKey.IDTask);
|
|
if (!string.IsNullOrEmpty(oldTask))
|
|
{
|
|
Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-------------> 接收重试任务 " + oldTask);
|
|
await ClearTaskError(long.Parse(oldTask));
|
|
var lastEnum = (await Redis.HMGetAsync<RedisChannelEnum>(RedisExpandKey.Task(oldTask), "LastEnum")).FirstOrDefault();
|
|
await InsertChannel(lastEnum, oldTask);
|
|
return;
|
|
}
|
|
if (Subscribe?.IsUnsubscribed == false)//排除重试机制后 多次接收任务导致内存泄露
|
|
return;
|
|
Subscribe = Redis.SubscribeList(RedisExpandKey.ChannelKey, async (taskId) =>
|
|
{
|
|
if (taskId is null) return;
|
|
Subscribe?.Dispose();//取消接收任务监听
|
|
//存储当前机器的任务
|
|
Redis.Set(RedisExpandKey.IDTask, taskId);
|
|
Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-------------> 接收到任务 " + taskId);
|
|
await InsertChannel(RedisChannelEnum.下载文件, taskId);
|
|
});
|
|
});
|
|
|
|
}
|
|
|
|
/// <summary>
|
|
/// 写入任务异常
|
|
/// </summary>
|
|
/// <param name="taskID"></param>
|
|
/// <param name="ex"></param>
|
|
/// <returns></returns>
|
|
public static async Task<bool> SetTaskErrorMessage(long taskID, Exception? ex)
|
|
{
|
|
var error = string.Empty;
|
|
if (ex != null)
|
|
{
|
|
//执行任务时出现异常
|
|
error = ex.Message + ex.StackTrace;
|
|
Console.WriteLine("====================[出现异常]====================");
|
|
Console.WriteLine(ex.Message);
|
|
Console.WriteLine(ex.StackTrace);
|
|
Console.WriteLine("==============================================");
|
|
//清除失败任务 重新接收任务
|
|
NewTaskAsync();
|
|
}
|
|
return await SetTaskError(taskID, error);
|
|
}
|
|
/// <summary>
|
|
/// 清除 任务的错误信息
|
|
/// </summary>
|
|
/// <param name="taskID"></param>
|
|
/// <returns></returns>
|
|
public static async Task<bool> ClearTaskError(long taskID) =>await SetTaskError(taskID, string.Empty);
|
|
/// <summary>
|
|
/// 修改任务的错误信息
|
|
/// </summary>
|
|
/// <param name="taskID"></param>
|
|
/// <param name="error"></param>
|
|
/// <returns></returns>
|
|
public static async Task<bool> SetTaskError(long taskID, string? error)
|
|
{
|
|
Redis.HMSet(RedisExpandKey.Task(taskID), "ErrorMessage", error);
|
|
return await DbScoped.Sugar.Updateable<VideoTask>()
|
|
.SetColumns(it => it.ErrorMessage == error)//SetColumns是可以叠加的 写2个就2个字段赋值
|
|
.Where(it => it.Id == taskID)
|
|
.ExecuteCommandAsync() == 1;
|
|
}
|
|
|
|
/// <summary>
|
|
/// 触发
|
|
/// </summary>
|
|
/// <param name="key"></param>
|
|
/// <param name="taskId"></param>
|
|
/// <param name="action"></param>
|
|
public static async Task TouchChannel(RedisChannelEnum key, string taskId, Func<string, Task> 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++)
|
|
{
|
|
Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-> 开始执行 " + key + " " + taskId);
|
|
try
|
|
{
|
|
Redis.HMSet(RedisExpandKey.Task(taskId), "LastEnum", key);
|
|
Redis.HMSet(RedisExpandKey.Task(taskId), "Progress", 0);
|
|
lock (Redis)
|
|
{
|
|
DbScoped.SugarScope.Updateable<VideoTask>()
|
|
.SetColumns(it => it.LastEnum == key)
|
|
.Where(it => it.Id == tID)
|
|
.ExecuteCommand();
|
|
|
|
}
|
|
await action(taskId);
|
|
return;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
Console.WriteLine("====================[出现异常]====================");
|
|
Console.WriteLine(ex.Message);
|
|
Console.WriteLine(ex.StackTrace);
|
|
Console.WriteLine("==============================================");
|
|
Thread.Sleep(1000);
|
|
Console.WriteLine(DateTime.Now.ToString("HH:mm:ss") + "-> 稍后后重试." + key + " " + taskId);
|
|
if (i+1== tryCount)
|
|
throw;
|
|
}
|
|
}
|
|
}
|
|
else
|
|
{
|
|
Console.WriteLine(key + " 任务函数 未实现");
|
|
}
|
|
}
|
|
|
|
}
|
|
}
|