feat(News): refactor news service and add sentiment tracking
This commit is contained in:
@@ -0,0 +1,438 @@
|
||||
using System.Collections.Concurrent;
|
||||
using System.IO;
|
||||
using System.Text.Json;
|
||||
using System.Text.Json.Serialization;
|
||||
using System.Text.RegularExpressions;
|
||||
using FinlyticAssets.Models;
|
||||
using FinlyticAssets.Util;
|
||||
using FinlyticCore.Dtos.News;
|
||||
using FinlyticCore.Util;
|
||||
using FinlyticNews.Entities;
|
||||
using FinlyticNews.Util;
|
||||
using Microsoft.Extensions.Configuration;
|
||||
using Microsoft.Extensions.DependencyInjection;
|
||||
using Microsoft.Extensions.Hosting;
|
||||
using Microsoft.Extensions.Logging;
|
||||
|
||||
namespace FinlyticNews.Services;
|
||||
|
||||
/// <summary>
|
||||
/// A background worker service that orchestrates link discovery, Playwright scraping,
|
||||
/// pre-filtering, n8n AI enrichment, database persistence, and MQTT broadcasts.
|
||||
/// </summary>
|
||||
public class NewsScraperBackgroundService : BackgroundService
|
||||
{
|
||||
/// <summary>
|
||||
/// Internal wrapper to associate compiled regex patterns with the unmodified AssetIndex record.
|
||||
/// </summary>
|
||||
private record CompiledAssetMatcher(
|
||||
AssetIndex Asset,
|
||||
string CoreName,
|
||||
Regex? WordRegex,
|
||||
Regex? CoreWordRegex
|
||||
);
|
||||
|
||||
private readonly IServiceScopeFactory _scopeFactory;
|
||||
private readonly ILogger<NewsScraperBackgroundService> _logger;
|
||||
private readonly NewsMqttClient _mqttClient;
|
||||
private readonly int _intervalMinutes;
|
||||
private readonly string _indexPath;
|
||||
|
||||
// In-Memory Cache for compiled asset matchers to prevent re-reading & re-compiling Regex
|
||||
private List<CompiledAssetMatcher>? _cachedAssetMatchers;
|
||||
private DateTime _lastIndexLoadTime = DateTime.MinValue;
|
||||
|
||||
public NewsScraperBackgroundService(
|
||||
IServiceScopeFactory scopeFactory,
|
||||
ILogger<NewsScraperBackgroundService> logger,
|
||||
NewsMqttClient mqttClient,
|
||||
IConfiguration configuration)
|
||||
{
|
||||
_scopeFactory = scopeFactory;
|
||||
_logger = logger;
|
||||
_mqttClient = mqttClient;
|
||||
|
||||
_intervalMinutes = configuration.GetValue<int>("ScrapingSettings:IntervalMinutes", 15);
|
||||
_indexPath = Path.Combine(Volumes.IndexRelativePath, "index.json");
|
||||
}
|
||||
|
||||
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] NewsScraperBackgroundService started. Interval: {Minutes} minutes.", "NewsChannel", _intervalMinutes);
|
||||
|
||||
while (!stoppingToken.IsCancellationRequested)
|
||||
{
|
||||
try
|
||||
{
|
||||
await RunScrapingCycleAsync(stoppingToken);
|
||||
}
|
||||
catch (Exception ex) when (ex is not OperationCanceledException)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] An unhandled exception occurred during news scraping cycle.", "NewsChannel");
|
||||
}
|
||||
|
||||
int intervalMinutes = _intervalMinutes;
|
||||
try
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var settingsDb = scope.ServiceProvider.GetService<ISettingsDbService>();
|
||||
if (settingsDb != null)
|
||||
{
|
||||
var settings = await settingsDb.GetSettingsAsync();
|
||||
if (settings?.ScrapingIntervalMinutes > 0)
|
||||
{
|
||||
intervalMinutes = settings.ScrapingIntervalMinutes;
|
||||
}
|
||||
}
|
||||
}
|
||||
catch { /* Ignore settings DB lookup failures */ }
|
||||
|
||||
var jitterSeconds = Random.Shared.Next(0, 300);
|
||||
var nextRunDelay = TimeSpan.FromMinutes(intervalMinutes) + TimeSpan.FromSeconds(jitterSeconds);
|
||||
_logger.LogInformation("[{Channel}] Scraping cycle completed. Next cycle in {Delay} (interval: {Minutes}m).", "NewsChannel", nextRunDelay, intervalMinutes);
|
||||
|
||||
try
|
||||
{
|
||||
await Task.Delay(nextRunDelay, stoppingToken);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] NewsScraperBackgroundService stopping.", "NewsChannel");
|
||||
}
|
||||
|
||||
private async Task RunScrapingCycleAsync(CancellationToken stoppingToken)
|
||||
{
|
||||
using var scope = _scopeFactory.CreateScope();
|
||||
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
||||
var discoveryService = scope.ServiceProvider.GetRequiredService<IArticleDiscoveryService>();
|
||||
var scraperService = scope.ServiceProvider.GetRequiredService<IPlaywrightScraperService>();
|
||||
var n8nService = scope.ServiceProvider.GetRequiredService<IN8nService>();
|
||||
|
||||
// Load pre-compiled asset index matchers for zero-latency pre-filtering
|
||||
var assetMatchers = await GetOrLoadAssetMatchersAsync();
|
||||
_logger.LogInformation("[{Channel}] Loaded {Count} asset index items for text pre-filtering.", "NewsChannel", assetMatchers.Count);
|
||||
|
||||
// 1. Scraping Retry Phase: query articles in status "Scraping" (failed Playwright runs)
|
||||
var failedArticles = await dbService.GetArticlesByStatusAsync("Scraping");
|
||||
if (failedArticles.Count > 0)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Found {Count} articles in status 'Scraping' that failed to scrape previously. Retrying...", "NewsChannel", failedArticles.Count);
|
||||
foreach (var article in failedArticles)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
await ProcessSingleArticleAsync(article, dbService, scraperService, n8nService, assetMatchers, stoppingToken);
|
||||
}
|
||||
}
|
||||
|
||||
// 2. Link Discovery Phase: query RSS feeds and listing pages
|
||||
var sources = await dbService.GetSourcesAsync();
|
||||
if (sources.Count == 0)
|
||||
{
|
||||
_logger.LogWarning("[{Channel}] No article sources configured in database. Skipping cycle.", "NewsChannel");
|
||||
return;
|
||||
}
|
||||
|
||||
foreach (var source in sources)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
|
||||
_logger.LogInformation("[{Channel}] Starting article link discovery for source: {SourceName} ({Url})", "NewsChannel", source.Name, source.Source);
|
||||
var discoveredArticles = await discoveryService.DiscoverLinksAsync(source.Source, source.Type, stoppingToken);
|
||||
|
||||
if (discoveredArticles == null || discoveredArticles.Count == 0)
|
||||
{
|
||||
_logger.LogDebug("No links discovered from source: {SourceName}", source.Name);
|
||||
continue;
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] Discovered {Count} potential article links from {SourceName}.", "NewsChannel", discoveredArticles.Count, source.Name);
|
||||
|
||||
foreach (var discovered in discoveredArticles)
|
||||
{
|
||||
if (stoppingToken.IsCancellationRequested) break;
|
||||
|
||||
// Idempotency Check & Deduplication
|
||||
var isDuplicate = await dbService.IsUrlDuplicateAsync(discovered.Url);
|
||||
if (isDuplicate)
|
||||
{
|
||||
_logger.LogDebug("Skipping duplicate article URL: {Url}", discovered.Url);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Register Initial Lock State in the database ("Pending")
|
||||
NewsArticleEntity? article;
|
||||
try
|
||||
{
|
||||
article = await dbService.CreatePendingArticleAsync(
|
||||
discovered.Url,
|
||||
discovered.Isins,
|
||||
discovered.Title,
|
||||
discovered.Summary,
|
||||
discovered.PublishedAt,
|
||||
discovered.Language);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Failed to register initial pending state for URL: {Url}. Skipping.", "NewsChannel", discovered.Url);
|
||||
continue;
|
||||
}
|
||||
|
||||
if (article == null || article.Id == Guid.Empty)
|
||||
{
|
||||
_logger.LogWarning("[{Channel}] Created pending article has invalid/empty ID for URL: {Url}. Skipping.", "NewsChannel", discovered.Url);
|
||||
continue;
|
||||
}
|
||||
|
||||
// Process single article pipeline
|
||||
await ProcessSingleArticleAsync(article, dbService, scraperService, n8nService, assetMatchers, stoppingToken);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private async Task ProcessSingleArticleAsync(
|
||||
NewsArticleEntity article,
|
||||
INewsDbService dbService,
|
||||
IPlaywrightScraperService scraperService,
|
||||
IN8nService n8nService,
|
||||
List<CompiledAssetMatcher> assetMatchers,
|
||||
CancellationToken stoppingToken)
|
||||
{
|
||||
try
|
||||
{
|
||||
// 3. Extraction with Headless Browser (Transitions to "Processing")
|
||||
await dbService.UpdateArticleStatusAsync(article.Id, "Processing");
|
||||
var (resolvedUrl, rawText) = await scraperService.ScrapeArticleAsync(article.SourceUrl);
|
||||
|
||||
if (string.IsNullOrWhiteSpace(rawText))
|
||||
{
|
||||
throw new InvalidOperationException("Scraping returned empty text body content.");
|
||||
}
|
||||
|
||||
// Update resolved URL if redirect occurred
|
||||
if (!string.Equals(resolvedUrl, article.SourceUrl, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Redirect detected. Initial: {OldUrl} -> Resolved: {NewUrl}", "NewsChannel", article.SourceUrl, resolvedUrl);
|
||||
if (await dbService.IsUrlDuplicateAsync(resolvedUrl))
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Redirected URL {ResolvedUrl} is a duplicate. Terminating processing.", "NewsChannel", resolvedUrl);
|
||||
await dbService.UpdateArticleStatusAsync(article.Id, "Duplicate");
|
||||
return;
|
||||
}
|
||||
|
||||
await dbService.UpdateArticleUrlAsync(article.Id, resolvedUrl);
|
||||
article.SourceUrl = resolvedUrl;
|
||||
}
|
||||
|
||||
// 4. Pre-filtering Assets (Optimized with Pre-Compiled Regex Patterns)
|
||||
var title = article.Title ?? string.Empty;
|
||||
var summary = article.Summary ?? string.Empty;
|
||||
|
||||
var preFilteredAssets = assetMatchers.Where(matcher =>
|
||||
{
|
||||
var asset = matcher.Asset;
|
||||
|
||||
// ISIN direct match
|
||||
if (rawText.Contains(asset.Isin, StringComparison.OrdinalIgnoreCase) ||
|
||||
title.Contains(asset.Isin, StringComparison.OrdinalIgnoreCase) ||
|
||||
summary.Contains(asset.Isin, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
// Fast regex word boundary check on Full Name
|
||||
if (matcher.WordRegex != null && (matcher.WordRegex.IsMatch(rawText) || matcher.WordRegex.IsMatch(title)))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
// Fast regex word boundary check on Core Name
|
||||
if (matcher.CoreName.Length >= 3 && matcher.CoreWordRegex != null &&
|
||||
(matcher.CoreWordRegex.IsMatch(rawText) || matcher.CoreWordRegex.IsMatch(title)))
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
})
|
||||
.Select(m => new FilteredAssetPayload(m.Asset.Name, m.Asset.Isin))
|
||||
.ToList();
|
||||
|
||||
if (preFilteredAssets.Count == 0)
|
||||
{
|
||||
_logger.LogInformation("[{Channel}] Pre-filtering: Article {Id} does not reference any known assets. Terminating pipeline.", "NewsChannel", article.Id);
|
||||
await dbService.UpdateArticleStatusAsync(article.Id, "Failed");
|
||||
return;
|
||||
}
|
||||
|
||||
_logger.LogInformation("[{Channel}] Pre-filtering matched {Count} assets for article {Id}.", "NewsChannel", preFilteredAssets.Count, article.Id);
|
||||
|
||||
// 5. Send to n8n Webhook Pipeline
|
||||
var n8nResponse = await n8nService.AnalyzeArticleAsync(rawText, preFilteredAssets, stoppingToken);
|
||||
if (n8nResponse == null)
|
||||
{
|
||||
throw new InvalidOperationException("n8n AI webhook execution returned null or failed.");
|
||||
}
|
||||
|
||||
// 6. Map and Save Completed Classification
|
||||
var matchedEntities = new List<MatchedAssetEntity>();
|
||||
foreach (var n8nAsset in n8nResponse.MatchedAssets)
|
||||
{
|
||||
var n8nCoreName = ExtractCoreAssetName(n8nAsset.Name);
|
||||
|
||||
var matchedIsin = preFilteredAssets.FirstOrDefault(fa =>
|
||||
fa.Name.Equals(n8nAsset.Name, StringComparison.OrdinalIgnoreCase) ||
|
||||
n8nAsset.Name.Contains(fa.Name, StringComparison.OrdinalIgnoreCase) ||
|
||||
(n8nCoreName.Length >= 3 && ExtractCoreAssetName(fa.Name).Equals(n8nCoreName, StringComparison.OrdinalIgnoreCase)))?.Isin;
|
||||
|
||||
if (string.IsNullOrWhiteSpace(matchedIsin))
|
||||
{
|
||||
matchedIsin = assetMatchers.FirstOrDefault(m =>
|
||||
m.Asset.Name.Equals(n8nAsset.Name, StringComparison.OrdinalIgnoreCase) ||
|
||||
n8nAsset.Name.Contains(m.Asset.Name, StringComparison.OrdinalIgnoreCase) ||
|
||||
(n8nCoreName.Length >= 3 && m.CoreName.Equals(n8nCoreName, StringComparison.OrdinalIgnoreCase)))?.Asset.Isin;
|
||||
}
|
||||
|
||||
if (!string.IsNullOrWhiteSpace(matchedIsin))
|
||||
{
|
||||
matchedEntities.Add(new MatchedAssetEntity
|
||||
{
|
||||
NewsArticleId = article.Id,
|
||||
Name = n8nAsset.Name,
|
||||
Isin = matchedIsin
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
var completedArticle = await dbService.SaveArticleClassificationAsync(article.Id, n8nResponse, matchedEntities);
|
||||
|
||||
if (completedArticle != null)
|
||||
{
|
||||
// 7. MQTT Broadcast (Sends completed article to downstream services)
|
||||
var dto = new NewsArticleDto
|
||||
{
|
||||
Id = completedArticle.Id,
|
||||
Title = completedArticle.Title,
|
||||
Author = completedArticle.Author,
|
||||
Summary = completedArticle.Summary,
|
||||
ContentRaw = completedArticle.ContentRaw,
|
||||
Language = completedArticle.Language,
|
||||
SourceUrl = completedArticle.SourceUrl,
|
||||
ScrapedAt = completedArticle.ScrapedAt,
|
||||
PublishedAt = completedArticle.PublishedAt,
|
||||
MatchedAssets = completedArticle.MatchedAssets.Select(m => new MatchedAssetDto
|
||||
{
|
||||
Name = m.Name,
|
||||
Isin = m.Isin
|
||||
}).ToList(),
|
||||
Status = completedArticle.Status
|
||||
};
|
||||
|
||||
await _mqttClient.BroadcastArticleAsync(dto);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Failed to complete processing pipeline for article: {Url}. Transitioning to 'Scraping' for next cycle retry.", "NewsChannel", article.SourceUrl);
|
||||
try
|
||||
{
|
||||
await dbService.UpdateArticleStatusAsync(article.Id, "Scraping");
|
||||
}
|
||||
catch { /* Suppress database secondary errors */ }
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns cached compiled asset matchers or parses the index file from disk if stale/missing.
|
||||
/// </summary>
|
||||
private async Task<List<CompiledAssetMatcher>> GetOrLoadAssetMatchersAsync()
|
||||
{
|
||||
if (_cachedAssetMatchers != null && (DateTime.UtcNow - _lastIndexLoadTime).TotalMinutes < 30)
|
||||
{
|
||||
return _cachedAssetMatchers;
|
||||
}
|
||||
|
||||
if (!File.Exists(_indexPath))
|
||||
{
|
||||
_logger.LogWarning("[{Channel}] Asset index file not found at: {Path}. Pre-filtering will match 0 assets.", "NewsChannel", _indexPath);
|
||||
return [];
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
await using var stream = File.OpenRead(_indexPath);
|
||||
|
||||
// Standard Deserialization for AssetIndex list
|
||||
var rawList = await JsonSerializer.DeserializeAsync<List<AssetIndex>>(stream);
|
||||
|
||||
if (rawList != null)
|
||||
{
|
||||
_cachedAssetMatchers = rawList.Select(asset =>
|
||||
{
|
||||
var coreName = ExtractCoreAssetName(asset.Name);
|
||||
return new CompiledAssetMatcher(
|
||||
Asset: asset,
|
||||
CoreName: coreName,
|
||||
WordRegex: BuildWordRegex(asset.Name),
|
||||
CoreWordRegex: BuildWordRegex(coreName)
|
||||
);
|
||||
}).ToList();
|
||||
|
||||
_lastIndexLoadTime = DateTime.UtcNow;
|
||||
return _cachedAssetMatchers;
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.LogError(ex, "[{Channel}] Failed to read or parse asset index file from {Path}.", "NewsChannel", _indexPath);
|
||||
}
|
||||
|
||||
return _cachedAssetMatchers ?? [];
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Helper to pre-compile Word Boundary Regex for an asset name.
|
||||
/// </summary>
|
||||
private static Regex? BuildWordRegex(string name)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(name)) return null;
|
||||
try
|
||||
{
|
||||
return new Regex($@"\b{Regex.Escape(name)}\b", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant | RegexOptions.Compiled);
|
||||
}
|
||||
catch
|
||||
{
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Extracts the core name of an asset by removing parenthetical metadata and corporate suffixes.
|
||||
/// </summary>
|
||||
private static string ExtractCoreAssetName(string name)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(name)) return string.Empty;
|
||||
|
||||
int parenIndex = name.IndexOf('(');
|
||||
if (parenIndex >= 0)
|
||||
{
|
||||
name = name[..parenIndex];
|
||||
}
|
||||
|
||||
name = name.Trim();
|
||||
|
||||
var suffixes = new[] { "Inc.", "Inc", "AG", "SE", "Co.", "Co", "Corp.", "Corp", "Ltd.", "Ltd", "plc", "GmbH", "SA", "NV", "Group" };
|
||||
foreach (var suffix in suffixes)
|
||||
{
|
||||
if (name.EndsWith(" " + suffix, StringComparison.OrdinalIgnoreCase))
|
||||
{
|
||||
name = name[..^suffix.Length].Trim();
|
||||
}
|
||||
}
|
||||
|
||||
return name;
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user