fix: stabilize bot polling and simplify tasks

This commit is contained in:
meoacgx committed 2025-12-22 19:07:31 +08:00
1 parent 3c1e54ec4f
commit 1758680f3d
14 files changed
+664 -148

No files matched your search

+31
View File
@@ -0,0 +1,31 @@
FROM mcr.microsoft.com/dotnet/sdk:8.0 AS build
WORKDIR /src
COPY . .
RUN dotnet restore "TelegramPanel.sln"
RUN dotnet publish "src/TelegramPanel.Web/TelegramPanel.Web.csproj" -c Release -o /app/publish --no-restore
# 说明:
# - 正常生产部署建议使用 mcr.microsoft.com/dotnet/aspnet:8.0 作为运行时镜像(更小)
# - 若当前网络环境导致 Docker/WSL 无法拉取 aspnet 镜像,可临时使用本文件(运行时使用 sdk 镜像)
FROM mcr.microsoft.com/dotnet/sdk:8.0 AS final
WORKDIR /app
ENV ASPNETCORE_URLS=http://+:5000
EXPOSE 5000
# 持久化目录:/data(通过 docker-compose 挂载)
# - 数据库:/data/telegram-panel.db
# - session:/data/sessions/
# - 本地配置:/data/appsettings.local.json(UI 保存 Telegram ApiId/ApiHash/同步开关等)
# - 后台密码:/data/admin_auth.json
RUN mkdir -p /data /data/sessions /data/logs \
&& rm -rf /app/logs \
&& ln -s /data/logs /app/logs \
&& ln -s /data/appsettings.local.json /app/appsettings.local.json || true
COPY --from=build /app/publish .
ENTRYPOINT ["dotnet", "TelegramPanel.Web.dll"]
+1 -2
View File
@@ -2,7 +2,7 @@ services:
telegram-panel:
build:
context: .
dockerfile: Dockerfile
dockerfile: ${TP_DOCKERFILE:-Dockerfile}
container_name: telegram-panel
restart: unless-stopped
ports:
@@ -27,4 +27,3 @@ services:
# 默认关闭 bot 轮询自动同步(需要时可手动同步)
Telegram__BotAutoSyncEnabled: "false"
@@ -23,6 +23,9 @@ public static class ServiceCollectionExtensions
services.AddScoped<AccountTelegramToolsService>();
services.AddScoped<BotTelegramService>();
// Bot API updates(getUpdates)统一轮询与分发:避免 409 Conflict
services.AddSingleton<BotUpdateHub>();
// 注册账号导入协调服务
services.AddScoped<AccountImportService>();
@@ -30,9 +30,19 @@ public class BotTelegramService
}
/// <summary>
/// 拉取 Bot API updates 并把 bot 新加入的频道写入数据库(同时确认 offset)。
/// 兼容入口:旧版曾在这里直接 getUpdates 拉取 my_chat_member。
/// 现在 getUpdates 已由 <see cref="BotUpdateHub"/> 统一轮询与分发,避免 409 Conflict 与 offset 争抢。
/// </summary>
public async Task<int> SyncBotChannelsAsync(int botId, CancellationToken cancellationToken)
public Task<int> SyncBotChannelsAsync(int botId, CancellationToken cancellationToken)
{
_logger.LogInformation("SyncBotChannelsAsync 已迁移为自动同步(BotUpdateHub),不再在此处拉取 updates:botId={BotId}", botId);
return Task.FromResult(0);
}
/// <summary>
/// 应用一批 my_chat_member updates,把 Bot 新加入/移除/升降权的频道写入数据库。
/// </summary>
public async Task<int> ApplyMyChatMemberUpdatesAsync(int botId, IReadOnlyList<JsonElement> updates, CancellationToken cancellationToken)
{
var bot = await _botManagement.GetBotAsync(botId)
?? throw new InvalidOperationException($"机器人不存在:{botId}");
@@ -40,38 +50,17 @@ public class BotTelegramService
if (!bot.IsActive)
throw new InvalidOperationException("该机器人已停用");
var token = bot.Token;
var offset = bot.LastUpdateId.HasValue ? bot.LastUpdateId.Value + 1 : (long?)null;
var allowedUpdates = JsonSerializer.Serialize(new[] { "my_chat_member" });
var result = await _api.CallAsync(token, "getUpdates", new Dictionary<string, string?>
{
["offset"] = offset?.ToString(),
["timeout"] = "0",
["limit"] = "100",
["allowed_updates"] = allowedUpdates
}, cancellationToken);
if (result.ValueKind != JsonValueKind.Array)
return 0;
// Telegram 会针对同一个 chat 连续产生多条 my_chat_member(例如 member -> administrator),
// 这里按 chat_id 去重并只应用“最后一次状态”,避免“新增两个”的错觉/重复写库。
var changesByChatId = new Dictionary<long, BotChatMemberChange>();
long? maxUpdateId = null;
foreach (var update in result.EnumerateArray())
foreach (var update in updates)
{
cancellationToken.ThrowIfCancellationRequested();
if (update.ValueKind != JsonValueKind.Object)
continue;
if (update.TryGetProperty("update_id", out var updateIdEl) && updateIdEl.TryGetInt64(out var updateId))
{
maxUpdateId = maxUpdateId.HasValue ? Math.Max(maxUpdateId.Value, updateId) : updateId;
}
// 只处理 my_chat_member:Bot 被加入/移除/升降权
if (!update.TryGetProperty("my_chat_member", out var myChatMember))
continue;
@@ -123,8 +112,6 @@ public class BotTelegramService
affected++;
}
if (maxUpdateId.HasValue)
bot.LastUpdateId = maxUpdateId.Value;
bot.LastSyncAt = DateTime.UtcNow;
await _botManagement.UpdateBotAsync(bot);
@@ -0,0 +1,408 @@
using System.Collections.Concurrent;
using System.Text.Json;
using System.Threading.Channels;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Logging;
using TelegramPanel.Data.Repositories;
namespace TelegramPanel.Core.Services.Telegram;
/// <summary>
/// Bot API 更新分发 Hub:
/// - 同一个 Bot Token 只允许一个 getUpdates 长轮询,否则会 409 Conflict
/// - 这里为每个 botId 维护一个单一轮询器,并把同一份 updates 广播给多个消费者(模块/后台服务)
/// </summary>
public sealed class BotUpdateHub : IAsyncDisposable
{
// 固定允许的更新类型:覆盖当前项目使用场景(转发/监听/入群事件)
private const string AllowedUpdatesJson = "[\"message\",\"edited_message\",\"channel_post\",\"edited_channel_post\",\"my_chat_member\"]";
private readonly IServiceScopeFactory _scopeFactory;
private readonly TelegramBotApiClient _botApi;
private readonly ILogger<BotUpdateHub> _logger;
private readonly SemaphoreSlim _gate = new(1, 1);
private readonly Dictionary<string, BotPoller> _pollersByToken = new(StringComparer.Ordinal);
public BotUpdateHub(
IServiceScopeFactory scopeFactory,
TelegramBotApiClient botApi,
ILogger<BotUpdateHub> logger)
{
_scopeFactory = scopeFactory;
_botApi = botApi;
_logger = logger;
}
public async Task<BotUpdateSubscription> SubscribeAsync(int botId, CancellationToken cancellationToken)
{
if (botId <= 0)
throw new ArgumentException("botId 无效", nameof(botId));
cancellationToken.ThrowIfCancellationRequested();
await _gate.WaitAsync(cancellationToken);
try
{
using var scope = _scopeFactory.CreateScope();
var botRepo = scope.ServiceProvider.GetRequiredService<IBotRepository>();
var bot = await botRepo.GetByIdAsync(botId);
if (bot == null)
throw new InvalidOperationException($"Bot 不存在:{botId}");
if (!bot.IsActive)
throw new InvalidOperationException("Bot 未启用");
var token = (bot.Token ?? "").Trim();
if (string.IsNullOrWhiteSpace(token))
throw new InvalidOperationException("Bot Token 为空");
if (!_pollersByToken.TryGetValue(token, out var poller))
{
poller = await BotPoller.CreateAsync(botId, token, bot.LastUpdateId, _scopeFactory, _botApi, _logger, cancellationToken);
_pollersByToken[token] = poller;
}
return poller.Subscribe();
}
finally
{
_gate.Release();
}
}
public async ValueTask DisposeAsync()
{
List<BotPoller> pollers;
await _gate.WaitAsync();
try
{
pollers = _pollersByToken.Values.ToList();
_pollersByToken.Clear();
}
finally
{
_gate.Release();
}
foreach (var p in pollers)
{
try { await p.DisposeAsync(); }
catch (Exception ex) { _logger.LogWarning(ex, "Dispose bot poller failed: {BotId}", p.BotId); }
}
}
public sealed class BotUpdateSubscription : IAsyncDisposable
{
private readonly Func<ValueTask> _dispose;
internal BotUpdateSubscription(int botId, ChannelReader<JsonElement> reader, Func<ValueTask> dispose)
{
BotId = botId;
Reader = reader;
_dispose = dispose;
}
public int BotId { get; }
public ChannelReader<JsonElement> Reader { get; }
public ValueTask DisposeAsync() => _dispose();
}
private sealed class BotPoller : IAsyncDisposable
{
private static readonly BoundedChannelOptions SubscriberChannelOptions = new(512)
{
SingleWriter = true,
SingleReader = true,
FullMode = BoundedChannelFullMode.DropOldest
};
private readonly int _persistBotId;
private readonly string _token;
private readonly IServiceScopeFactory _scopeFactory;
private readonly TelegramBotApiClient _botApi;
private readonly ILogger _logger;
private readonly CancellationTokenSource _cts = new();
private readonly Task _loopTask;
private readonly object _subscribersLock = new();
private readonly Dictionary<Guid, Channel<JsonElement>> _subscribers = new();
private long _nextOffset;
public int BotId => _persistBotId;
private BotPoller(
int persistBotId,
string token,
long nextOffset,
IServiceScopeFactory scopeFactory,
TelegramBotApiClient botApi,
ILogger logger)
{
_persistBotId = persistBotId;
_token = token;
_nextOffset = nextOffset;
_scopeFactory = scopeFactory;
_botApi = botApi;
_logger = logger;
_loopTask = Task.Run(() => RunAsync(_cts.Token));
}
public static async Task<BotPoller> CreateAsync(
int botId,
string token,
long? lastUpdateId,
IServiceScopeFactory scopeFactory,
TelegramBotApiClient botApi,
ILogger logger,
CancellationToken cancellationToken)
{
// 初始化 offset:
// - 如果数据库里已有 LastUpdateId,则从其后开始
// - 否则快进到最新,避免冷启动回放历史消息造成刷屏
long nextOffset;
if (lastUpdateId.HasValue)
{
nextOffset = lastUpdateId.Value + 1;
}
else
{
nextOffset = await FastForwardOffsetAsync(botId, token, scopeFactory, botApi, logger, cancellationToken);
}
return new BotPoller(botId, token, nextOffset, scopeFactory, botApi, logger);
}
public BotUpdateSubscription Subscribe()
{
var id = Guid.NewGuid();
var ch = Channel.CreateBounded<JsonElement>(SubscriberChannelOptions);
lock (_subscribersLock)
{
_subscribers[id] = ch;
}
return new BotUpdateSubscription(_persistBotId, ch.Reader, async () =>
{
Channel<JsonElement>? removed = null;
lock (_subscribersLock)
{
if (_subscribers.Remove(id, out var existing))
removed = existing;
}
if (removed != null)
{
try { removed.Writer.TryComplete(); }
catch { /* ignore */ }
}
await ValueTask.CompletedTask;
});
}
private async Task RunAsync(CancellationToken cancellationToken)
{
// 给启动期一点喘息
try { await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken); }
catch (OperationCanceledException) { return; }
var conflictStreak = 0;
while (!cancellationToken.IsCancellationRequested)
{
try
{
// 没有订阅者时也继续拉取并确认 offset,避免积压导致后续刷屏;
// 同时还能保证 Bot 频道自动同步等后台能力可工作。
var updates = await _botApi.CallAsync(_token, "getUpdates", new Dictionary<string, string?>
{
["offset"] = _nextOffset.ToString(),
["timeout"] = "25",
["limit"] = "100",
["allowed_updates"] = AllowedUpdatesJson
}, cancellationToken);
if (updates.ValueKind != JsonValueKind.Array)
continue;
long? maxUpdateId = null;
foreach (var update in updates.EnumerateArray())
{
cancellationToken.ThrowIfCancellationRequested();
if (!TryGetUpdateId(update, out var updateId))
continue;
maxUpdateId = maxUpdateId.HasValue ? Math.Max(maxUpdateId.Value, updateId) : updateId;
Broadcast(update);
}
if (maxUpdateId.HasValue)
{
_nextOffset = maxUpdateId.Value + 1;
await SaveLastUpdateIdAsync(maxUpdateId.Value, cancellationToken);
}
conflictStreak = 0;
}
catch (OperationCanceledException) when (cancellationToken.IsCancellationRequested)
{
break;
}
catch (InvalidOperationException ex) when (ex.Message.Contains("getUpdates (409)", StringComparison.OrdinalIgnoreCase))
{
// 典型原因:
// - 同一个 bot token 还有别的实例在 long-poll
// - 或者本进程有 bug 造成并发(理论上 Hub 已避免)
conflictStreak++;
var backoffSeconds = Math.Min(60, 2 * conflictStreak);
_logger.LogWarning(ex, "Bot getUpdates 409 conflict: botId={BotId}(请确保该 Bot Token 仅有一个实例在轮询 getUpdates)", _persistBotId);
await Task.Delay(TimeSpan.FromSeconds(backoffSeconds), cancellationToken);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Bot update poll loop failed: botId={BotId}", _persistBotId);
await Task.Delay(TimeSpan.FromSeconds(2), cancellationToken);
}
}
}
private void Broadcast(JsonElement update)
{
List<Channel<JsonElement>> targets;
lock (_subscribersLock)
{
if (_subscribers.Count == 0)
return;
targets = _subscribers.Values.ToList();
}
foreach (var ch in targets)
{
try
{
// Clone:JsonElement 的生命周期绑定 JsonDocument
ch.Writer.TryWrite(update.Clone());
}
catch (Exception ex)
{
_logger.LogDebug(ex, "Broadcast update failed: botId={BotId}", _persistBotId);
}
}
}
private static bool TryGetUpdateId(JsonElement update, out long updateId)
{
updateId = 0;
if (!update.TryGetProperty("update_id", out var el))
return false;
if (el.ValueKind != JsonValueKind.Number)
return false;
return el.TryGetInt64(out updateId);
}
private async Task SaveLastUpdateIdAsync(long updateId, CancellationToken cancellationToken)
{
using var scope = _scopeFactory.CreateScope();
var botRepo = scope.ServiceProvider.GetRequiredService<IBotRepository>();
var bot = await botRepo.GetByIdAsync(_persistBotId);
if (bot == null)
return;
if (!bot.LastUpdateId.HasValue || updateId > bot.LastUpdateId.Value)
{
bot.LastUpdateId = updateId;
await botRepo.UpdateAsync(bot);
}
}
private static async Task<long> FastForwardOffsetAsync(
int botId,
string token,
IServiceScopeFactory scopeFactory,
TelegramBotApiClient botApi,
ILogger logger,
CancellationToken cancellationToken)
{
long offset = 0;
// 尝试最多 20 次(2000 条)以“清空积压”,避免首次启用模块直接刷历史。
for (var i = 0; i < 20; i++)
{
cancellationToken.ThrowIfCancellationRequested();
var updates = await botApi.CallAsync(token, "getUpdates", new Dictionary<string, string?>
{
["offset"] = offset.ToString(),
["timeout"] = "0",
["limit"] = "100",
["allowed_updates"] = AllowedUpdatesJson
}, cancellationToken);
if (updates.ValueKind != JsonValueKind.Array)
break;
long? maxUpdateId = null;
foreach (var u in updates.EnumerateArray())
{
if (!TryGetUpdateId(u, out var id))
continue;
maxUpdateId = maxUpdateId.HasValue ? Math.Max(maxUpdateId.Value, id) : id;
}
if (!maxUpdateId.HasValue)
break;
offset = maxUpdateId.Value + 1;
}
// 写入数据库(用于下次启动直接从最新开始)
try
{
using var scope = scopeFactory.CreateScope();
var botRepo = scope.ServiceProvider.GetRequiredService<IBotRepository>();
var bot = await botRepo.GetByIdAsync(botId);
if (bot != null && offset > 0)
{
var last = offset - 1;
if (!bot.LastUpdateId.HasValue || last > bot.LastUpdateId.Value)
{
bot.LastUpdateId = last;
await botRepo.UpdateAsync(bot);
}
}
}
catch (Exception ex)
{
logger.LogWarning(ex, "FastForwardOffset persistence failed: botId={BotId}", botId);
}
return offset;
}
public async ValueTask DisposeAsync()
{
_cts.Cancel();
try { await _loopTask; } catch { /* ignore */ }
lock (_subscribersLock)
{
foreach (var ch in _subscribers.Values)
{
try { ch.Writer.TryComplete(); } catch { /* ignore */ }
}
_subscribers.Clear();
}
_cts.Dispose();
}
}
}
@@ -1,4 +1,5 @@
using System.Net.Http;
using System.Net.Http.Headers;
using System.Text;
using System.Text.Json;
using Microsoft.Extensions.Logging;
@@ -118,6 +119,76 @@ public class TelegramBotApiClient
return result.Clone();
}
/// <summary>
/// 调用 Bot API 并上传多个文件(使用 multipart/form-data)。
/// 典型用途:sendMediaGroup(attach://file1...)。
/// </summary>
public async Task<JsonElement> CallWithFilesAsync(
string token,
string method,
IReadOnlyDictionary<string, string?> parameters,
IReadOnlyDictionary<string, (Stream Stream, string FileName, string? ContentType)> files,
CancellationToken cancellationToken)
{
token = (token ?? string.Empty).Trim();
if (string.IsNullOrWhiteSpace(token))
throw new InvalidOperationException("Bot Token 为空");
method = (method ?? string.Empty).Trim();
if (string.IsNullOrWhiteSpace(method))
throw new InvalidOperationException("method 为空");
if (files == null || files.Count == 0)
throw new InvalidOperationException("files 为空");
var url = $"https://api.telegram.org/bot{Uri.EscapeDataString(token)}/{Uri.EscapeDataString(method)}";
using var content = new MultipartFormDataContent();
foreach (var (key, value) in parameters)
{
if (!string.IsNullOrWhiteSpace(key) && value != null)
content.Add(new StringContent(value), key);
}
foreach (var (name, file) in files)
{
if (string.IsNullOrWhiteSpace(name))
continue;
var stream = file.Stream ?? throw new InvalidOperationException($"files[{name}] stream 为空");
if (stream.CanSeek)
stream.Position = 0;
var fileName = (file.FileName ?? "file").Trim();
if (string.IsNullOrWhiteSpace(fileName))
fileName = "file";
var streamContent = new StreamContent(stream);
streamContent.Headers.ContentType = new MediaTypeHeaderValue(string.IsNullOrWhiteSpace(file.ContentType) ? "application/octet-stream" : file.ContentType);
content.Add(streamContent, name, fileName);
}
using var resp = await _http.PostAsync(url, content, cancellationToken);
var json = await resp.Content.ReadAsStringAsync(cancellationToken);
using var doc = JsonDocument.Parse(json);
var root = doc.RootElement;
var ok = root.TryGetProperty("ok", out var okEl) && okEl.ValueKind == JsonValueKind.True;
if (!ok)
{
var code = root.TryGetProperty("error_code", out var codeEl) ? codeEl.GetInt32() : 0;
var desc = root.TryGetProperty("description", out var descEl) ? descEl.GetString() : "未知错误";
throw new InvalidOperationException($"Bot API 调用失败:{method} ({code}) {desc}");
}
if (!root.TryGetProperty("result", out var result))
throw new InvalidOperationException($"Bot API 返回缺少 result:{method}");
return result.Clone();
}
private static string BuildUrl(string token, string method, IReadOnlyDictionary<string, string?> query)
{
var sb = new StringBuilder();
@@ -1,6 +1,7 @@
using System.Collections.Concurrent;
using System.Linq;
using System.Reflection;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.Logging;
using TelegramPanel.Core.Interfaces;
using WTelegram;
@@ -15,11 +16,13 @@ public class TelegramClientPool : ITelegramClientPool, IDisposable
{
private readonly ConcurrentDictionary<int, Client> _clients = new();
private readonly ConcurrentDictionary<int, SemaphoreSlim> _locks = new();
private readonly IConfiguration _configuration;
private readonly ILogger<TelegramClientPool> _logger;
private bool _disposed;
public TelegramClientPool(ILogger<TelegramClientPool> logger)
public TelegramClientPool(IConfiguration configuration, ILogger<TelegramClientPool> logger)
{
_configuration = configuration;
_logger = logger;
}
@@ -61,6 +64,12 @@ public class TelegramClientPool : ITelegramClientPool, IDisposable
phoneNumber = NormalizePhone(phoneNumber);
string Config(string what)
{
var proxyServer = (_configuration["Telegram:Proxy:Server"] ?? "").Trim();
var proxyPort = (_configuration["Telegram:Proxy:Port"] ?? "").Trim();
var proxyUsername = (_configuration["Telegram:Proxy:Username"] ?? "").Trim();
var proxyPassword = (_configuration["Telegram:Proxy:Password"] ?? "").Trim();
var proxySecret = (_configuration["Telegram:Proxy:Secret"] ?? "").Trim();
return what switch
{
"api_id" => apiId.ToString(),
@@ -69,6 +78,17 @@ public class TelegramClientPool : ITelegramClientPool, IDisposable
"session_key" => string.IsNullOrWhiteSpace(sessionKey) ? null! : sessionKey,
"phone_number" => string.IsNullOrWhiteSpace(phoneNumber) ? null! : phoneNumber,
"user_id" => userId.HasValue && userId.Value > 0 ? userId.Value.ToString() : null!,
// 代理(可选):用于 Telegram MTProto 连接(WTelegramClient)。
// 常见场景:宿主机能直连 Telegram,但 Docker/服务器环境需要代理。
// - SOCKS5/HTTP:配置 server/port;如需认证再填 username/password
// - MTProto:额外配置 secret
"proxy_server" => string.IsNullOrWhiteSpace(proxyServer) ? null! : proxyServer,
"proxy_port" => string.IsNullOrWhiteSpace(proxyPort) ? null! : proxyPort,
"proxy_username" => string.IsNullOrWhiteSpace(proxyUsername) ? null! : proxyUsername,
"proxy_password" => string.IsNullOrWhiteSpace(proxyPassword) ? null! : proxyPassword,
"proxy_secret" => string.IsNullOrWhiteSpace(proxySecret) ? null! : proxySecret,
_ => null! // 使用 null! 抑制警告,这是 WTelegramClient 的预期行为
};
}
@@ -13,7 +13,11 @@ public static class ServiceCollectionExtensions
{
// 注册数据库上下文
services.AddDbContext<AppDbContext>(options =>
options.UseSqlite(connectionString));
options.UseSqlite(connectionString, sqlite =>
{
// 避免 Include 多个集合导航时的“笛卡尔爆炸”与性能警告
sqlite.UseQuerySplittingBehavior(QuerySplittingBehavior.SplitQuery);
}));
// 注册所有Repository
services.AddScoped<IAccountRepository, AccountRepository>();
@@ -261,7 +261,10 @@
try
{
var count = await BotTelegram.SyncBotChannelsAsync(selectedBotId, CancellationToken.None);
Snackbar.Add($"同步完成:影响 {count} 个频道", Severity.Success);
if (count > 0)
Snackbar.Add($"同步完成:影响 {count} 个频道", Severity.Success);
else
Snackbar.Add("已启用秒级自动同步:手动同步无需再拉取 updates", Severity.Info);
await LoadBotData();
}
catch (Exception ex)
@@ -53,10 +53,10 @@
{
<MudAlert Severity="Severity.Warning" Variant="Variant.Outlined" Dense="true" Class="mb-3">
<MudText Typo="Typo.subtitle2">检测到模块贡献冲突(已自动忽略部分定义):</MudText>
<MudList T="string" Dense="true">
<MudList T="System.String" Dense="true">
@foreach (var d in Contributions.Diagnostics)
{
<MudListItem T="string">@d</MudListItem>
<MudListItem T="System.String">@d</MudListItem>
}
</MudList>
</MudAlert>
@@ -3,60 +3,14 @@
@inject TelegramPanel.Web.Modules.ModuleContributionRegistry Contributions
@inject ISnackbar Snackbar
@inject IDialogService DialogService
@inject NavigationManager Navigation
@using TelegramPanel.Data.Entities
@using TelegramPanel.Core.BatchTasks
@using TelegramPanel.Web.Modules
@implements IDisposable
<PageTitle>任务中心 - Telegram Panel</PageTitle>
<MudText Typo="Typo.h4" Class="mb-4">任务中心</MudText>
@if (ConfigurableTasks.Count > 0)
{
<MudCard Class="mb-4">
<MudCardContent>
<MudText Typo="Typo.h6" Class="mb-2">持续任务(可配置)</MudText>
<MudText Typo="Typo.caption" Class="mud-text-secondary mb-3">
这类任务通常是模块提供的“常驻后台能力”,不占用批量任务队列;在这里可直接打开配置窗口编辑。
</MudText>
<MudTable Items="@ConfigurableTasks" Dense="true" Hover="true" Breakpoint="Breakpoint.Sm">
<HeaderContent>
<MudTh>任务</MudTh>
<MudTh>归属</MudTh>
<MudTh>入口</MudTh>
<MudTh>操作</MudTh>
</HeaderContent>
<RowTemplate>
<MudTd DataLabel="任务">
<MudText Typo="Typo.subtitle2">@context.DisplayName</MudText>
@if (!string.IsNullOrWhiteSpace(context.Description))
{
<MudText Typo="Typo.caption" Class="mud-text-secondary">@context.Description</MudText>
}
<MudText Typo="Typo.caption" Class="mud-text-secondary">@context.TaskType</MudText>
</MudTd>
<MudTd DataLabel="归属">
<MudChip T="string" Size="Size.Small" Color="@GetCategoryColorByCategory(context.Category)">
@GetCategoryDisplayName(context.Category)
</MudChip>
</MudTd>
<MudTd DataLabel="入口">
<MudText Typo="Typo.caption">@context.CreateRoute</MudText>
</MudTd>
<MudTd DataLabel="操作">
<MudButton Variant="Variant.Text" Color="Color.Primary" OnClick="@(() => OpenTaskConfig(context))">
编辑
</MudButton>
</MudTd>
</RowTemplate>
</MudTable>
</MudCardContent>
</MudCard>
}
<MudCard>
<MudCardContent>
<MudTable Items="@filteredTasks" Hover="true" Breakpoint="Breakpoint.Sm" Loading="@loading">
@@ -381,61 +335,4 @@
if (result is { Canceled: false })
await LoadTasks();
}
private IReadOnlyList<TelegramPanel.Modules.ModuleTaskDefinition> ConfigurableTasks =>
Contributions.Tasks
.Select(x => x.Definition)
.Where(x => !IsSystemCategory(x.Category))
.Where(x => !string.IsNullOrWhiteSpace(x.CreateRoute))
.OrderBy(x => x.Order)
.ThenBy(x => x.DisplayName)
.ToList();
private static bool IsSystemCategory(string? category)
=> string.Equals((category ?? "").Trim(), "system", StringComparison.OrdinalIgnoreCase);
private async Task OpenTaskConfig(TelegramPanel.Modules.ModuleTaskDefinition def)
{
var route = (def.CreateRoute ?? "").Trim();
if (route.Length == 0)
return;
if (TryParseExtRoute(route, out var moduleId, out var pageKey))
{
var options = new DialogOptions { CloseOnEscapeKey = true, MaxWidth = MaxWidth.ExtraLarge, FullWidth = true };
var parameters = new DialogParameters
{
["ModuleId"] = moduleId,
["PageKey"] = pageKey
};
DialogService.Show<TelegramPanel.Web.Components.Dialogs.ModulePageDialog>("任务配置", parameters, options);
await Task.CompletedTask;
return;
}
// fallback:不是模块扩展页面则直接跳转
Navigation.NavigateTo(route);
}
private static bool TryParseExtRoute(string route, out string moduleId, out string pageKey)
{
moduleId = "";
pageKey = "";
route = (route ?? "").Trim();
if (!route.StartsWith("/ext/", StringComparison.OrdinalIgnoreCase))
return false;
var parts = route.Split('/', StringSplitOptions.RemoveEmptyEntries | StringSplitOptions.TrimEntries);
if (parts.Length < 3)
return false;
if (!string.Equals(parts[0], "ext", StringComparison.OrdinalIgnoreCase))
return false;
moduleId = parts[1];
pageKey = parts[2];
return moduleId.Length > 0 && pageKey.Length > 0;
}
}
+30 -1
View File
@@ -12,6 +12,7 @@ using TelegramPanel.Data;
using TelegramPanel.Web.Modules;
using TelegramPanel.Web.Services;
using Microsoft.Extensions.Logging;
using Microsoft.AspNetCore.DataProtection;
// 诊断:对某个目录下的 *.json/*.session 做一次“可转换/可校验”检查(不写数据库)
// 用法:dotnet run --project src/TelegramPanel.Web -- --diag-session-dir "D:/path/to/dir"
@@ -248,6 +249,27 @@ builder.Services.AddRazorComponents()
// MudBlazor
builder.Services.AddMudServices();
// DataProtection keys 持久化:避免容器重建/重启后出现 antiforgery token 无法解密
try
{
var configuredKeysPath = (builder.Configuration["DataProtection:KeysPath"] ?? "").Trim();
var keysPath = configuredKeysPath;
if (string.IsNullOrWhiteSpace(keysPath))
{
keysPath = Directory.Exists("/data")
? "/data/keys"
: Path.Combine(builder.Environment.ContentRootPath, "data-protection-keys");
}
Directory.CreateDirectory(keysPath);
builder.Services.AddDataProtection()
.PersistKeysToFileSystem(new DirectoryInfo(keysPath));
}
catch (Exception ex)
{
Console.Error.WriteLine($"Failed to configure DataProtection keys persistence: {ex.Message}");
}
// 反向代理支持(宝塔/Nginx/Caddy 等)
// 让应用正确识别外部访问的 Host/Proto,避免重定向到 http://localhost/...
builder.Services.Configure<ForwardedHeadersOptions>(options =>
@@ -543,7 +565,14 @@ if (!app.Environment.IsDevelopment())
app.UseForwardedHeaders();
app.UseHttpsRedirection();
// 仅在存在 HTTPS 端口/端点时启用重定向;否则会产生 “Failed to determine the https port for redirect.” 噪声
var httpsPort = app.Configuration["ASPNETCORE_HTTPS_PORT"];
var urls = app.Configuration["ASPNETCORE_URLS"] ?? "";
var hasHttpsEndpoint = !string.IsNullOrWhiteSpace(httpsPort)
|| urls.Contains("https://", StringComparison.OrdinalIgnoreCase)
|| !string.IsNullOrWhiteSpace(app.Configuration["Kestrel:Endpoints:Https:Url"]);
if (hasHttpsEndpoint)
app.UseHttpsRedirection();
// 静态文件(包括 MudBlazor 等 NuGet 包的静态 Web 资源)
app.UseStaticFiles();
@@ -1,6 +1,7 @@
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using System.Text.Json;
using TelegramPanel.Core.Services;
using TelegramPanel.Core.Services.Telegram;
@@ -14,13 +15,16 @@ public class BotAutoSyncBackgroundService : BackgroundService
private readonly IServiceScopeFactory _scopeFactory;
private readonly ILogger<BotAutoSyncBackgroundService> _logger;
private readonly IConfiguration _configuration;
private readonly BotUpdateHub _updateHub;
public BotAutoSyncBackgroundService(
IServiceScopeFactory scopeFactory,
BotUpdateHub updateHub,
IConfiguration configuration,
ILogger<BotAutoSyncBackgroundService> logger)
{
_scopeFactory = scopeFactory;
_updateHub = updateHub;
_configuration = configuration;
_logger = logger;
}
@@ -35,22 +39,24 @@ public class BotAutoSyncBackgroundService : BackgroundService
return;
}
// 秒级轮询(近似“监听”效果);可通过配置调整
var seconds = _configuration.GetValue("Telegram:BotAutoSyncIntervalSeconds", 5);
// 这里的 interval 用于:刷新 bot 列表 + 批量 drain 已收集的 updates(避免每条 update 都写 DB)
var seconds = _configuration.GetValue("Telegram:BotAutoSyncIntervalSeconds", 2);
if (seconds < 2) seconds = 2;
if (seconds > 60) seconds = 60;
var interval = TimeSpan.FromSeconds(seconds);
_logger.LogInformation("Bot auto sync started, interval {IntervalSeconds} seconds", seconds);
_logger.LogInformation("Bot auto sync started (via BotUpdateHub), interval {IntervalSeconds} seconds", seconds);
// 延迟一点,避免与启动时 DB 迁移抢资源
await Task.Delay(TimeSpan.FromSeconds(5), stoppingToken);
var subscriptions = new Dictionary<int, BotUpdateHub.BotUpdateSubscription>();
while (!stoppingToken.IsCancellationRequested)
{
try
{
await SyncOnceAsync(stoppingToken);
await SyncOnceAsync(subscriptions, stoppingToken);
}
catch (OperationCanceledException) when (stoppingToken.IsCancellationRequested)
{
@@ -63,9 +69,17 @@ public class BotAutoSyncBackgroundService : BackgroundService
await Task.Delay(interval, stoppingToken);
}
foreach (var sub in subscriptions.Values)
{
try { await sub.DisposeAsync(); }
catch { /* ignore */ }
}
}
private async Task SyncOnceAsync(CancellationToken cancellationToken)
private async Task SyncOnceAsync(
Dictionary<int, BotUpdateHub.BotUpdateSubscription> subscriptions,
CancellationToken cancellationToken)
{
using var scope = _scopeFactory.CreateScope();
var botManagement = scope.ServiceProvider.GetRequiredService<BotManagementService>();
@@ -75,20 +89,69 @@ public class BotAutoSyncBackgroundService : BackgroundService
if (bots.Count == 0)
return;
// 1) 确保订阅已建立(每个 botId 一个订阅即可)
var aliveBotIds = bots.Select(b => b.Id).ToHashSet();
foreach (var bot in bots)
{
cancellationToken.ThrowIfCancellationRequested();
if (subscriptions.ContainsKey(bot.Id))
continue;
try
{
var count = await botTelegram.SyncBotChannelsAsync(bot.Id, cancellationToken);
if (count > 0)
_logger.LogInformation("Bot auto sync: bot {BotId} synced {Count} channels", bot.Id, count);
else
_logger.LogDebug("Bot auto sync: bot {BotId} synced 0 channels", bot.Id);
subscriptions[bot.Id] = await _updateHub.SubscribeAsync(bot.Id, cancellationToken);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Bot auto sync failed for bot {BotId}", bot.Id);
_logger.LogWarning(ex, "Bot auto sync subscribe failed for bot {BotId}", bot.Id);
}
}
// 2) 清理已停用/已删除 bot 的订阅
foreach (var staleId in subscriptions.Keys.Where(id => !aliveBotIds.Contains(id)).ToList())
{
if (subscriptions.Remove(staleId, out var sub))
{
try { await sub.DisposeAsync(); }
catch { /* ignore */ }
}
}
// 3) 批量 drain:只取 my_chat_member,按 botId 分批应用(去重在 BotTelegramService 内部做)
foreach (var bot in bots)
{
cancellationToken.ThrowIfCancellationRequested();
if (!subscriptions.TryGetValue(bot.Id, out var sub))
continue;
var batch = new List<JsonElement>();
while (sub.Reader.TryRead(out var update))
{
if (update.ValueKind == System.Text.Json.JsonValueKind.Object
&& update.TryGetProperty("my_chat_member", out _))
{
batch.Add(update);
if (batch.Count >= 200)
break;
}
}
if (batch.Count == 0)
continue;
try
{
var count = await botTelegram.ApplyMyChatMemberUpdatesAsync(bot.Id, batch, cancellationToken);
if (count > 0)
_logger.LogInformation("Bot auto sync: bot {BotId} applied {Count} updates", bot.Id, count);
else
_logger.LogDebug("Bot auto sync: bot {BotId} applied 0 updates", bot.Id);
}
catch (Exception ex)
{
_logger.LogWarning(ex, "Bot auto sync apply failed for bot {BotId}", bot.Id);
}
}
}
+1
View File
@@ -48,6 +48,7 @@ docker run --rm `
-w "/src" `
mcr.microsoft.com/dotnet/sdk:8.0 `
dotnet publish "$projectContainer" -c Release -o "$publishContainer" /p:UseAppHost=false
if ($LASTEXITCODE -ne 0) { throw "dotnet publish 失败(退出码:$LASTEXITCODE)" }
$stagingLib = Join-Path $stagingHost "lib"
New-Item -ItemType Directory -Force -Path $stagingLib | Out-Null