feat(news): add scraper adapters, article deduplication, blocklist service, and remove tracked publish artifacts
This commit is contained in:
@@ -1,8 +1,6 @@
|
||||
using System;
|
||||
using System.Collections.Generic;
|
||||
using System.IO;
|
||||
using System.Linq;
|
||||
using System.Text.Json;
|
||||
using System.Threading;
|
||||
using System.Threading.Tasks;
|
||||
using FinlyticCore.Dtos;
|
||||
@@ -12,7 +10,6 @@ using FinlyticCore.Dtos.Settings;
|
||||
using FinlyticCore.Models;
|
||||
using FinlyticCore.Services;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticNews.Database;
|
||||
using FinlyticNews.Entities;
|
||||
using FinlyticNews.Services;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
@@ -23,7 +20,7 @@ using Microsoft.Extensions.Logging;
|
||||
namespace FinlyticNews.Util;
|
||||
|
||||
/// <summary>
|
||||
/// A managed MQTT client for broadcasting completed news articles and responding to RPC requests.
|
||||
/// Managed MQTT client for broadcasting completed news articles and handling typed RPC requests for FinlyticNews.
|
||||
/// </summary>
|
||||
public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
{
|
||||
@@ -44,12 +41,7 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
/// <inheritdoc />
|
||||
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"] ?? "FinlyticNews")}_{Guid.NewGuid()}"
|
||||
};
|
||||
var config = MqttConfiguration.FromConfiguration(_configuration, "FinlyticNews");
|
||||
|
||||
_logger.LogInformation("Starting News MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
|
||||
await ConnectAsync(config);
|
||||
@@ -65,22 +57,22 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
/// <inheritdoc />
|
||||
protected override async Task OnConnectedAsync()
|
||||
{
|
||||
_logger.LogInformation("News MQTT client connected. Subscribing to RPC topics...");
|
||||
_logger.LogInformation("News MQTT client connected. Registering RPC topic subscriptions...");
|
||||
|
||||
await SubscribeAsync("services/request/news_Get/#");
|
||||
await SubscribeAsync("services/request/news_GetById/#");
|
||||
await SubscribeAsync("services/request/news_GetPending/#");
|
||||
await SubscribeAsync("services/request/news_UpdateStatus/#");
|
||||
await SubscribeAsync("services/request/news_settings_GetAll/#");
|
||||
await SubscribeAsync("services/request/news_settings_Update/#");
|
||||
await SubscribeAsync("services/request/health_Ping/#");
|
||||
await SubscribeAsync("services/config/updated/#");
|
||||
await SubscribeAsync(MqttTopics.ResponseWildcard);
|
||||
await SubscribeRpcAsync<DailyNewsRequest, List<NewsArticleDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGet), HandleNewsGetRpcAsync);
|
||||
await SubscribeRpcAsync<ArticleRequest, NewsArticleDto?>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGetById), HandleNewsGetByIdRpcAsync);
|
||||
await SubscribeRpcAsync<object, List<NewsArticleDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsGetPending), HandleNewsGetPendingRpcAsync);
|
||||
await SubscribeRpcAsync<UpdateNewsStatusRequest, UpdateNewsStatusResponse>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsUpdateStatus), HandleNewsUpdateStatusRpcAsync);
|
||||
await SubscribeRpcAsync<object, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsSettingsGetAll), HandleSettingsGetAllRpcAsync);
|
||||
await SubscribeRpcAsync<Dictionary<string, object?>, List<DynamicSettingDto>>(MqttTopics.RequestFilter(MqttTopics.Channels.NewsSettingsUpdate), HandleSettingsUpdateRpcAsync);
|
||||
await SubscribeAsync<object>(MqttTopics.RequestFilter(MqttTopics.Channels.HealthPing), HandleHealthPingRpcAsync);
|
||||
|
||||
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
|
||||
{
|
||||
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticNews", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await PublishAsync("finlytic/logs/FinlyticNews", logDto);
|
||||
await PublishAsync(MqttTopics.Logs("FinlyticNews"), logDto);
|
||||
}
|
||||
};
|
||||
}
|
||||
@@ -90,317 +82,126 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
/// </summary>
|
||||
public async Task BroadcastArticleAsync(NewsArticleDto article)
|
||||
{
|
||||
const string topic = "services/news/completed";
|
||||
const string topic = MqttTopics.NewsCompleted;
|
||||
_logger.LogInformation("Broadcasting completed article to MQTT topic: {Topic}. ID: {Id}", topic, article.Id);
|
||||
await PublishAsync(topic, article);
|
||||
|
||||
var firstIsin = article.MatchedAssets.FirstOrDefault()?.Isin;
|
||||
if (!string.IsNullOrWhiteSpace(firstIsin))
|
||||
{
|
||||
string isinTopic = $"finlytic/news/stream/{firstIsin.Trim().ToLowerInvariant()}";
|
||||
string isinTopic = MqttTopics.NewsStream(firstIsin);
|
||||
await PublishAsync(isinTopic, article);
|
||||
}
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override async Task OnMessageReceivedAsync(string topic, string payload)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(topic)) return;
|
||||
|
||||
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
if (topic.EndsWith("FinlyticNews", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
await OnConfigUpdatedAsync(payload);
|
||||
}
|
||||
return;
|
||||
}
|
||||
|
||||
var segments = topic.Split('/');
|
||||
if (segments.Length < 4) return;
|
||||
|
||||
var channel = segments[2];
|
||||
var correlationId = segments[^1];
|
||||
|
||||
switch (channel)
|
||||
{
|
||||
case "news_Get":
|
||||
await OnGetNewsAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "news_GetById":
|
||||
await OnGetNewsByIdAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "news_GetPending":
|
||||
await OnGetPendingNewsAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "news_UpdateStatus":
|
||||
await OnUpdateNewsStatusAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "news_settings_GetAll":
|
||||
await OnSettingsGetAllAsync(correlationId);
|
||||
break;
|
||||
|
||||
case "news_settings_Update":
|
||||
await OnSettingsUpdateAsync(payload, correlationId);
|
||||
break;
|
||||
|
||||
case "health_Ping":
|
||||
await OnHealthPingAsync(segments, correlationId);
|
||||
break;
|
||||
|
||||
default:
|
||||
_logger.LogDebug("Received unhandled RPC channel: {Channel}", channel);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
private async Task OnGetNewsAsync(string payload, string correlationId)
|
||||
private async Task<List<NewsArticleDto>> HandleNewsGetRpcAsync(DailyNewsRequest? req, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "Received RPC news_Get request. Correlation: {CorrelationId}", correlationId);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "Received RPC news_Get request [CorrelationId: {CorrelationId}]", correlationId);
|
||||
|
||||
int limit = 20;
|
||||
int offset = 0;
|
||||
string? isin = null;
|
||||
DateTime? date = null;
|
||||
string? status = null;
|
||||
string? searchQuery = null;
|
||||
int limit = req?.Limit > 0 ? req.Limit : 20;
|
||||
int offset = req?.Offset >= 0 ? req.Offset : 0;
|
||||
string? isin = !string.IsNullOrWhiteSpace(req?.Isin) ? req.Isin : null;
|
||||
string? status = req?.Status;
|
||||
string? searchQuery = req?.Query;
|
||||
DateTime? date = req?.Date;
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(payload))
|
||||
if (req?.HasSentiment == true && string.IsNullOrEmpty(status))
|
||||
{
|
||||
try
|
||||
{
|
||||
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.DailyNewsRequest);
|
||||
if (req != null)
|
||||
{
|
||||
limit = req.Limit > 0 ? req.Limit : 20;
|
||||
offset = req.Offset >= 0 ? req.Offset : 0;
|
||||
isin = !string.IsNullOrWhiteSpace(req.Isin) ? req.Isin : null;
|
||||
status = req.Status;
|
||||
searchQuery = req.Query;
|
||||
date = req.Date;
|
||||
|
||||
if (req.HasSentiment == true && string.IsNullOrEmpty(status))
|
||||
{
|
||||
status = "Analyzed";
|
||||
}
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, ex, "Failed to parse DailyNewsRequest payload on news_Get");
|
||||
}
|
||||
status = "Analyzed";
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
var articles = await dbService.GetFilteredNewsAsync(limit, offset, isin, date, status, searchQuery);
|
||||
var dtos = (await Task.WhenAll(articles.Select(a => MapToDtoAsync(a)))).ToList();
|
||||
|
||||
string responseTopic = $"services/response/news_Get/{correlationId}";
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "Publishing RPC response to {ResponseTopic} with {Count} articles.", responseTopic, dtos.Count);
|
||||
await PublishAsync(responseTopic, dtos);
|
||||
return (await Task.WhenAll(articles.Select(a => MapToDtoAsync(a)))).ToList();
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "Failed to compile RPC response for news_Get");
|
||||
try
|
||||
{
|
||||
string responseTopic = $"services/response/news_Get/{correlationId}";
|
||||
await PublishAsync(responseTopic, new List<NewsArticleDto>());
|
||||
}
|
||||
catch { }
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
private async Task OnGetNewsByIdAsync(string payload, string correlationId)
|
||||
private async Task<NewsArticleDto?> HandleNewsGetByIdRpcAsync(ArticleRequest? req, string correlationId)
|
||||
{
|
||||
string responseTopic = $"services/response/news_GetById/{correlationId}";
|
||||
if (string.IsNullOrWhiteSpace(payload))
|
||||
var targetIdStr = req?.ArticleId ?? req?.Id;
|
||||
if (Guid.TryParse(targetIdStr, out var articleId))
|
||||
{
|
||||
await PublishAsync(responseTopic, (object?)null);
|
||||
return;
|
||||
}
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
var article = await dbService.GetArticleByIdAsync(articleId);
|
||||
|
||||
try
|
||||
{
|
||||
var request = JsonSerializer.Deserialize<ArticleRequest>(payload, FinlyticJsonSerializerContext.Default.ArticleRequest);
|
||||
var targetIdStr = request?.ArticleId ?? request?.Id;
|
||||
|
||||
if (Guid.TryParse(targetIdStr, out var articleId))
|
||||
if (article != null)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
var article = await dbService.GetArticleByIdAsync(articleId);
|
||||
|
||||
if (article != null)
|
||||
{
|
||||
var dto = await MapToDtoAsync(article);
|
||||
await PublishAsync(responseTopic, dto);
|
||||
return;
|
||||
}
|
||||
return await MapToDtoAsync(article);
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
|
||||
await PublishAsync(responseTopic, (object?)null);
|
||||
return null;
|
||||
}
|
||||
|
||||
private async Task OnGetPendingNewsAsync(string payload, string correlationId)
|
||||
private async Task<List<NewsArticleDto>> HandleNewsGetPendingRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
int limit = 10;
|
||||
if (!string.IsNullOrWhiteSpace(payload))
|
||||
{
|
||||
try
|
||||
{
|
||||
using var doc = JsonDocument.Parse(payload);
|
||||
if (doc.RootElement.TryGetProperty("limit", out var limitProp) && limitProp.TryGetInt32(out var parsedLimit))
|
||||
{
|
||||
limit = Math.Min(parsedLimit, 10);
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
|
||||
var pendingArticles = await dbService.GetArticlesByStatusAsync("Pending");
|
||||
var dtos = (await Task.WhenAll(pendingArticles.Take(limit).Select(a => MapToDtoAsync(a)))).ToList();
|
||||
|
||||
string responseTopic = $"services/response/news_GetPending/{correlationId}";
|
||||
await PublishAsync(responseTopic, dtos);
|
||||
var pendingArticles = await dbService.GetArticlesByStatusAsync("Completed");
|
||||
return (await Task.WhenAll(pendingArticles.Take(10).Select(a => MapToDtoAsync(a)))).ToList();
|
||||
}
|
||||
catch
|
||||
{
|
||||
return [];
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
|
||||
private async Task OnUpdateNewsStatusAsync(string payload, string correlationId)
|
||||
private async Task<UpdateNewsStatusResponse> HandleNewsUpdateStatusRpcAsync(UpdateNewsStatusRequest? req, string correlationId)
|
||||
{
|
||||
UpdateNewsStatusResponse response;
|
||||
try
|
||||
if (req != null && req.Id != Guid.Empty)
|
||||
{
|
||||
var request = JsonSerializer.Deserialize<UpdateNewsStatusRequest>(payload, FinlyticJsonSerializerContext.Default.UpdateNewsStatusRequest);
|
||||
if (request != null && request.Id != Guid.Empty)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
|
||||
await dbService.UpdateArticleStatusAsync(request.Id, request.Status);
|
||||
response = new UpdateNewsStatusResponse(true, "Status updated successfully.");
|
||||
}
|
||||
else
|
||||
{
|
||||
response = new UpdateNewsStatusResponse(false, "Invalid payload.");
|
||||
}
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
await dbService.UpdateArticleStatusAsync(req.Id, req.Status);
|
||||
return new UpdateNewsStatusResponse(true, "Status updated successfully.");
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
response = new UpdateNewsStatusResponse(false, ex.Message);
|
||||
}
|
||||
|
||||
string responseTopic = $"services/response/news_UpdateStatus/{correlationId}";
|
||||
await PublishAsync(responseTopic, response);
|
||||
return new UpdateNewsStatusResponse(false, "Invalid payload.");
|
||||
}
|
||||
|
||||
private async Task OnSettingsGetAllAsync(string correlationId)
|
||||
private async Task<List<DynamicSettingDto>> HandleSettingsGetAllRpcAsync(object? _, string correlationId)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_GetAll] Retrieving all dynamic settings via reflection [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
{
|
||||
var settings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/news_settings_GetAll/{correlationId}";
|
||||
|
||||
await PublishAsync(responseTopic, settings);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_GetAll] Published {Count} settings to '{ResponseTopic}'", settings.Count, responseTopic);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticNews] [Settings_GetAll] Failed to retrieve settings.");
|
||||
}
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_GetAll] Retrieving service dynamic settings [CorrelationId: {CorrelationId}]", correlationId);
|
||||
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
}
|
||||
|
||||
private async Task OnSettingsUpdateAsync(string payload, string correlationId)
|
||||
private async Task<List<DynamicSettingDto>> HandleSettingsUpdateRpcAsync(Dictionary<string, object?>? updates, string correlationId)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(payload)) return;
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
||||
var settingsService = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_Update] Processing settings update RPC [CorrelationId: {CorrelationId}]", correlationId);
|
||||
try
|
||||
if (updates != null && updates.Count > 0)
|
||||
{
|
||||
Dictionary<string, object?>? updates = null;
|
||||
try
|
||||
{
|
||||
updates = JsonSerializer.Deserialize<Dictionary<string, object?>>(payload);
|
||||
}
|
||||
catch
|
||||
{
|
||||
var list = JsonSerializer.Deserialize<List<DynamicSettingDto>>(payload);
|
||||
if (list != null)
|
||||
{
|
||||
updates = new Dictionary<string, object?>();
|
||||
foreach (var item in list) updates[item.Key] = item.Value;
|
||||
}
|
||||
}
|
||||
|
||||
if (updates != null && updates.Count > 0)
|
||||
{
|
||||
await settingsService.UpdateSettingsAsync(updates);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
||||
}
|
||||
|
||||
var currentSettings = await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
var responseTopic = $"services/response/news_settings_Update/{correlationId}";
|
||||
await PublishAsync(responseTopic, currentSettings);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
await finlyticLogger.LogErrorAsync(SettingKeys.MqttChannel, ex, "[FinlyticNews] [Settings_Update] Failed to update settings.");
|
||||
await settingsService.UpdateSettingsAsync(updates);
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.MqttChannel, "[FinlyticNews] [Settings_Update] Successfully updated {Count} settings in database and cache.", updates.Count);
|
||||
}
|
||||
return await settingsService.GetAllRegisteredSettingsAsync(new[] { typeof(SettingKeys) });
|
||||
}
|
||||
|
||||
private async Task OnConfigUpdatedAsync(string payload)
|
||||
private async Task HandleHealthPingRpcAsync(object? _, string topic, string correlationId)
|
||||
{
|
||||
try
|
||||
if (topic.Contains("FinlyticNews", StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
using var doc = JsonDocument.Parse(payload);
|
||||
if (doc.RootElement.TryGetProperty("settings", out var settingsProp))
|
||||
{
|
||||
var dict = JsonSerializer.Deserialize<Dictionary<string, object?>>(settingsProp.GetRawText());
|
||||
if (dict != null && dict.Count > 0)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
||||
await settings.UpdateSettingsAsync(dict);
|
||||
}
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
|
||||
private async Task OnHealthPingAsync(string[] segments, string correlationId)
|
||||
{
|
||||
bool isForMe = segments.Length >= 5 && segments[3].Equals("FinlyticNews", StringComparison.OrdinalIgnoreCase);
|
||||
if (isForMe)
|
||||
{
|
||||
string respTopic = $"services/response/health_Ping/{correlationId}";
|
||||
string respTopic = MqttTopics.ResponseTopic(MqttTopics.Channels.HealthPing, correlationId);
|
||||
await PublishAsync(respTopic, new ServiceHealthResponse("FinlyticNews", "Online", DateTime.UtcNow, "Connected"));
|
||||
|
||||
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var finlyticLogger = scope.ServiceProvider.GetRequiredService<IFinlyticLogger<NewsMqttClient>>();
|
||||
await finlyticLogger.LogInfoAsync(SettingKeys.HealthPingChannel, "Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].", correlationId);
|
||||
@@ -408,7 +209,11 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Maps a NewsArticleEntity to a NewsArticleDto while performing zero-latency disk lookups for sentiment summaries.
|
||||
/// Maps a NewsArticleEntity to a NewsArticleDto, enriching it with the FinBERT sentiment result for the
|
||||
/// article by querying FinlyticSentiment over the <see cref="MqttTopics.Channels.SentimentGetArticle"/> RPC
|
||||
/// channel. This previously read a legacy on-disk cache at <c>data/summaries/articles/*.json</c> and
|
||||
/// <c>data/summaries/isin/*.json</c>, but nothing in this repository writes those files anymore (FinlyticSentiment
|
||||
/// persists results to its own database) - the disk lookup was silently degrading Sentiment to always-null.
|
||||
/// </summary>
|
||||
private async Task<NewsArticleDto> MapToDtoAsync(NewsArticleEntity a)
|
||||
{
|
||||
@@ -419,55 +224,27 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
|
||||
|
||||
try
|
||||
{
|
||||
var targetId = a.Id.ToString();
|
||||
IsinAnalysisEntry? sentimentEntry = null;
|
||||
|
||||
var articlePath = Path.Combine(Directory.GetCurrentDirectory(), "data", "summaries", "articles", $"{targetId}.json");
|
||||
if (File.Exists(articlePath))
|
||||
if (IsConnected)
|
||||
{
|
||||
try
|
||||
var targetId = a.Id.ToString();
|
||||
var sentimentEntry = await SendRpcRequestAsync<IsinAnalysisEntry, ArticleRequest>(
|
||||
MqttTopics.Channels.SentimentGetArticle,
|
||||
new ArticleRequest(targetId, targetId),
|
||||
TimeSpan.FromSeconds(3));
|
||||
|
||||
if (sentimentEntry?.FinbertResult != null)
|
||||
{
|
||||
var json = await File.ReadAllTextAsync(articlePath);
|
||||
sentimentEntry = JsonSerializer.Deserialize<IsinAnalysisEntry>(json, FinlyticJsonSerializerContext.Default.IsinAnalysisEntry);
|
||||
finbertResult = sentimentEntry.FinbertResult;
|
||||
sentimentLabel = finbertResult.Label;
|
||||
sentimentScore = finbertResult.CompoundScore;
|
||||
confidence = finbertResult.Confidence;
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
|
||||
if (sentimentEntry == null && a.MatchedAssets != null && a.MatchedAssets.Count > 0)
|
||||
{
|
||||
foreach (var asset in a.MatchedAssets)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(asset.Isin)) continue;
|
||||
var isinPath = Path.Combine(Directory.GetCurrentDirectory(), "data", "summaries", "isin", $"{asset.Isin.Trim()}.json");
|
||||
|
||||
if (File.Exists(isinPath))
|
||||
{
|
||||
try
|
||||
{
|
||||
var json = await File.ReadAllTextAsync(isinPath);
|
||||
var isinDoc = JsonSerializer.Deserialize<IsinSentimentSummaryDto>(json, FinlyticJsonSerializerContext.Default.IsinSentimentSummaryDto);
|
||||
var match = isinDoc?.Analyses?.FirstOrDefault(entry => string.Equals(entry.Article?.ArticleId?.Trim(), targetId, StringComparison.OrdinalIgnoreCase));
|
||||
|
||||
if (match != null)
|
||||
{
|
||||
sentimentEntry = match;
|
||||
break;
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (sentimentEntry?.FinbertResult != null)
|
||||
{
|
||||
finbertResult = sentimentEntry.FinbertResult;
|
||||
sentimentLabel = finbertResult.Label;
|
||||
sentimentScore = finbertResult.CompoundScore;
|
||||
confidence = finbertResult.Confidence;
|
||||
}
|
||||
}
|
||||
catch { }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogWarning(ex, "[NewsMqttClient] Failed to fetch sentiment for article {ArticleId} via '{Channel}' RPC.", a.Id, MqttTopics.Channels.SentimentGetArticle);
|
||||
}
|
||||
|
||||
return new NewsArticleDto
|
||||
{
|
||||
|
||||
Reference in New Issue
Block a user