478 lines
20 KiB
C#
478 lines
20 KiB
C#
using System.Text.Json;
|
|
using FinlyticCore.Dtos;
|
|
using FinlyticCore.Dtos.News;
|
|
using FinlyticCore.Dtos.Sentiment;
|
|
using FinlyticCore.Models;
|
|
using FinlyticCore.Util;
|
|
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 requesting pending news articles and updating sentiment results.
|
|
/// </summary>
|
|
public class SentimentMqttClient : ManagedMqttClient, IHostedService
|
|
{
|
|
private readonly ILogger<SentimentMqttClient> _logger;
|
|
private readonly IConfiguration _configuration;
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
|
|
/// <summary>
|
|
/// Initializes a new instance of the <see cref="SentimentMqttClient"/> class.
|
|
/// </summary>
|
|
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 = new MqttConfiguration
|
|
{
|
|
Host = _configuration["MQTT:Host"] ?? _configuration["MQTT__Host"] ?? "localhost",
|
|
Port = Convert.ToInt32(_configuration["MQTT:Port"] ?? _configuration["MQTT__Port"] ?? "1883"),
|
|
ClientId =
|
|
$"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticSentiment")}_{Guid.NewGuid()}"
|
|
};
|
|
|
|
_logger.LogInformation("[{Channel}] Starting Sentiment MQTT client. Host: {Host}, ClientId: {ClientId}",
|
|
"SentimentChannel", 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("[{Channel}] Stopping Sentiment MQTT client and disconnecting.", "SentimentChannel");
|
|
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(
|
|
"[{Channel}] Sentiment MQTT client connected. Subscribing to RPC response and broadcast topics...",
|
|
"SentimentChannel");
|
|
await SubscribeAsync("services/response/#");
|
|
await SubscribeAsync("services/news/completed");
|
|
await SubscribeAsync("services/request/sentiment_GetArticle/#");
|
|
await SubscribeAsync("services/request/sentiment_GetIsin/#");
|
|
await SubscribeAsync("services/request/sentiment_Analyze/#");
|
|
await SubscribeAsync("services/request/health_Ping/#");
|
|
await SubscribeAsync("services/config/updated/#");
|
|
}
|
|
|
|
/// <inheritdoc />
|
|
protected override async Task OnMessageReceivedAsync(string topic, string payload)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(topic)) return;
|
|
|
|
// 1. Config update events
|
|
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
if (topic.EndsWith("FinlyticSentiment", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnConfigUpdatedAsync(payload);
|
|
}
|
|
|
|
return;
|
|
}
|
|
|
|
if (topic.Equals("services/news/completed", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnNewsCompletedAsync(payload);
|
|
return;
|
|
}
|
|
|
|
// Extract correlationId from topic suffix (e.g. services/request/sentiment_GetArticle/{correlationId})
|
|
var lastSlash = topic.LastIndexOf('/');
|
|
if (lastSlash < 0 || lastSlash >= topic.Length - 1) return;
|
|
|
|
var correlationId = topic.Substring(lastSlash + 1);
|
|
|
|
// 2. Dispatch to specific channel handlers
|
|
if (topic.StartsWith("services/request/sentiment_GetArticle", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnSentimentGetArticleAsync(payload, correlationId);
|
|
}
|
|
else if (topic.StartsWith("services/request/sentiment_GetIsin", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnSentimentGetIsinAsync(payload, correlationId);
|
|
}
|
|
else if (topic.StartsWith("services/request/sentiment_Analyze", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnSentimentAnalyzeAsync(payload, correlationId);
|
|
}
|
|
else if (topic.StartsWith("services/request/health_Ping", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await OnHealthPingAsync(topic, correlationId);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles dynamic service config update events.
|
|
/// </summary>
|
|
private async Task OnConfigUpdatedAsync(string payload)
|
|
{
|
|
_logger.LogInformation("[{Channel}] [SentimentMqttClient] Received config update event for FinlyticSentiment.",
|
|
"SentimentChannel");
|
|
try
|
|
{
|
|
using var doc = JsonDocument.Parse(payload);
|
|
if (doc.RootElement.TryGetProperty("settings", out var settingsProp))
|
|
{
|
|
var dict = JsonSerializer.Deserialize(settingsProp.GetRawText(), typeof(Dictionary<string, string>),
|
|
FinlyticJsonSerializerContext.Default) as Dictionary<string, string>;
|
|
if (dict != null && dict.Count > 0)
|
|
{
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
|
|
await settingsDb.UpdateSettingsFromDictionaryAsync(dict);
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Successfully persisted {Count} updated settings for FinlyticSentiment.",
|
|
"SentimentChannel", dict.Count);
|
|
}
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Error processing MQTT config update event.",
|
|
"SentimentChannel");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles health_Ping RPC requests.
|
|
/// </summary>
|
|
private async Task OnHealthPingAsync(string topic, string correlationId)
|
|
{
|
|
if (topic.Contains("FinlyticSentiment", StringComparison.OrdinalIgnoreCase) ||
|
|
!topic.Contains("/", StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
string respTopic = $"services/response/health_Ping/{correlationId}";
|
|
await PublishAsync(respTopic,
|
|
new ServiceHealthResponse("FinlyticSentiment", "Online", DateTime.UtcNow, "Connected"));
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].",
|
|
"SentimentChannel", correlationId);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles broadcasted articles on services/news/completed.
|
|
/// </summary>
|
|
private async Task OnNewsCompletedAsync(string payload)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(payload)) return;
|
|
try
|
|
{
|
|
var article =
|
|
JsonSerializer.Deserialize(payload, typeof(NewsArticleDto), FinlyticJsonSerializerContext.Default) as
|
|
NewsArticleDto;
|
|
if (article != null && article.Id != Guid.Empty && OnArticleReceived != null)
|
|
{
|
|
_logger.LogInformation(
|
|
"[{Channel}] Received real-time article broadcast on services/news/completed: {Title} (ID: {Id})",
|
|
"SentimentChannel", article.Title, article.Id);
|
|
await OnArticleReceived.Invoke(article);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Error parsing broadcasted article on services/news/completed.",
|
|
"SentimentChannel");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles sentiment_GetArticle RPC requests using source-generated DTO deserialization.
|
|
/// </summary>
|
|
private async Task OnSentimentGetArticleAsync(string payload, string correlationId)
|
|
{
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Processing RPC sentiment_GetArticle request [CorrelationId: {CorrelationId}]",
|
|
"SentimentChannel", correlationId);
|
|
if (string.IsNullOrWhiteSpace(payload)) return;
|
|
|
|
try
|
|
{
|
|
var request =
|
|
JsonSerializer.Deserialize(payload, typeof(ArticleRequest), FinlyticJsonSerializerContext.Default) as
|
|
ArticleRequest;
|
|
var articleId = request?.ArticleId ?? request?.Id;
|
|
|
|
if (string.IsNullOrWhiteSpace(articleId))
|
|
{
|
|
_logger.LogWarning("[{Channel}] [SentimentMqttClient] Missing articleId in request payload.",
|
|
"SentimentChannel");
|
|
return;
|
|
}
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var storageService = scope.ServiceProvider.GetRequiredService<ISentimentStorageService>();
|
|
var sentimentEntry = await storageService.GetArticleSentimentAsync(articleId);
|
|
|
|
string responseTopic = $"services/response/sentiment_GetArticle/{correlationId}";
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_GetArticle response for article {ArticleId} to {ResponseTopic}",
|
|
"SentimentChannel", articleId, responseTopic);
|
|
await PublishAsync(responseTopic, sentimentEntry);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex,
|
|
"[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetArticle RPC request.",
|
|
"SentimentChannel");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles sentiment_GetIsin RPC requests using source-generated DTO deserialization.
|
|
/// </summary>
|
|
private async Task OnSentimentGetIsinAsync(string payload, string correlationId)
|
|
{
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Processing RPC sentiment_GetIsin request [CorrelationId: {CorrelationId}]",
|
|
"SentimentChannel", correlationId);
|
|
if (string.IsNullOrWhiteSpace(payload)) return;
|
|
|
|
try
|
|
{
|
|
var request =
|
|
JsonSerializer.Deserialize(payload, typeof(IsinRequest), FinlyticJsonSerializerContext.Default) as
|
|
IsinRequest;
|
|
var isin = request?.Isin;
|
|
|
|
if (string.IsNullOrWhiteSpace(isin))
|
|
{
|
|
_logger.LogWarning("[{Channel}] [SentimentMqttClient] Missing ISIN in request payload.",
|
|
"SentimentChannel");
|
|
return;
|
|
}
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var storageService = scope.ServiceProvider.GetRequiredService<ISentimentStorageService>();
|
|
var isinSummary = await storageService.GetIsinSummaryAsync(isin);
|
|
|
|
string responseTopic = $"services/response/sentiment_GetIsin/{correlationId}";
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_GetIsin response for ISIN {Isin} to {ResponseTopic}",
|
|
"SentimentChannel", isin, responseTopic);
|
|
await PublishAsync(responseTopic, isinSummary);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_GetIsin RPC request.",
|
|
"SentimentChannel");
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Handles manual/forced sentiment_Analyze RPC requests using existing analyzer and storage services.
|
|
/// </summary>
|
|
private async Task OnSentimentAnalyzeAsync(string payload, string correlationId)
|
|
{
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Processing RPC sentiment_Analyze request [CorrelationId: {CorrelationId}]",
|
|
"SentimentChannel", correlationId);
|
|
string responseTopic = $"services/response/sentiment_Analyze/{correlationId}";
|
|
|
|
if (string.IsNullOrWhiteSpace(payload))
|
|
{
|
|
await PublishAsync(responseTopic, (object?)null);
|
|
return;
|
|
}
|
|
|
|
try
|
|
{
|
|
var request =
|
|
JsonSerializer.Deserialize(payload, typeof(AnalyzeSentimentRequest),
|
|
FinlyticJsonSerializerContext.Default) as AnalyzeSentimentRequest;
|
|
|
|
if (request == null ||
|
|
(string.IsNullOrWhiteSpace(request.ArticleId) && string.IsNullOrWhiteSpace(request.Isin)))
|
|
{
|
|
_logger.LogWarning(
|
|
"[{Channel}] [SentimentMqttClient] Missing ArticleId or Isin in sentiment_Analyze request payload.",
|
|
"SentimentChannel");
|
|
await PublishAsync(responseTopic, (object?)null);
|
|
return;
|
|
}
|
|
|
|
using var scope = _scopeFactory.CreateScope();
|
|
var storageService = scope.ServiceProvider.GetRequiredService<ISentimentStorageService>();
|
|
var analyzerService = scope.ServiceProvider.GetRequiredService<IFinBertAnalyzerService>();
|
|
|
|
object? result = null;
|
|
|
|
// Fall 1: Manuelle Analyse für einen einzelnen Artikel
|
|
if (!string.IsNullOrWhiteSpace(request.ArticleId))
|
|
{
|
|
var cleanArticleId = request.ArticleId.Trim();
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Processing article analysis for ArticleId: {ArticleId} (ForceReload: {ForceReload})",
|
|
"SentimentChannel", cleanArticleId, request.ForceReload);
|
|
|
|
IsinAnalysisEntry? existingEntry = null;
|
|
|
|
if (!request.ForceReload)
|
|
{
|
|
existingEntry = await storageService.GetArticleSentimentAsync(cleanArticleId);
|
|
}
|
|
|
|
if (existingEntry != null)
|
|
{
|
|
result = existingEntry;
|
|
}
|
|
else
|
|
{
|
|
// Artikel per RPC von FinlyticNews abfragen
|
|
if (Guid.TryParse(cleanArticleId, out var articleGuid))
|
|
{
|
|
var article = await SendRpcRequestAsync<NewsArticleDto, ArticleRequest>(
|
|
"news_GetById",
|
|
new ArticleRequest(cleanArticleId, cleanArticleId),
|
|
TimeSpan.FromSeconds(8));
|
|
|
|
if (article != null)
|
|
{
|
|
var finbertResult = await analyzerService.AnalyzeArticleAsync(article);
|
|
|
|
// 🎯 NULL-CHECK: Falls Analyse fehlschlägt/null liefert -> abbrechen
|
|
if (finbertResult == null)
|
|
{
|
|
_logger.LogWarning(
|
|
"[{Channel}] [SentimentMqttClient] FinBERT analysis returned NULL for article {ArticleId}. Aborting manual analysis.",
|
|
"SentimentChannel", cleanArticleId);
|
|
await PublishAsync(responseTopic, (object?)null);
|
|
return;
|
|
}
|
|
|
|
// Speichern & Summaries aktualisieren
|
|
await storageService.SaveArticleSentimentAsync(article, finbertResult);
|
|
|
|
if (article.MatchedAssets != null && article.MatchedAssets.Count > 0)
|
|
{
|
|
foreach (var asset in article.MatchedAssets)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(asset.Isin)) continue;
|
|
|
|
await storageService.UpdateIsinSummaryAsync(
|
|
asset.Isin,
|
|
asset.Name,
|
|
"General",
|
|
article,
|
|
finbertResult);
|
|
|
|
await storageService.UpdateSectorSummaryAsync(
|
|
"General",
|
|
asset.Isin,
|
|
article.Id.ToString(),
|
|
finbertResult);
|
|
}
|
|
}
|
|
|
|
// FinlyticNews über Re-Analyse informieren
|
|
await UpdateArticleStatusAsync(article.Id, "Analyzed");
|
|
|
|
result = await storageService.GetArticleSentimentAsync(cleanArticleId);
|
|
}
|
|
else
|
|
{
|
|
_logger.LogWarning(
|
|
"[{Channel}] [SentimentMqttClient] Could not retrieve article {ArticleId} from FinlyticNews for re-analysis.",
|
|
"SentimentChannel", cleanArticleId);
|
|
}
|
|
}
|
|
}
|
|
}
|
|
// Fall 2: ISIN Gesamtsummary anfordern
|
|
else if (!string.IsNullOrWhiteSpace(request.Isin))
|
|
{
|
|
var cleanIsin = request.Isin.Trim();
|
|
_logger.LogInformation("[{Channel}] [SentimentMqttClient] Fetching sentiment summary for ISIN: {Isin}",
|
|
"SentimentChannel", cleanIsin);
|
|
|
|
result = await storageService.GetIsinSummaryAsync(cleanIsin);
|
|
}
|
|
|
|
_logger.LogInformation(
|
|
"[{Channel}] [SentimentMqttClient] Publishing RPC sentiment_Analyze response to {ResponseTopic}",
|
|
"SentimentChannel", responseTopic);
|
|
await PublishAsync(responseTopic, result);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] [SentimentMqttClient] Failed to process sentiment_Analyze RPC request.",
|
|
"SentimentChannel");
|
|
await PublishAsync(responseTopic, (object?)null);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// Requests pending news articles from FinlyticNews via MQTT RPC.
|
|
/// </summary>
|
|
/// <param name="limit">The maximum number of articles to request (capped at 10).</param>
|
|
/// <returns>A list of pending news article DTOs.</returns>
|
|
public async Task<List<NewsArticleDto>> GetPendingArticlesAsync(int limit = 10)
|
|
{
|
|
try
|
|
{
|
|
var payload = new LimitRequest(Math.Min(limit, 10));
|
|
var articles =
|
|
await SendRpcRequestAsync<List<NewsArticleDto>, LimitRequest>("news_GetPending", payload,
|
|
TimeSpan.FromSeconds(10));
|
|
return articles ?? [];
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_GetPending.", "SentimentChannel");
|
|
}
|
|
|
|
return [];
|
|
}
|
|
|
|
/// <summary>
|
|
/// Dispatches an RPC request to update the article status in FinlyticNews (e.g. to "Analyzed").
|
|
/// </summary>
|
|
/// <param name="id">The article identifier.</param>
|
|
/// <param name="status">The target status string (default "Analyzed").</param>
|
|
/// <returns>True if the status update succeeded; otherwise, false.</returns>
|
|
public async Task<bool> UpdateArticleStatusAsync(Guid id, string status = "Analyzed")
|
|
{
|
|
try
|
|
{
|
|
var request = new UpdateNewsStatusRequest(id, status);
|
|
var response =
|
|
await SendRpcRequestAsync<UpdateNewsStatusResponse, UpdateNewsStatusRequest>("news_UpdateStatus",
|
|
request, TimeSpan.FromSeconds(8));
|
|
return response?.Success ?? false;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
_logger.LogError(ex, "[{Channel}] Error executing MQTT RPC for news_UpdateStatus (ID: {Id}).",
|
|
"SentimentChannel", id);
|
|
}
|
|
|
|
return false;
|
|
}
|
|
} |