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 long DefaultMaxImageBytes = 10 * 1024 * 1024;
private const float DefaultCompletionThreshold = 80;
private const float DefaultCommunityScoreThreshold = 90;
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("期刊任务消息内容为空");
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 referenceAnswers = await client.Queryable()
.Where(a => taskIds.Contains(a.JournalPageTaskId) && !a.IsDeleted)
.ToListAsync(cancellationToken);
var referenceAnswerMap = referenceAnswers
.GroupBy(a => a.JournalPageTaskId)
.ToDictionary(g => g.Key, g => g.ToList());
var page = await client.Queryable()
.Where(p => p.Id == data.PageId && !p.IsDeleted)
.FirstAsync(cancellationToken);
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;
}
referenceAnswerMap.TryGetValue(task.Id, out var taskReferenceAnswers);
var scoreResult = await ScoreQuestionAsync(task, question, taskReferenceAnswers ?? [], cancellationToken);
answerContexts.Add(new JournalAnswerContext(
BuildAnswerEntity(data, question, task, page, scoreResult, GetCompletionThreshold()),
question,
task,
scoreResult));
}
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(
JournalPageTask task,
Question question,
List referenceAnswers,
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 answerImages = await BuildAnswerImageContentsAsync(question, cancellationToken);
if (answerImages.Count == 0)
{
throw new InvalidOperationException($"题目 {question.Id} 缺少答案图片");
}
var referenceAnswerImages = await BuildReferenceAnswerImageContentsAsync(task.Id, referenceAnswers, cancellationToken);
var content = new List