292 lines
14 KiB
C#
292 lines
14 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Text.Json;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Dtos.News;
|
|
using FinlyticCore.Dtos.Sentiment;
|
|
using FinlyticCore.Dtos.Settings;
|
|
using FinlyticCore.Models;
|
|
using FinlyticCore.Services;
|
|
using FinlyticCore.Util;
|
|
using FinlyticSentiment.Entities;
|
|
using FinlyticSentiment.Services;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
using Microsoft.Extensions.Logging;
|
|
|
|
namespace FinlyticSentiment.Util;
|
|
|
|
/// <summary>
|
|
/// Managed MQTT client for sentiment evaluations and RPC queries.
|
|
/// </summary>
|
|
public class SentimentMqttClient : ManagedMqttClient, IHostedService
|
|
{
|
|
private readonly ILogger<SentimentMqttClient> _logger;
|
|
private readonly IConfiguration _configuration;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
|
|
public SentimentMqttClient(
|
|
ILogger<SentimentMqttClient> logger,
|
|
IConfiguration configuration,
|
|
IServiceScopeFactory scopeFactory) : base(logger)
|
|
{
|
|
_logger = logger;
|
|
_configuration = configuration;
|
|
_scopeFactory = scopeFactory;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Starts the MQTT client and connects to the broker.
|
|
/// </summary>
|
|
public async Task StartAsync(CancellationToken cancellationToken)
|
|
{
|
|
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticSentiment");
|
|
|
|
_logger.LogInformation("Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
|
await ConnectAsync(config);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Stops the MQTT client and disconnects from the broker.
|
|
/// </summary>
|
|
public async Task StopAsync(CancellationToken cancellationToken)
|
|
{
|
|
_logger.LogInformation("Stopping Sentiment MQTT client and disconnecting.");
|
|
await DisconnectAsync();
|
|
}
|
|
|
|
/// <summary>
|
|
/// Event triggered when a real-time article is broadcasted on services/news/completed.
|
|
/// </summary>
|
|
public event Func<NewsArticleDto, Task>? OnArticleReceived;
|
|
|
|
/// <inheritdoc />
|
|
protected override async Task OnConnectedAsync()
|
|
{
|
|
_logger.LogInformation("Sentiment MQTT client connected. Registering RPC topic subscriptions...");
|
|
|
|
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
|
await SubscribeRpcAsync<GetSentimentByIsinRequest, IsinSentimentSummaryDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetIsin), HandleSentimentGetIsinRpcAsync);
|
|
await SubscribeRpcAsync<GetSectorSentimentRequest, SectorSentimentSummaryDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetSector), HandleSentimentGetSectorRpcAsync);
|
|
await SubscribeRpcAsync<ArticleRequest, IsinAnalysisEntry?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetArticle), HandleSentimentGetArticleRpcAsync);
|
|
await SubscribeRpcAsync<PaginatedRequest, List<CompanySentimentSummaryEntity>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentGetAll), HandleSentimentGetAllRpcAsync);
|
|
await SubscribeRpcAsync<JsonElement, IsinAnalysisEntry?>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentAnalyze), HandleSentimentAnalyzeRpcAsync);
|
|
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
|
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.SentimentSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
|
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
|
await SubscribeAsync<NewsArticleDto>(MqttTopics.NewsCompleted, HandleNewsCompletedBroadcastAsync);
|
|
|
|
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
|
{
|
|
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticSentiment", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await PublishAsync(MqttTopics.Logs("FinlyticSentiment"), logDto);
|
|
}
|
|
};
|
|
}
|
|
|
|
/// <summary>
|
|
/// Requests pending news articles from FinlyticNews via MQTT RPC.
|
|
/// </summary>
|
|
public async Task<List<NewsArticleDto>> GetPendingArticlesAsync(int limit = 10)
|
|
{
|
|
try
|
|
{
|
|
var req = new LimitRequest(limit);
|
|
var response = await SendRpcRequestAsync<List<NewsArticleDto>, LimitRequest>(
|
|
MqttTopics.Channels.NewsGetPending,
|
|
req,
|
|
TimeSpan.FromSeconds(5)
|
|
);
|
|
return response ?? [];
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Failed to retrieve pending articles from FinlyticNews via news_GetPending");
|
|
return [];
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Updates the article processing status in FinlyticNews.
|
|
/// </summary>
|
|
public async Task UpdateArticleStatusAsync(Guid articleId, string status)
|
|
{
|
|
try
|
|
{
|
|
var req = new UpdateNewsStatusRequest(articleId, status);
|
|
await SendRpcRequestAsync<UpdateNewsStatusResponse, UpdateNewsStatusRequest>(
|
|
MqttTopics.Channels.NewsUpdateStatus,
|
|
req,
|
|
TimeSpan.FromSeconds(5)
|
|
);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "Failed to update article status for {ArticleId} to '{Status}'", articleId, status);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Broadcasts an updated sentiment result for an asset over MQTT.
|
|
/// </summary>
|
|
public async Task BroadcastSentimentResultAsync(string isin, IsinSentimentSummaryDto summary)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(isin) || summary == null) return;
|
|
await PublishAsync(MqttTopics.SentimentStream(isin), summary);
|
|
}
|
|
|
|
private async Task<IsinSentimentSummaryDto?> HandleSentimentGetIsinRpcAsync(GetSentimentByIsinRequest? req, string correlationId)
|
|
{
|
|
var targetIsin = req?.Isin;
|
|
if (string.IsNullOrWhiteSpace(targetIsin)) return null;
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetIsin for {Isin} [CorrelationId: {CorrelationId}]", targetIsin, correlationId);
|
|
return await dbService.GetIsinSummaryDtoAsync(targetIsin);
|
|
}
|
|
|
|
private async Task<SectorSentimentSummaryDto?> HandleSentimentGetSectorRpcAsync(GetSectorSentimentRequest? req, string correlationId)
|
|
{
|
|
var sector = req?.Sector;
|
|
if (string.IsNullOrWhiteSpace(sector)) return null;
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetSector for {Sector} [CorrelationId: {CorrelationId}]", sector, correlationId);
|
|
return await dbService.GetSectorSentimentAsync(sector);
|
|
}
|
|
|
|
private async Task<IsinAnalysisEntry?> HandleSentimentGetArticleRpcAsync(ArticleRequest? req, string correlationId)
|
|
{
|
|
var targetId = req?.ArticleId ?? req?.Id;
|
|
if (string.IsNullOrWhiteSpace(targetId) || !Guid.TryParse(targetId, out var articleId))
|
|
return null;
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetArticle for {ArticleId} [CorrelationId: {CorrelationId}]", articleId, correlationId);
|
|
return await dbService.GetArticleSentimentEntryAsync(articleId);
|
|
}
|
|
|
|
private async Task<List<CompanySentimentSummaryEntity>> HandleSentimentGetAllRpcAsync(PaginatedRequest? req, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_GetAll [CorrelationId: {CorrelationId}]", correlationId);
|
|
return await dbService.GetAllCompanySentimentsAsync(req?.Limit ?? 50, req?.Offset ?? 0);
|
|
}
|
|
|
|
private async Task<IsinAnalysisEntry?> HandleSentimentAnalyzeRpcAsync(JsonElement rawPayload, string correlationId)
|
|
{
|
|
if (rawPayload.ValueKind == JsonValueKind.Undefined || rawPayload.ValueKind == JsonValueKind.Null) return null;
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var analyzer = scope.ServiceProvider.GetRequiredService<IFinBertAnalyzerService>();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<ISentimentDbService>();
|
|
|
|
NewsArticleDto? article = null;
|
|
try
|
|
{
|
|
var rawText = rawPayload.GetRawText();
|
|
if (rawPayload.TryGetProperty("contentRaw", out _) || rawPayload.TryGetProperty("sourceUrl", out _))
|
|
{
|
|
article = JsonSerializer.Deserialize<NewsArticleDto>(rawText);
|
|
}
|
|
else if (rawPayload.TryGetProperty("articleId", out var articleIdProp))
|
|
{
|
|
var articleIdStr = articleIdProp.GetString();
|
|
if (Guid.TryParse(articleIdStr, out var parsedGuid))
|
|
{
|
|
article = await SendRpcRequestAsync<NewsArticleDto, ArticleRequest>(
|
|
MqttTopics.Channels.NewsGetById,
|
|
new ArticleRequest(articleIdStr, articleIdStr),
|
|
TimeSpan.FromSeconds(5)
|
|
);
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await finlyticLogger.LogWarningAsync(SettingKeys.SentimentChannel, ex, "[SentimentMqttClient] Failed to deserialize payload in sentiment_Analyze");
|
|
}
|
|
|
|
if (article == null || article.Id == Guid.Empty) return null;
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.SentimentChannel, "[SentimentMqttClient] Processing RPC sentiment_Analyze for {ArticleId} [CorrelationId: {CorrelationId}]", article.Id, correlationId);
|
|
var result = await analyzer.AnalyzeArticleAsync(article);
|
|
|
|
if (result != null && article.MatchedAssets != null)
|
|
{
|
|
foreach (var asset in article.MatchedAssets)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(asset.Isin)) continue;
|
|
await dbService.SaveArticleSentimentAsync(article.Id, asset.Isin, asset.Name, null, article.PublishedAt, result);
|
|
}
|
|
|
|
return await dbService.GetArticleSentimentEntryAsync(article.Id);
|
|
}
|
|
|
|
return null;
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_GetAll] Retrieving service dynamic settings [CorrelationId: {CorrelationId}]", correlationId);
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
|
if (updates != null && updates.Count > 0)
|
|
{
|
|
await settingsService.UpdateSettingsAsync(updates);
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticSentiment] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
|
}
|
|
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
|
}
|
|
|
|
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
|
{
|
|
if (topic.Contains("FinlyticSentiment", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
|
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticSentiment", "Online", DateTime.UtcNow, "Connected"));
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<SentimentMqttClient>>();
|
|
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
|
}
|
|
}
|
|
|
|
private async Task HandleNewsCompletedBroadcastAsync(NewsArticleDto? article, string topic, string correlationId)
|
|
{
|
|
if (article != null && OnArticleReceived != null)
|
|
{
|
|
await OnArticleReceived.Invoke(article);
|
|
}
|
|
}
|
|
} |