diff --git a/LMS.Common/Extensions/BeijingTimeExtension.cs b/LMS.Common/Extensions/BeijingTimeExtension.cs index 8a9bf02..864cb2c 100644 --- a/LMS.Common/Extensions/BeijingTimeExtension.cs +++ b/LMS.Common/Extensions/BeijingTimeExtension.cs @@ -13,14 +13,40 @@ } /// - /// 将UTC时间转换为北京时间 + /// 智能转换时间为北京时间 + /// 如果是UTC时间则转换，否则直接返回 /// - /// - /// - public static DateTime TransferUtcToBeijingTime(DateTime utcTime) + /// 输入的时间 + /// 北京时间 + public static DateTime TransferUtcToBeijingTime(DateTime dateTime) { - return TimeZoneInfo.ConvertTimeFromUtc(utcTime, - TimeZoneInfo.FindSystemTimeZoneById("China Standard Time")); + // 只有UTC时间才需要转换 + if (dateTime.Kind == DateTimeKind.Utc) + { + try + { + // 优先使用系统时区信息 + return TimeZoneInfo.ConvertTimeFromUtc(dateTime, + TimeZoneInfo.FindSystemTimeZoneById("China Standard Time")); + } + catch (TimeZoneNotFoundException) + { + try + { + // Linux系统可能使用这个ID + return TimeZoneInfo.ConvertTimeFromUtc(dateTime, + TimeZoneInfo.FindSystemTimeZoneById("Asia/Shanghai")); + } + catch (TimeZoneNotFoundException) + { + // 找不到时区就手动加8小时 + return dateTime.AddHours(8); + } + } + } + + // 非UTC时间直接返回 + return dateTime; } } } diff --git a/LMS.Common/LMS.Common.csproj b/LMS.Common/LMS.Common.csproj index 67198ff..94d9115 100644 --- a/LMS.Common/LMS.Common.csproj +++ b/LMS.Common/LMS.Common.csproj @@ -8,6 +8,8 @@ + + diff --git a/LMS.DAO/ApplicationDbContext.cs b/LMS.DAO/ApplicationDbContext.cs index 1c9a317..7698f77 100644 --- a/LMS.DAO/ApplicationDbContext.cs +++ b/LMS.DAO/ApplicationDbContext.cs @@ -1,6 +1,7 @@  using LMS.Repository.DB; +using LMS.Repository.MJPackage; using LMS.Repository.Models.DB; using Microsoft.AspNetCore.Identity.EntityFrameworkCore; using Microsoft.EntityFrameworkCore; @@ -46,6 +47,12 @@ namespace LMS.DAO public DbSet DataInfo { get; set; } + public DbSet MJApiTokens { get; set; } + + public DbSet MJApiTokenUsage { get; set; } + + public DbSet MJApiTasks { get; set; } + protected override void OnModelCreating(ModelBuilder modelBuilder) { base.OnModelCreating(modelBuilder); @@ -81,6 +88,30 @@ namespace LMS.DAO ) .HasColumnType("json"); // 指定MySQL字段类型为JSON }); + + modelBuilder.Entity(entity => + { + entity.ToTable("MJApiTokens"); + entity.Property(e => e.Token).IsRequired().HasMaxLength(64); + entity.Property(e => e.DailyLimit).HasDefaultValue(0); + entity.Property(e => e.TotalLimit).HasDefaultValue(0); + entity.Property(e => e.ConcurrencyLimit).HasDefaultValue(1); + entity.Property(e => e.CreatedAt).HasDefaultValueSql("CURRENT_TIMESTAMP"); + + entity.HasIndex(e => e.Token).IsUnique(); + entity.HasIndex(e => e.ExpiresAt); + }); + + modelBuilder.Entity(entity => + { + entity.ToTable("MJApiTokenUsage"); + entity.HasKey(e => new { e.TokenId, e.Date }); + + entity.HasOne() + .WithMany() + .HasForeignKey(e => e.TokenId) + .OnDelete(DeleteBehavior.Cascade); + }); } } } diff --git a/LMS.Repository/DB/MJApiTasks.cs b/LMS.Repository/DB/MJApiTasks.cs new file mode 100644 index 0000000..1a7bb45 --- /dev/null +++ b/LMS.Repository/DB/MJApiTasks.cs @@ -0,0 +1,41 @@ +using LMS.Repository.MJPackage; +using System.ComponentModel.DataAnnotations; + +namespace LMS.Repository.DB +{ + public class MJApiTasks + { + [Key] + public string TaskId { get; set; } + public string Token { get; set; } + public long TokenId { get; set; } + public DateTime StartTime { get; set; } + public DateTime? EndTime { get; set; } + public string Status { get; set; } + public string ThirdPartyTaskId { get; set; } // 第三方任务ID + public string? Properties { get; set; } + } + + public class MJApiTaskCollection : MJApiTasks + { + public long TokenId { get; set; } + public string Token { get; set; } + } + + public class MJTaskStatus + { + public const string NOT_START = "NOT_START"; + + public const string SUBMITTED = "SUBMITTED"; + + public const string IN_PROGRESS = "IN_PROGRESS"; + + public const string FAILURE = "FAILURE"; + + public const string SUCCESS = "SUCCESS"; + + public const string MODAL = "MODAL"; + + public const string CANCEL = "CANCEL"; + } +} diff --git a/LMS.Repository/DB/MJApiTokenUsage.cs b/LMS.Repository/DB/MJApiTokenUsage.cs new file mode 100644 index 0000000..7870089 --- /dev/null +++ b/LMS.Repository/DB/MJApiTokenUsage.cs @@ -0,0 +1,22 @@ +using LMS.Common.Extensions; +using System.ComponentModel.DataAnnotations; + +namespace LMS.Repository.DB +{ + public class MJApiTokenUsage + { + [Key] + public long TokenId { get; set; } + + [Key] + public DateTime Date { get; set; } + + public int DailyUsage { get; set; } = 0; + + public int TotalUsage { get; set; } = 0; + + public DateTime LastActivityAt { get; set; } = BeijingTimeExtension.GetBeijingTime(); + + public string? HistoryUse { get; set; } = null; + } +} diff --git a/LMS.Repository/DB/MJApiTokens.cs b/LMS.Repository/DB/MJApiTokens.cs new file mode 100644 index 0000000..947d1ac --- /dev/null +++ b/LMS.Repository/DB/MJApiTokens.cs @@ -0,0 +1,29 @@ +using LMS.Common.Extensions; +using System.ComponentModel.DataAnnotations; +using System.ComponentModel.DataAnnotations.Schema; + +namespace LMS.Repository.DB +{ + public class MJApiTokens + { + [Key] + [DatabaseGenerated(DatabaseGeneratedOption.Identity)] + public long Id { get; set; } + + [Required] + [StringLength(64)] + public required string Token { get; set; } + [Required] + public required string UseToken { get; set; } // 实际使用的Token + + public int DailyLimit { get; set; } = 0; + + public int TotalLimit { get; set; } = 0; + + public int ConcurrencyLimit { get; set; } = 1; + + public DateTime CreatedAt { get; set; } = BeijingTimeExtension.GetBeijingTime(); + + public DateTime? ExpiresAt { get; set; } + } +} diff --git a/LMS.Repository/MJPackage/AddOrModifyTokenModel.cs b/LMS.Repository/MJPackage/AddOrModifyTokenModel.cs new file mode 100644 index 0000000..34a57a2 --- /dev/null +++ b/LMS.Repository/MJPackage/AddOrModifyTokenModel.cs @@ -0,0 +1,43 @@ +using System.ComponentModel.DataAnnotations; + +namespace LMS.Repository.MJPackage +{ + public class AddOrModifyTokenModel + { + /// + /// token + /// + [Required] + [StringLength(64)] + public required string Token { get; set; } + + /// + /// 实际使用的Token + /// + public required string UseToken { get; set; } + + /// + /// 日限制 + /// + [Required] + public required int DailyLimit { get; set; } + + /// + /// 总限制 + /// + [Required] + public required int TotalLimit { get; set; } + + /// + /// 并发限制 + /// + [Required] + public required int ConcurrencyLimit { get; set; } + + /// + /// 使用天数 + /// + [Required] + public required int UseDayCount { get; set; } + } +} diff --git a/LMS.Repository/MJPackage/MJSubmitImageModel.cs b/LMS.Repository/MJPackage/MJSubmitImageModel.cs new file mode 100644 index 0000000..baaad93 --- /dev/null +++ b/LMS.Repository/MJPackage/MJSubmitImageModel.cs @@ -0,0 +1,74 @@ +namespace LMS.Repository.MJPackage +{ + public class MJSubmitImageModel + { + /// + /// bot 类型，mj(默认)或niji + /// MID_JOURNEY | 枚举值: NIJI_JOURNEY + /// + public string? BotType { get; set; } + + /// + /// 提示词。 + /// + public string Prompt { get; set; } + + /// + /// 垫图base64数组。 + /// + public List? Base64Array { get; set; } + + /// + /// 账号过滤 + /// + public AccountFilter? AccountFilter { get; set; } + + /// + /// 自定义参数。 + /// + public string? State { get; set; } + + /// + /// 回调地址, 为空时使用全局notifyHook。 + /// + public string? NotifyHook { get; set; } + } + + public class AccountFilter + { + /// + /// 过滤指定实例的账号 + /// + public string? InstanceId { get; set; } + + /// + /// 账号模式 RELAX | FAST | TURBO + /// + public List? Modes { get; set; } = new List(); + + /// + /// 账号是否 remix（Midjourney Remix） + /// + public bool? Remix { get; set; } + + /// + /// 账号是否 remix（Nijiourney Remix） + /// + public bool? NijiRemix { get; set; } + + /// + /// 账号过滤时，remix 自动提交视为账号的 remix 为 false + /// + public bool? RemixAutoConsidered { get; set; } + } + + /// + /// 生成速度模式枚举. + /// + public enum GenerationSpeedMode + { + RELAX, + FAST, + TURBO + } +} diff --git a/LMS.Repository/MJPackage/MJTaskCallbackModel.cs b/LMS.Repository/MJPackage/MJTaskCallbackModel.cs new file mode 100644 index 0000000..ccab6e7 --- /dev/null +++ b/LMS.Repository/MJPackage/MJTaskCallbackModel.cs @@ -0,0 +1,39 @@ +using LMS.Repository.DB; +using System.Text.Json.Serialization; + +namespace LMS.Repository.MJPackage +{ + public class MJTaskCallbackModel + { + public string Id { get; set; } + + public string? Action { get; set; } + + public MJTaskStatus? Status { get; set; } + + public string? Prompt { get; set; } + + public string? PromptEn { get; set; } + + public string? Description { get; set; } + + public long? SubmitTime { get; set; } + + public long? StartTime { get; set; } + + public long? FinishTime { get; set; } + + public string? Progress { get; set; } + + public string? ImageUrl { get; set; } + + public string? FailReason { get; set; } + + public ResponseProperties? Properties { get; set; } + } + + public class ResponseProperties + { + public string? FinalPrompt { get; set; } + } +} \ No newline at end of file diff --git a/LMS.Repository/MJPackage/SyncResult.cs b/LMS.Repository/MJPackage/SyncResult.cs new file mode 100644 index 0000000..9e73cb1 --- /dev/null +++ b/LMS.Repository/MJPackage/SyncResult.cs @@ -0,0 +1,18 @@ +namespace LMS.Repository.MJPackage +{ + // Models/SyncResult.cs + public class SyncResult + { + public int TotalTokenCount { get; set; } + public int ActiveTokenCount { get; set; } + public int RecordsUpdated { get; set; } + } + + public class TaskStatistics + { + public int TotalTasks { get; set; } = 0; + public int CompletedTasks { get; set; } = 0; + public int FailedTasks { get; set; } = 0; + public int InProgressTasks { get; set; } = 0; + } +} diff --git a/LMS.Repository/MJPackage/TokenCacheItem.cs b/LMS.Repository/MJPackage/TokenCacheItem.cs new file mode 100644 index 0000000..b8f61d2 --- /dev/null +++ b/LMS.Repository/MJPackage/TokenCacheItem.cs @@ -0,0 +1,29 @@ +using LMS.Common.Extensions; +using LMS.Repository.DB; + +namespace LMS.Repository.MJPackage +{ + // Models/TokenCacheItem.cs + public class TokenCacheItem + { + public long Id { get; set; } + public string Token { get; set; } + public string UseToken { get; set; } // 实际请求使用的Token + public int DailyLimit { get; set; } + public int TotalLimit { get; set; } + public int ConcurrencyLimit { get; set; } // 新增：并发限制 + public DateTime CreatedAt { get; set; } + public DateTime? ExpiresAt { get; set; } + public int DailyUsage { get; set; } + public int TotalUsage { get; set; } + public DateTime LastActivityTime { get; set; } = BeijingTimeExtension.GetBeijingTime(); + public string? HistoryUse { get; set; } // 历史使用记录 + + public int CurrentlyExecuting { get; set; } = 0; + } + + public class TokenAndTaskCollection : TokenCacheItem + { + public List TaskCollections { get; set; } = []; + } +} diff --git a/LMS.Repository/MJPackage/TokenCacheStats.cs b/LMS.Repository/MJPackage/TokenCacheStats.cs new file mode 100644 index 0000000..e453c96 --- /dev/null +++ b/LMS.Repository/MJPackage/TokenCacheStats.cs @@ -0,0 +1,11 @@ +namespace LMS.Repository.MJPackage +{ + public class TokenCacheStats + { + public int TotalTokens { get; set; } + public int ActiveTokens { get; set; } + public int InactiveTokens { get; set; } + public int TotalDailyUsage { get; set; } + public int TotalUsage { get; set; } + } +} diff --git a/LMS.Repository/MJPackage/TokenQueryResult.cs b/LMS.Repository/MJPackage/TokenQueryResult.cs new file mode 100644 index 0000000..743596e --- /dev/null +++ b/LMS.Repository/MJPackage/TokenQueryResult.cs @@ -0,0 +1,20 @@ +namespace LMS.Repository.MJPackage +{ + // 查询结果映射类 + public class TokenQueryResult + { + public long Id { get; set; } + public string Token { get; set; } + public string UseToken { get; set; } // 实际请求使用的Token + public int DailyLimit { get; set; } + public int TotalLimit { get; set; } + public int ConcurrencyLimit { get; set; } + public DateTime CreatedAt { get; set; } + public DateTime? ExpiresAt { get; set; } + public int DailyUsage { get; set; } + public int TotalUsage { get; set; } + public DateTime LastActivityTime { get; set; } + + public string? HistoryUse { get; set; } // 历史使用记录 + } +} diff --git a/LMS.Repository/MJPackage/TokenUsageData.cs b/LMS.Repository/MJPackage/TokenUsageData.cs new file mode 100644 index 0000000..fc8e853 --- /dev/null +++ b/LMS.Repository/MJPackage/TokenUsageData.cs @@ -0,0 +1,12 @@ +namespace LMS.Repository.MJPackage +{ + public class TokenUsageData + { + public long TokenId { get; set; } + public DateTime Date { get; set; } + public int DailyUsage { get; set; } + public int TotalUsage { get; set; } + public DateTime LastActivityTime { get; set; } + public string? HistoryUse { get; set; } + } +} diff --git a/LMS.Tools/MJPackage/ITaskConcurrencyManager.cs b/LMS.Tools/MJPackage/ITaskConcurrencyManager.cs new file mode 100644 index 0000000..6a67758 --- /dev/null +++ b/LMS.Tools/MJPackage/ITaskConcurrencyManager.cs @@ -0,0 +1,19 @@ +using LMS.Repository.DB; + + +namespace LMS.Tools.MJPackage +{ + // Services/ITaskConcurrencyManager.cs + public interface ITaskConcurrencyManager + { + Task CreateTaskAsync(string token, string thirdPartyTaskId); + Task UpdateTaskInDatabase(MJApiTasks mJApiTasks); + Task GetTaskInfoAsync(string taskId); + + Task GetTaskInfoByThirdPartyIdAsync(string taskId); + + Task> GetRunningTasksAsync(string token = null); + Task<(int maxConcurrency, int running, int available)> GetConcurrencyStatusAsync(string token); + Task CleanupTimeoutTasksAsync(TimeSpan timeout); + } +} diff --git a/LMS.Tools/MJPackage/ITaskService.cs b/LMS.Tools/MJPackage/ITaskService.cs new file mode 100644 index 0000000..3543ba7 --- /dev/null +++ b/LMS.Tools/MJPackage/ITaskService.cs @@ -0,0 +1,10 @@ +using LMS.Repository.DB; +using Microsoft.AspNetCore.Mvc; + +namespace LMS.Tools.MJPackage +{ + public interface ITaskService + { + Task?> FetchTaskAsync(MJApiTasks mJApiTasks); + } +} diff --git a/LMS.Tools/MJPackage/ITokenService.cs b/LMS.Tools/MJPackage/ITokenService.cs new file mode 100644 index 0000000..f7b39c0 --- /dev/null +++ b/LMS.Tools/MJPackage/ITokenService.cs @@ -0,0 +1,22 @@ +using LMS.Repository.DB; +using LMS.Repository.MJPackage; + +namespace LMS.Tools.MJPackage +{ + public interface ITokenService + { + Task GetTokenAsync(string token); + + Task GetDatabaseTokenAsync(string token, bool hasHistory = false); + + Task GetMJapiTokenByIdAsync(long tokenId); + + Task ResetDailyUsage(); + + void IncrementUsage(string token); + + Task LoadOriginTokenAsync(); + + Task GetOriginToken(); + } +} diff --git a/LMS.Tools/MJPackage/TaskConcurrencyManager.cs b/LMS.Tools/MJPackage/TaskConcurrencyManager.cs new file mode 100644 index 0000000..5e714c0 --- /dev/null +++ b/LMS.Tools/MJPackage/TaskConcurrencyManager.cs @@ -0,0 +1,216 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.DB; +using LMS.Repository.MJPackage; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using System.Collections.Concurrent; +using System.Text; + +namespace LMS.Tools.MJPackage +{ + public class TaskConcurrencyManager : ITaskConcurrencyManager + { + private readonly ConcurrentDictionary _activeTasks = new(); + private readonly ConcurrentDictionary _thirdPartyTaskMap = new(); // ThirdPartyTaskId -> TaskId + private readonly TokenUsageTracker _usageTracker; + private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; + private readonly ApplicationDbContext _dbContext; + private readonly ITokenService _tokenService; + + public TaskConcurrencyManager( + TokenUsageTracker usageTracker, + IServiceScopeFactory scopeFactory, + ILogger logger, + ApplicationDbContext dbContext, + ITokenService tokenService) + { + _usageTracker = usageTracker; + _scopeFactory = scopeFactory; + _logger = logger; + _dbContext = dbContext; + _tokenService = tokenService; + } + + /// + /// 尝试开始新任务（获取并发许可） + /// + public async Task CreateTaskAsync( + string token, + string thirdPartyTaskId) + { + try + { + TokenCacheItem? tokenConfig = await _tokenService.GetTokenAsync(token); + if (tokenConfig == null || string.IsNullOrWhiteSpace(tokenConfig.UseToken)) + { + _logger.LogWarning($"无效的Token: {token}"); + return; + } + // 创建任务信息 + var taskId = Guid.NewGuid().ToString("N"); + var mJApiTasks = new MJApiTasks + { + TaskId = taskId, + Token = token, + TokenId = tokenConfig.Id, + StartTime = BeijingTimeExtension.GetBeijingTime(), + Status = MJTaskStatus.NOT_START, + ThirdPartyTaskId = thirdPartyTaskId, + Properties = null + }; + + // 5. 持久化任务信息到数据库 + await SaveTaskToDatabase(mJApiTasks); + } + catch (Exception ex) + { + _logger.LogError(ex, $"开始任务时发生错误: Token={token}"); + } + } + + + /// + /// 获取任务信息 + /// + public async Task GetTaskInfoAsync(string taskId) + { + if (_activeTasks.TryGetValue(taskId, out var taskInfo)) + { + return taskInfo; + } + + // 如果内存中没有，尝试从数据库加载 + return await LoadTaskFromDatabase(taskId); + } + + /// + /// 通过第三方ID获取数据 + /// + /// + /// + public async Task GetTaskInfoByThirdPartyIdAsync(string thirdPartyId) + { + if (string.IsNullOrWhiteSpace(thirdPartyId)) + { + _logger.LogWarning("第三方任务ID为空"); + return null; + } + MJApiTasks? mJApiTasks = await _dbContext.MJApiTasks.FirstOrDefaultAsync(x => x.ThirdPartyTaskId == thirdPartyId); + return mJApiTasks; + } + + /// + /// 获取运行中的任务列表 + /// + public async Task> GetRunningTasksAsync(string token = null) + { + var runningTasks = _activeTasks.Values + .Where(t => t.Status != MJTaskStatus.SUCCESS && t.Status != MJTaskStatus.FAILURE && t.Status != MJTaskStatus.CANCEL) + .Where(t => string.IsNullOrEmpty(token) || t.Token == token) + .OrderBy(t => t.StartTime) + .ToList(); + + _logger.LogDebug($"当前运行中的任务数: {runningTasks.Count}" + (string.IsNullOrEmpty(token) ? "" : $", Token={token}")); + + return await Task.FromResult(runningTasks); + } + + /// + /// 获取Token的并发状态 + /// + public async Task<(int maxConcurrency, int running, int available)> GetConcurrencyStatusAsync(string token) + { + var status = _usageTracker.GetConcurrencyStatus(token); + return await Task.FromResult((status.maxCount, status.currentlyExecuting, status.available)); + } + + /// + /// 清理超时任务 + /// + public async Task CleanupTimeoutTasksAsync(TimeSpan timeout) + { + _logger.LogInformation($"开始清理超时任务，超时阈值: {timeout.TotalMinutes}分钟"); + + var cutoffTime = BeijingTimeExtension.GetBeijingTime() - timeout; + var timeoutTasks = _activeTasks.Values + .Where(t => t.StartTime < cutoffTime && t.Status != MJTaskStatus.SUCCESS && t.Status != MJTaskStatus.FAILURE && t.Status != MJTaskStatus.CANCEL) + .ToList(); + + _logger.LogInformation($"发现 {timeoutTasks.Count} 个超时任务"); + + foreach (var task in timeoutTasks) + { + _logger.LogWarning($"清理超时任务: TaskId={task.TaskId}, Token={task.Token}, 开始时间={task.StartTime:yyyy-MM-dd HH:mm:ss}"); + _usageTracker.ReleaseConcurrencyPermit(task.Token); + } + } + + /// + /// 保存任务到数据库 + /// + private async Task SaveTaskToDatabase(MJApiTasks mJApiTasks) + { + try + { + await _dbContext.MJApiTasks.AddAsync(mJApiTasks); + await _dbContext.SaveChangesAsync(); + _logger.LogInformation($"任务已保存到数据库: TaskId={mJApiTasks.TaskId}, Token={mJApiTasks.Token}"); + } + catch (Exception ex) + { + _logger.LogError(ex, $"保存任务到数据库失败: TaskId={mJApiTasks.TaskId}"); + } + } + + /// + /// 更新数据库中的任务状态 + /// + public async Task UpdateTaskInDatabase(MJApiTasks mJApiTasks) + { + try + { + MJApiTasks? apiTasks = await _dbContext.MJApiTasks.FirstOrDefaultAsync(x => x.ThirdPartyTaskId == mJApiTasks.ThirdPartyTaskId); + if (apiTasks == null) + { + _logger.LogWarning($"未找到任务: TaskId={mJApiTasks.TaskId}"); + return; + } + apiTasks.Status = mJApiTasks.Status; + apiTasks.EndTime = mJApiTasks.EndTime; + apiTasks.Properties = mJApiTasks.Properties; + _dbContext.MJApiTasks.Update(apiTasks); + await _dbContext.SaveChangesAsync(); + } + catch (Exception ex) + { + _logger.LogError(ex, $"更新任务状态到数据库失败: TaskId={mJApiTasks.TaskId}"); + } + } + + /// + /// 从数据库加载任务 + /// + private async Task LoadTaskFromDatabase(string taskId) + { + try + { + MJApiTasks? mJApiTasks = await _dbContext.MJApiTasks.FirstOrDefaultAsync(x => x.TaskId == taskId); + if (mJApiTasks == null) + { + _logger.LogWarning($"未找到任务: TaskId={taskId}"); + return null; + } + + return mJApiTasks; + } + catch (Exception ex) + { + _logger.LogError(ex, $"从数据库加载任务失败: TaskId={taskId}"); + return null; + } + } + } +} diff --git a/LMS.Tools/MJPackage/TaskService.cs b/LMS.Tools/MJPackage/TaskService.cs new file mode 100644 index 0000000..4fa0e91 --- /dev/null +++ b/LMS.Tools/MJPackage/TaskService.cs @@ -0,0 +1,147 @@ + +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.DB; +using Microsoft.AspNetCore.Mvc; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using System.Net; +using System.Net.Sockets; +using System.Threading.Tasks; + +namespace LMS.Tools.MJPackage +{ + public class TaskService(ITokenService tokenService, ILogger logger, ApplicationDbContext dbContext, ITaskConcurrencyManager taskConcurrencyManager) : ITaskService + { + private readonly ITokenService _tokenService = tokenService; + private readonly ILogger _logger = logger; + private readonly ApplicationDbContext _dbContext = dbContext; + private readonly ITaskConcurrencyManager _taskConcurrencyManager = taskConcurrencyManager; + + public async Task?> FetchTaskAsync(MJApiTasks mJApiTasks) + { + try + { + // 获取UseToken，先尝试 Token，再尝试 TokenId + var tokenConfig = await _tokenService.GetTokenAsync(mJApiTasks.Token); + string useToken = string.Empty; + + if (tokenConfig == null) + { + // Token 没找到 尝试用 TokenId 查找 + MJApiTokens? mJApiTokens = await _tokenService.GetMJapiTokenByIdAsync(mJApiTasks.TokenId); + if (mJApiTokens == null) + { + return null; + } + useToken = mJApiTokens.UseToken; + } + else + { + useToken = tokenConfig.UseToken; + } + + if (string.IsNullOrWhiteSpace(useToken)) + { + _logger.LogInformation($"Token is empty for task ID: {mJApiTasks.TaskId}"); + return null; + } + + // 尝试备用API + var backupResult = await TryBackupApiAsync(mJApiTasks.ThirdPartyTaskId, useToken); + + if (string.IsNullOrWhiteSpace(backupResult)) + { + // 没有找到数据 + _logger.LogInformation($"备用API没有返回数据，TaskId: {mJApiTasks.TaskId}"); + return null; + } + + var properties = new Dictionary(); + try + { + // 不为空 开始解析数据 + properties = JsonConvert.DeserializeObject>(backupResult); + } + catch (JsonException ex) + { + _logger.LogError($"解析备用API返回数据失败: {ex.Message}"); + return null; + } + + if (properties == null) + { + _logger.LogInformation($"备用API返回数据为空，TaskId: {mJApiTasks.TaskId}"); + return null; + } + return properties; + } + catch (Exception ex) + { + // 记录异常日志 + _logger.LogError($"Error fetching task: {ex.Message}"); + return null; + } + } + private async Task TryBackupApiAsync(string id, string useToken) + { + const string backupUrlTemplate = "https://api.laitool.cc/mj/task/{0}/fetch"; + const int maxRetries = 3; + const int baseDelayMs = 1000; + + using var client = new HttpClient(); + client.DefaultRequestHeaders.Add("Authorization", "sk-" + useToken); + client.Timeout = TimeSpan.FromSeconds(30); + + var backupUrl = string.Format(backupUrlTemplate, id); + + for (int attempt = 1; attempt <= maxRetries; attempt++) + { + try + { + var response = await client.GetAsync(backupUrl); + var content = await response.Content.ReadAsStringAsync(); + // 判断请求是不是报错 + if (!response.IsSuccessStatusCode) + { + _logger.LogWarning("备用API调用返回错误状态码，TaskId: {TaskId}, Attempt: {Attempt}, StatusCode: {StatusCode}", + id, attempt, response.StatusCode); + return null; + } + return content; + } + catch (Exception ex) when (IsRetriableException(ex)) + { + if (attempt < maxRetries) + { + var delay = baseDelayMs * (int)Math.Pow(2, attempt - 1); + _logger.LogWarning(ex, "备用API调用失败，TaskId: {TaskId}, Attempt: {Attempt}, 将在{Delay}ms后重试", + id, attempt, delay); + await Task.Delay(delay); + } + else + { + _logger.LogError(ex, "备用API调用最终失败，TaskId: {TaskId}, MaxAttempts: {MaxAttempts}", + id, maxRetries); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "备用API调用发生不可重试异常，TaskId: {TaskId}, Attempt: {Attempt}", + id, attempt); + break; + } + } + + return null; + } + + private static bool IsRetriableException(Exception ex) + { + return ex is HttpRequestException || + ex is TaskCanceledException || + ex is SocketException; + } + + } +} diff --git a/LMS.Tools/MJPackage/TaskStatusCheckService.cs b/LMS.Tools/MJPackage/TaskStatusCheckService.cs new file mode 100644 index 0000000..9049353 --- /dev/null +++ b/LMS.Tools/MJPackage/TaskStatusCheckService.cs @@ -0,0 +1,114 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.DB; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using Quartz; + +namespace LMS.Tools.MJPackage +{ + [DisallowConcurrentExecution] + public class TaskStatusCheckService(ITokenService tokenService, ApplicationDbContext dbContext, ILogger logger, ITaskService taskService, ITaskConcurrencyManager taskConcurrencyManager) : IJob + { + private readonly ITokenService _tokenService = tokenService; + private readonly ApplicationDbContext _dbContext = dbContext; + private readonly ILogger _logger = logger; + private readonly ITaskService _taskService = taskService; + private readonly ITaskConcurrencyManager _taskConcurrencyManager = taskConcurrencyManager; + + public async Task Execute(IJobExecutionContext context) + { + _logger.LogInformation($"开始检查TASK信息 - 检查间隔: 5 分钟，同步加载 原始请求的Token！"); + + var startTime = BeijingTimeExtension.GetBeijingTime(); + try + { + // 强制同步数据库数据 + await _tokenService.LoadOriginTokenAsync(); + + // 检查Task状态和返回值 + // 获取所有超过五分钟没有完成的人物 + List tasks = await _dbContext.MJApiTasks.Where(t => t.Status != MJTaskStatus.CANCEL && t.Status != MJTaskStatus.SUCCESS && t.Status != MJTaskStatus.FAILURE && t.StartTime < BeijingTimeExtension.GetBeijingTime().AddMinutes(-5)).ToListAsync(); + + if (tasks.Count == 0) + { + _logger.LogInformation("没有需要检查的任务！"); + return; + } + + // 开始每个请求 + foreach (MJApiTasks task in tasks) + { + try + { + Dictionary? properties = await _taskService.FetchTaskAsync(task); + // 没有找到数据的 + if (properties == null) + { + // 没有找到数据 直接把任务失败 + task.Status = MJTaskStatus.FAILURE; + var newProperties = new + { + failReason = "任务丢失或未找到" + }; + task.EndTime = BeijingTimeExtension.GetBeijingTime(); + task.Properties = JsonConvert.SerializeObject(newProperties); + } + else + { + // 尝试获取状态字段 + string status = MJTaskStatus.SUBMITTED; + if (properties.TryGetValue("status", out var statusElement)) + { + status = statusElement.ToString() ?? MJTaskStatus.SUBMITTED; + } + else if (properties.TryGetValue("Status", out var statusElementCap)) + { + status = statusElementCap.ToString() ?? MJTaskStatus.SUBMITTED; + } + task.Status = status; + + if (status == MJTaskStatus.SUCCESS || status == MJTaskStatus.FAILURE || status == MJTaskStatus.CANCEL) + { + // 当前任务已经被释放过了 + // 开始修改数据 + task.EndTime = BeijingTimeExtension.GetBeijingTime(); + task.Properties = JsonConvert.SerializeObject(properties); + } + else + { + // 任务还在处理中 + task.EndTime = null; // 处理中没有结束时间 + task.Properties = JsonConvert.SerializeObject(properties); + } + } + // 开始修改数据 + await _taskConcurrencyManager.UpdateTaskInDatabase(task); + } + catch (Exception ex) + { + // 报错 + _logger.LogError(ex, "检查任务 {TaskId} 时发生错误", task.TaskId); + task.Status = MJTaskStatus.FAILURE; + var newProperties = new + { + failReason = "任务报错" + }; + task.EndTime = BeijingTimeExtension.GetBeijingTime(); + task.Properties = JsonConvert.SerializeObject(newProperties); + // 开始修改数据 + await _taskConcurrencyManager.UpdateTaskInDatabase(task); + } + } + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + _logger.LogInformation($"Task状态检查完成，影响的Task {tasks.Count}，耗时: {duration}ms", duration.TotalMilliseconds); + } + catch (Exception ex) + { + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + _logger.LogError(ex, "Token同步失败，耗时: {Duration}ms", duration.TotalMilliseconds); + } + } + } +} diff --git a/LMS.Tools/MJPackage/TokenResetService.cs b/LMS.Tools/MJPackage/TokenResetService.cs new file mode 100644 index 0000000..62eb84e --- /dev/null +++ b/LMS.Tools/MJPackage/TokenResetService.cs @@ -0,0 +1,24 @@ +using LMS.Common.Extensions; +using Microsoft.Extensions.Logging; +using Quartz; +using System; +using System.Collections.Generic; +using System.Linq; +using System.Text; +using System.Threading.Tasks; + +namespace LMS.Tools.MJPackage +{ + [DisallowConcurrentExecution] + public class TokenResetService(ITokenService tokenService, ILogger logger) : IJob + { + private readonly ITokenService _tokenService = tokenService; + private readonly ILogger _logger = logger; + public async Task Execute(IJobExecutionContext context) + { + _logger.LogInformation("开始每天重置 Token 日使用 统计数据， 执行时间 " + BeijingTimeExtension.GetBeijingTime()); + await _tokenService.ResetDailyUsage(); + } + + } +} diff --git a/LMS.Tools/MJPackage/TokenService.cs b/LMS.Tools/MJPackage/TokenService.cs new file mode 100644 index 0000000..2de2ce1 --- /dev/null +++ b/LMS.Tools/MJPackage/TokenService.cs @@ -0,0 +1,352 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.DB; +using LMS.Repository.MJPackage; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.Caching.Memory; +using Microsoft.Extensions.Logging; +using Newtonsoft.Json; +using System.Data; +using System.Runtime.CompilerServices; + +namespace LMS.Tools.MJPackage +{ + public class TokenService( + ApplicationDbContext dbContext, + IMemoryCache memoryCache, + TokenUsageTracker usageTracker, + ILogger logger) : ITokenService + { + private readonly ApplicationDbContext _dbContext = dbContext; + private readonly IMemoryCache _memoryCache = memoryCache; + private readonly TokenUsageTracker _usageTracker = usageTracker; + private readonly ILogger _logger = logger; + + /// + /// 从数据库获取token + /// + /// + /// + public async Task GetDatabaseTokenAsync(string token, bool hasHistory = false) + { + try + { + var today = BeijingTimeExtension.GetBeijingTime().Date; + // 使用EF Core的FromSqlRaw执行原生SQL + var dbResult = await _dbContext.Database + .SqlQuery($@" + SELECT + t.Id, t.Token, t.DailyLimit, t.TotalLimit, t.ConcurrencyLimit, + t.CreatedAt, t.ExpiresAt, t.UseToken, + COALESCE(u.DailyUsage, 0) as DailyUsage, + COALESCE(u.TotalUsage, 0) as TotalUsage, + COALESCE(u.HistoryUse, '') as HistoryUse, + COALESCE(u.LastActivityAt, t.CreatedAt) as LastActivityTime + FROM MJApiTokens t + LEFT JOIN MJApiTokenUsage u ON t.Id = u.TokenId + WHERE t.Token = {token}") + .FirstOrDefaultAsync(); + + if (dbResult == null) + { + return null; + } + + // 3. 转换为TokenCacheItem + var tokenItem = new TokenCacheItem + { + Id = dbResult.Id, + Token = dbResult.Token, + UseToken = dbResult.UseToken ?? string.Empty, // 确保UseToken不为null + DailyLimit = dbResult.DailyLimit, + TotalLimit = dbResult.TotalLimit, + ConcurrencyLimit = dbResult.ConcurrencyLimit, + CreatedAt = dbResult.CreatedAt, + ExpiresAt = dbResult.ExpiresAt, + DailyUsage = dbResult.DailyUsage, + TotalUsage = dbResult.TotalUsage, + LastActivityTime = dbResult.LastActivityTime, + HistoryUse = hasHistory ? dbResult.HistoryUse : string.Empty + }; + return tokenItem; + } + catch (Exception ex) + { + _logger.LogError(ex, $"从数据库获取Token时发生错误: {token}"); + throw; + } + } + + public async Task GetMJapiTokenByIdAsync(long tokenId) + { + try + { + MJApiTokens? mJApiTokens = await _dbContext.MJApiTokens + .AsNoTracking() + .FirstOrDefaultAsync(t => t.Id == tokenId); + return mJApiTokens; + } + catch (Exception ex) + { + _logger.LogError(ex, $"获取Token ID {tokenId} 时发生错误"); + throw; + } + } + + /// + /// 异步获取Token信息，先从缓存获取，缓存未命中则从数据库加载 + /// + /// Token字符串 + /// Token缓存项，未找到返回null + public async Task GetTokenAsync(string token) + { + _logger.LogDebug($"开始获取Token: {token}"); + + // 1. 检查内存缓存 + if (_usageTracker.TryGetToken(token, out var cacheItem)) + { + _logger.LogDebug($"Token从内存缓存中获取成功: {token}"); + return cacheItem; + } + + // 2. 从数据库加载 - 使用EF Core原生SQL查询 + _logger.LogDebug($"Token不在缓存中，从数据库加载: {token}"); + + try + { + TokenCacheItem? tokenItem = await GetDatabaseTokenAsync(token); + if (tokenItem == null) + { + _logger.LogWarning($"Token未找到: {token}"); + return null; + } + + // 更新最后活动时间，从数据库中获取得话 设置最后活跃时间为当前时间 + tokenItem.LastActivityTime = BeijingTimeExtension.GetBeijingTime(); + // 4. 加入内存缓存 + _usageTracker.AddOrUpdateToken(tokenItem); + + // 5. 设置内存缓存 (30分钟) + _memoryCache.Set($"Token_{token}", tokenItem, TimeSpan.FromMinutes(30)); + + _logger.LogInformation($"Token从数据库加载成功: {token}, ID: {tokenItem.Id}, 日限制: {tokenItem.DailyLimit}, 并发限制: {tokenItem.ConcurrencyLimit}"); + return tokenItem; + } + catch (Exception ex) + { + _logger.LogError(ex, $"从数据库获取Token时发生错误: {token}"); + throw; + } + } + + /// + /// 增加Token使用量 + /// + /// Token字符串 + public void IncrementUsage(string token) + { + _logger.LogDebug($"递增Token使用量: {token}"); + _usageTracker.IncrementUsage(token); + } + + public async Task LoadOriginTokenAsync() + { + // 没找到 从数据库中获取 + Options? oprions = await _dbContext.Options.Where(x => x.Key == "MJPackageOriginToken").FirstOrDefaultAsync(); + if (oprions == null) + { + _logger.LogWarning("未找到原始Token配置"); + return string.Empty; + } + + // 处理数据 + string originToken = oprions.GetValueObject() ?? string.Empty; + if (string.IsNullOrWhiteSpace(originToken)) + { + _logger.LogWarning("未找到原始Token配置"); + return string.Empty; + } + _usageTracker.OriginToken = originToken; + return originToken; + } + + public async Task GetOriginToken() + { + // 缓存中就有 直接返回 + if (!string.IsNullOrWhiteSpace(_usageTracker.OriginToken)) + { + return _usageTracker.OriginToken; + } + // 缓存中没有 从数据库中获取 + return await LoadOriginTokenAsync(); + } + + + /// + /// 重置Token的使用数据 + /// + /// + public async Task ResetDailyUsage() + { + var startTime = BeijingTimeExtension.GetBeijingTime(); + try + { + // 批量重置数据库数据 + int totalTokenCount = await BatchResetTokenDailyUsage(); + + // 删除不活跃的token + var (act, nact) = _usageTracker.RemoveNotActiveTokens(TimeSpan.FromMinutes(5)); + + // 重置缓存中的数据 + _usageTracker.ResetDailyUsage(); + + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + _logger.LogInformation($"Token日使用量重置完成: {totalTokenCount} 个Token, 活跃Token: {act}, 耗时: {duration.TotalMilliseconds}ms"); + } + catch (Exception ex) + { + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + _logger.LogError(ex, "Token同步失败，耗时: {Duration}ms", duration.TotalMilliseconds); + } + } + + + /// + /// 批量重置当日使用限制 + /// + /// + private async Task BatchResetTokenDailyUsage() + { + var beijingTime = BeijingTimeExtension.GetBeijingTime(); + + _logger.LogInformation($"重置token日限制：开始批量重置 - 北京时间: {beijingTime:yyyy-MM-dd HH:mm:ss}"); + + // 修复SQL查询 - 只查询有使用记录且需要重置的Token + string sql = @" + SELECT + t.Id, t.Token, t.DailyLimit, t.TotalLimit, t.ConcurrencyLimit, + t.CreatedAt, t.ExpiresAt, t.UseToken, + COALESCE(u.DailyUsage, 0) as DailyUsage, + COALESCE(u.HistoryUse, '') as HistoryUse, + COALESCE(u.TotalUsage, 0) as TotalUsage, + COALESCE(u.LastActivityAt, t.CreatedAt) as LastActivityTime + FROM MJApiTokens t + LEFT JOIN MJApiTokenUsage u ON t.Id = u.TokenId + WHERE u.DailyUsage > 0 + AND (t.ExpiresAt IS NULL OR t.ExpiresAt > UTC_TIMESTAMP())"; + + var dbResult = await _dbContext.Database + .SqlQuery(FormattableStringFactory.Create(sql)) + .ToListAsync(); + + if (dbResult.Count == 0) + { + _logger.LogInformation("重置token日限制：没有需要重置的token"); + return 0; + } + + _logger.LogInformation($"找到 {dbResult.Count} 个需要重置的Token"); + + // 统计重置前的总使用量 + var totalDailyUsageBeforeReset = dbResult.Sum(x => x.DailyUsage); + _logger.LogInformation($"重置前总日使用量: {totalDailyUsageBeforeReset}"); + + var updatedCount = 0; + const int batchSize = 100; // 分批处理，避免内存过大 + + // 分批处理Token重置 + for (int batchStart = 0; batchStart < dbResult.Count; batchStart += batchSize) + { + var batch = dbResult.Skip(batchStart).Take(batchSize).ToList(); + var batchTokenIds = batch.Select(x => x.Id).ToList(); + + // 批量查询当前批次的使用记录 + var tokenUsageList = await _dbContext.MJApiTokenUsage + .Where(x => batchTokenIds.Contains(x.TokenId)) + .ToListAsync(); + + if (!tokenUsageList.Any()) + { + _logger.LogWarning($"批次 {batchStart / batchSize + 1}: 没有找到使用记录"); + continue; + } + + // 使用事务确保数据一致性 + using var transaction = await _dbContext.Database.BeginTransactionAsync(); + try + { + foreach (var tokenUsage in tokenUsageList) + { + var tokenInfo = batch.FirstOrDefault(x => x.Id == tokenUsage.TokenId); + if (tokenInfo == null || tokenUsage.DailyUsage == 0) + continue; + + // 处理历史记录 + ProcessHistoryAndResetUsage(tokenUsage, tokenInfo); + } + // 批量保存 + var batchUpdated = await _dbContext.SaveChangesAsync(); + await transaction.CommitAsync(); + + updatedCount += batchUpdated; + _logger.LogInformation($"批次 {batchStart / batchSize + 1} 完成: 处理 {batch.Count} 个Token，更新 {batchUpdated} 条记录"); + } + catch (Exception ex) + { + await transaction.RollbackAsync(); + _logger.LogError(ex, $"批次 {batchStart / batchSize + 1} 重置失败"); + throw; + } + } + + _logger.LogInformation($"✅ 批量重置完成 - 总共更新 {updatedCount} 条记录"); + _logger.LogInformation($"📊 重置统计 - 重置前日使用量: {totalDailyUsageBeforeReset} → 重置后: 0"); + + return updatedCount; + } + + /// + /// 处理历史记录并重置使用量 + /// + private void ProcessHistoryAndResetUsage(MJApiTokenUsage tokenUsage, TokenQueryResult tokenInfo) + { + try + { + // 解析现有历史记录 + List historyList; + try + { + historyList = string.IsNullOrEmpty(tokenUsage.HistoryUse) + ? [] + : JsonConvert.DeserializeObject>(tokenUsage.HistoryUse) ?? new List(); + } + catch (JsonException ex) + { + _logger.LogWarning(ex, $"Token {tokenInfo.Token} 历史记录JSON解析失败，将创建新的历史记录"); + historyList = []; + } + + // 添加当前记录到历史 + historyList.Add(new MJApiTokenUsage + { + TokenId = tokenUsage.TokenId, + Date = BeijingTimeExtension.GetBeijingTime().Date.AddDays(-1), + DailyUsage = tokenUsage.DailyUsage, + TotalUsage = tokenUsage.TotalUsage, + LastActivityAt = tokenUsage.LastActivityAt, + HistoryUse = "" + }); + + // 重置使用量 + tokenUsage.DailyUsage = 0; + tokenUsage.HistoryUse = JsonConvert.SerializeObject(historyList); + + _logger.LogDebug($"Token {tokenInfo.Token} 重置: 日使用量 {tokenUsage.DailyUsage} → 0, 历史记录数: {historyList.Count}"); + } + catch (Exception ex) + { + _logger.LogError(ex, $"处理Token {tokenInfo.Token} 的历史记录时发生错误"); + } + } + } +} diff --git a/LMS.Tools/MJPackage/TokenSyncService.cs b/LMS.Tools/MJPackage/TokenSyncService.cs new file mode 100644 index 0000000..d62bb3b --- /dev/null +++ b/LMS.Tools/MJPackage/TokenSyncService.cs @@ -0,0 +1,198 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.MJPackage; +using Microsoft.EntityFrameworkCore; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Logging; +using Quartz; +using System.Data; +using System.Text; + +namespace LMS.Tools.MJPackage +{ + [DisallowConcurrentExecution] + public class TokenSyncService : IJob + { + private readonly TokenUsageTracker _usageTracker; + private readonly IServiceProvider _serviceProvider; + private readonly ILogger _logger; + + // 活动阈值：5分钟内有活动的Token才同步 + private readonly TimeSpan _activityThreshold = TimeSpan.FromMinutes(5); + + public TokenSyncService( + TokenUsageTracker usageTracker, + IServiceProvider serviceProvider, + ILogger logger) + { + _usageTracker = usageTracker; + _serviceProvider = serviceProvider; + _logger = logger; + } + + public async Task Execute(IJobExecutionContext context) + { + _logger.LogInformation($"开始同步Token信息 - 同步间隔: 30 秒, 活动阈值: {_activityThreshold.TotalMinutes}分钟 (使用EF Core)"); + + var startTime = BeijingTimeExtension.GetBeijingTime(); + try + { + var syncResult = await SyncActiveTokensToDatabase(); + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + + if (syncResult.ActiveTokenCount > 0) + { + _logger.LogInformation( + "Token同步完成: {ActiveTokens}/{TotalTokens} 个活跃Token已同步, 耗时: {Duration}ms, 更新记录: {RecordsUpdated}", + syncResult.ActiveTokenCount, + syncResult.TotalTokenCount, + duration.TotalMilliseconds, + syncResult.RecordsUpdated); + } + else + { + _logger.LogDebug( + "Token同步跳过: 无活跃Token (总计: {TotalTokens}, 耗时: {Duration}ms)", + syncResult.TotalTokenCount, + duration.TotalMilliseconds); + } + } + catch (Exception ex) + { + var duration = BeijingTimeExtension.GetBeijingTime() - startTime; + _logger.LogError(ex, "Token同步失败，耗时: {Duration}ms", duration.TotalMilliseconds); + } + } + + /// + /// 同步活跃Token数据到数据库 + /// + /// 同步结果 + private async Task SyncActiveTokensToDatabase() + { + // 先 删除10分钟内不活跃得Token + var (act, nact) = _usageTracker.RemoveNotActiveTokens(TimeSpan.FromMinutes(10)); + _logger.LogInformation($"删除不活跃的 Token 数 {nact}，删除后活跃 Token 数：{act}，判断不活跃时间：{10} 分钟"); + + // 1. 获取活跃Token（最近5分钟内有活动的Token） + var activeTokens = _usageTracker.GetActiveTokens(_activityThreshold).ToList(); + var totalTokens = _usageTracker.GetAllTokens().Count(); + + if (!activeTokens.Any()) + { + _logger.LogInformation("0 条活跃Token，跳过同步！"); + return new SyncResult + { + TotalTokenCount = totalTokens, + ActiveTokenCount = 0, + RecordsUpdated = 0 + }; + } + + // 2. 创建数据库上下文 + using var scope = _serviceProvider.CreateScope(); + var dbContext = scope.ServiceProvider.GetRequiredService(); + + var today = BeijingTimeExtension.GetBeijingTime().Date; + var recordsUpdated = 0; + + // 3. 构造批量数据 + var batchData = activeTokens.Select(token => new TokenUsageData + { + TokenId = token.Id, + Date = today, + DailyUsage = token.DailyUsage, + TotalUsage = token.TotalUsage, + LastActivityTime = token.LastActivityTime + }).ToList(); + + // 4. 使用EF Core事务批量更新数据库 + using var transaction = await dbContext.Database.BeginTransactionAsync(); + try + { + recordsUpdated = await BatchUpdateTokenUsageWithEfCore(dbContext, batchData); + await transaction.CommitAsync(); + + _logger.LogDebug( + "批量更新完成: {RecordsUpdated} 条活跃Token记录已更新，跳过 {SkippedCount} 条非活跃Token", + recordsUpdated, + totalTokens - activeTokens.Count); + } + catch (Exception ex) + { + await transaction.RollbackAsync(); + _logger.LogError(ex, "数据库事务失败，已回滚"); + throw; + } + + return new SyncResult + { + TotalTokenCount = totalTokens, + ActiveTokenCount = activeTokens.Count, + RecordsUpdated = recordsUpdated + }; + } + + /// + /// 使用EF Core批量更新Token使用数据 + /// + /// 数据库上下文 + /// 批量数据 + /// 更新的记录数 + private async Task BatchUpdateTokenUsageWithEfCore(ApplicationDbContext dbContext, List batchData) + { + int batchSize = 500; + if (!batchData.Any()) return 0; + + var recordsUpdated = 0; + + // 分批处理 + for (int i = 0; i < batchData.Count; i += batchSize) + { + var batch = batchData.Skip(i).Take(batchSize).ToList(); + + // 构建真正的批量 SQL + var sqlBuilder = new StringBuilder(); + var parameters = new List(); + + sqlBuilder.AppendLine("INSERT INTO MJApiTokenUsage (TokenId, Date, DailyUsage, TotalUsage, LastActivityAt) VALUES "); + + // 为每条记录构建 VALUES 子句 + for (int j = 0; j < batch.Count; j++) + { + if (j > 0) sqlBuilder.Append(", "); + + var paramIndex = j * 5; + sqlBuilder.Append($"({{{paramIndex}}}, {{{paramIndex + 1}}}, {{{paramIndex + 2}}}, {{{paramIndex + 3}}}, {{{paramIndex + 4}}})"); + + parameters.AddRange(new object[] + { + batch[j].TokenId, + batch[j].Date, + batch[j].DailyUsage, + batch[j].TotalUsage, + batch[j].LastActivityTime + }); + } + + sqlBuilder.AppendLine(@" + ON DUPLICATE KEY UPDATE + Date = VALUES(Date), + DailyUsage = VALUES(DailyUsage), + TotalUsage = VALUES(TotalUsage), + LastActivityAt = VALUES(LastActivityAt)"); + + // 一次性执行整个批次 + var affectedRows = await dbContext.Database.ExecuteSqlRawAsync( + sqlBuilder.ToString(), + parameters.ToArray()); + + recordsUpdated += affectedRows; + + _logger.LogInformation($"批量更新完成: 批次 {i / batchSize + 1}, 记录数: {batch.Count}, 影响行数: {affectedRows}"); + } + + return recordsUpdated; + } + } +} diff --git a/LMS.Tools/MJPackage/TokenUsageTracker.cs b/LMS.Tools/MJPackage/TokenUsageTracker.cs new file mode 100644 index 0000000..dfc716b --- /dev/null +++ b/LMS.Tools/MJPackage/TokenUsageTracker.cs @@ -0,0 +1,504 @@ +using LMS.Common.Extensions; +using LMS.Repository.DB; +using LMS.Repository.MJPackage; +using Microsoft.Extensions.Logging; +using System.Collections.Concurrent; + +namespace LMS.Tools.MJPackage +{ + public class TokenUsageTracker + { + private readonly ConcurrentDictionary _tokenCache = new(); + private readonly ConcurrentDictionary> _concurrencyControllers = new(); + private readonly ReaderWriterLockSlim _cacheLock = new(LockRecursionPolicy.SupportsRecursion); + private string _originToken = string.Empty; + + private readonly ILogger _logger; + + public TokenUsageTracker(ILogger logger) + { + _logger = logger; + _logger.LogInformation("TokenUsageTracker服务已初始化"); + } + + /// + /// 并发控制器 - 支持平滑调整并发限制 + /// + private class ConcurrencyController + { + private SemaphoreSlim _semaphore; + private int _maxCount; + private int _currentlyExecuting; + private readonly object _lock = new object(); + private readonly ILogger _logger; + + public ConcurrencyController(int initialLimit, ILogger logger) + { + _maxCount = initialLimit; + _semaphore = new SemaphoreSlim(initialLimit, initialLimit); + _currentlyExecuting = 0; + _logger = logger; + } + + /// + /// 获取当前最大并发数 + /// + public int MaxCount => _maxCount; + + /// + /// 获取当前正在执行的任务数 + /// + public int CurrentlyExecuting => _currentlyExecuting; + + /// + /// 获取当前可用的并发槽位 + /// + public int AvailableCount => _semaphore.CurrentCount; + + /// + /// 等待获取执行许可 + /// + public async Task WaitAsync(string token) + { + var acquired = await _semaphore.WaitAsync(0); + if (acquired) + { + lock (_lock) + { + _currentlyExecuting++; + } + _logger.LogInformation($"Token获取并发许可: {token}, 当前执行中: {_currentlyExecuting}/{_maxCount}"); + } + return acquired; + } + + /// + /// 释放执行许可 + /// + public void Release(string token) + { + lock (_lock) + { + if (_currentlyExecuting > 0) + { + _currentlyExecuting--; + _semaphore.Release(); + _logger.LogInformation($"Token释放并发许可: {token}, 当前执行中: {_currentlyExecuting}/{_maxCount}"); + } + else + { + _logger.LogWarning($"Token释放并发许可: {token}, 尝试释放许可但当前执行数已为0: {token}"); + } + } + } + + /// + /// 平滑调整并发限制 + /// + public bool AdjustLimitAsync(int newLimit, string token) + { + if (newLimit <= 0) + { + throw new ArgumentException("并发限制必须大于0", nameof(newLimit)); + } + + lock (_lock) + { + if (_maxCount == newLimit) + { + return false; // 无需调整 + } + + var oldLimit = _maxCount; + _maxCount = newLimit; + + if (newLimit > oldLimit) + { + // 扩大并发限制：释放额外的许可 + var additionalPermits = newLimit - oldLimit; + for (int i = 0; i < additionalPermits; i++) + { + _semaphore.Release(); + } + + _logger.LogInformation($"Token并发限制已扩大: {token}, {oldLimit} -> {newLimit}, 当前执行: {_currentlyExecuting}"); + } + else + { + // 缩小并发限制：等待现有任务完成 + var excessExecuting = _currentlyExecuting - newLimit; + if (excessExecuting > 0) + { + _logger.LogWarning($"Token并发限制缩小但有超额任务: {token}, {oldLimit} -> {newLimit}, 超额: {excessExecuting}, 将等待任务自然完成"); + } + else + { + _logger.LogInformation($"Token并发限制已缩小: {token}, {oldLimit} -> {newLimit}, 当前执行: {_currentlyExecuting}"); + } + } + + return true; + } + } + + /// + /// 销毁资源 + /// + public void Dispose() + { + _semaphore?.Dispose(); + } + } + + /// + /// 尝试从缓存中获取Token + /// + public bool TryGetToken(string token, out TokenCacheItem cacheItem) + { + + var found = _tokenCache.TryGetValue(token, out cacheItem); + if (found) + { + _logger.LogDebug($"从缓存中找到Token: {token}"); + } + return found; + + } + + /// + /// 添加或更新Token到缓存（支持平滑并发限制调整） + /// + public void AddOrUpdateTokenAsync(TokenCacheItem tokenItem) + { + _cacheLock.EnterWriteLock(); + try + { + _tokenCache[tokenItem.Token] = tokenItem; + + // 获取或创建并发控制器 + var lazyController = _concurrencyControllers.GetOrAdd( + tokenItem.Token, + _ => new Lazy(() => + new ConcurrencyController(tokenItem.ConcurrencyLimit, _logger))); + + var controller = lazyController.Value; + // 平滑调整并发限制 + var adjusted = controller.AdjustLimitAsync(tokenItem.ConcurrencyLimit, tokenItem.Token); + + if (adjusted) + { + _logger.LogInformation($"Token并发限制已调整: {tokenItem.Token}, 新限制: {tokenItem.ConcurrencyLimit}"); + } + + _logger.LogDebug($"Token已添加到缓存: {tokenItem.Token}, 日限制: {tokenItem.DailyLimit}, 总限制: {tokenItem.TotalLimit}, 并发限制: {tokenItem.ConcurrencyLimit}"); + } + finally + { + _cacheLock.ExitWriteLock(); + } + } + + /// + /// 同步版本（保持向后兼容） + /// + public void AddOrUpdateToken(TokenCacheItem tokenItem) + { + // 使用异步版本，但同步等待 + AddOrUpdateTokenAsync(tokenItem); + } + + /// + /// 增加Token使用量 + /// + public void IncrementUsage(string token) + { + _cacheLock.EnterUpgradeableReadLock(); + try + { + if (_tokenCache.TryGetValue(token, out var cacheItem)) + { + _cacheLock.EnterWriteLock(); + try + { + cacheItem.DailyUsage++; + cacheItem.TotalUsage++; + cacheItem.LastActivityTime = BeijingTimeExtension.GetBeijingTime(); + + _logger.LogDebug($"Token使用量已更新: {token}, 今日使用: {cacheItem.DailyUsage}, 总使用: {cacheItem.TotalUsage}"); + } + finally + { + _cacheLock.ExitWriteLock(); + } + } + else + { + _logger.LogWarning($"尝试更新未缓存的Token使用量: {token}"); + } + } + finally + { + _cacheLock.ExitUpgradeableReadLock(); + } + } + + /// + /// 获取Token的并发控制器 + /// + public async Task WaitForConcurrencyPermitAsync(string token) + { + + if (_concurrencyControllers.TryGetValue(token, out var controller)) + { + return await controller.Value.WaitAsync(token); + } + + _logger.LogWarning($"未找到Token的并发控制器: {token}"); + return false; + } + + /// + /// 释放Token的并发许可 + /// + public void ReleaseConcurrencyPermit(string token) + { + + if (_concurrencyControllers.TryGetValue(token, out var controller)) + { + controller.Value.Release(token); + } + else + { + _logger.LogWarning($"未找到Token的并发控制器无法释放: {token}"); + } + + } + + /// + /// 获取Token的并发状态信息 + /// + public (int maxCount, int currentlyExecuting, int available) GetConcurrencyStatus(string token) + { + _cacheLock.EnterReadLock(); + try + { + if (_concurrencyControllers.TryGetValue(token, out var controller)) + { + return (controller.Value.MaxCount, controller.Value.CurrentlyExecuting, controller.Value.AvailableCount); + } + return (0, 0, 0); + } + finally + { + _cacheLock.ExitReadLock(); + } + } + + + /// + /// 获取活跃Token列表 + /// + public List GetActiveTokens(TimeSpan activityThreshold) + { + _cacheLock.EnterReadLock(); + try + { + var cutoffTime = BeijingTimeExtension.GetBeijingTime() - activityThreshold; + var activeTokens = _tokenCache.Values + .Where(t => t.LastActivityTime > cutoffTime) + .ToList(); + + _logger.LogDebug($"找到 {activeTokens.Count} 个活跃Token (阈值: {activityThreshold.TotalMinutes} 分钟)"); + return activeTokens; + } + finally + { + _cacheLock.ExitReadLock(); + } + } + + /// + /// 移除不活跃的Token + /// + /// 活跃时间阈值 + /// 移除的Token数量 + public (int activateTokenCount, int notActivateTokenCount) RemoveNotActiveTokens(TimeSpan activityThreshold) + { + _cacheLock.EnterWriteLock(); + try + { + var cutoffTime = BeijingTimeExtension.GetBeijingTime() - activityThreshold; + var initialCount = _tokenCache.Count; + + // 找出需要移除的不活跃Token + var tokensToRemove = _tokenCache + .Where(kvp => kvp.Value.LastActivityTime <= cutoffTime) + .Select(kvp => kvp.Key) + .ToList(); + + if (tokensToRemove.Count == 0) + { + _logger.LogDebug($"没有找到需要移除的不活跃Token (阈值: {activityThreshold.TotalMinutes} 分钟)"); + return (initialCount, 0); + } + + // 移除不活跃的Token缓存 + var removedCount = 0; + foreach (var tokenKey in tokensToRemove) + { + if (_tokenCache.TryRemove(tokenKey, out var removedToken)) + { + // 同时清理对应的并发控制器 + if (_concurrencyControllers.TryRemove(tokenKey, out var controller)) + { + // 如果并发控制器已经被创建，需要释放资源 + if (controller.IsValueCreated) + { + controller.Value.Dispose(); + } + } + removedCount++; + _logger.LogDebug($"移除不活跃Token: {tokenKey}, 最后活跃时间: {removedToken.LastActivityTime:yyyy-MM-dd HH:mm:ss}"); + } + } + + _logger.LogInformation($"清理不活跃Token完成: 移除 {removedCount} 个Token (阈值: {activityThreshold.TotalMinutes} 分钟), 剩余: {_tokenCache.Count} 个"); + + return (_tokenCache.Count, removedCount); + } + finally + { + _cacheLock.ExitWriteLock(); + } + } + + /// + /// 移除指定的token + /// + /// + /// + public int RemoveToken(string token) + { + _cacheLock.EnterWriteLock(); + try + { + // 找出需要移除的不活跃Token + var tokensToRemove = _tokenCache + .Where(kvp => kvp.Value.Token == token) + .Select(kvp => kvp.Key) + .ToList(); + + if (tokensToRemove.Count == 0) + { + // 没有找到 + return 0; + } + + // 移除不活跃的Token缓存 + var removedCount = 0; + foreach (var tokenKey in tokensToRemove) + { + if (_tokenCache.TryRemove(tokenKey, out var removedToken)) + { + // 同时清理对应的并发控制器 + if (_concurrencyControllers.TryRemove(tokenKey, out var controller)) + { + // 如果并发控制器已经被创建，需要释放资源 + if (controller.IsValueCreated) + { + controller.Value.Dispose(); + } + } + removedCount++; + } + } + return _tokenCache.Count; + } + finally + { + _cacheLock.ExitWriteLock(); + } + } + + /// + /// 获取所有Token列表 + /// + public IEnumerable GetAllTokens() + { + _cacheLock.EnterReadLock(); + try + { + return _tokenCache.Values.ToList(); + } + finally + { + _cacheLock.ExitReadLock(); + } + } + + /// + /// 获取缓存统计信息（包含并发状态） + /// + public TokenCacheStats GetCacheStats() + { + _cacheLock.EnterReadLock(); + try + { + var now = BeijingTimeExtension.GetBeijingTime(); + var tokens = _tokenCache.Values.ToList(); + + var stats = new TokenCacheStats + { + TotalTokens = tokens.Count, + ActiveTokens = tokens.Count(t => t.LastActivityTime > now.AddMinutes(-5)), + InactiveTokens = tokens.Count(t => t.LastActivityTime <= now.AddMinutes(-5)), + TotalDailyUsage = tokens.Sum(t => t.DailyUsage), + TotalUsage = tokens.Sum(t => t.TotalUsage) + }; + + return stats; + } + finally + { + _cacheLock.ExitReadLock(); + } + } + + /// + /// 重置所有Token的日使用量 + /// + public void ResetDailyUsage() + { + _cacheLock.EnterWriteLock(); + try + { + // 重置缓存中的使用量 + foreach (var token in _tokenCache.Values) + { + token.DailyUsage = 0; + } + } + finally + { + _cacheLock.ExitWriteLock(); + } + } + + public string OriginToken + { + get => _originToken; + set + { + if (!string.IsNullOrWhiteSpace(value)) + { + _originToken = value; + } + else + { + // 如果尝试设置为空值，记录警告日志，可能请求原始的请求不可用 + _logger.LogWarning("尝试设置OriginToken为空值，可能请求原始的请求不可用！！"); + } + } + } + } +} diff --git a/LMS.service/Configuration/QuartzTaskSchedulerConfig.cs b/LMS.service/Configuration/QuartzTaskSchedulerConfig.cs index 7d16d2f..0b9cb14 100644 --- a/LMS.service/Configuration/QuartzTaskSchedulerConfig.cs +++ b/LMS.service/Configuration/QuartzTaskSchedulerConfig.cs @@ -1,62 +1,100 @@ -using LMS.Tools.TaskScheduler; +using LMS.Tools.MJPackage; +using LMS.Tools.TaskScheduler; using Quartz; -namespace LMS.service.Configuration + +public static class QuartzTaskSchedulerConfig { - public static class QuartzTaskSchedulerConfig + public static void AddQuartzTaskSchedulerService(this IServiceCollection services) { - public static void AddQuartzTaskSchedulerService(this IServiceCollection services) + services.AddQuartz(q => { - // 注册 Quartz 服务 - services.AddQuartz(q => + // 时区配置 + var chinaTimeZone = GetChinaTimeZone(); + + // 每月任务配置 + ConfigureMonthlyTask(q, chinaTimeZone); + + // 每日任务配置 + ConfigureDailyTask(q, chinaTimeZone); + + // 每30秒任务配置 + ConfigureThirtySecondTask(q, chinaTimeZone); + + ConfigureFiveMinuteTask(q, chinaTimeZone); + }); + + services.AddQuartzHostedService(options => + { + options.WaitForJobsToComplete = true; + }); + + // 注册作业类 + services.AddTransient(); + services.AddTransient(); + services.AddTransient(); + services.AddTransient(); + } + + private static TimeZoneInfo GetChinaTimeZone() + { + try + { + return TimeZoneInfo.FindSystemTimeZoneById("China Standard Time"); + } + catch + { + try { - - // 配置作业 - var jobKey = new JobKey("MonthlyTask", "DefaultGroup"); - - // 方法1：通过配置属性设置时区 - // 获取中国时区 - TimeZoneInfo chinaTimeZone; - try - { - // 尝试获取 Windows 时区名称 - chinaTimeZone = TimeZoneInfo.FindSystemTimeZoneById("China Standard Time"); - } - catch - { - try - { - // 尝试获取 Linux 时区名称 - chinaTimeZone = TimeZoneInfo.FindSystemTimeZoneById("Asia/Shanghai"); - } - catch - { - // 如果都不可用，使用 UTC+8 - chinaTimeZone = TimeZoneInfo.CreateCustomTimeZone( - "China_Custom", - new TimeSpan(8, 0, 0), - "China Custom Time", - "China Standard Time"); - } - } - - // 添加作业 - q.AddJob(opts => opts.WithIdentity(jobKey)); - - // 添加触发器 - 每月1号凌晨0点执行 - q.AddTrigger(opts => opts - .ForJob(jobKey) - .WithIdentity("MonthlyTaskTrigger", "DefaultGroup") - .WithCronSchedule("0 0 0 1 * ?")); // 每月1号凌晨0点 - }); - - // 添加 Quartz 托管服务 - services.AddQuartzHostedService(options => + return TimeZoneInfo.FindSystemTimeZoneById("Asia/Shanghai"); + } + catch { - options.WaitForJobsToComplete = true; - }); - - // 注册作业 - services.AddTransient(); + return TimeZoneInfo.CreateCustomTimeZone( + "China_Custom", + new TimeSpan(8, 0, 0), + "China Custom Time", + "China Standard Time"); + } } } -} + + private static void ConfigureMonthlyTask(IServiceCollectionQuartzConfigurator q, TimeZoneInfo timeZone) + { + var jobKey = new JobKey("MonthlyTask", "DefaultGroup"); + q.AddJob(opts => opts.WithIdentity(jobKey)); + q.AddTrigger(opts => opts + .ForJob(jobKey) + .WithIdentity("MonthlyTaskTrigger", "DefaultGroup") + .WithCronSchedule("0 0 0 1 * ?", x => x.InTimeZone(timeZone))); + } + + private static void ConfigureDailyTask(IServiceCollectionQuartzConfigurator q, TimeZoneInfo timeZone) + { + var jobKey = new JobKey("DailyTask", "DefaultGroup"); + q.AddJob(opts => opts.WithIdentity(jobKey)); + q.AddTrigger(opts => opts + .ForJob(jobKey) + .WithIdentity("DailyTaskTrigger", "DefaultGroup") + .WithCronSchedule("0 10 0 * * ?", x => x.InTimeZone(timeZone))); // 每天凌晨0点10分执行 + } + + private static void ConfigureThirtySecondTask(IServiceCollectionQuartzConfigurator q, TimeZoneInfo timeZone) + { + var jobKey = new JobKey("ThirtySecondTask", "DefaultGroup"); + q.AddJob(opts => opts.WithIdentity(jobKey)); + q.AddTrigger(opts => opts + .ForJob(jobKey) + .WithIdentity("ThirtySecondTaskTrigger", "DefaultGroup") + .WithCronSchedule("*/30 * * * * ?", x => x.InTimeZone(timeZone))); // 每30秒执行一次 + } + + private static void ConfigureFiveMinuteTask(IServiceCollectionQuartzConfigurator q, TimeZoneInfo timeZone) + { + var jobKey = new JobKey("FiveMinuteTask", "DefaultGroup"); + q.AddJob(opts => opts.WithIdentity(jobKey)); + q.AddTrigger(opts => opts + .ForJob(jobKey) + .WithIdentity("FiveMinuteTaskTrigger", "DefaultGroup") + .WithCronSchedule("0 */5 * * * ?", x => x.InTimeZone(timeZone))); // 每5分钟执行一次 + } +} \ No newline at end of file diff --git a/LMS.service/Configuration/ServiceConfiguration.cs b/LMS.service/Configuration/ServiceConfiguration.cs index 945d580..44249a4 100644 --- a/LMS.service/Configuration/ServiceConfiguration.cs +++ b/LMS.service/Configuration/ServiceConfiguration.cs @@ -5,12 +5,14 @@ using LMS.DAO.UserDAO; using LMS.service.Configuration.InitConfiguration; using LMS.service.Extensions.Mail; using LMS.service.Service; +using LMS.service.Service.MJPackage; using LMS.service.Service.Other; using LMS.service.Service.PermissionService; using LMS.service.Service.PromptService; using LMS.service.Service.RoleService; using LMS.service.Service.SoftwareService; using LMS.service.Service.UserService; +using LMS.Tools.MJPackage; namespace Lai_server.Configuration { @@ -38,6 +40,8 @@ namespace Lai_server.Configuration services.AddScoped(); services.AddScoped(); services.AddScoped(); + services.AddScoped(); + services.AddScoped(); // 注入 DAO services.AddScoped(); @@ -53,6 +57,20 @@ namespace Lai_server.Configuration // 添加分布式缓存(用于存储验证码) services.AddDistributedMemoryCache(); + + // 注册自定义服务 + services.AddSingleton(); + services.AddScoped(); + services.AddScoped(); + services.AddScoped(); + + + // 注册后台服务 + services.AddHostedService(); + services.AddHostedService(); + + //services.AddHostedService(); + //services.AddHostedService(); } } } diff --git a/LMS.service/Controllers/MJPackageController.cs b/LMS.service/Controllers/MJPackageController.cs new file mode 100644 index 0000000..f4815c4 --- /dev/null +++ b/LMS.service/Controllers/MJPackageController.cs @@ -0,0 +1,115 @@ +using Betalgo.Ranul.OpenAI.ObjectModels.RealtimeModels; +using LMS.Repository.MJPackage; +using LMS.service.Extensions.Attributes; +using LMS.service.Service.MJPackage; +using LMS.Tools.MJPackage; +using Microsoft.AspNetCore.Mvc; +using Newtonsoft.Json; +using System.Text; +using System.Text.Json; + +namespace LMS.service.Controllers +{ + [Route("api/[controller]")] + [ApiController] + public class MJPackageController(TokenUsageTracker usageTracker, ITaskConcurrencyManager taskConcurrencyManager, ILogger logger, IMJPackageService mJPackageService) : ControllerBase + { + private readonly TokenUsageTracker _usageTracker = usageTracker; + private readonly ILogger _logger = logger; + private readonly ITaskConcurrencyManager _taskConcurrencyManager = taskConcurrencyManager; + private readonly IMJPackageService _mJPackageService = mJPackageService; + + [HttpPost("mj/submit/imagine")] + [RateLimit] + public async Task Imagine([FromBody] MJSubmitImageModel model) + { + + string token = (string)(HttpContext.Items["UseToken"] ?? string.Empty); + string requestToken = (string)(HttpContext.Items["RequestToken"] ?? string.Empty); + if (string.IsNullOrWhiteSpace(token)) + { + return Unauthorized("API token is empty"); + } + if (string.IsNullOrWhiteSpace(requestToken)) + { + return Unauthorized("API token is empty"); + } + + using HttpClient client = new HttpClient(); + client.DefaultRequestHeaders.Add("Authorization", "Bearer sk-" + token); + + model.NotifyHook = "https://lms.laitool.cn/api/MJPackage/mj/mj-notify-hook"; + string body = JsonConvert.SerializeObject(model); + + client.Timeout = Timeout.InfiniteTimeSpan; + string mjUrl = "https://api.laitool.cc/mj/submit/imagine"; + var response = await client.PostAsync(mjUrl, new StringContent(body, Encoding.UTF8, "application/json")); + // 读取响应内容 + string content = await response.Content.ReadAsStringAsync(); + + // 直接返回原始响应，包括状态码和内容 + var res = new ContentResult + { + Content = content, + ContentType = response.Content.Headers.ContentType?.ToString() ?? "application/json", + StatusCode = (int)response.StatusCode + }; + // 判断请求任务的状态，判断是不是需要释放 + if (res.StatusCode != 200 && res.StatusCode != 201) + { + // 释放 + _usageTracker.ReleaseConcurrencyPermit(requestToken); + _logger.LogInformation($"请求失败，Token并发许可已释放: {token}, 状态码: {res.StatusCode}"); + return res; + } + + // 判断是不是提交成功，并且记录请求返回的ID + // 把 Content 转换为 匿名 对象 + var result = JsonConvert.DeserializeAnonymousType(content, new + { + code = 1, + description = "提交成功", + result = 1320098173412546 + }); + if (result == null) + { + // 失败 + _usageTracker.ReleaseConcurrencyPermit(requestToken); + _logger.LogInformation($"请求失败，返回的请求体为空，Token并发许可已释放: {token}, 状态码: {res.StatusCode}"); + return res; + } + else + { + if (result.code != 1 && result.code != 22) + { + _usageTracker.ReleaseConcurrencyPermit(requestToken); + _logger.LogInformation($"请求失败，code: {result.code}，Token并发许可已释放: {token}, 状态码: {res.StatusCode}"); + return res; + } + } + // 开始写入任务 + await _taskConcurrencyManager.CreateTaskAsync(requestToken, result.result.ToString()); + + return res; + } + + [HttpGet("mj/task/{id}/fetch")] + public async Task> Fetch(string id) + { + string token = HttpContext.Request.Headers.Authorization.ToString() + .Replace("Bearer ", "", StringComparison.OrdinalIgnoreCase) + .Trim(); + + var res = await _mJPackageService.FetchTaskAsync(id, token); + return res; + } + + [HttpPost("mj/mj-notify-hook")] + //[Route("mjPackage/mj/mj-notify-hook")] + public async Task MJNotifyHook([FromBody] JsonElement model) + { + return await _mJPackageService.MJNotifyHookAsync(model); + } + } +} + diff --git a/LMS.service/Controllers/TokenManagementController.cs b/LMS.service/Controllers/TokenManagementController.cs new file mode 100644 index 0000000..00891f7 --- /dev/null +++ b/LMS.service/Controllers/TokenManagementController.cs @@ -0,0 +1,212 @@ +using LMS.Common.Extensions; +using LMS.Repository.DB; +using LMS.Repository.DTO; +using LMS.Repository.MJPackage; +using LMS.service.Service.MJPackage; +using Microsoft.AspNetCore.Authorization; +using Microsoft.AspNetCore.Mvc; +using System.ComponentModel.DataAnnotations; + +namespace LMS.service.Controllers +{ + [Route("api/[controller]/[action]")] + [ApiController] + public class TokenManagementController : ControllerBase + { + private readonly ITokenManagementService _tokenManagementService; + private readonly ILogger _logger; + + public TokenManagementController( + ITokenManagementService tokenManagementService, + ILogger logger) + { + _tokenManagementService = tokenManagementService; + _logger = logger; + } + + #region 用户-使用Token查询对应的任务 + + [HttpGet("{token}")] + public async Task>>> QueryTokenTaskCollection(string token, [Required] int page, [Required] int pageSize, string? thirdPartyTaskId) + { + return await _tokenManagementService.QueryTokenTaskCollection(token, page, pageSize, thirdPartyTaskId); + } + + #endregion + + #region 用户-查询Token是不是存在，返回简单数据 + + [HttpGet("{token}")] + public async Task>> GetTokenItem(string token) + { + return await _tokenManagementService.GetTokenItem(token); + } + + #endregion + + + #region 管理员-新增Token + + [HttpPost] + [Authorize] + public async Task>> AddToken([FromBody] AddOrModifyTokenModel model) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.AddToken(requestUserId, model); + } + + #endregion + + #region 管理员-修改token + + [HttpPost("{tokenId}")] + [Authorize] + public async Task>> ModifyToken(long tokenId, [FromBody] AddOrModifyTokenModel model) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.ModifyToken(requestUserId, tokenId, model); + } + + + #endregion + + #region 管理员-删除token + + [HttpDelete("{tokenId}")] + [Authorize] + public async Task>> DeleteToken(long tokenId) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.DeleteToken(requestUserId, tokenId); + } + + #endregion + + #region 管理员-获取所有Token + + [HttpGet] + [Authorize] + public async Task>>> QueryTokenCollection([Required] int page, [Required] int pageSize, string? token, long? tokenId, bool? efficient) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.QueryTokenCollection(page, pageSize, token, tokenId, efficient, requestUserId); + } + + #endregion + + #region 管理员-获取指定ID得Token + + [HttpGet("{tokenId}")] + [Authorize] + public async Task>> QueryTokenById(long tokenId) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.QueryTokenById(tokenId, requestUserId); + } + + #endregion + + #region 管理员-获取所有的任务 + + [HttpGet] + [Authorize] + public async Task>>> QueryTaskCollection([Required] int page, [Required] int pageSize, string? thirdPartyTaskId, string? token, long? tokenId) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.QueryTaskCollection(requestUserId, page, pageSize, thirdPartyTaskId, token, tokenId); + } + + #endregion + + #region 管理员-删除小于指定时间的任务 + + [HttpDelete("{timestamp}")] + [Authorize] + public async Task>> DeleteMJTaskEarlyTimestamp(long timestamp) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.DeleteMJTaskEarlyTimestamp(requestUserId, timestamp); + } + + #endregion + + #region 管理员-删除指定ID的任务 + + [HttpDelete("{taskId}")] + [Authorize] + public async Task>> DeleteMJTask(string taskId) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.DeleteMJTask(requestUserId, taskId); + } + + #endregion + + #region 管理员-手动刷新token缓存 + + /// + /// 手动刷新Token缓存 + /// + /// Token字符串 + /// 刷新结果 + [HttpPost("{token}")] + [Authorize] + public async Task>> RefreshToken(string token) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.RefreshToken(requestUserId, token); + } + #endregion + + #region 管理员-获取活跃的token + + [HttpGet] + [Authorize] + public async Task>>> GetActiveTokens([FromQuery] int minutes = 5) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.GetActiveTokens(requestUserId, minutes); + } + + #endregion + + #region 管理员-移除不活跃的token + + [HttpGet] + [Authorize] + public async Task>> RemoveNotActiveTokens([FromQuery] int minutes = 5) + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.RemoveNotActiveTokens(requestUserId, minutes); + } + + #endregion + + #region 内存统计接口 + + /// + /// 健康检查端点 + /// + /// 系统健康状态 + [HttpGet] + [Authorize] + public async Task>> GetHealth() + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.GetHealth(requestUserId); + } + + #endregion + + #region 任务统计 + [HttpGet] + [Authorize] + public async Task>> GetDayTaskStatistics() + { + long requestUserId = ConvertExtension.ObjectToLong(HttpContext.Items["UserId"] ?? 0); + return await _tokenManagementService.GetDayTaskStatistics(requestUserId); + } + + #endregion + } +} diff --git a/LMS.service/Controllers/UserController.cs b/LMS.service/Controllers/UserController.cs index 890fb49..8328c2a 100644 --- a/LMS.service/Controllers/UserController.cs +++ b/LMS.service/Controllers/UserController.cs @@ -80,7 +80,7 @@ namespace LMS.service.Controllers HttpOnly = true, Secure = true, // 如果使用 HTTPS SameSite = SameSiteMode.None, - Expires = DateTime.UtcNow.AddDays(7), + Expires = BeijingTimeExtension.GetBeijingTime().AddDays(7), }; Response.Cookies.Append("refreshToken", ((LoginResponse)apiResponse.Data).RefreshToken, cookieOptions); return apiResponse; diff --git a/LMS.service/Extensions/Attributes/RateLimitAttribute.cs b/LMS.service/Extensions/Attributes/RateLimitAttribute.cs new file mode 100644 index 0000000..70859ef --- /dev/null +++ b/LMS.service/Extensions/Attributes/RateLimitAttribute.cs @@ -0,0 +1,176 @@ +using LMS.Common.Extensions; +using LMS.Tools.MJPackage; +using Microsoft.AspNetCore.Mvc; +using Microsoft.AspNetCore.Mvc.Filters; + +namespace LMS.service.Extensions.Attributes +{ + [AttributeUsage(AttributeTargets.Method)] + public class RateLimitAttribute : ActionFilterAttribute, IAsyncActionFilter + { + private string _token; + private DateTime _startTime; + private bool _concurrencyAcquired = false; + + public RateLimitAttribute() + { + } + + public override async Task OnActionExecutionAsync(ActionExecutingContext context, ActionExecutionDelegate next) + { + _startTime = BeijingTimeExtension.GetBeijingTime(); + + // 从服务容器获取需要的服务 + var serviceProvider = context.HttpContext.RequestServices; + var logger = serviceProvider.GetRequiredService>(); + var tokenService = serviceProvider.GetRequiredService(); + var usageTracker = serviceProvider.GetRequiredService(); + + try + { + // 1. 获取Token + _token = context.HttpContext.Request.Headers.Authorization.ToString() + .Replace("Bearer ", "", StringComparison.OrdinalIgnoreCase) + .Trim(); + + if (string.IsNullOrEmpty(_token)) + { + logger.LogWarning($"请求缺少Token, IP: {context.HttpContext.Connection.RemoteIpAddress}"); + context.Result = new UnauthorizedObjectResult("Missing API token"); + return; + } + + // 2. 获取Token配置 + var tokenConfig = await tokenService.GetTokenAsync(_token); + if (tokenConfig == null) + { + context.Result = new ObjectResult("Invalid token") + { + StatusCode = StatusCodes.Status403Forbidden + }; + return; + } + + // 3. 检查Token是不是到期 + if (tokenConfig.ExpiresAt != null && tokenConfig.ExpiresAt < BeijingTimeExtension.GetBeijingTime()) + { + context.Result = new ObjectResult("expired token") + { + StatusCode = StatusCodes.Status403Forbidden + }; + return; + } + + // 4. 判断当前Token得上一次使用时间是否超过了10分钟，超过了重新从数据库获取 + if (tokenConfig.LastActivityTime < BeijingTimeExtension.GetBeijingTime().AddMinutes(-10)) + { + logger.LogInformation($"Token {_token} 上次活动时间超过10分钟，重新从数据库获取配置"); + tokenConfig = await tokenService.GetDatabaseTokenAsync(_token); + if (tokenConfig == null) + { + context.Result = new ObjectResult("Invalid or expired token") + { + StatusCode = StatusCodes.Status403Forbidden + }; + return; + } + } + + // 5. 检查日使用限制 + if (tokenConfig.DailyLimit > 0 && tokenConfig.DailyUsage >= tokenConfig.DailyLimit) + { + logger.LogWarning($"Token日限制已达上限: {_token}, 当前使用: {tokenConfig.DailyUsage}, 限制: {tokenConfig.DailyLimit}"); + context.Result = new ObjectResult("Daily limit exceeded") + { + StatusCode = StatusCodes.Status403Forbidden + }; + return; + } + + // 6. 检查总使用限制 + if (tokenConfig.TotalLimit > 0 && tokenConfig.TotalUsage >= tokenConfig.TotalLimit) + { + logger.LogWarning($"Token总限制已达上限: {_token}, 当前使用: {tokenConfig.TotalUsage}, 限制: {tokenConfig.TotalLimit}"); + context.Result = new ObjectResult("Total limit exceeded") + { + StatusCode = StatusCodes.Status403Forbidden + }; + return; + } + + // 7. 并发控制 + var (maxCount, currentlyExecuting, available) = usageTracker.GetConcurrencyStatus(_token); + logger.LogInformation($"Token并发状态: {_token}, 最大: {maxCount}, 执行中: {currentlyExecuting}, 可用: {available}"); + + // 等待获取并发许可 + _concurrencyAcquired = await usageTracker.WaitForConcurrencyPermitAsync(_token); + if (!_concurrencyAcquired) + { + logger.LogInformation($"Token并发限制超出: {_token}, 并发限制: {tokenConfig.ConcurrencyLimit}"); + context.Result = new ObjectResult($"Concurrency limit exceeded (max: {tokenConfig.ConcurrencyLimit})") + { + StatusCode = StatusCodes.Status429TooManyRequests + }; + return; + } + + logger.LogInformation($"Token验证成功，开始处理请求: {_token}, 并发限制: {tokenConfig.ConcurrencyLimit}"); + + if (string.IsNullOrWhiteSpace(tokenConfig.UseToken)) + { + context.Result = new ObjectResult($"Token Error") + { + StatusCode = StatusCodes.Status401Unauthorized + }; + return; + } + + // 将新token存储在HttpContext.Items中 + context.HttpContext.Items["UseToken"] = tokenConfig.UseToken; + context.HttpContext.Items["RequestToken"] = _token; + + // 执行 Action + var executedContext = await next(); + + // 6. 更新使用计数 (仅成功请求) + if (executedContext.HttpContext.Response.StatusCode < 400) + { + tokenService.IncrementUsage(_token); + + var duration = BeijingTimeExtension.GetBeijingTime() - _startTime; + logger.LogInformation($"请求处理成功: Token={_token}, 状态码={executedContext.HttpContext.Response.StatusCode}, 耗时={duration.TotalMilliseconds}ms"); + } + else + { + logger.LogWarning($"请求处理失败: Token={_token}, 状态码={executedContext.HttpContext.Response.StatusCode}"); + } + } + catch (Exception ex) + { + // 在异常情况下也要释放并发许可 + if (_concurrencyAcquired) + { + + usageTracker.ReleaseConcurrencyPermit(_token); + + } + logger.LogError(ex, $"处理Token请求时发生错误: {_token}，已释放Token许可！"); + context.Result = new ObjectResult("Internal server error") + { + StatusCode = StatusCodes.Status500InternalServerError + }; + } + finally + { + // 7. 释放并发许可 + //if (_concurrencyAcquired) + //{ + // usageTracker.ReleaseConcurrencyPermit(_token); + + // var newStatus = usageTracker.GetConcurrencyStatus(_token); + // logger.LogInformation($"Token并发许可已释放: {_token}, 执行中: {newStatus.currentlyExecuting}, 可用: {newStatus.available}"); + //} + } + } + } +} diff --git a/LMS.service/Program.cs b/LMS.service/Program.cs index 54117ee..16b7090 100644 --- a/LMS.service/Program.cs +++ b/LMS.service/Program.cs @@ -5,6 +5,7 @@ using LMS.Repository.Models.DB; using LMS.service.Configuration; using LMS.service.Configuration.InitConfiguration; using LMS.service.Extensions.Middleware; +using LMS.Tools.MJPackage; using Microsoft.AspNetCore.Identity; using Microsoft.EntityFrameworkCore; using Serilog; @@ -89,9 +90,6 @@ builder.Services.AddEndpointsApiExplorer(); // עSwagger builder.Services.AddSwaggerService(); -builder.Services.AddHostedService(); -builder.Services.AddHostedService(); - var app = builder.Build(); @@ -120,6 +118,7 @@ app.MapControllers(); // ڹܵʹIPм app.UseIpRateLimiting(); +// Ӷ̬Ȩ޵м app.UseMiddleware(); app.UseEndpoints(endpoints => { diff --git a/LMS.service/Service/ForwardWordService.cs b/LMS.service/Service/ForwardWordService.cs index 8a4e242..99d1f56 100644 --- a/LMS.service/Service/ForwardWordService.cs +++ b/LMS.service/Service/ForwardWordService.cs @@ -115,6 +115,16 @@ public class ForwardWordService(ApplicationDbContext context) throw new Exception("参数错误"); } + // 判断请求的url是不是满足条件 + if (string.IsNullOrEmpty(request.GptUrl)) + { + throw new Exception("请求的url为空"); + } + if (!request.GptUrl.StartsWith("https://ark.cn-beijing.volces.com") && !request.GptUrl.StartsWith("https://api.moonshot.cn") && !request.GptUrl.StartsWith("https://laitool.net") && !request.GptUrl.StartsWith("https://api.laitool.cc") && !request.GptUrl.StartsWith("https://laitool.cc")) + { + throw new Exception("请求的url不合法"); + } + // 获取提示词预设 Prompt? prompt = await _context.Prompt.FirstOrDefaultAsync(x => x.PromptTypeId == request.PromptTypeId && x.Id == request.PromptId); if (prompt == null) diff --git a/LMS.service/Service/MJPackage/IMJPackageService.cs b/LMS.service/Service/MJPackage/IMJPackageService.cs new file mode 100644 index 0000000..99bcfa9 --- /dev/null +++ b/LMS.service/Service/MJPackage/IMJPackageService.cs @@ -0,0 +1,18 @@ +using Microsoft.AspNetCore.Mvc; +using System.Text.Json; + +namespace LMS.service.Service.MJPackage +{ + public interface IMJPackageService + { + /// + /// mj 得 /mj/task/{id}/fetch 转发接口得实现 + /// 内含重试请求，全部重试失败 会尝试充数据库中读取消息 + /// + /// + /// + /// + Task> FetchTaskAsync(string id, string token); + Task MJNotifyHookAsync(JsonElement model); + } +} diff --git a/LMS.service/Service/MJPackage/ITokenManagementService.cs b/LMS.service/Service/MJPackage/ITokenManagementService.cs new file mode 100644 index 0000000..3e54a0f --- /dev/null +++ b/LMS.service/Service/MJPackage/ITokenManagementService.cs @@ -0,0 +1,26 @@ +using LMS.Repository.DB; +using LMS.Repository.DTO; +using LMS.Repository.MJPackage; +using Microsoft.AspNetCore.Mvc; + +namespace LMS.service.Service.MJPackage +{ + public interface ITokenManagementService + { + Task>>> QueryTokenTaskCollection(string token, int page, int pageSize, string? thirdPartyTaskId); + Task>> GetTokenItem(string token); + Task>> AddToken(long requestUserId, AddOrModifyTokenModel model); + Task>> ModifyToken(long requestUserId, long tokenId, AddOrModifyTokenModel model); + Task>> DeleteToken(long requestUserId, long tokenId); + Task>>> QueryTaskCollection(long requestUserId, int page, int pageSize, string? thirdPartyTaskId, string? token, long? tokenId); + Task>>> QueryTokenCollection(int page, int pageSize, string? token, long? tokenId, bool? efficient, long requestUserId); + Task>> DeleteMJTaskEarlyTimestamp(long requestUserId, long timestamp); + Task>> DeleteMJTask(long requestUserId, string taskId); + Task>> RefreshToken(long requestUserId, string token); + Task>>> GetActiveTokens(long requestUserId, int minutes); + Task>> RemoveNotActiveTokens(long requestUserId, int minutes); + Task>> GetHealth(long requestUserId); + Task>> QueryTokenById(long tokenId, long requestUserId); + Task>> GetDayTaskStatistics(long requestUserId); + } +} diff --git a/LMS.service/Service/MJPackage/MJPackageService.cs b/LMS.service/Service/MJPackage/MJPackageService.cs new file mode 100644 index 0000000..40f801c --- /dev/null +++ b/LMS.service/Service/MJPackage/MJPackageService.cs @@ -0,0 +1,295 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.Repository.DB; +using LMS.Tools.MJPackage; +using Microsoft.AspNetCore.Mvc; +using Microsoft.EntityFrameworkCore; +using System.Net.Sockets; +using System.Text.Json; +using static Betalgo.Ranul.OpenAI.ObjectModels.StaticValues.AssistantsStatics.MessageStatics; + +namespace LMS.service.Service.MJPackage +{ + public class MJPackageService(ILogger logger, ApplicationDbContext dbContext, TokenUsageTracker usageTracker, ITaskConcurrencyManager taskConcurrencyManager, ITokenService tokenService) : IMJPackageService + { + private readonly ILogger _logger = logger; + private readonly ApplicationDbContext _dbContext = dbContext; + private readonly TokenUsageTracker _usageTracker = usageTracker; + private readonly ITaskConcurrencyManager _taskConcurrencyManager = taskConcurrencyManager; + private readonly ITokenService _tokenService = tokenService; + + public async Task> FetchTaskAsync(string id, string token) + { + + // 参数验证 + var validationResult = ValidateParameters(id, token); + if (validationResult != null) return validationResult; + + // 获取UseToken + // 2. 获取Token配置 + var tokenConfig = await _tokenService.GetTokenAsync(token); + if (tokenConfig == null || string.IsNullOrWhiteSpace(tokenConfig.UseToken)) + { + return new UnauthorizedObjectResult(new { error = "无效或过期的Token" }); + } + + // 尝试原始API + var originResult = await TryOriginApiAsync(id); + if (originResult != null) return originResult; + + // 尝试备用API + var backupResult = await TryBackupApiAsync(id, tokenConfig.UseToken); + if (backupResult != null) return backupResult; + + // 从数据库获取缓存数据 + return await GetTaskFromDatabaseAsync(id, token); + } + + + #region MJ Package 的回调处理 + + public async Task MJNotifyHookAsync(JsonElement model) + { + try + { + string rawJson = model.GetRawText(); + + // 尝试获取ID字段 + string mjId = string.Empty; + if (model.TryGetProperty("id", out var idElement)) + { + mjId = idElement.ToString(); + } + else if (model.TryGetProperty("Id", out var idElementCap)) + { + mjId = idElementCap.ToString(); + } + + if (string.IsNullOrWhiteSpace(mjId)) + { + _logger.LogWarning("MJNotifyHook: 接收到的回调数据中缺少ID"); + return new BadRequestObjectResult("缺少ID"); + } + + // 获取任务 + var mjTask = await _taskConcurrencyManager.GetTaskInfoByThirdPartyIdAsync(mjId); + if (mjTask == null) + { + return new NotFoundObjectResult($"未找到ID为 {mjId} 的任务"); + } + + // 尝试获取状态字段 + string status = MJTaskStatus.SUBMITTED; + + if (model.TryGetProperty("status", out var statusElement)) + { + status = statusElement.GetString() ?? MJTaskStatus.SUBMITTED; + } + else if (model.TryGetProperty("Status", out var statusElementCap)) + { + status = statusElementCap.GetString() ?? MJTaskStatus.SUBMITTED; + } + + MJApiTasks mJApiTasks = new() + { + TaskId = mjTask.TaskId, + Token = mjTask.Token, + Status = status, + StartTime = mjTask.StartTime, + EndTime = null, + ThirdPartyTaskId = mjId, + Properties = rawJson // 或者直接存储 model + }; + + if (mjTask.Status == MJTaskStatus.SUCCESS || mjTask.Status == MJTaskStatus.FAILURE || mjTask.Status == MJTaskStatus.CANCEL) + { + // 当前任务已经被释放过了 + // 开始修改数据 + mJApiTasks.EndTime = BeijingTimeExtension.GetBeijingTime(); + await _taskConcurrencyManager.UpdateTaskInDatabase(mJApiTasks); + + return new OkObjectResult(null); + } + + if (status == MJTaskStatus.SUCCESS || status == MJTaskStatus.FAILURE || status == MJTaskStatus.CANCEL) + { + mJApiTasks.EndTime = BeijingTimeExtension.GetBeijingTime(); + _usageTracker.ReleaseConcurrencyPermit(mjTask.Token); + } + + // 开始修改数据 + await _taskConcurrencyManager.UpdateTaskInDatabase(mJApiTasks); + + return new OkObjectResult(null); + } + catch (Exception ex) + { + _logger.LogError(ex, "MJNotifyHook 处理回调数据时发生异常"); + return new StatusCodeResult(StatusCodes.Status500InternalServerError); + } + } + + #endregion + + #region 私有辅助方法 + + private static ActionResult? ValidateParameters(string id, string token) + { + if (string.IsNullOrWhiteSpace(id)) + return new BadRequestObjectResult(new { error = "任务ID不能为空" }); + + if (string.IsNullOrWhiteSpace(token)) + return new UnauthorizedObjectResult(new { error = "缺少授权Token" }); + + return null; + } + + private async Task?> TryOriginApiAsync(string id) + { + const string originUrl = "https://mjapi.bzu.cn/mj/task/{0}/fetch"; + + // 判断 原始token 不存在 直接 返回空 + string orginToken = await _tokenService.GetOriginToken(); + if (string.IsNullOrWhiteSpace(orginToken)) + { + return null; + } + + try + { + using var client = CreateHttpClient(orginToken, false); + var response = await client.GetAsync(string.Format(originUrl, id)); + var content = await response.Content.ReadAsStringAsync(); + if ((int)response.StatusCode == 204 || string.IsNullOrWhiteSpace(content)) + { + throw new Exception("原始API返回204 No Content或空内容"); + } + + return new ContentResult + { + Content = content, + ContentType = response.Content.Headers.ContentType?.ToString() ?? "application/json", + StatusCode = (int)response.StatusCode + }; + } + catch (Exception ex) + { + _logger.LogWarning(ex, "原始API调用失败，TaskId: {TaskId}，准备尝试备用API", id); + return null; + } + } + + private static bool IsRetriableException(Exception ex) + { + return ex is HttpRequestException || + ex is TaskCanceledException || + ex is SocketException; + } + + private async Task?> TryBackupApiAsync(string id, string useToken) + { + const string backupUrlTemplate = "https://api.laitool.cc/mj/task/{0}/fetch"; + const int maxRetries = 3; + const int baseDelayMs = 1000; + + using var client = CreateHttpClient($"Bearer sk-{useToken}", true); + var backupUrl = string.Format(backupUrlTemplate, id); + + for (int attempt = 1; attempt <= maxRetries; attempt++) + { + try + { + var response = await client.GetAsync(backupUrl); + var content = await response.Content.ReadAsStringAsync(); + + _logger.LogInformation("备用API调用成功，TaskId: {TaskId}, Attempt: {Attempt}, StatusCode: {StatusCode}", + id, attempt, response.StatusCode); + + return new ContentResult + { + Content = content, + ContentType = response.Content.Headers.ContentType?.ToString() ?? "application/json", + StatusCode = (int)response.StatusCode + }; + } + catch (Exception ex) when (IsRetriableException(ex)) + { + if (attempt < maxRetries) + { + var delay = baseDelayMs * (int)Math.Pow(2, attempt - 1); + _logger.LogWarning(ex, "备用API调用失败，TaskId: {TaskId}, Attempt: {Attempt}, 将在{Delay}ms后重试", + id, attempt, delay); + await Task.Delay(delay); + } + else + { + _logger.LogError(ex, "备用API调用最终失败，TaskId: {TaskId}, MaxAttempts: {MaxAttempts}", + id, maxRetries); + } + } + catch (Exception ex) + { + _logger.LogError(ex, "备用API调用发生不可重试异常，TaskId: {TaskId}, Attempt: {Attempt}", + id, attempt); + break; + } + } + + return null; + } + + + private async Task> GetTaskFromDatabaseAsync(string id, string token) + { + try + { + // 这里需要根据您的数据库结构调整 + // 假设您有一个任务表存储MJ任务的状态信息 + var taskFromDb = await _dbContext.MJApiTasks // 替换为您实际的表名 + .Where(x => x.ThirdPartyTaskId == id && x.Token == token) // 确保用户只能访问自己的任务 + .FirstOrDefaultAsync(); + + if (taskFromDb != null && !string.IsNullOrWhiteSpace(taskFromDb.Properties)) + { + return new OkObjectResult(taskFromDb.Properties); + } + else + { + return new ObjectResult(new + { + error = "服务暂时不可用", + message = "无法获取任务状态，MJ服务连接失败且本地无缓存数据", + ThirdPartyTaskId = id, + }) + { + StatusCode = 502 + }; + } + } + catch (Exception dbEx) + { + _logger.LogError(dbEx, $"从数据库获取任务数据时发生异常，TaskId: {id}"); + + return new ObjectResult(new + { + error = "系统异常", + message = "获取任务状态失败，请稍后重试", + ThirdPartyTaskId = id + }) + { + StatusCode = 500 + }; + } + } + + private static HttpClient CreateHttpClient(string authorization, bool isBearerToken) + { + var client = new HttpClient(); + client.DefaultRequestHeaders.Add("Authorization", authorization); + client.Timeout = TimeSpan.FromSeconds(30); + return client; + } + + #endregion + } +} diff --git a/LMS.service/Service/MJPackage/TokenManagementService.cs b/LMS.service/Service/MJPackage/TokenManagementService.cs new file mode 100644 index 0000000..d473bc9 --- /dev/null +++ b/LMS.service/Service/MJPackage/TokenManagementService.cs @@ -0,0 +1,878 @@ +using LMS.Common.Extensions; +using LMS.DAO; +using LMS.DAO.UserDAO; +using LMS.Repository.DB; +using LMS.Repository.DTO; +using LMS.Repository.MJPackage; +using LMS.Tools.MJPackage; +using Microsoft.AspNetCore.Mvc; +using Microsoft.EntityFrameworkCore; +using System.Data; +using System.Diagnostics; +using static LMS.Common.Enums.ResponseCodeEnum; + +namespace LMS.service.Service.MJPackage +{ + public class TokenManagementService( + ApplicationDbContext dbContext, + TokenUsageTracker usageTracker, + ITokenService tokenService, + UserBasicDao userBasicDao, + ILogger logger) : ITokenManagementService + { + + private readonly ApplicationDbContext _dbContext = dbContext; + private readonly TokenUsageTracker _usageTracker = usageTracker; + private readonly ITokenService _tokenService = tokenService; + private readonly ILogger _logger = logger; + private readonly UserBasicDao _userBasicDao = userBasicDao; + + #region 使用Token查询对应的任务-用户 + /// + /// 查询任务集合 通过参数 + /// + /// + /// + /// + /// + /// + public async Task>>> QueryTokenTaskCollection(string token, int page, int pageSize, string? thirdPartyTaskId) + { + try + { + if (string.IsNullOrWhiteSpace(token)) + { + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不能为空"); + } + + TokenCacheItem? tokenCache = await _tokenService.GetDatabaseTokenAsync(token, true); + if (tokenCache == null) + { + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在或已过期"); + } + + // 处理token数据 + TokenAndTaskCollection tokenCacheItem = new() + { + Id = tokenCache.Id, + Token = tokenCache.Token, + DailyLimit = tokenCache.DailyLimit, + TotalLimit = tokenCache.TotalLimit, + ConcurrencyLimit = tokenCache.ConcurrencyLimit, + CreatedAt = tokenCache.CreatedAt, + ExpiresAt = tokenCache.ExpiresAt, + DailyUsage = tokenCache.DailyUsage, + TotalUsage = tokenCache.TotalUsage, + LastActivityTime = tokenCache.LastActivityTime + }; + + + var (maxCount, currentlyExecuting, available) = _usageTracker.GetConcurrencyStatus(tokenCache.Token); + tokenCacheItem.CurrentlyExecuting = currentlyExecuting; + // 开始处理 task 数据 + IQueryable query = _dbContext.MJApiTasks.Where(x => x.TokenId == tokenCacheItem.Id); + + if (!string.IsNullOrWhiteSpace(thirdPartyTaskId)) + { + query = query.Where(x => x.ThirdPartyTaskId == thirdPartyTaskId); + } + + int total = await query.CountAsync(); + List mJApiTasks = await query.OrderByDescending(x => x.StartTime).Skip((page - 1) * pageSize).Take(pageSize).ToListAsync(); + + // 处理任务集合 + // 将某个属性设置为空值 + foreach (var task in mJApiTasks) + { + task.Token = "****"; + } + + tokenCacheItem.TaskCollections = mJApiTasks; + + return APIResponseModel>.CreateSuccessResponseModel(ResponseCode.Success, new CollectionResponse + { + Total = total, + Collection = [tokenCacheItem], + Current = page + }); + + } + catch (Exception ex) + { + return APIResponseModel>.CreateErrorResponseModel(Common.Enums.ResponseCodeEnum.ResponseCode.SystemError, ex.Message); + } + } + + #endregion + + #region 查询Token是不是存在-用户 + /// + /// 获取Token的数据 + /// + /// + /// + public async Task>> GetTokenItem(string token) + { + try + { + if (string.IsNullOrWhiteSpace(token)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不能为空"); + } + + TokenCacheItem? tokenCache = await _tokenService.GetDatabaseTokenAsync(token, false); + if (tokenCache == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在"); + } + + var (maxCount, currentlyExecuting, available) = _usageTracker.GetConcurrencyStatus(tokenCache.Token); + tokenCache.CurrentlyExecuting = currentlyExecuting; + tokenCache.UseToken = "*******************"; + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, tokenCache); + } + catch (Exception ex) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, ex.Message); + } + } + + #endregion + + #region 新增Token + + public async Task>> AddToken(long requestUserId, AddOrModifyTokenModel model) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + // 是 超级管理员 直接添加数据就行 + if (string.IsNullOrWhiteSpace(model.Token)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不能为空"); + } + + if (string.IsNullOrWhiteSpace(model.UseToken)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "使用Token不能为空"); + } + + // 判断token是不是存在 + MJApiTokens? exitMJApiTokens = await _dbContext.MJApiTokens + .AsNoTracking() + .FirstOrDefaultAsync(x => x.Token == model.Token || x.UseToken == model.UseToken); + if (exitMJApiTokens != null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token或使用Token已存在，请检查数据！"); + } + + if (model.DailyLimit < 0 || model.TotalLimit < 0 || model.ConcurrencyLimit < 0 || model.UseDayCount < 0) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "限制参数不能小于0"); + } + + MJApiTokens mJApiTokens = new() + { + Token = model.Token, + UseToken = model.UseToken, + DailyLimit = model.DailyLimit, + TotalLimit = model.TotalLimit, + ConcurrencyLimit = model.ConcurrencyLimit, + CreatedAt = BeijingTimeExtension.GetBeijingTime(), + ExpiresAt = BeijingTimeExtension.GetBeijingTime().AddDays(model.UseDayCount) + }; + + // 开始新增 + await _dbContext.MJApiTokens.AddAsync(mJApiTokens); + await _dbContext.SaveChangesAsync(); + string message = $"Token添加成功: {model.Token}, 日限制: {model.DailyLimit}, 总限制: {model.TotalLimit}, 并发限制: {model.ConcurrencyLimit}，有效期: {mJApiTokens.ExpiresAt}"; + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, message); + } + catch (Exception ex) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, ex.Message); + } + } + + #endregion + + #region 修改Token + + public async Task>> ModifyToken(long requestUserId, long tokenId, AddOrModifyTokenModel model) + { + try + { + if (string.IsNullOrWhiteSpace(model.Token)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不能为空"); + } + if (string.IsNullOrWhiteSpace(model.UseToken)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "使用Token不能为空"); + } + + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + // 是 超级管理员 直接修改数据就行 + MJApiTokens? mJApiTokens = await _dbContext.MJApiTokens + .Where(x => x.Id == tokenId).FirstOrDefaultAsync(); + + if (mJApiTokens == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在"); + } + + // 判断当前传入的token 是不是已经再别的地方使用了 + MJApiTokens? exitMJApiTokens = await _dbContext.MJApiTokens + .AsNoTracking() + .FirstOrDefaultAsync(x => (x.Token == model.Token || x.UseToken == model.UseToken) && x.Id != tokenId); + if (exitMJApiTokens != null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token或者是使用Token已存在，请检查数据"); + } + + + // 开始修改 + mJApiTokens.Token = model.Token; + mJApiTokens.UseToken = model.UseToken; + + if (model.DailyLimit > 0) + { + mJApiTokens.DailyLimit = model.DailyLimit; + } + if (model.TotalLimit > 0) + { + mJApiTokens.TotalLimit = model.TotalLimit; + } + if (model.ConcurrencyLimit > 0) + { + mJApiTokens.ConcurrencyLimit = model.ConcurrencyLimit; + } + + if (model.UseDayCount > 0) + { + // 不是 -1 就需要重新设置到期时间 + mJApiTokens.ExpiresAt = mJApiTokens.CreatedAt.AddDays(model.UseDayCount); + } + + _dbContext.MJApiTokens.Update(mJApiTokens); + await _dbContext.SaveChangesAsync(); + + // 刷新一下内存中的限制 + await RefreshTokenFromDatabaseAsync(model.Token); + + string message = $"Token修改成功: {model.Token}, 日限制: {mJApiTokens.DailyLimit}, 总限制: {mJApiTokens.TotalLimit}, 并发限制: {mJApiTokens.ConcurrencyLimit}，有效期: {mJApiTokens.ExpiresAt}，请注意：修改Token后，之前的Token将失效。"; + + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, message); + } + catch (Exception ex) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, ex.Message); + } + } + + #endregion + + #region 删除Token + + public async Task>> DeleteToken(long requestUserId, long tokenId) + { + var transaction = await _dbContext.Database.BeginTransactionAsync(); + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + // 是 超级管理员 直接删除数据就行 + // 先删除 使用数据 + MJApiTokenUsage? tokenUsage = await _dbContext.MJApiTokenUsage + .Where(x => x.TokenId == tokenId) + .FirstOrDefaultAsync(); + if (tokenUsage != null) + { + _dbContext.MJApiTokenUsage.Remove(tokenUsage); + } + + // 再删除 Token 数据 + MJApiTokens? mJApiTokens = await _dbContext.MJApiTokens + .Where(x => x.Id == tokenId).FirstOrDefaultAsync(); + if (mJApiTokens == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在"); + } + _dbContext.MJApiTokens.Remove(mJApiTokens); + await _dbContext.SaveChangesAsync(); + await transaction.CommitAsync(); + // 将内存中的token移除 + + _usageTracker.RemoveToken(mJApiTokens.Token); + _logger.LogInformation($"Token删除成功: {mJApiTokens.Token}，ID: {tokenId}"); + string message = $"Token删除成功: {mJApiTokens.Token}，ID: {tokenId}，请注意：删除Token后，之前的Token将失效。"; + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, message); + } + catch (Exception ex) + { + await transaction.RollbackAsync(); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, ex.Message); + } + } + + #endregion + + #region 查询所有的任务，可以带参数 + + public async Task>>> QueryTaskCollection(long requestUserId, int page, int pageSize, string? thirdPartyTaskId, string? token, long? tokenId) + { + try + { + // 权限检查 + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + // 构建查询条件 + IQueryable query = _dbContext.MJApiTasks.AsQueryable(); + + // 根据不同参数组合构建查询 + if (tokenId.HasValue) + { + query = query.Where(x => x.TokenId == tokenId); + } + + if (!string.IsNullOrWhiteSpace(token)) + { + // 如果只提供了token字符串 + query = query.Where(x => x.Token.Contains(token)); + } + + // 添加第三方任务ID过滤 + if (!string.IsNullOrWhiteSpace(thirdPartyTaskId)) + { + query = query.Where(x => x.ThirdPartyTaskId == thirdPartyTaskId); + } + + // 获取总数 + int total = await query.CountAsync(); + + if (total == 0) + { + return APIResponseModel>.CreateSuccessResponseModel( + ResponseCode.Success, new CollectionResponse + { + Total = 0, + Collection = new List(), + Current = page + }); + } + + // 分页查询任务数据 + var tasks = await query + .OrderByDescending(x => x.StartTime) + .Skip((page - 1) * pageSize) + .Take(pageSize) + .ToListAsync(); + + // 构建返回结果 + var result = tasks.Select(task => new MJApiTaskCollection + { + TaskId = task.TaskId, + Token = task.Token, + TokenId = task.TokenId, + StartTime = task.StartTime, + EndTime = task.EndTime, + Status = task.Status, + ThirdPartyTaskId = task.ThirdPartyTaskId, + Properties = task.Properties, + }).ToList(); + + return APIResponseModel>.CreateSuccessResponseModel( + ResponseCode.Success, new CollectionResponse + { + Total = total, + Collection = result, // 直接赋值，不要用数组包装 + Current = page + }); + } + catch (Exception ex) + { + _logger?.LogError(ex, "查询任务集合时发生错误 - 用户ID: {UserId}, Token: {Token}, TokenId: {TokenId}", + requestUserId, token, tokenId); + + return APIResponseModel>.CreateErrorResponseModel( + ResponseCode.SystemError, "查询失败，请稍后重试"); + } + } + + #endregion + + #region 查询所有的Token， 可以带参数 + public async Task>>> QueryTokenCollection(int page, int pageSize, string? token, long? tokenId, bool? efficient, long requestUserId) + { + try + { + // 权限检查 + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + var currentUtcTime = BeijingTimeExtension.GetBeijingTime(); // 2025-06-09 12:20:40 + + // 构建基础查询 + var query = from t in _dbContext.MJApiTokens + join u in _dbContext.MJApiTokenUsage on t.Id equals u.TokenId into tokenUsage + from usage in tokenUsage.DefaultIfEmpty() + select new + { + Token = t, + Usage = usage + }; + + // 应用过滤条件 + if (!string.IsNullOrWhiteSpace(token)) + { + query = query.Where(x => x.Token.Token == token); + } + + if (tokenId.HasValue) + { + query = query.Where(x => x.Token.Id == tokenId.Value); + } + + if (efficient.HasValue) + { + if (efficient.Value) + { + // 查询有效得token + query = query.Where(x => + (x.Token.ExpiresAt == null || x.Token.ExpiresAt > currentUtcTime)); + } + else + { + // 查询无效或不活跃的Token + query = query.Where(x => + (x.Token.ExpiresAt != null && x.Token.ExpiresAt <= currentUtcTime)); + } + } + + // 获取总数 + var total = await query.CountAsync(); + + if (total == 0) + { + return APIResponseModel>.CreateSuccessResponseModel( + ResponseCode.Success, new CollectionResponse + { + Total = 0, + Collection = [], + Current = page + }); + } + + // 分页查询并投影到TokenCacheItem + var tokenItems = await query + .OrderByDescending(x => x.Token.CreatedAt) + .Skip((page - 1) * pageSize) + .Take(pageSize) + .Select(x => new TokenCacheItem + { + Id = x.Token.Id, + Token = x.Token.Token, + UseToken = x.Token.UseToken, + DailyLimit = x.Token.DailyLimit, + TotalLimit = x.Token.TotalLimit, + ConcurrencyLimit = x.Token.ConcurrencyLimit, + CreatedAt = x.Token.CreatedAt, + ExpiresAt = x.Token.ExpiresAt, + DailyUsage = x.Usage != null ? x.Usage.DailyUsage : 0, + TotalUsage = x.Usage != null ? x.Usage.TotalUsage : 0, + LastActivityTime = x.Usage != null ? x.Usage.LastActivityAt : x.Token.CreatedAt, + HistoryUse = x.Usage != null ? x.Usage.HistoryUse : "" + }) + .ToListAsync(); + + for (int i = 0; i < tokenItems.Count; i++) + { + var tokenItem = tokenItems[i]; + var (maxCount, currentlyExecuting, available) = usageTracker.GetConcurrencyStatus(tokenItem.Token); + tokenItems[i].CurrentlyExecuting = currentlyExecuting; + } + + _logger?.LogInformation($"✅ Token查询完成, 总数: {total}, 当前页: {page}, 页大小: {pageSize}, 返回: {tokenItems.Count} 条"); + + return APIResponseModel>.CreateSuccessResponseModel( + ResponseCode.Success, new CollectionResponse + { + Total = total, + Collection = tokenItems, + Current = page + }); + } + catch (Exception ex) + { + _logger?.LogError(ex, "❌ 查询Token集合时发生错误 - 用户: qiang-lo, UTC时间: 2025-06-09 12:20:40"); + return APIResponseModel>.CreateErrorResponseModel( + ResponseCode.SystemError, "查询失败，请稍后重试"); + } + } + + #endregion + + + #region 删除小于指定时间的任务 + + public async Task>> DeleteMJTaskEarlyTimestamp(long requestUserId, long timestamp) + { + const string operationName = "删除早期MJ任务"; + try + { + // 1. 参数验证 + if (!IsValidTimestamp(timestamp)) + { + _logger.LogWarning($"{operationName} - 无效的时间戳: {timestamp}"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "时间戳格式错误"); + } + + // 2. 权限检查 + if (!await _userBasicDao.CheckUserIsSuperAdmin(requestUserId)) + { + _logger.LogWarning($"{operationName} - 用户 {requestUserId} 无权限执行此操作"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + // 3. 时间戳转换 + DateTime targetDateTime = DateTimeOffset.FromUnixTimeMilliseconds(timestamp).DateTime; + _logger.LogInformation($"{operationName} - 开始删除早于 {targetDateTime:yyyy-MM-dd HH:mm:ss} 的任务"); + + DateTime beijingTargetDateTime = targetDateTime.AddHours(8); // 转换为北京时间 + // 4. 使用批量删除，避免加载到内存 + int deletedCount = await _dbContext.MJApiTasks + .Where(x => x.StartTime <= beijingTargetDateTime) + .ExecuteDeleteAsync(); // 使用 EF Core 7+ 的批量删除 + + // 5. 记录结果 + string resultMessage = deletedCount == 0 + ? "没有找到符合条件的任务" + : $"成功删除 {deletedCount} 个任务"; + + _logger.LogInformation($"{operationName} - {resultMessage}，目标时间: {targetDateTime:yyyy-MM-dd HH:mm:ss}"); + + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, resultMessage); + } + catch (ArgumentOutOfRangeException ex) + { + _logger.LogError(ex, $"{operationName} - 时间戳转换失败: {timestamp}"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "时间戳格式错误"); + } + catch (Exception ex) + { + _logger.LogError(ex, $"{operationName} - 执行失败，时间戳: {timestamp}"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, "系统错误，请稍后重试"); + } + } + + #endregion + + #region 删除指定ID的任务 + + public async Task>> DeleteMJTask(long requestUserId, string taskId) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + // 1. 参数验证 + if (string.IsNullOrWhiteSpace(taskId)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "任务ID不能为空"); + } + // 2. 查找任务 + var task = await _dbContext.MJApiTasks + .Where(x => x.TaskId == taskId) + .FirstOrDefaultAsync(); + if (task == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "任务不存在"); + } + // 3. 删除任务 + _dbContext.MJApiTasks.Remove(task); + await _dbContext.SaveChangesAsync(); + _logger.LogInformation($"删除指定ID的任务成功，任务ID: {taskId}"); + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, "任务删除成功"); + } + catch (Exception ex) + { + _logger.LogError(ex, $"删除指定ID的任务失败，任务ID: {taskId}, 失败原因：{ex.Message}"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, "删除任务失败，请稍后重试"); + } + } + + #endregion + + #region 手动刷新Token + + public async Task>> RefreshToken(long requestUserId, string token) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + // 1. 参数验证 + if (string.IsNullOrWhiteSpace(token)) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不能为空"); + } + TokenCacheItem tokenCache = await RefreshTokenFromDatabaseAsync(token); + if (tokenCache == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在"); + } + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, $"Token刷新成功: {token}, 并发限制: {tokenCache.ConcurrencyLimit}，日出图限制: {tokenCache.DailyLimit}, 当前日出图总数: {tokenCache.DailyUsage}, 总出图量: {tokenCache.TotalUsage}"); + } + catch (Exception ex) + { + string message = $"手动刷新Token失败，Token: {token}, 失败原因：{ex.Message}"; + _logger.LogError(ex, message); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, message); + } + } + + #endregion + + #region 获取活跃的Token + + public async Task>>> GetActiveTokens(long requestUserId, int minutes) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + var threshold = TimeSpan.FromMinutes(minutes); + List activeTokens = _usageTracker.GetActiveTokens(threshold); + return APIResponseModel>.CreateSuccessResponseModel(ResponseCode.Success, new CollectionResponse + { + Total = activeTokens.Count, + Collection = activeTokens, + Current = 1 // 这里可以设置为1，因为我们只返回一页 + }); + } + catch (Exception ex) + { + string message = $"获取活跃的Token失败，错误信息：{ex.Message}"; + _logger.LogError(ex, message); + return APIResponseModel>.CreateErrorResponseModel(ResponseCode.SystemError, message); + } + } + + #endregion + + #region 移除不活跃的Token + + public async Task>> RemoveNotActiveTokens(long requestUserId, int minutes) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + var threshold = TimeSpan.FromMinutes(minutes); + var (activateTokenCount, notActivateTokenCount) = _usageTracker.RemoveNotActiveTokens(threshold); + string message = $"删除不活跃得 Token 数: {notActivateTokenCount}，阈值: {minutes}分钟"; + _logger.LogInformation(message); + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, message); + } + catch (Exception ex) + { + string message = $"移除不活跃的Token失败，错误信息：{ex.Message}"; + _logger.LogError(ex, message); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, message); + } + } + + #endregion + + #region 健康检查和统计接口 + public async Task>> GetHealth(long requestUserId) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + var stats = _usageTracker.GetCacheStats(); + var now = BeijingTimeExtension.GetBeijingTime(); + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, new + { + Status = "Healthy", + Timestamp = now, + CacheStats = stats, + Uptime = now - Process.GetCurrentProcess().StartTime.ToUniversalTime() + }); + } + catch (Exception ex) + { + _logger.LogError(ex, "获取系统健康状态失败"); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, "获取系统健康状态失败"); + } + } + #endregion + + #region 获取指定ID得Token + + /// + /// 获取指定ID得Token + /// + /// + /// + /// + public async Task>> QueryTokenById(long tokenId, long requestUserId) + { + try + { + + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + + MJApiTokens? mjApiTokens = await _dbContext.MJApiTokens + .AsNoTracking() + .Where(x => x.Id == tokenId) + .FirstOrDefaultAsync(); + if (mjApiTokens == null) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.ParameterError, "Token不存在"); + } + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, mjApiTokens); + } + catch (Exception ex) + { + _logger.LogError(ex, "查询Token失败，TokenId: {TokenId}, 错误信息: {Message}", tokenId, ex.Message); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, "查询Token失败，请稍后重试"); + } + } + + #endregion + + #region 获取日统计数据 + + public async Task>> GetDayTaskStatistics(long requestUserId) + { + try + { + bool isSuperAdmin = await _userBasicDao.CheckUserIsSuperAdmin(requestUserId); + if (!isSuperAdmin) + { + return APIResponseModel.CreateErrorResponseModel(ResponseCode.NotPermissionAction); + } + // 获取当前日期 + DateTime today = BeijingTimeExtension.GetBeijingTime().Date; // 获取今天的日期，不包含时间部分 + // 查询今天的任务 + var tasks = await _dbContext.MJApiTasks + .Where(x => x.StartTime.Date == today) + .OrderByDescending(x => x.StartTime) + .ToListAsync(); + if (tasks.Count == 0) + { + return APIResponseModel.CreateSuccessResponseModel( + ResponseCode.Success, new TaskStatistics()); + } + + // 统计任务数量 + int totalTasks = tasks.Count; + // 统计成功任务数量 + int successfulTasks = tasks.Count(x => x.Status == MJTaskStatus.SUCCESS); + // 统计失败任务数量 + int failedTasks = tasks.Count(x => x.Status == MJTaskStatus.FAILURE || x.Status == MJTaskStatus.CANCEL); + + // 剩下的都是再执行的 + int inProgressTasks = totalTasks - successfulTasks - failedTasks; + return APIResponseModel.CreateSuccessResponseModel(ResponseCode.Success, new TaskStatistics + { + TotalTasks = totalTasks, + CompletedTasks = successfulTasks, + FailedTasks = failedTasks, + InProgressTasks = inProgressTasks + }); + } + catch (Exception ex) + { + _logger.LogError(ex, "获取日统计数据失败，错误信息: {Message}", ex.Message); + return APIResponseModel.CreateErrorResponseModel(ResponseCode.SystemError, "获取日统计数据失败，请稍后重试"); + } + } + + #endregion + + + /// + /// 从数据库重新加载Token到内存缓存（平滑更新） + /// + private async Task RefreshTokenFromDatabaseAsync(string token) + { + _logger.LogDebug($"从数据库平滑刷新Token到内存: {token}"); + + try + { + TokenCacheItem? tokenItem = await _tokenService.GetDatabaseTokenAsync(token); + if (tokenItem == null) + { + // 将内存的中的这个token删掉 + _usageTracker.RemoveToken(token); + throw new Exception($"Token不存在"); + } + + // 3. 平滑更新到内存缓存（这里会自动处理并发限制的平滑调整） + _usageTracker.AddOrUpdateTokenAsync(tokenItem); + + _logger.LogInformation($"Token平滑刷新成功: {token}, 并发限制: {tokenItem.ConcurrencyLimit}"); + return tokenItem; + } + catch (Exception ex) + { + _logger.LogError(ex, $"平滑刷新Token失败: {token} ，" + ex.Message); + throw; + } + } + + + // 辅助方法：验证时间戳 + private static bool IsValidTimestamp(long timestamp) + { + // 检查时间戳是否在合理范围内 + // Unix时间戳最小值（1970-01-01）和最大值（约2038年或更远） + const long minTimestamp = 0; + const long maxTimestamp = 253402300799999; // 9999-12-31的毫秒时间戳 + + return timestamp >= minTimestamp && timestamp <= maxTimestamp; + } + } +} diff --git a/LMS.service/Service/MachineService.cs b/LMS.service/Service/MachineService.cs index a42b98c..c91ee7d 100644 --- a/LMS.service/Service/MachineService.cs +++ b/LMS.service/Service/MachineService.cs @@ -9,8 +9,6 @@ using LMS.Repository.Models.DB; using Microsoft.AspNetCore.Identity; using Microsoft.AspNetCore.Mvc; using Microsoft.EntityFrameworkCore; -using System.Collections.Generic; -using System.Linq; using static LMS.Common.Enums.MachineEnum; using static LMS.Common.Enums.ResponseCodeEnum; using static LMS.Repository.DTO.MachineDto; diff --git a/LMS.service/Service/OptionsService.cs b/LMS.service/Service/OptionsService.cs index 384649d..a01f6cb 100644 --- a/LMS.service/Service/OptionsService.cs +++ b/LMS.service/Service/OptionsService.cs @@ -1,25 +1,22 @@ using AutoMapper; +using LinqKit; using LMS.Common.Dictionary; using LMS.Common.Enums; +using LMS.Common.Extensions; using LMS.Common.Templates; using LMS.DAO; using LMS.DAO.UserDAO; -using LMS.Repository.DB; using LMS.Repository.DTO; +using LMS.Repository.DTO.OptionDto; using LMS.Repository.Models.DB; using LMS.Repository.Options; using LMS.service.Extensions.Mail; using Microsoft.AspNetCore.Identity; using Microsoft.AspNetCore.Mvc; using Microsoft.EntityFrameworkCore; -using Microsoft.Extensions.Options; -using System.Linq; +using System.Linq.Dynamic.Core; using static LMS.Common.Enums.ResponseCodeEnum; using Options = LMS.Repository.DB.Options; -using System.Linq.Dynamic.Core; -using LinqKit; -using LMS.Repository.DTO.OptionDto; -using LMS.Common.Extensions; namespace LMS.service.Service { diff --git a/LMS.service/appsettings.json b/LMS.service/appsettings.json index 524319f..4c9cd7d 100644 --- a/LMS.service/appsettings.json +++ b/LMS.service/appsettings.json @@ -68,6 +68,6 @@ } ] }, - "Version": "1.1.1", + "Version": "1.1.2", "AllowedHosts": "*" } diff --git a/SQL/v1.1.2/MJApiTasks.sql b/SQL/v1.1.2/MJApiTasks.sql new file mode 100644 index 0000000..82d35f5 --- /dev/null +++ b/SQL/v1.1.2/MJApiTasks.sql @@ -0,0 +1,41 @@ +/* + Navicat Premium Dump SQL + + Source Server : 亿速云（国内） + Source Server Type : MySQL + Source Server Version : 80018 (8.0.18) + Source Host : yisurds-66dc0b453c05d4.rds.ysydb1.com:14080 + Source Schema : LMS_TEST + + Target Server Type : MySQL + Target Server Version : 80018 (8.0.18) + File Encoding : 65001 + + Date: 11/06/2025 13:55:05 +*/ + +SET NAMES utf8mb4; +SET FOREIGN_KEY_CHECKS = 0; + +-- ---------------------------- +-- Table structure for MJApiTasks +-- ---------------------------- +DROP TABLE IF EXISTS `MJApiTasks`; +CREATE TABLE `MJApiTasks` ( + `TaskId` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL, + `Token` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL, + `StartTime` datetime NOT NULL, + `EndTime` datetime NULL DEFAULT NULL, + `Status` varchar(32) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL DEFAULT '0' COMMENT '0=Pending, 1=NotStart, 2=Running, 3=Completed, 4=Failed, 5=Timeout', + `ThirdPartyTaskId` varchar(100) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL DEFAULT NULL, + `Properties` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL, + `CreatedAt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP, + `UpdatedAt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`TaskId`) USING BTREE, + INDEX `idx_token`(`Token` ASC) USING BTREE, + INDEX `idx_third_party_task_id`(`ThirdPartyTaskId` ASC) USING BTREE, + INDEX `idx_status`(`Status` ASC) USING BTREE, + INDEX `idx_start_time`(`StartTime` ASC) USING BTREE +) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci COMMENT = '异步任务表' ROW_FORMAT = Dynamic; + +SET FOREIGN_KEY_CHECKS = 1; diff --git a/SQL/v1.1.2/MJApiTokenUsage.sql b/SQL/v1.1.2/MJApiTokenUsage.sql new file mode 100644 index 0000000..8b1f7f7 --- /dev/null +++ b/SQL/v1.1.2/MJApiTokenUsage.sql @@ -0,0 +1,36 @@ +/* + Navicat Premium Dump SQL + + Source Server : 亿速云（国内） + Source Server Type : MySQL + Source Server Version : 80018 (8.0.18) + Source Host : yisurds-66dc0b453c05d4.rds.ysydb1.com:14080 + Source Schema : LMS_TEST + + Target Server Type : MySQL + Target Server Version : 80018 (8.0.18) + File Encoding : 65001 + + Date: 11/06/2025 13:55:58 +*/ + +SET NAMES utf8mb4; +SET FOREIGN_KEY_CHECKS = 0; + +-- ---------------------------- +-- Table structure for MJApiTokenUsage +-- ---------------------------- +DROP TABLE IF EXISTS `MJApiTokenUsage`; +CREATE TABLE `MJApiTokenUsage` ( + `TokenId` bigint(20) NOT NULL, + `Date` date NOT NULL, + `DailyUsage` int(11) NOT NULL DEFAULT 0 COMMENT '当日使用量', + `TotalUsage` int(11) NOT NULL DEFAULT 0 COMMENT '累计使用量', + `LastActivityAt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '最后活动时间', + `HistoryUse` longtext CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NULL COMMENT '使用记录', + PRIMARY KEY (`TokenId`) USING BTREE, + INDEX `IdxLastActivity`(`LastActivityAt` ASC) USING BTREE, + CONSTRAINT `MJApiTokenUsage_ibfk_1` FOREIGN KEY (`TokenId`) REFERENCES `MJApiTokens` (`Id`) ON DELETE CASCADE ON UPDATE RESTRICT +) ENGINE = InnoDB CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci COMMENT = '令牌使用统计表' ROW_FORMAT = Dynamic; + +SET FOREIGN_KEY_CHECKS = 1; diff --git a/SQL/v1.1.2/MJApiTokens.sql b/SQL/v1.1.2/MJApiTokens.sql new file mode 100644 index 0000000..746ca7e --- /dev/null +++ b/SQL/v1.1.2/MJApiTokens.sql @@ -0,0 +1,37 @@ +/* + Navicat Premium Dump SQL + + Source Server : 亿速云（国内） + Source Server Type : MySQL + Source Server Version : 80018 (8.0.18) + Source Host : yisurds-66dc0b453c05d4.rds.ysydb1.com:14080 + Source Schema : LMS_TEST + + Target Server Type : MySQL + Target Server Version : 80018 (8.0.18) + File Encoding : 65001 + + Date: 11/06/2025 13:55:48 +*/ + +SET NAMES utf8mb4; +SET FOREIGN_KEY_CHECKS = 0; + +-- ---------------------------- +-- Table structure for MJApiTokens +-- ---------------------------- +DROP TABLE IF EXISTS `MJApiTokens`; +CREATE TABLE `MJApiTokens` ( + `Id` bigint(20) NOT NULL AUTO_INCREMENT, + `Token` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT 'API令牌', + `DailyLimit` int(11) NULL DEFAULT 0 COMMENT '日限制(0=无限制)', + `TotalLimit` int(11) NULL DEFAULT 0 COMMENT '总限制(0=无限制)', + `ConcurrencyLimit` int(11) NOT NULL DEFAULT 1 COMMENT '并发限制', + `CreatedAt` datetime NOT NULL DEFAULT CURRENT_TIMESTAMP COMMENT '创建时间', + `ExpiresAt` datetime NULL DEFAULT NULL COMMENT '到期时间(NULL=永不过期)', + PRIMARY KEY (`Id`) USING BTREE, + UNIQUE INDEX `UkToken`(`Token` ASC) USING BTREE, + INDEX `IdxExpires`(`ExpiresAt` ASC) USING BTREE +) ENGINE = InnoDB AUTO_INCREMENT = 9 CHARACTER SET = utf8mb4 COLLATE = utf8mb4_0900_ai_ci COMMENT = 'API令牌表' ROW_FORMAT = Dynamic; + +SET FOREIGN_KEY_CHECKS = 1;