334 lines
16 KiB
C#
334 lines
16 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Linq;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using FinlyticCore.Dtos.News;
|
|
using FinlyticCore.Services;
|
|
using FinlyticNews.Entities;
|
|
using FinlyticNews.Util;
|
|
using Microsoft.Extensions.DependencyInjection;
|
|
using Microsoft.Extensions.Hosting;
|
|
|
|
namespace FinlyticNews.Services;
|
|
|
|
/// <summary>
|
|
/// Background worker service orchestrating link discovery, Playwright scraping,
|
|
/// non-AI deduplication, in-memory asset matching, plausibility validation, and MQTT broadcasts.
|
|
/// </summary>
|
|
public class NewsScraperBackgroundService : BackgroundService
|
|
{
|
|
private readonly IServiceScopeFactory _scopeFactory;
|
|
private readonly IFinlyticLogger<NewsScraperBackgroundService> _finlyticLogger;
|
|
private readonly NewsMqttClient _mqttClient;
|
|
private readonly INewsBlocklistService _blocklistService;
|
|
private readonly IArticleDeduplicationService _deduplicationService;
|
|
private readonly IAssetMatcherService _assetMatcherService;
|
|
private readonly IAssetValidationService _assetValidationService;
|
|
|
|
public NewsScraperBackgroundService(
|
|
IServiceScopeFactory scopeFactory,
|
|
IFinlyticLogger<NewsScraperBackgroundService> finlyticLogger,
|
|
NewsMqttClient mqttClient,
|
|
INewsBlocklistService blocklistService,
|
|
IArticleDeduplicationService deduplicationService,
|
|
IAssetMatcherService assetMatcherService,
|
|
IAssetValidationService assetValidationService)
|
|
{
|
|
_scopeFactory = scopeFactory;
|
|
_finlyticLogger = finlyticLogger;
|
|
_mqttClient = mqttClient;
|
|
_blocklistService = blocklistService;
|
|
_deduplicationService = deduplicationService;
|
|
_assetMatcherService = assetMatcherService;
|
|
_assetValidationService = assetValidationService;
|
|
}
|
|
|
|
protected override async Task ExecuteAsync(CancellationToken stoppingToken)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] NewsScraperBackgroundService started.");
|
|
|
|
// Initialize in-memory blocklist and deduplication cache on startup
|
|
try
|
|
{
|
|
await _blocklistService.InitializeAsync(stoppingToken);
|
|
await _deduplicationService.InitializeAsync(stoppingToken);
|
|
await _assetMatcherService.ReloadIndexAsync(stoppingToken);
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] Error during startup service initialization.");
|
|
}
|
|
|
|
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] Unhandled exception in 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(Math.Max(1, 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 settings = scope.ServiceProvider.GetRequiredService<ISettingsService>();
|
|
|
|
var maxArticlesPerFeed = await settings.GetSettingAsync(SettingKeys.MaxArticlesPerFeed, stoppingToken);
|
|
|
|
// Process any previously interrupted articles
|
|
var pendingScraping = await dbService.GetArticlesByStatusAsync("Scraping");
|
|
if (pendingScraping.Count > 0)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Retrying {Count} pending scraping articles...", pendingScraping.Count);
|
|
foreach (var article in pendingScraping)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) return;
|
|
await ProcessSingleArticleAsync(article, scraperService, dbService, stoppingToken);
|
|
}
|
|
}
|
|
|
|
var sources = await dbService.GetSourcesAsync();
|
|
if (sources.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogWarningAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] No article sources configured in database.");
|
|
return;
|
|
}
|
|
|
|
foreach (var source in sources)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) break;
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Discovering articles from source: {Name} ({Url})", source.Name, source.Source);
|
|
var discoveredArticles = await discoveryService.DiscoverLinksAsync(source.Source, source.Type, stoppingToken);
|
|
|
|
if (discoveredArticles == null || discoveredArticles.Count == 0)
|
|
{
|
|
continue;
|
|
}
|
|
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Discovered {Count} potential article links from {Name}.", discoveredArticles.Count, source.Name);
|
|
var toProcess = discoveredArticles.Take(maxArticlesPerFeed > 0 ? maxArticlesPerFeed : 20);
|
|
|
|
foreach (var discovered in toProcess)
|
|
{
|
|
if (stoppingToken.IsCancellationRequested) break;
|
|
|
|
// 1. Fast Blocklist & Duplicate check before creating DB entry
|
|
if (_blocklistService.IsBlocked(discovered.Url))
|
|
{
|
|
await _finlyticLogger.LogDebugAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Skipping blocked article URL: {Url}", discovered.Url);
|
|
continue;
|
|
}
|
|
|
|
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 create pending article for {Url}", discovered.Url);
|
|
continue;
|
|
}
|
|
|
|
if (article == null || article.Id == Guid.Empty) continue;
|
|
|
|
await ProcessSingleArticleAsync(article, scraperService, dbService, stoppingToken);
|
|
}
|
|
}
|
|
}
|
|
|
|
private async Task ProcessSingleArticleAsync(
|
|
NewsArticleEntity article,
|
|
IPlaywrightScraperService scraperService,
|
|
INewsDbService dbService,
|
|
CancellationToken stoppingToken)
|
|
{
|
|
try
|
|
{
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Processing");
|
|
|
|
// 1. Playwright Headless Scrape & Redirect Resolution
|
|
var (resolvedUrl, scrapeResult) = await scraperService.ScrapeArticleAsync(article.SourceUrl);
|
|
|
|
// 2. Redirect Handling
|
|
if (!string.IsNullOrWhiteSpace(resolvedUrl) && !resolvedUrl.Equals(article.SourceUrl, StringComparison.OrdinalIgnoreCase))
|
|
{
|
|
if (_blocklistService.IsBlocked(resolvedUrl) || await dbService.IsUrlDuplicateAsync(resolvedUrl))
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Redirected URL {Url} is already blocked/duplicate.", resolvedUrl);
|
|
await _blocklistService.BlockUrlAsync(article.SourceUrl, "RedirectToDuplicate", stoppingToken);
|
|
await dbService.DeleteArticleAsync(article.Id);
|
|
return;
|
|
}
|
|
|
|
await dbService.UpdateArticleUrlAsync(article.Id, resolvedUrl);
|
|
article.SourceUrl = resolvedUrl;
|
|
}
|
|
|
|
// 3. Content Validity Check
|
|
var title = !string.IsNullOrWhiteSpace(scrapeResult.Title) ? scrapeResult.Title : article.Title;
|
|
var content = scrapeResult.TextContent;
|
|
|
|
if (string.IsNullOrWhiteSpace(content) || content.Length < 80 || string.IsNullOrWhiteSpace(title))
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Article {Id} has empty or insufficient content ({Length} chars). Adding to blocklist.", article.Id, content?.Length ?? 0);
|
|
await _blocklistService.BlockUrlAsync(article.SourceUrl, "EmptyOrInvalidContent", stoppingToken);
|
|
await dbService.DeleteArticleAsync(article.Id);
|
|
return;
|
|
}
|
|
|
|
var publishedAt = scrapeResult.PublishedAt ?? article.PublishedAt;
|
|
|
|
// 4. Non-AI Content & Title Deduplication Check (SimHash + N-Gram)
|
|
var dupCheck = _deduplicationService.CheckDuplicate(title, content, publishedAt);
|
|
if (dupCheck.IsDuplicate)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.DeduplicationChannel, "[NewsScraperBackgroundService] Article {Id} identified as duplicate of {OriginalId} (Reason: {Reason}). Blocking URL.", article.Id, dupCheck.DuplicateOfArticleId?.ToString() ?? "Unknown", dupCheck.Reason ?? "Duplicate content");
|
|
await _blocklistService.BlockUrlAsync(article.SourceUrl, $"Duplicate:{dupCheck.DuplicateOfArticleId}", stoppingToken);
|
|
await dbService.DeleteArticleAsync(article.Id);
|
|
return;
|
|
}
|
|
|
|
// 5. In-Memory Asset Matching (ISIN Regex, Name Tokenizer, Suffix Trimming)
|
|
var candidateMatches = _assetMatcherService.MatchAssets(title, content, article.SourceUrl);
|
|
if (candidateMatches.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.MatcherChannel, "[NewsScraperBackgroundService] No candidate assets matched for article {Id} ('{Title}'). Blocking URL.", article.Id, title);
|
|
await _blocklistService.BlockUrlAsync(article.SourceUrl, "NoMatchedAssets", stoppingToken);
|
|
await dbService.DeleteArticleAsync(article.Id);
|
|
return;
|
|
}
|
|
|
|
// 6. Multi-Criteria Asset Validation (Sector Clustering, Title Weighting, Existence)
|
|
var validatedMatches = await _assetValidationService.ValidateCandidateAssetsAsync(candidateMatches, title, content, stoppingToken);
|
|
if (validatedMatches.Count == 0)
|
|
{
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.MatcherChannel, "[NewsScraperBackgroundService] All candidate matches pruned during validation for article {Id}. Blocking URL.", article.Id);
|
|
await _blocklistService.BlockUrlAsync(article.SourceUrl, "NoValidatedAssets", stoppingToken);
|
|
await dbService.DeleteArticleAsync(article.Id);
|
|
return;
|
|
}
|
|
|
|
// 7. Prepare Matched Asset Entities
|
|
var matchedEntities = validatedMatches.Select(m => new MatchedAssetEntity
|
|
{
|
|
Id = Guid.NewGuid(),
|
|
NewsArticleId = article.Id,
|
|
Isin = m.Isin,
|
|
Name = m.Name
|
|
}).ToList();
|
|
|
|
// 8. Generate Clean Summary & Excerpt
|
|
var summary = scrapeResult.Excerpt;
|
|
if (string.IsNullOrWhiteSpace(summary))
|
|
{
|
|
summary = content.Length > 280 ? content[..280] + "..." : content;
|
|
}
|
|
|
|
// 9. Persist Completed Article
|
|
var updatedArticle = await dbService.SaveProcessedArticleAsync(
|
|
id: article.Id,
|
|
title: title,
|
|
author: scrapeResult.Author,
|
|
summary: summary,
|
|
contentRaw: content,
|
|
language: scrapeResult.Language ?? article.Language ?? "de",
|
|
publishedAt: publishedAt,
|
|
titleHash: dupCheck.TitleHash,
|
|
simHash: dupCheck.SimHash,
|
|
matchedAssets: matchedEntities
|
|
);
|
|
|
|
// 10. Register in deduplication memory cache
|
|
_deduplicationService.RegisterArticle(article.Id, dupCheck.TitleHash, dupCheck.SimHash, title, publishedAt);
|
|
|
|
// 11. Broadcast via MQTT
|
|
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);
|
|
await _finlyticLogger.LogInfoAsync(SettingKeys.NewsChannel, "[NewsScraperBackgroundService] Successfully processed and broadcasted article {Id} ('{Title}') with {Count} matched assets.", article.Id, title, matchedEntities.Count);
|
|
}
|
|
}
|
|
catch (Exception ex)
|
|
{
|
|
await _finlyticLogger.LogErrorAsync(SettingKeys.NewsChannel, ex, "[NewsScraperBackgroundService] Exception processing article {Url}. Flagging for next cycle retry.", article.SourceUrl);
|
|
await dbService.UpdateArticleStatusAsync(article.Id, "Scraping");
|
|
}
|
|
}
|
|
} |