Kleidukos/Finlytic
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(news): dynamic settings, IFinlyticLogger, live log streaming, and EF migration

Browse Source
This commit is contained in:
Kleidukos
2026-08-15 21:30:01 +02:00
parent a1f2b888f6
commit a94c36a878
12 changed files with 800 additions and 555 deletions
Download Patch File Download Diff File Expand all files Collapse all files
FinlyticNews/Util/NewsMqttClient.cs
+122 -90
View File
@@ -1,10 +1,18 @@
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;
using FinlyticCore.Dtos.News;
using FinlyticCore.Dtos.Sentiment;
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;
@@ -40,12 +48,10 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
{
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()}"
ClientId = $"{(_configuration["MQTT:ClientId"] ?? _configuration["MQTT__ClientId"] ?? "FinlyticNews")}_{Guid.NewGuid()}"
};
_logger.LogInformation("Starting News MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host,
config.ClientId);
_logger.LogInformation("Starting News MQTT client. Host: {Host}, ClientId: {ClientId}", config.Host, config.ClientId);
await ConnectAsync(config);
}
@@ -61,13 +67,22 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
{
_logger.LogInformation("News MQTT client connected. Subscribing to RPC topics...");
// ZUSAMMENGELEGT: Unified News Fetching (news_Get deckt news_GetDaily mit ab)
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/#");
FinlyticCore.Services.FinlyticLogBroadcaster.OnLogPublished = async (logDto) =>
{
if (IsConnected && string.Equals(logDto.ServiceName, "FinlyticNews", StringComparison.OrdinalIgnoreCase))
{
await PublishAsync("finlytic/logs/FinlyticNews", logDto);
}
};
}
/// <summary>
@@ -75,12 +90,10 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
/// </summary>
public async Task BroadcastArticleAsync(NewsArticleDto article)
{
// 1. Primärer System-Broadcast
const string topic = "services/news/completed";
_logger.LogInformation("Broadcasting completed article to MQTT topic: {Topic}. ID: {Id}", topic, article.Id);
await PublishAsync(topic, article);
// 2. Zielgerichteter ISIN-Stream für Echtzeit-Frontend-Feeds
var firstIsin = article.MatchedAssets.FirstOrDefault()?.Isin;
if (!string.IsNullOrWhiteSpace(firstIsin))
{
@@ -94,14 +107,12 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
{
if (string.IsNullOrWhiteSpace(topic)) return;
// 1. System Config Updates
if (topic.StartsWith("services/config/updated", StringComparison.OrdinalIgnoreCase))
{
if (topic.EndsWith("FinlyticNews", StringComparison.OrdinalIgnoreCase))
{
await OnConfigUpdatedAsync(payload);
}
return;
}
@@ -111,7 +122,6 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
var channel = segments[2];
var correlationId = segments[^1];
// 2. Unified RPC Dispatching via Switch
switch (channel)
{
case "news_Get":
@@ -130,6 +140,14 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
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;
@@ -142,7 +160,10 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
private async Task OnGetNewsAsync(string payload, string correlationId)
{
_logger.LogInformation("Received RPC news_Get request. Correlation: {CorrelationId}", 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);
int limit = 20;
int offset = 0;
@@ -155,9 +176,7 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
{
try
{
// Direktes Deserialisieren über das DailyNewsRequest-DTO
var req = JsonSerializer.Deserialize(payload, FinlyticJsonSerializerContext.Default.DailyNewsRequest);
if (req != null)
{
limit = req.Limit > 0 ? req.Limit : 20;
@@ -167,7 +186,6 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
searchQuery = req.Query;
date = req.Date;
// Status anpassen, falls HasSentiment gesetzt ist
if (req.HasSentiment == true && string.IsNullOrEmpty(status))
{
status = "Analyzed";
@@ -176,44 +194,35 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Failed to parse DailyNewsRequest payload on news_Get");
await finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, ex, "Failed to parse DailyNewsRequest payload on news_Get");
}
}
try
{
using var scope = _scopeFactory.CreateScope();
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}";
_logger.LogInformation("Publishing RPC response to {ResponseTopic} with {Count} articles.", responseTopic,
dtos.Count);
await finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "Publishing RPC response to {ResponseTopic} with {Count} articles.", responseTopic, dtos.Count);
await PublishAsync(responseTopic, dtos);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to compile RPC response for news_Get");
// Antworte mit leerer Liste, um RPC-Timeouts im Gateway zu vermeiden
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
{
}
catch { }
}
}
private async Task OnGetNewsByIdAsync(string payload, string correlationId)
{
_logger.LogInformation("Received RPC news_GetById request. Correlation: {CorrelationId}", correlationId);
string responseTopic = $"services/response/news_GetById/{correlationId}";
if (string.IsNullOrWhiteSpace(payload))
{
await PublishAsync(responseTopic, (object?)null);
@@ -222,9 +231,7 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
try
{
var request =
JsonSerializer.Deserialize<ArticleRequest>(payload,
FinlyticJsonSerializerContext.Default.ArticleRequest);
var request = JsonSerializer.Deserialize<ArticleRequest>(payload, FinlyticJsonSerializerContext.Default.ArticleRequest);
var targetIdStr = request?.ArticleId ?? request?.Id;
if (Guid.TryParse(targetIdStr, out var articleId))
@@ -241,33 +248,25 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
}
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to execute RPC news_GetById");
}
catch { }
await PublishAsync(responseTopic, (object?)null);
}
private async Task OnGetPendingNewsAsync(string payload, string correlationId)
{
_logger.LogInformation("Received RPC news_GetPending request. Correlation: {CorrelationId}", 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))
if (doc.RootElement.TryGetProperty("limit", out var limitProp) && limitProp.TryGetInt32(out var parsedLimit))
{
limit = Math.Min(parsedLimit, 10);
}
}
catch
{
}
catch { }
}
try
@@ -279,25 +278,17 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
var dtos = (await Task.WhenAll(pendingArticles.Take(limit).Select(a => MapToDtoAsync(a)))).ToList();
string responseTopic = $"services/response/news_GetPending/{correlationId}";
_logger.LogInformation("Publishing RPC news_GetPending response to {ResponseTopic} with {Count} articles.",
responseTopic, dtos.Count);
await PublishAsync(responseTopic, dtos);
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to publish RPC pending news response");
}
catch { }
}
private async Task OnUpdateNewsStatusAsync(string payload, string correlationId)
{
_logger.LogInformation("Received RPC news_UpdateStatus request. Correlation: {CorrelationId}", correlationId);
UpdateNewsStatusResponse response;
try
{
var request = JsonSerializer.Deserialize<UpdateNewsStatusRequest>(payload,
FinlyticJsonSerializerContext.Default.UpdateNewsStatusRequest);
var request = JsonSerializer.Deserialize<UpdateNewsStatusRequest>(payload, FinlyticJsonSerializerContext.Default.UpdateNewsStatusRequest);
if (request != null && request.Id != Guid.Empty)
{
using var scope = _scopeFactory.CreateScope();
@@ -313,7 +304,6 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
}
catch (Exception ex)
{
_logger.LogError(ex, "Failed to execute RPC news_UpdateStatus");
response = new UpdateNewsStatusResponse(false, ex.Message);
}
@@ -321,42 +311,99 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
await PublishAsync(responseTopic, response);
}
private async Task OnSettingsGetAllAsync(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.");
}
}
private async Task OnSettingsUpdateAsync(string payload, 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
{
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.");
}
}
private async Task OnConfigUpdatedAsync(string payload)
{
_logger.LogInformation("Received config update event for FinlyticNews.");
try
{
using var doc = JsonDocument.Parse(payload);
if (doc.RootElement.TryGetProperty("settings", out var settingsProp))
{
var dict = JsonSerializer.Deserialize<Dictionary<string, string>>(settingsProp.GetRawText());
var dict = JsonSerializer.Deserialize<Dictionary<string, object?>>(settingsProp.GetRawText());
if (dict != null && dict.Count > 0)
{
using var scope = _scopeFactory.CreateScope();
var settingsDb = scope.ServiceProvider.GetRequiredService<ISettingsDbService>();
await settingsDb.UpdateSettingsFromDictionaryAsync(dict);
_logger.LogInformation("Successfully persisted {Count} updated settings.", dict.Count);
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
await settings.UpdateSettingsAsync(dict);
}
}
}
catch (Exception ex)
{
_logger.LogError(ex, "Error processing MQTT config update event");
}
catch { }
}
private async Task OnHealthPingAsync(string[] segments, string correlationId)
{
// Zerlegt den Topic-Pfad z. B. services/request/health_Ping/FinlyticNews/{correlationId}
bool isForMe = segments.Length >= 5 && segments[3].Equals("FinlyticNews", StringComparison.OrdinalIgnoreCase);
if (isForMe)
{
string respTopic = $"services/response/health_Ping/{correlationId}";
await PublishAsync(respTopic,
new ServiceHealthResponse("FinlyticNews", "Online", DateTime.UtcNow, "Connected"));
_logger.LogInformation("Responded to live health_Ping RPC request [CorrelationId: {CorrelationId}].",
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);
}
}
@@ -375,41 +422,31 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
var targetId = a.Id.ToString();
IsinAnalysisEntry? sentimentEntry = null;
// 1. Snappy Local Disk Check for Article File
var articlePath = Path.Combine(Directory.GetCurrentDirectory(), "data", "summaries", "articles",
$"{targetId}.json");
var articlePath = Path.Combine(Directory.GetCurrentDirectory(), "data", "summaries", "articles", $"{targetId}.json");
if (File.Exists(articlePath))
{
try
{
var json = await File.ReadAllTextAsync(articlePath);
sentimentEntry = JsonSerializer.Deserialize<IsinAnalysisEntry>(json,
FinlyticJsonSerializerContext.Default.IsinAnalysisEntry);
}
catch
{
sentimentEntry = JsonSerializer.Deserialize<IsinAnalysisEntry>(json, FinlyticJsonSerializerContext.Default.IsinAnalysisEntry);
}
catch { }
}
// 2. ISIN Summary File Fallback
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");
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));
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)
{
@@ -417,9 +454,7 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
break;
}
}
catch
{
}
catch { }
}
}
}
@@ -432,10 +467,7 @@ public class NewsMqttClient : ManagedMqttClient, IHostedService
confidence = finbertResult.Confidence;
}
}
catch (Exception ex)
{
_logger.LogTrace(ex, "[MapToDtoAsync] Sentiment fetch skipped for article {Id}", a.Id);
}
catch { }
return new NewsArticleDto
{