批量任务支持后台静默执行

2025-12-19 21:00:06 +08:00
1 parent c48ba87197
commit 66cb199d4a
@@ -305,59 +305,15 @@
                Failed = 0,
                Config = System.Text.Json.JsonSerializer.Serialize(new
                {
-                    ChannelIds = targetChannels.Select(c => c.TelegramId).ToList(),
-                    ChannelTitles = targetChannels.Select(c => c.Title).ToList(),
-                    Rights = rights.ToString(),
+                    ChannelTelegramIds = targetChannels.Select(c => c.TelegramId).ToList(),
+                    Usernames = usernameList,
+                    Rights = (int)rights,
                    Title = adminTitle,
-                    ExecutedByAccountId = selectedAccountId
+                    PreferredExecuteAccountId = selectedAccountId
                })
            };
            var createdTask = await TaskManagement.CreateTaskAsync(task);
-            await TaskManagement.StartTaskAsync(createdTask.Id);
-
-            var totalSuccess = 0;
-            var totalFailed = 0;
-
-            foreach (var channel in targetChannels)
-            {
-                var executeAccountId = await ChannelManagement.ResolveExecuteAccountIdAsync(channel, selectedAccountId > 0 ? selectedAccountId : null);
-                if (executeAccountId is not > 0)
-                    throw new InvalidOperationException($"频道“{channel.Title}”没有可用的执行账号：请先【同步数据】或手动选择一个执行账号");
-                var adminRequests = usernameList.Select(u => new AdminRequest(u, rights, adminTitle)).ToList();
-
-                List<SetAdminResult> setResults;
-                try
-                {
-                    setResults = await ChannelService.BatchSetAdminsAsync(executeAccountId.Value, channel.TelegramId, adminRequests);
-                }
-                catch (Exception ex)
-                {
-                    setResults = usernameList.Select(u => new SetAdminResult(u, false, ex.Message)).ToList();
-                }
-
-                foreach (var r in setResults)
-                {
-                    results.Add(new AdminSetOutcome(
-                        ChannelTelegramId: channel.TelegramId,
-                        ChannelTitle: channel.Title,
-                        Username: r.Username,
-                        Success: r.Success,
-                        Error: r.Error,
-                        ExecutedByAccountId: executeAccountId.Value
-                    ));
-                }
-
-                var successCount = setResults.Count(r => r.Success);
-                var failedCount = setResults.Count - successCount;
-                totalSuccess += successCount;
-                totalFailed += failedCount;
-
-                await TaskManagement.UpdateTaskProgressAsync(createdTask.Id, totalSuccess, totalFailed);
-                StateHasChanged();
-            }
-
-            await TaskManagement.CompleteTaskAsync(createdTask.Id, true);
-            Snackbar.Add($"设置完成：成功 {totalSuccess}，失败 {totalFailed}（频道：{targetChannels.Count}）", totalFailed == 0 ? Severity.Success : Severity.Warning);
+            Snackbar.Add($"已提交后台任务 #{createdTask.Id}（批量设置管理员）。请到【任务中心】查看进度。", Severity.Success);
        }
        catch (Exception ex)
        {
@@ -261,56 +261,14 @@
                Failed = 0,
                Config = System.Text.Json.JsonSerializer.Serialize(new
                {
-                    ChannelIds = targetChannels.Select(c => c.TelegramId).ToList(),
-                    ChannelTitles = targetChannels.Select(c => c.Title).ToList(),
+                    ChannelTelegramIds = targetChannels.Select(c => c.TelegramId).ToList(),
+                    Usernames = usernameList,
                    DelayMs = delayMs,
-                    ExecutedByAccountId = selectedAccountId
+                    PreferredExecuteAccountId = selectedAccountId
                })
            };
            var createdTask = await TaskManagement.CreateTaskAsync(task);
-            await TaskManagement.StartTaskAsync(createdTask.Id);
-
-            var totalSuccess = 0;
-            var totalFailed = 0;
-
-            foreach (var channel in targetChannels)
-            {
-                var executeAccountId = await ChannelManagement.ResolveExecuteAccountIdAsync(channel, selectedAccountId > 0 ? selectedAccountId : null);
-                if (executeAccountId is not > 0)
-                    throw new InvalidOperationException($"频道“{channel.Title}”没有可用的执行账号：请先【同步数据】或手动选择一个执行账号");
-
-                List<InviteResult> results;
-                try
-                {
-                    results = await ChannelService.BatchInviteUsersAsync(executeAccountId.Value, channel.TelegramId, usernameList, delayMs);
-                }
-                catch (Exception ex)
-                {
-                    results = usernameList.Select(u => new InviteResult(u, false, ex.Message)).ToList();
-                }
-
-                foreach (var r in results)
-                {
-                    inviteResults.Add(new InviteOutcome(
-                        ChannelTelegramId: channel.TelegramId,
-                        ChannelTitle: channel.Title,
-                        Username: r.Username,
-                        Success: r.Success,
-                        Error: r.Error,
-                        ExecutedByAccountId: executeAccountId.Value
-                    ));
-                }
-
-                var successCount = results.Count(r => r.Success);
-                var failedCount = results.Count - successCount;
-                totalSuccess += successCount;
-                totalFailed += failedCount;
-                await TaskManagement.UpdateTaskProgressAsync(createdTask.Id, totalSuccess, totalFailed);
-                StateHasChanged();
-            }
-
-            await TaskManagement.CompleteTaskAsync(createdTask.Id, true);
-            Snackbar.Add($"邀请完成：成功 {totalSuccess}，失败 {totalFailed}（频道：{targetChannels.Count}）", totalFailed == 0 ? Severity.Success : Severity.Warning);
+            Snackbar.Add($"已提交后台任务 #{createdTask.Id}（批量邀请）。请到【任务中心】查看进度。", Severity.Success);
        }
        catch (Exception ex)
        {
@@ -3,6 +3,7 @@
@inject ISnackbar Snackbar
@inject IDialogService DialogService
@using TelegramPanel.Data.Entities
+@implements IDisposable

 <PageTitle>任务中心 - Telegram Panel</PageTitle>

@@ -14,6 +15,8 @@
            <ToolBarContent>
                <MudText Typo="Typo.h6">任务列表 (@tasks.Count)</MudText>
                <MudSpacer />
+                <MudIconButton Icon="@Icons.Material.Filled.Refresh" Color="Color.Default" Size="Size.Small"
+                               Disabled="@loading" OnClick="@(async (MouseEventArgs _) => await LoadTasks())" />
                <MudSelect @bind-Value="filterStatus" Label="状态筛选" Variant="Variant.Outlined" Dense="true" Style="min-width: 120px;">
                    <MudSelectItem Value='@("all")'>全部</MudSelectItem>
                    <MudSelectItem Value='@("pending")'>待执行</MudSelectItem>
@@ -81,12 +84,33 @@
    private bool loading = true;
    private string filterStatus = "all";

+    private PeriodicTimer? refreshTimer;
+    private CancellationTokenSource? refreshCts;
+
    private IEnumerable<BatchTask> filteredTasks =>
        filterStatus == "all" ? tasks : tasks.Where(t => t.Status == filterStatus);

    protected override async Task OnInitializedAsync()
    {
        await LoadTasks();
+
+        refreshCts = new CancellationTokenSource();
+        refreshTimer = new PeriodicTimer(TimeSpan.FromSeconds(3));
+        _ = Task.Run(async () =>
+        {
+            try
+            {
+                while (await refreshTimer.WaitForNextTickAsync(refreshCts.Token))
+                {
+                    if (tasks.Any(t => t.Status is "pending" or "running"))
+                        await InvokeAsync(LoadTasks);
+                }
+            }
+            catch (OperationCanceledException)
+            {
+                // ignore
+            }
+        });
    }

    private async Task LoadTasks()
@@ -107,6 +131,23 @@
        }
    }

+    public void Dispose()
+    {
+        try
+        {
+            refreshCts?.Cancel();
+        }
+        catch
+        {
+            // ignore
+        }
+        finally
+        {
+            refreshTimer?.Dispose();
+            refreshCts?.Dispose();
+        }
+    }
+
    private double GetProgress(BatchTask task)
    {
        if (task.Total == 0) return 0;
@@ -63,6 +63,7 @@ builder.Services.AddTelegramPanelData(connectionString);
 builder.Services.AddTelegramPanelCore();
 builder.Services.AddScoped<AccountExportService>();
 builder.Services.AddScoped<DataSyncService>();
+builder.Services.AddHostedService<BatchTaskBackgroundService>();
 builder.Services.AddHostedService<BotAutoSyncBackgroundService>();
 builder.Services.AddHttpClient<TelegramBotApiClient>();

@@ -0,0 +1,263 @@
+using System.Text.Json;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using TelegramPanel.Core.Interfaces;
+using TelegramPanel.Core.Services;
+using TelegramPanel.Data.Entities;
+
+namespace TelegramPanel.Web.Services;
+
+/// <summary>
+/// 批量任务后台执行器：从数据库拉取 pending 任务并在后台静默运行。
+/// </summary>
+public sealed class BatchTaskBackgroundService : BackgroundService
+{
+    private readonly IServiceScopeFactory _scopeFactory;
+    private readonly ILogger<BatchTaskBackgroundService> _logger;
+    private readonly IConfiguration _configuration;
+
+    public BatchTaskBackgroundService(
+        IServiceScopeFactory scopeFactory,
+        IConfiguration configuration,
+        ILogger<BatchTaskBackgroundService> logger)
+    {
+        _scopeFactory = scopeFactory;
+        _configuration = configuration;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
+    {
+        var enabled = _configuration.GetValue("BatchTasks:Enabled", true);
+        if (!enabled)
+        {
+            _logger.LogInformation("Batch task runner disabled (BatchTasks:Enabled=false)");
+            return;
+        }
+
+        var seconds = _configuration.GetValue("BatchTasks:PollIntervalSeconds", 2);
+        if (seconds < 1) seconds = 1;
+        if (seconds > 30) seconds = 30;
+        var interval = TimeSpan.FromSeconds(seconds);
+
+        _logger.LogInformation("Batch task runner started, interval {IntervalSeconds} seconds", seconds);
+
+        // 延迟一点，避免与启动时 DB 迁移抢资源
+        await Task.Delay(TimeSpan.FromSeconds(3), stoppingToken);
+
+        while (!stoppingToken.IsCancellationRequested)
+        {
+            try
+            {
+                await TryRunOneAsync(stoppingToken);
+            }
+            catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
+            {
+                // ignore
+            }
+            catch (Exception ex)
+            {
+                _logger.LogWarning(ex, "Batch task runner loop failed");
+            }
+
+            await Task.Delay(interval, stoppingToken);
+        }
+    }
+
+    private async Task TryRunOneAsync(CancellationToken cancellationToken)
+    {
+        using var scope = _scopeFactory.CreateScope();
+        var taskManagement = scope.ServiceProvider.GetRequiredService<BatchTaskManagementService>();
+        var channelManagement = scope.ServiceProvider.GetRequiredService<ChannelManagementService>();
+        var channelService = scope.ServiceProvider.GetRequiredService<IChannelService>();
+
+        var pending = (await taskManagement.GetTasksByStatusAsync("pending"))
+            .OrderBy(t => t.CreatedAt)
+            .FirstOrDefault();
+
+        if (pending == null)
+            return;
+
+        await taskManagement.StartTaskAsync(pending.Id);
+        _logger.LogInformation("Batch task started: {TaskId} {TaskType}", pending.Id, pending.TaskType);
+
+        var completed = pending.Completed;
+        var failed = pending.Failed;
+
+        try
+        {
+            switch (pending.TaskType)
+            {
+                case "invite":
+                    (completed, failed) = await RunInviteAsync(pending, taskManagement, channelManagement, channelService, completed, failed, cancellationToken);
+                    break;
+                case "set_admin":
+                    (completed, failed) = await RunSetAdminAsync(pending, taskManagement, channelManagement, channelService, completed, failed, cancellationToken);
+                    break;
+                default:
+                    failed = pending.Total == 0 ? 1 : pending.Total;
+                    await taskManagement.UpdateTaskProgressAsync(pending.Id, completed, failed);
+                    await taskManagement.CompleteTaskAsync(pending.Id, success: false);
+                    _logger.LogWarning("Unsupported batch task type: {TaskType} (task {TaskId})", pending.TaskType, pending.Id);
+                    return;
+            }
+
+            // 如果任务被用户取消（当前实现：Cancel 会把状态写成 failed），则不覆盖它
+            var latest = await taskManagement.GetTaskAsync(pending.Id);
+            if (latest != null && latest.Status != "running")
+                return;
+
+            await taskManagement.CompleteTaskAsync(pending.Id, success: failed == 0);
+            _logger.LogInformation("Batch task completed: {TaskId} {TaskType} (ok={Ok}, completed={Completed}, failed={Failed})",
+                pending.Id, pending.TaskType, failed == 0, completed, failed);
+        }
+        catch (Exception ex)
+        {
+            _logger.LogWarning(ex, "Batch task failed: {TaskId} {TaskType}", pending.Id, pending.TaskType);
+            try
+            {
+                await taskManagement.UpdateTaskProgressAsync(pending.Id, completed, failed == 0 ? 1 : failed);
+                await taskManagement.CompleteTaskAsync(pending.Id, success: false);
+            }
+            catch
+            {
+                // ignore secondary failures
+            }
+        }
+    }
+
+    private static async Task<(int completed, int failed)> RunInviteAsync(
+        BatchTask task,
+        BatchTaskManagementService taskManagement,
+        ChannelManagementService channelManagement,
+        IChannelService channelService,
+        int completed,
+        int failed,
+        CancellationToken cancellationToken)
+    {
+        if (string.IsNullOrWhiteSpace(task.Config))
+            throw new InvalidOperationException("任务配置为空");
+
+        var cfg = JsonSerializer.Deserialize<InviteTaskConfig>(task.Config);
+        if (cfg == null || cfg.ChannelTelegramIds.Count == 0 || cfg.Usernames.Count == 0)
+            throw new InvalidOperationException("任务配置缺少 ChannelTelegramIds 或 Usernames");
+
+        foreach (var channelTelegramId in cfg.ChannelTelegramIds)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            var latest = await taskManagement.GetTaskAsync(task.Id);
+            if (latest != null && latest.Status != "running")
+                return (completed, failed);
+
+            var channel = await channelManagement.GetChannelByTelegramIdAsync(channelTelegramId);
+            if (channel == null)
+            {
+                failed += cfg.Usernames.Count;
+                await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+                continue;
+            }
+
+            var executeAccountId = await channelManagement.ResolveExecuteAccountIdAsync(channel, cfg.PreferredExecuteAccountId > 0 ? cfg.PreferredExecuteAccountId : null);
+            if (executeAccountId is not > 0)
+            {
+                failed += cfg.Usernames.Count;
+                await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+                continue;
+            }
+
+            List<InviteResult> results;
+            try
+            {
+                results = await channelService.BatchInviteUsersAsync(executeAccountId.Value, channelTelegramId, cfg.Usernames, cfg.DelayMs);
+            }
+            catch (Exception ex)
+            {
+                results = cfg.Usernames.Select(u => new InviteResult(u, false, ex.Message)).ToList();
+            }
+
+            completed += results.Count(r => r.Success);
+            failed += results.Count(r => !r.Success);
+            await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+        }
+
+        return (completed, failed);
+    }
+
+    private static async Task<(int completed, int failed)> RunSetAdminAsync(
+        BatchTask task,
+        BatchTaskManagementService taskManagement,
+        ChannelManagementService channelManagement,
+        IChannelService channelService,
+        int completed,
+        int failed,
+        CancellationToken cancellationToken)
+    {
+        if (string.IsNullOrWhiteSpace(task.Config))
+            throw new InvalidOperationException("任务配置为空");
+
+        var cfg = JsonSerializer.Deserialize<SetAdminTaskConfig>(task.Config);
+        if (cfg == null || cfg.ChannelTelegramIds.Count == 0 || cfg.Usernames.Count == 0)
+            throw new InvalidOperationException("任务配置缺少 ChannelTelegramIds 或 Usernames");
+
+        var rights = (AdminRights)cfg.Rights;
+        var adminRequests = cfg.Usernames.Select(u => new AdminRequest(u, rights, cfg.Title)).ToList();
+
+        foreach (var channelTelegramId in cfg.ChannelTelegramIds)
+        {
+            cancellationToken.ThrowIfCancellationRequested();
+
+            var latest = await taskManagement.GetTaskAsync(task.Id);
+            if (latest != null && latest.Status != "running")
+                return (completed, failed);
+
+            var channel = await channelManagement.GetChannelByTelegramIdAsync(channelTelegramId);
+            if (channel == null)
+            {
+                failed += cfg.Usernames.Count;
+                await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+                continue;
+            }
+
+            var executeAccountId = await channelManagement.ResolveExecuteAccountIdAsync(channel, cfg.PreferredExecuteAccountId > 0 ? cfg.PreferredExecuteAccountId : null);
+            if (executeAccountId is not > 0)
+            {
+                failed += cfg.Usernames.Count;
+                await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+                continue;
+            }
+
+            List<SetAdminResult> results;
+            try
+            {
+                results = await channelService.BatchSetAdminsAsync(executeAccountId.Value, channelTelegramId, adminRequests);
+            }
+            catch (Exception ex)
+            {
+                results = cfg.Usernames.Select(u => new SetAdminResult(u, false, ex.Message)).ToList();
+            }
+
+            completed += results.Count(r => r.Success);
+            failed += results.Count(r => !r.Success);
+            await taskManagement.UpdateTaskProgressAsync(task.Id, completed, failed);
+        }
+
+        return (completed, failed);
+    }
+
+    private sealed record InviteTaskConfig(
+        List<long> ChannelTelegramIds,
+        List<string> Usernames,
+        int DelayMs,
+        int PreferredExecuteAccountId
+    );
+
+    private sealed record SetAdminTaskConfig(
+        List<long> ChannelTelegramIds,
+        List<string> Usernames,
+        int Rights,
+        string Title,
+        int PreferredExecuteAccountId
+    );
+}