427 lines
19 KiB
C#
427 lines
19 KiB
C#
using System;
|
|
using System.Collections.Concurrent;
|
|
using System.Collections.Generic;
|
|
using System.IO;
|
|
using System.Linq;
|
|
using System.Text.Json;
|
|
using System.Text.RegularExpressions;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticAssets.Models;
|
|
using FinlyticAssets.Util;
|
|
using FinlyticCore.Dtos.News;
|
|
using FinlyticCore.Services;
|
|
using FinlyticCore.Util;
|
|
using FinlyticNews.Entities;
|
|
using FinlyticNews.Util;
|
|
using Microsoft.Extensions.Configuration;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
|
|
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
|
|
{
|
|
private record CompiledAssetMatcher(
|
|
AssetIndex Asset,
|
|
string CoreName,
|
|
Regex? WordRegex,
|
|
Regex? CoreWordRegex
|
|
);
|
|
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly IFinlyticLogger<NewsScraperBackgroundService> _finlyticLogger;
|
|
private readonly NewsMqttClient _mqttClient;
|
|
private readonly string _indexPath;
|
|
|
|
private List<CompiledAssetMatcher>? _cachedAssetMatchers;
|
|
private DateTime _lastIndexLoadTime = DateTime.MinValue;
|
|
|
|
public NewsScraperBackgroundService(
|
|
IServiceScopeFactory scopeFactory,
|
|
IFinlyticLogger<NewsScraperBackgroundService> finlyticLogger,
|
|
NewsMqttClient mqttClient,
|
|
IConfiguration configuration)
|
|
{
|
|
_scopeFactory = scopeFactory;
|
|
_finlyticLogger = finlyticLogger;
|
|
_mqttClient = mqttClient;
|
|
_indexPath = Path.Combine(Volumes.IndexRelativePath, "index.json");
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] NewsScraperBackgroundService started.");
|
|
|
|
while (!stoppingToken.IsCancellationRequested)
|
|
{
|
|
try
|
|
{
|
|
await using var scope = _scopeFactory.CreateAsyncScope();
|
|
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
bool enabled = await settings.GetSettingAsync(SettingKeys.EnableAutoScraping, stoppingToken);
|
|
|
|
if (enabled)
|
|
{
|
|
await RunScrapingCycleAsync(stoppingToken);
|
|
}
|
|
else
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Auto-scraping is disabled via settings.");
|
|
}
|
|
}
|
|
catch (Exception ex) when (ex is not OperationCanceledException)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] An unhandled exception occurred during news scraping cycle.");
|
|
}
|
|
|
|
int intervalMinutes = 15;
|
|
try
|
|
{
|
|
await using var scope = _scopeFactory.CreateAsyncScope();
|
|
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
intervalMinutes = await settings.GetSettingAsync(SettingKeys.ScrapeIntervalMinutes, stoppingToken);
|
|
}
|
|
catch { }
|
|
|
|
var jitterSeconds = Random.Shared.Next(0, 60);
|
|
var nextRunDelay = TimeSpan.FromMinutes(intervalMinutes) + TimeSpan.FromSeconds(jitterSeconds);
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Scraping cycle completed. Next cycle in {Delay} (interval: {Minutes}m).", nextRunDelay, intervalMinutes);
|
|
|
|
try
|
|
{
|
|
await Task.Delay(nextRunDelay, stoppingToken);
|
|
}
|
|
catch (OperationCanceledException)
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] NewsScraperBackgroundService stopping.");
|
|
}
|
|
|
|
private async Task RunScrapingCycleAsync(CancellationToken stoppingToken)
|
|
{
|
|
await using var scope = _scopeFactory.CreateAsyncScope();
|
|
var dbService = scope.ServiceProvider.GetRequiredService<INewsDbService>();
|
|
var discoveryService = scope.ServiceProvider.GetRequiredService<IArticleDiscoveryService>();
|
|
var scraperService = scope.ServiceProvider.GetRequiredService<IPlaywrightScraperService>();
|
|
var n8nService = scope.ServiceProvider.GetRequiredService<IN8nService>();
|
|
var settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
var maxArticlesPerFeed = await settings.GetSettingAsync(SettingKeys.MaxArticlesPerFeed, stoppingToken);
|
|
|
|
var assetMatchers = await GetOrLoadAssetMatchersAsync();
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Loaded {Count} asset index items for text pre-filtering.", assetMatchers.Count);
|
|
|
|
var failedArticles = await dbService.GetArticlesByStatusAsync("Scraping");
|
|
if (failedArticles.Count > 0)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Found {Count} articles in status 'Scraping' that failed to scrape previously. Retrying...", failedArticles.Count);
|
|
foreach (var article in failedArticles)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) return;
|
|
await ProcessSingleArticleAsync(article, scraperService, n8nService, dbService, assetMatchers, stoppingToken);
|
|
}
|
|
}
|
|
|
|
var sources = await dbService.GetSourcesAsync();
|
|
if (sources.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] No article sources configured in database. Skipping cycle.");
|
|
return;
|
|
}
|
|
|
|
foreach (var source in sources)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) break;
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Starting article link discovery for source: {SourceName} ({Url})", source.Name, source.Source);
|
|
var discoveredArticles = await discoveryService.DiscoverLinksAsync(source.Source, source.Type, stoppingToken);
|
|
|
|
if (discoveredArticles == null || discoveredArticles.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogDebugAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] No links discovered from source: {SourceName}", source.Name);
|
|
continue;
|
|
}
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Discovered {Count} potential article links from {SourceName}.", discoveredArticles.Count, source.Name);
|
|
|
|
var toProcess = discoveredArticles.Take(maxArticlesPerFeed > 0 ? maxArticlesPerFeed : 20);
|
|
|
|
foreach (var discovered in toProcess)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) break;
|
|
|
|
if (await dbService.IsUrlDuplicateAsync(discovered.Url))
|
|
{
|
|
await _finlyticLogger.LogDebugAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Skipping duplicate article URL: {Url}", discovered.Url);
|
|
continue;
|
|
}
|
|
|
|
NewsArticleEntity article;
|
|
try
|
|
{
|
|
article = await dbService.CreatePendingArticleAsync(
|
|
discovered.Url,
|
|
discovered.Isins,
|
|
discovered.Title,
|
|
discovered.Summary,
|
|
discovered.PublishedAt,
|
|
discovered.Language
|
|
);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] Failed to register initial pending state for URL: {Url}. Skipping.", discovered.Url);
|
|
continue;
|
|
}
|
|
|
|
if (article.Id == Guid.Empty)
|
|
{
|
|
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Created pending article has invalid/empty ID for URL: {Url}. Skipping.", discovered.Url);
|
|
continue;
|
|
}
|
|
|
|
await ProcessSingleArticleAsync(article, scraperService, n8nService, dbService, assetMatchers, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ProcessSingleArticleAsync(
|
|
NewsArticleEntity article,
|
|
IPlaywrightScraperService scraperService,
|
|
IN8nService n8nService,
|
|
INewsDbService dbService,
|
|
List<CompiledAssetMatcher> assetMatchers,
|
|
CancellationToken stoppingToken)
|
|
{
|
|
try
|
|
{
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Processing");
|
|
|
|
var (resolvedUrl, rawContent) = await scraperService.ScrapeArticleAsync(article.SourceUrl);
|
|
|
|
if (!string.IsNullOrWhiteSpace(resolvedUrl) &&
|
|
!resolvedUrl.Equals(article.SourceUrl, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Redirect detected. Initial: {OldUrl} -> Resolved: {NewUrl}", article.SourceUrl, resolvedUrl);
|
|
if (await dbService.IsUrlDuplicateAsync(resolvedUrl))
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Redirected URL {ResolvedUrl} is a duplicate. Terminating processing.", resolvedUrl);
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Failed");
|
|
return;
|
|
}
|
|
|
|
await dbService.UpdateArticleUrlAsync(article.Id, resolvedUrl);
|
|
article.SourceUrl = resolvedUrl;
|
|
}
|
|
|
|
if (string.IsNullOrWhiteSpace(rawContent) || rawContent.Length < 60)
|
|
{
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Failed");
|
|
return;
|
|
}
|
|
|
|
var discoveredIsins = article.MatchedAssets.Select(m => m.Isin).Where(i => !string.IsNullOrEmpty(i)).ToList();
|
|
var preFilteredAssets = PreFilterAssets(rawContent, article.Title, assetMatchers, discoveredIsins);
|
|
|
|
if (preFilteredAssets.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Pre-filtering: Article {Id} does not reference any known assets. Terminating pipeline.", article.Id);
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Failed");
|
|
return;
|
|
}
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Pre-filtering matched {Count} assets for article {Id}.", preFilteredAssets.Count, article.Id);
|
|
|
|
var n8nResponse = await n8nService.AnalyzeArticleAsync(rawContent, preFilteredAssets, stoppingToken);
|
|
if (n8nResponse == null)
|
|
{
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Failed");
|
|
return;
|
|
}
|
|
|
|
var matchedEntities = new List<MatchedAssetEntity>();
|
|
if (n8nResponse.MatchedAssets != null && n8nResponse.MatchedAssets.Count > 0)
|
|
{
|
|
foreach (var asset in n8nResponse.MatchedAssets)
|
|
{
|
|
var preMatch = preFilteredAssets.FirstOrDefault(p => p.Name.Equals(asset.Name, StringComparison.OrdinalIgnoreCase));
|
|
var isin = preMatch?.Isin ?? asset.Ticker ?? "";
|
|
if (string.IsNullOrWhiteSpace(isin)) continue;
|
|
|
|
matchedEntities.Add(new MatchedAssetEntity
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
NewsArticleId = article.Id,
|
|
Isin = isin.Trim().ToUpperInvariant(),
|
|
Name = !string.IsNullOrWhiteSpace(asset.Name) ? asset.Name.Trim() : isin.Trim().ToUpperInvariant()
|
|
});
|
|
}
|
|
}
|
|
|
|
if (matchedEntities.Count == 0)
|
|
{
|
|
foreach (var preMatch in preFilteredAssets)
|
|
{
|
|
matchedEntities.Add(new MatchedAssetEntity
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
NewsArticleId = article.Id,
|
|
Isin = preMatch.Isin,
|
|
Name = preMatch.Name
|
|
});
|
|
}
|
|
}
|
|
|
|
var updatedArticle = await dbService.SaveArticleClassificationAsync(article.Id, n8nResponse, matchedEntities);
|
|
|
|
if (updatedArticle != null)
|
|
{
|
|
var dto = new NewsArticleDto
|
|
{
|
|
Id = updatedArticle.Id,
|
|
Title = updatedArticle.Title,
|
|
Author = updatedArticle.Author,
|
|
Summary = updatedArticle.Summary,
|
|
ContentRaw = updatedArticle.ContentRaw,
|
|
Language = updatedArticle.Language,
|
|
SourceUrl = updatedArticle.SourceUrl,
|
|
ScrapedAt = updatedArticle.ScrapedAt,
|
|
PublishedAt = updatedArticle.PublishedAt,
|
|
Status = updatedArticle.Status,
|
|
MatchedAssets = updatedArticle.MatchedAssets.Select(m => new MatchedAssetDto
|
|
{
|
|
Name = m.Name,
|
|
Isin = m.Isin
|
|
}).ToList()
|
|
};
|
|
|
|
await _mqttClient.BroadcastArticleAsync(dto);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] Failed to complete processing pipeline for article: {Url}. Transitioning to 'Scraping' for next cycle retry.", article.SourceUrl);
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Scraping");
|
|
}
|
|
}
|
|
|
|
private List<FilteredAssetPayload> PreFilterAssets(
|
|
string content,
|
|
string? title,
|
|
List<CompiledAssetMatcher> assetMatchers,
|
|
List<string>? priorityIsins = null)
|
|
{
|
|
if (assetMatchers.Count == 0) return new List<FilteredAssetPayload>();
|
|
|
|
var fullText = (title != null ? title + " " + content : content);
|
|
var matched = new Dictionary<string, FilteredAssetPayload>(StringComparer.OrdinalIgnoreCase);
|
|
|
|
if (priorityIsins != null && priorityIsins.Count > 0)
|
|
{
|
|
foreach (var isin in priorityIsins)
|
|
{
|
|
var match = assetMatchers.FirstOrDefault(m => string.Equals(m.Asset.Isin, isin, StringComparison.OrdinalIgnoreCase));
|
|
if (match != null && !matched.ContainsKey(match.Asset.Isin))
|
|
{
|
|
matched[match.Asset.Isin] = new FilteredAssetPayload(match.Asset.Name, match.Asset.Isin);
|
|
}
|
|
}
|
|
}
|
|
|
|
foreach (var m in assetMatchers)
|
|
{
|
|
if (matched.ContainsKey(m.Asset.Isin)) continue;
|
|
|
|
if (fullText.Contains(m.Asset.Isin, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
matched[m.Asset.Isin] = new FilteredAssetPayload(m.Asset.Name, m.Asset.Isin);
|
|
continue;
|
|
}
|
|
|
|
if (m.WordRegex != null && m.WordRegex.IsMatch(fullText))
|
|
{
|
|
matched[m.Asset.Isin] = new FilteredAssetPayload(m.Asset.Name, m.Asset.Isin);
|
|
continue;
|
|
}
|
|
|
|
if (m.CoreWordRegex != null && m.CoreWordRegex.IsMatch(fullText))
|
|
{
|
|
matched[m.Asset.Isin] = new FilteredAssetPayload(m.Asset.Name, m.Asset.Isin);
|
|
}
|
|
}
|
|
|
|
return matched.Values.ToList();
|
|
}
|
|
|
|
private async Task<List<CompiledAssetMatcher>> GetOrLoadAssetMatchersAsync()
|
|
{
|
|
if (_cachedAssetMatchers != null && (DateTime.UtcNow - _lastIndexLoadTime).TotalMinutes < 60)
|
|
{
|
|
return _cachedAssetMatchers;
|
|
}
|
|
|
|
if (!File.Exists(_indexPath))
|
|
{
|
|
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Asset index file not found at: {Path}. Pre-filtering will match 0 assets.", _indexPath);
|
|
return new List<CompiledAssetMatcher>();
|
|
}
|
|
|
|
try
|
|
{
|
|
using var stream = File.OpenRead(_indexPath);
|
|
var indexList = await JsonSerializer.DeserializeAsync<List<AssetIndex>>(stream);
|
|
|
|
if (indexList == null || indexList.Count == 0)
|
|
{
|
|
return new List<CompiledAssetMatcher>();
|
|
}
|
|
|
|
var compiled = new List<CompiledAssetMatcher>(indexList.Count);
|
|
foreach (var asset in indexList)
|
|
{
|
|
if (string.IsNullOrWhiteSpace(asset.Name) || string.IsNullOrWhiteSpace(asset.Isin))
|
|
continue;
|
|
|
|
var rawName = asset.Name.Trim();
|
|
var coreName = ExtractCoreName(rawName);
|
|
|
|
Regex? wordRegex = null;
|
|
if (rawName.Length >= 4)
|
|
{
|
|
wordRegex = new Regex($@"\b{Regex.Escape(rawName)}\b", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant | RegexOptions.Compiled);
|
|
}
|
|
|
|
Regex? coreWordRegex = null;
|
|
if (!string.IsNullOrWhiteSpace(coreName) && coreName.Length >= 4 && !coreName.Equals(rawName, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
coreWordRegex = new Regex($@"\b{Regex.Escape(coreName)}\b", RegexOptions.IgnoreCase | RegexOptions.CultureInvariant | RegexOptions.Compiled);
|
|
}
|
|
|
|
compiled.Add(new CompiledAssetMatcher(asset, coreName, wordRegex, coreWordRegex));
|
|
}
|
|
|
|
_cachedAssetMatchers = compiled;
|
|
_lastIndexLoadTime = DateTime.UtcNow;
|
|
return _cachedAssetMatchers;
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] Failed to read or parse asset index file from {Path}.", _indexPath);
|
|
return _cachedAssetMatchers ?? new List<CompiledAssetMatcher>();
|
|
}
|
|
}
|
|
|
|
private static string ExtractCoreName(string rawName)
|
|
{
|
|
var cleaned = Regex.Replace(rawName, @"\b(AG|SE|SA|NV|PLC|INC|CORP|LLC|GMBH|CO|KG|HOLDING|GROUP|CLASS\s+[A-Z])\b", "", RegexOptions.IgnoreCase);
|
|
return cleaned.Trim(' ', '.', ',', '-');
|
|
}
|
|
} |