IBKRTrader/Modules/CongressTrading/Workers/CongressHistoryImportWorker.cs

using IBKRTrader.Core.Database;
using IBKRTrader.Core.Logging;
using IBKRTrader.Modules.CongressTrading.Database;
using IBKRTrader.Modules.CongressTrading.Scraper;
using IBKRTrader.Core.Workers;

namespace IBKRTrader.Modules.CongressTrading.Workers;

/// <summary>
/// Einmaliger History-Import – übersteht beliebig viele Neustarts.
///
/// Fortschritts-Persistenz via core_settings:
///   ct.history_import_done   = "true"  → abgeschlossen, Worker skippt sich selbst
///   ct.history_import_page   = "N"     → nächste zu verarbeitende Seite
///   ct.history_import_total  = "N"     → Gesamtseitenanzahl (gecacht)
///
/// Alle Trade-Daten (Datum, Typ, Value, Ticker, Mitglied) kommen direkt
/// aus dem Listen-JSON – kein separates Detail-Scraping notwendig.
/// </summary>
public class CongressHistoryImportWorker : WorkerBase
{
    private readonly CapitolTradesScraper _scraper;
    private readonly CongressRepository   _repo;

    private const string DoneFlag    = "ct.history_import_done";
    private const string ProgressKey = "ct.history_import_page";
    private const string TotalKey    = "ct.history_import_total";

    public override string     Name   => "CT-HistoryImport";
    public override string     Module => "CT";
    public override WorkerType Type   => WorkerType.Worker;

    protected override TimeSpan? Interval => TimeSpan.FromDays(365);

    public CongressHistoryImportWorker(
        LoggingService       logger,
        DatabaseService      db,
        CapitolTradesScraper scraper,
        CongressRepository   repo)
        : base(logger, db)
    {
        _scraper  = scraper;
        _repo     = repo;
        Info.Info = "Bereit – prüfe Fortschritt beim Start...";
    }

    protected override async Task ExecuteAsync(CancellationToken ct)
    {
        // ── 1. Bereits vollständig abgeschlossen? ────────────────────────────
        if (await _repo.GetSettingAsync(DoneFlag) == "true")
        {
            Logger.Info(Module, "CT History-Import: bereits abgeschlossen.");
            Info.Info = "✓ Abgeschlossen";
            return;
        }

        // ── 2. Fortschritt aus DB laden ──────────────────────────────────────
        var pageStr   = await _repo.GetSettingAsync(ProgressKey);
        var startPage = pageStr != null && int.TryParse(pageStr, out var sp) ? sp : 1;

        var totalStr  = await _repo.GetSettingAsync(TotalKey);
        int totalPages;

        if (totalStr != null && int.TryParse(totalStr, out var tp) && tp > 0)
        {
            totalPages = tp;
            Logger.Info(Module,
                $"▶ Fortsetze History-Import: Seite {startPage} von {totalPages} (Gesamtanzahl aus Cache)");
        }
        else
        {
            Logger.Info(Module, "Ermittle Gesamtseitenanzahl von capitoltrades.com...");
            totalPages = await _scraper.GetTotalPagesAsync(ct);
            await _repo.SetSettingAsync(TotalKey, totalPages.ToString());
            Logger.Info(Module,
                $"▶ Starte History-Import: {totalPages} Seiten gefunden. Beginne bei Seite {startPage}.");
        }


        // ── 3. Seiten paginieren ─────────────────────────────────────────────
        var knownMembers  = await _repo.GetAllMemberBioIdsAsync();
        var importedTotal = 0;

        for (var page = startPage; page <= totalPages; page++)
        {
            ct.ThrowIfCancellationRequested();

            Info.Info = $"Seite {page}/{totalPages} – {importedTotal} neue Trades";

            var (trades, memberIds) = await _scraper.ScrapeTradesPageAsync(page, ct);

            if (trades.Count == 0)
            {
                Logger.Warn(Module,
                    $"[{page}/{totalPages}] Leere Seite – Seite wird übersprungen.");
                await _repo.SetSettingAsync(ProgressKey, (page + 1).ToString());
                continue;
            }

            // Neue Kongressmitglieder anlegen (Daten kommen aus MemberSnapshot)
            foreach (var trade in trades.Where(t => t.MemberSnapshot != null))
            {
                var snapshot = trade.MemberSnapshot!;
                if (knownMembers.Contains(snapshot.BioId)) continue;

                await _repo.UpsertMemberAsync(snapshot);
                knownMembers.Add(snapshot.BioId);
                Logger.Info(Module,
                    $"  + {snapshot.Name} ({snapshot.BioId}) {snapshot.Party}/{snapshot.State}");
            }

            // Trades einfügen (INSERT IGNORE – Duplikate werden automatisch ignoriert)
            foreach (var trade in trades)
            {
                ct.ThrowIfCancellationRequested();
                if (await _repo.TradeExistsAsync(trade.TradeId)) continue;
                await _repo.InsertTradeAsync(trade, detailsFetched: true);
                importedTotal++;
            }

            // ─── Fortschritt nach jeder erfolgreichen Seite persistieren ───
            await _repo.SetSettingAsync(ProgressKey, (page + 1).ToString());

            Logger.Info(Module,
                $"[{page}/{totalPages}] {trades.Count} Trades, {importedTotal} neu gesamt.");
        }

        // ── 4. Abschluss ─────────────────────────────────────────────────────
        await _repo.SetSettingAsync(DoneFlag, "true");

        var totalInDb   = await _repo.GetTradeCountAsync();
        var membersInDb = await _repo.GetMemberCountAsync();

        Logger.Info(Module,
            $"=== History-Import abgeschlossen: {totalInDb} Trades, {membersInDb} Mitglieder ===");
        Info.Info = $"✓ Fertig – {totalInDb} Trades, {membersInDb} Mitglieder";
    }
}