using QYZH.InteractiveMagazine.Infrastructure.OSS;
using QYZH.InteractiveMagazine.Models.Entity;
using QYZH.InteractiveMagazine.Models.Enum;
using SqlSugar;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using System.Text.Json.Serialization;
namespace QYZH.InteractiveMagazine.WorkService.Consumers;
///
/// 期刊任务接收消费者
///
public class JournalTaskReceiveConsumer(
ILogger logger,
IConfiguration configuration,
IServiceScopeFactory scopeFactory,
IHttpClientFactory httpClientFactory,
OssService ossService) : IQueueConsumer
{
private const int DefaultAiScoreMaxRetryCount = 3;
private const int DefaultAiScoreRetryDelayMilliseconds = 1000;
private const int DefaultAiMaxConcurrency = 1;
private const long DefaultMaxImageBytes = 10 * 1024 * 1024;
private const float DefaultCompletionThreshold = 80;
private const float DefaultCommunityScoreThreshold = 90;
private const string AiProcessingMessage = "AI批阅中,请稍后";
private static readonly object AiSemaphoreLock = new();
private static SemaphoreSlim? aiSemaphore;
private static int aiSemaphoreLimit;
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) && !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 groupTaskIds = groupTasks.Select(t => t.Id).Distinct().ToList();
var referenceAnswers = await client.Queryable()
.Where(a => groupTaskIds.Contains(a.JournalPageTaskId) && !a.IsDeleted)
.ToListAsync(cancellationToken);
var referenceAnswerMap = referenceAnswers
.GroupBy(a => a.JournalPageTaskId)
.ToDictionary(g => g.Key, g => g.ToList());
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 existingAnswers = await client.Queryable()
.Where(a => a.UserId == data.UserId && groupTaskIds.Contains(a.JournalPageTaskId) && !a.IsDeleted)
.ToListAsync(cancellationToken);
var existingAnswerMap = existingAnswers
.GroupBy(a => a.JournalPageTaskId)
.ToDictionary(g => g.Key, g => g.OrderByDescending(a => a.UpdatedAt ?? a.CreatedAt).First());
var questionContexts = new List();
var answerContexts = 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;
}
if (!task.NeedAiProcess)
{
logger.LogInformation("期刊任务配置为人工批改,跳过AI评分,TaskId: {TaskId}, UserId: {UserId}", task.Id, data.UserId);
continue;
}
pageMap.TryGetValue(task.JournalPageId, out var taskPage);
questionContexts.Add(new JournalQuestionContext(question, task, taskPage));
}
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;
}
if (IsCompletedSameAnswer(scoreUnit, persistedContext.Task, data, existingAnswerMap))
{
logger.LogInformation("期刊任务作答已完成且内容未变化,跳过AI评分,UserId: {UserId}, GroupId: {GroupId}, TaskIds: {TaskIds}",
data.UserId, scoreUnit.GroupId, string.Join(",", scoreUnit.Questions.Select(q => q.Task.Id)));
continue;
}
try
{
var scoreResult = await ScoreQuestionAsync(scoreUnit, referenceAnswerMap, cancellationToken);
var normalizedResult = NormalizeScoreResult(scoreResult, persistedContext.Task);
answerContexts.Add(new JournalAnswerContext(
BuildAnswerEntity(data, scoreUnit, persistedContext.Question, persistedContext.Task, persistedContext.Page, normalizedResult, GetCompletionThreshold()),
persistedContext.Question,
persistedContext.Task,
normalizedResult));
LogSkippedGroupTasks(scoreUnit, persistedContext.Task);
}
catch (OperationCanceledException)
{
throw;
}
catch (Exception ex)
{
logger.LogWarning(ex, "期刊任务AI评分失败,已标记为处理中,UserId: {UserId}, GroupId: {GroupId}, TaskIds: {TaskIds}",
data.UserId, scoreUnit.GroupId, string.Join(",", scoreUnit.Questions.Select(q => q.Task.Id)));
var failureResult = BuildFailureScoreResult();
answerContexts.Add(new JournalAnswerContext(
BuildAnswerEntity(data, scoreUnit, persistedContext.Question, persistedContext.Task, persistedContext.Page, failureResult, GetCompletionThreshold()),
persistedContext.Question,
persistedContext.Task,
failureResult));
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);
}
else
{
await client.Insertable(BuildAnswerSnapshot(existing)).ExecuteCommandAsync(cancellationToken);
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);
}
await InsertCommunityMessageIfNeededAsync(client, context, 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 async Task ScoreQuestionAsync(
JournalScoreUnit scoreUnit,
Dictionary> referenceAnswerMap,
CancellationToken cancellationToken)
{
var apiKey = configuration["AiChat:ApiKey"];
var baseUrl = configuration["AiChat:BaseUrl"];
var model = configuration["AiChat:Model"];
var timeoutSeconds = configuration.GetValue("AiChat:TimeoutSeconds");
var maxTokens = configuration.GetValue("AiChat:MaxTokens");
var temperature = configuration.GetValue("AiChat:Temperature");
if (string.IsNullOrWhiteSpace(apiKey) || string.IsNullOrWhiteSpace(baseUrl) || string.IsNullOrWhiteSpace(model))
{
throw new InvalidOperationException("AI聊天服务配置不完整,请检查 AiChat 配置节点");
}
var content = new List