diff --git a/src/TelegramPanel.Web/Components/Pages/BatchAdmins.razor b/src/TelegramPanel.Web/Components/Pages/BatchAdmins.razor index 5cadbbf..2127836 100644 --- a/src/TelegramPanel.Web/Components/Pages/BatchAdmins.razor +++ b/src/TelegramPanel.Web/Components/Pages/BatchAdmins.razor @@ -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 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) { diff --git a/src/TelegramPanel.Web/Components/Pages/BatchInvite.razor b/src/TelegramPanel.Web/Components/Pages/BatchInvite.razor index 361a330..565f0a4 100644 --- a/src/TelegramPanel.Web/Components/Pages/BatchInvite.razor +++ b/src/TelegramPanel.Web/Components/Pages/BatchInvite.razor @@ -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 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) { diff --git a/src/TelegramPanel.Web/Components/Pages/Tasks.razor b/src/TelegramPanel.Web/Components/Pages/Tasks.razor index 14ab4e6..6162a0b 100644 --- a/src/TelegramPanel.Web/Components/Pages/Tasks.razor +++ b/src/TelegramPanel.Web/Components/Pages/Tasks.razor @@ -3,6 +3,7 @@ @inject ISnackbar Snackbar @inject IDialogService DialogService @using TelegramPanel.Data.Entities +@implements IDisposable 任务中心 - Telegram Panel @@ -14,6 +15,8 @@ 任务列表 (@tasks.Count) + 全部 待执行 @@ -81,12 +84,33 @@ private bool loading = true; private string filterStatus = "all"; + private PeriodicTimer? refreshTimer; + private CancellationTokenSource? refreshCts; + private IEnumerable 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; diff --git a/src/TelegramPanel.Web/Program.cs b/src/TelegramPanel.Web/Program.cs index f4148d9..68302a5 100644 --- a/src/TelegramPanel.Web/Program.cs +++ b/src/TelegramPanel.Web/Program.cs @@ -63,6 +63,7 @@ builder.Services.AddTelegramPanelData(connectionString); builder.Services.AddTelegramPanelCore(); builder.Services.AddScoped(); builder.Services.AddScoped(); +builder.Services.AddHostedService(); builder.Services.AddHostedService(); builder.Services.AddHttpClient(); diff --git a/src/TelegramPanel.Web/Services/BatchTaskBackgroundService.cs b/src/TelegramPanel.Web/Services/BatchTaskBackgroundService.cs new file mode 100644 index 0000000..9a14c44 --- /dev/null +++ b/src/TelegramPanel.Web/Services/BatchTaskBackgroundService.cs @@ -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; + +/// +/// 批量任务后台执行器：从数据库拉取 pending 任务并在后台静默运行。 +/// +public sealed class BatchTaskBackgroundService : BackgroundService +{ + private readonly IServiceScopeFactory _scopeFactory; + private readonly ILogger _logger; + private readonly IConfiguration _configuration; + + public BatchTaskBackgroundService( + IServiceScopeFactory scopeFactory, + IConfiguration configuration, + ILogger 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(); + var channelManagement = scope.ServiceProvider.GetRequiredService(); + var channelService = scope.ServiceProvider.GetRequiredService(); + + 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(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 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(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 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 ChannelTelegramIds, + List Usernames, + int DelayMs, + int PreferredExecuteAccountId + ); + + private sealed record SetAdminTaskConfig( + List ChannelTelegramIds, + List Usernames, + int Rights, + string Title, + int PreferredExecuteAccountId + ); +}