using QYZH.InteractiveMagazine.Models.Entity; using QYZH.InteractiveMagazine.Models.Enum; using SqlSugar; using System.Text; using System.Text.Json; namespace QYZH.InteractiveMagazine.WorkService.Consumers; /// /// 期刊任务接收消费者 /// public class JournalTaskReceiveConsumer( ILogger logger, IConfiguration configuration, IServiceScopeFactory scopeFactory) : IQueueConsumer { private const float DefaultCompletionThreshold = 80; private const string AiProcessingMessage = "AI批阅中,请稍后"; public string Exchange => "ex.journal"; public string QueueName => "mq.journal.task.receive"; public string RoutingKey => "rk.journal.task.receive"; public async Task HandleAsync(byte[] body, CancellationToken cancellationToken = default) { var message = Encoding.UTF8.GetString(body); logger.LogInformation("收到期刊任务消息: {Message}", message); using var scope = scopeFactory.CreateScope(); var client = scope.ServiceProvider.GetRequiredService(); var data = JsonSerializer.Deserialize(message, new JsonSerializerOptions { PropertyNameCaseInsensitive = true }) ?? throw new InvalidOperationException("期刊任务消息内容为空"); data.Normalize(); if (data.Questions == null || data.Questions.Length == 0) { logger.LogWarning("期刊任务消息没有题目,UserId: {UserId}, JournalId: {JournalId}, PageId: {PageId}", data.UserId, data.JournalId, data.PageId); return; } var taskIds = data.Questions.Select(q => q.Id).Distinct().ToList(); var tasks = await client.Queryable() .Where(t => taskIds.Contains(t.Id) && !t.IsDeleted) .ToListAsync(cancellationToken); var taskMap = tasks.ToDictionary(t => t.Id); var groupIds = tasks.Select(t => t.GroupId > 0 ? t.GroupId : t.Id).Distinct().ToList(); var groupTasks = await client.Queryable() .Where(t => (groupIds.Contains(t.GroupId) || groupIds.Contains(t.Id)) && !t.IsDeleted) .ToListAsync(cancellationToken); var persistedTaskMap = groupTasks .GroupBy(t => t.GroupId > 0 ? t.GroupId : t.Id) .ToDictionary( g => g.Key, g => g.FirstOrDefault(t => t.Id == g.Key) ?? g.OrderBy(t => t.Id).First()); var pageIds = groupTasks.Select(t => t.JournalPageId).Append(data.PageId).Distinct().ToList(); var pages = await client.Queryable() .Where(p => pageIds.Contains(p.Id) && !p.IsDeleted) .ToListAsync(cancellationToken); var pageMap = pages.ToDictionary(p => p.Id); var questionContexts = new List(); foreach (var question in data.Questions) { if (!taskMap.TryGetValue(question.Id, out var task)) { logger.LogWarning("未找到期刊任务,TaskId: {TaskId}, UserId: {UserId}", question.Id, data.UserId); continue; } pageMap.TryGetValue(task.JournalPageId, out var taskPage); questionContexts.Add(new JournalQuestionContext(question, task, taskPage)); } var answerContexts = new List(); foreach (var scoreUnit in BuildScoreUnits(questionContexts)) { var persistedContext = BuildPersistedContext(scoreUnit, persistedTaskMap, pageMap); if (persistedContext == null) { logger.LogWarning("期刊任务上传单元缺少可入库任务,UserId: {UserId}, GroupId: {GroupId}", data.UserId, scoreUnit.GroupId); continue; } var processingResult = BuildFailureScoreResult(); answerContexts.Add(new JournalAnswerContext( BuildAnswerEntity(data, scoreUnit, persistedContext.Question, persistedContext.Task, persistedContext.Page, processingResult, GetCompletionThreshold()), persistedContext.Question, persistedContext.Task, processingResult)); LogSkippedGroupTasks(scoreUnit, persistedContext.Task); } if (answerContexts.Count == 0) { logger.LogWarning("期刊任务消息没有可入库的答题记录,UserId: {UserId}, JournalId: {JournalId}, PageId: {PageId}", data.UserId, data.JournalId, data.PageId); return; } client.Ado.BeginTran(); try { foreach (var context in answerContexts) { var answer = context.Answer; var existing = await client.Queryable() .Where(a => a.UserId == answer.UserId && a.JournalPageTaskId == answer.JournalPageTaskId && !a.IsDeleted) .FirstAsync(cancellationToken); if (existing == null) { await client.Insertable(answer).ExecuteCommandAsync(cancellationToken); continue; } answer.Id = existing.Id; answer.CreatedBy = existing.CreatedBy; answer.CreatedAt = existing.CreatedAt; answer.UpdatedBy = answer.UserId.ToString(); answer.UpdatedAt = DateTime.Now; await client.Updateable(answer) .IgnoreColumns(a => new { a.CreatedBy, a.CreatedAt }) .Where(a => a.Id == existing.Id) .ExecuteCommandAsync(cancellationToken); } client.Ado.CommitTran(); } catch { client.Ado.RollbackTran(); throw; } logger.LogInformation("期刊任务上传数据保存完成,UserId: {UserId}, JournalId: {JournalId}, PageId: {PageId}, Count: {Count}", data.UserId, data.JournalId, data.PageId, answerContexts.Count); } public Task OnErrorAsync(byte[] body, Exception exception) { var message = Encoding.UTF8.GetString(body); logger.LogError(exception, "处理期刊任务消息失败: {Message}", message); return Task.CompletedTask; } private static JournalAnswerScoreResult BuildFailureScoreResult() { return new JournalAnswerScoreResult { Completion = 0, Result = AiProcessingMessage }; } private static List BuildScoreUnits(List contexts) { return contexts .GroupBy(c => c.Task.GroupId > 0 ? c.Task.GroupId : c.Task.Id) .Select(g => new JournalScoreUnit( g.Key, g.OrderBy(c => c.Page?.PageNum ?? 0) .ThenBy(c => ParseTaskNo(c.Task.No)) .ThenBy(c => c.Task.Id) .ToList())) .ToList(); } private static JournalQuestionContext? BuildPersistedContext( JournalScoreUnit scoreUnit, Dictionary persistedTaskMap, Dictionary pageMap) { if (scoreUnit.Questions.Count == 0) { return null; } var sourceContext = scoreUnit.Questions.FirstOrDefault(q => q.Task.Id == scoreUnit.GroupId) ?? scoreUnit.Questions.First(); var persistedTask = persistedTaskMap.TryGetValue(scoreUnit.GroupId, out var task) ? task : sourceContext.Task; pageMap.TryGetValue(persistedTask.JournalPageId, out var persistedPage); return new JournalQuestionContext(sourceContext.Question, persistedTask, persistedPage); } private static int ParseTaskNo(string? taskNo) { return int.TryParse(taskNo, out var no) ? no : int.MaxValue; } private void LogSkippedGroupTasks(JournalScoreUnit scoreUnit, JournalPageTask persistedTask) { var skippedTaskIds = scoreUnit.Questions .Select(q => q.Task.Id) .Where(id => id != persistedTask.Id) .Distinct() .ToList(); if (skippedTaskIds.Count == 0) { return; } logger.LogInformation("期刊跨页题评分结果仅保存到主任务,GroupId: {GroupId}, PersistedTaskId: {PersistedTaskId}, SkippedTaskIds: {SkippedTaskIds}", scoreUnit.GroupId, persistedTask.Id, string.Join(",", skippedTaskIds)); } private static string SerializeAnswerUrls(JournalScoreUnit scoreUnit) { var answerUrls = scoreUnit.Questions .SelectMany(q => q.Question.AnswerUrl ?? []) .Where(url => !string.IsNullOrWhiteSpace(url)) .Distinct() .ToArray(); return JsonSerializer.Serialize(answerUrls); } private static DateTime GetAnswerStartTime(JournalScoreUnit scoreUnit) { var startTimes = scoreUnit.Questions .Select(q => q.Question.AnswerStartTime) .Where(t => t != default) .ToList(); return startTimes.Count == 0 ? default : startTimes.Min(); } private static DateTime GetAnswerEndTime(JournalScoreUnit scoreUnit) { var endTimes = scoreUnit.Questions .Select(q => q.Question.AnswerEndTime) .Where(t => t != default) .ToList(); return endTimes.Count == 0 ? default : endTimes.Max(); } private static int GetAnswerSeconds(JournalScoreUnit scoreUnit) { return scoreUnit.Questions.Sum(q => Math.Max(0, q.Question.AnswerTime)); } private static int GetBreakCount(JournalScoreUnit scoreUnit) { return scoreUnit.Questions.Sum(q => Math.Max(0, q.Question.BreakCount)); } private static string SerializeBreakTimes(JournalScoreUnit scoreUnit) { var breakTimes = scoreUnit.Questions .SelectMany(q => q.Question.BreakTimes ?? []) .ToList(); return JsonSerializer.Serialize(breakTimes); } private static JournalPageTaskUserAnswer BuildAnswerEntity( QuestionData data, JournalScoreUnit scoreUnit, Question question, JournalPageTask task, JournalPage? page, JournalAnswerScoreResult scoreResult, float completionThreshold) { var now = DateTime.Now; var growthPoint = Math.Max(0, scoreResult.GrowthPoint); var points = Math.Max(0, scoreResult.Points); var answerStatus = scoreResult.Completion >= completionThreshold ? UserAnswerStatusEnum.Complete : UserAnswerStatusEnum.Processing; return new JournalPageTaskUserAnswer { JournalId = data.JournalId, JournalPageId = data.PageId, JournalPageTaskId = task.Id, JournalPageTaskGroupId = task.GroupId, UserId = data.UserId, Result = scoreResult.Result, Points = points, GrowthPoints = growthPoint, Score = Math.Max(0, scoreResult.Score), Comprehension = Math.Max(0, scoreResult.Comprehension), Judgment = Math.Max(0, scoreResult.Judgment), Expression = Math.Max(0, scoreResult.Expression), Persuasiveness = Math.Max(0, scoreResult.Persuasiveness), QuestionAnswerUrl = question.Url, AnswerUrl = SerializeAnswerUrls(scoreUnit), PageAnswerUrl = data.PageAnswerUrl, Revision = 0, AnswerStartTime = GetAnswerStartTime(scoreUnit), AnswerEndTime = GetAnswerEndTime(scoreUnit), AnswerSeconds = GetAnswerSeconds(scoreUnit), ImageRecognition = 0, JournalPageNum = page?.PageNum ?? 0, Modify = 0, LastTag = 0, DotPageNum = page?.PageNum ?? 0, PageResultUrl = string.Empty, Type = ((int)task.Type).ToString(), DotPageNo = page?.PageNo ?? string.Empty, PageAnswerDotUrl = string.Empty, BreakCount = GetBreakCount(scoreUnit), BreakTimes = SerializeBreakTimes(scoreUnit), AssignmentStatus = (int)AssignmentStatusEnum.UnAssignmented, Status = (int)answerStatus, CreatedBy = data.UserId.ToString(), CreatedAt = data.CreatedTime == default ? now : data.CreatedTime, UpdatedBy = data.UserId.ToString(), UpdatedAt = now }; } private float GetCompletionThreshold() { var threshold = configuration.GetValue("AiChat:CompletionThreshold"); if (threshold <= 0) { threshold = DefaultCompletionThreshold; } return threshold; } private static string TrimResult(string? result) { if (string.IsNullOrWhiteSpace(result)) { return string.Empty; } return result.Length <= 50 ? result : result[..50]; } } /// /// 期刊题目作答消息 /// public class QuestionData { /// /// 用户ID /// public long UserId { get; set; } /// /// 学生ID /// public long StudentId { get; set; } /// /// 期刊ID /// public long JournalId { get; set; } /// /// 作业ID /// public long HomeworkId { get; set; } /// /// 书籍ID /// public long BookId { get; set; } /// /// 页ID /// public long PageId { get; set; } /// /// 页面作答图片地址 /// public string PageAnswerUrl { get; set; } = string.Empty; /// /// 题目作答列表 /// public Question[] Questions { get; set; } = []; /// /// 创建时间 /// public DateTime CreatedTime { get; set; } /// /// 兼容新版作答消息字段 /// public void Normalize() { if (UserId <= 0) { UserId = StudentId; } if (JournalId <= 0) { JournalId = BookId > 0 ? BookId : HomeworkId; } } } /// /// 题目作答数据 /// public class Question { /// /// 题目ID /// public long Id { get; set; } /// /// 题目图片地址 /// public string Url { get; set; } = string.Empty; /// /// 答案图片地址 /// public string[] AnswerUrl { get; set; } = []; /// /// 作答开始时间 /// public DateTime AnswerStartTime { get; set; } /// /// 作答结束时间 /// public DateTime AnswerEndTime { get; set; } /// /// 作答耗时秒数 /// public int AnswerTime { get; set; } /// /// 中断次数 /// public int BreakCount { get; set; } /// /// 中断记录 /// public List BreakTimes { get; set; } = []; } /// /// 中断时间记录 /// public class BreakTime { /// /// 中断时间 /// public DateTime Time { get; set; } /// /// 等待时间 /// public long WaitTime { get; set; } } /// /// AI评分结果 /// public class JournalAnswerScoreResult { /// /// 题目得分 /// public float Score { get; set; } /// /// 成长值 /// public int GrowthPoint { get; set; } /// /// 积分 /// public int Points { get; set; } /// /// 理解力评分 /// public float Comprehension { get; set; } /// /// 判断力评分 /// public float Judgment { get; set; } /// /// 表达力评分 /// public float Expression { get; set; } /// /// 说服力评分 /// public float Persuasiveness { get; set; } /// /// 瀹屾垚搴? /// public float Completion { get; set; } /// /// 50字内评语 /// public string Result { get; set; } = string.Empty; } public record JournalAnswerContext( JournalPageTaskUserAnswer Answer, Question Question, JournalPageTask Task, JournalAnswerScoreResult ScoreResult); public record JournalQuestionContext( Question Question, JournalPageTask Task, JournalPage? Page); public record JournalScoreUnit( long GroupId, List Questions); public record AnswerImageContent(string DataUrl); /// /// AI评分不可重试异常。 /// public class NonRetryAiScoreException(string message) : Exception(message);