carrydela/ProxyPool
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

frist

Browse Source
This commit is contained in:
dela
2026-01-31 22:53:12 +08:00
commit bc639cf460
30 changed files with 6836 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
internal/telegram/bot.go Normal file
View File

@@ -0,0 +1,197 @@
package telegram
import (
"context"
"log/slog"
"sync"
"time"
"proxyrotator/internal/config"
"proxyrotator/internal/store"
tele "gopkg.in/telebot.v3"
)
// Bot Telegram Bot 管理器
type Bot struct {
mu sync.RWMutex
bot *tele.Bot
cfg *config.Config
store store.ProxyStore
scheduler *Scheduler
notifier *Notifier
running bool
stopChan chan struct{}
}
// Status Bot 状态
type Status struct {
Running bool `json:"running"`
Connected bool `json:"connected"`
Username string `json:"username,omitempty"`
}
// NewBot 创建 Bot 实例
func NewBot(cfg *config.Config, proxyStore store.ProxyStore) *Bot {
return &Bot{
cfg: cfg,
store: proxyStore,
stopChan: make(chan struct{}),
}
}
// Start 启动 Bot
func (b *Bot) Start(ctx context.Context) error {
b.mu.Lock()
defer b.mu.Unlock()
if b.running {
return nil
}
if b.cfg.TelegramBotToken == "" {
slog.Info("telegram bot token not configured, skipping")
return nil
}
return b.startInternal()
}
// startInternal 内部启动（需要持有锁）
func (b *Bot) startInternal() error {
pref := tele.Settings{
Token: b.cfg.TelegramBotToken,
Poller: &tele.LongPoller{Timeout: 10 * time.Second},
}
bot, err := tele.NewBot(pref)
if err != nil {
slog.Error("failed to create telegram bot", "error", err)
return err
}
b.bot = bot
b.notifier = NewNotifier(bot, b.cfg.TelegramNotifyChatID)
b.scheduler = NewScheduler(b.store, b.notifier, b.cfg)
// 注册命令处理器
b.registerCommands(b.cfg.TelegramAdminIDs)
// 启动调度器
b.scheduler.Start()
// 注册命令菜单
commands := []tele.Command{
{Text: "stats", Description: "查看代理池统计"},
{Text: "groups", Description: "查看分组统计"},
{Text: "get", Description: "获取可用代理 (默认1个，如 /get 5)"},
{Text: "import", Description: "导入代理 (如 /import groupname)"},
{Text: "test", Description: "触发测活 (如 /test groupname)"},
{Text: "purge", Description: "清理死代理"},
{Text: "help", Description: "显示帮助信息"},
}
if err := bot.SetCommands(commands); err != nil {
slog.Warn("failed to set bot commands", "error", err)
}
// 启动 Bot
b.stopChan = make(chan struct{})
go func() {
slog.Info("telegram bot started", "username", bot.Me.Username)
bot.Start()
}()
b.running = true
return nil
}
// Stop 停止 Bot
func (b *Bot) Stop() {
b.mu.Lock()
defer b.mu.Unlock()
b.stopInternal()
}
// stopInternal 内部停止（需要持有锁）
func (b *Bot) stopInternal() {
if !b.running {
return
}
if b.scheduler != nil {
b.scheduler.Stop()
}
if b.bot != nil {
b.bot.Stop()
slog.Info("telegram bot stopped")
}
close(b.stopChan)
b.running = false
}
// Status 获取 Bot 状态
func (b *Bot) Status() Status {
b.mu.RLock()
defer b.mu.RUnlock()
status := Status{
Running: b.running,
}
if b.bot != nil && b.running {
status.Connected = true
status.Username = b.bot.Me.Username
}
return status
}
// TriggerTest 手动触发测活
func (b *Bot) TriggerTest(ctx context.Context) error {
b.mu.RLock()
defer b.mu.RUnlock()
if b.scheduler == nil {
return nil
}
return b.scheduler.RunTest(ctx)
}
// registerCommands 注册命令
func (b *Bot) registerCommands(adminIDs []int64) {
// 管理员权限中间件
adminOnly := func(next tele.HandlerFunc) tele.HandlerFunc {
return func(c tele.Context) error {
if len(adminIDs) == 0 {
return next(c)
}
userID := c.Sender().ID
for _, id := range adminIDs {
if id == userID {
return next(c)
}
}
return c.Send("⛔ 无权限访问")
}
}
// 创建命令处理器
cmds := NewCommands(b.store, b.scheduler)
b.bot.Handle("/start", adminOnly(cmds.HandleStart))
b.bot.Handle("/help", adminOnly(cmds.HandleHelp))
b.bot.Handle("/stats", adminOnly(cmds.HandleStats))
b.bot.Handle("/groups", adminOnly(cmds.HandleGroups))
b.bot.Handle("/get", adminOnly(cmds.HandleGet))
b.bot.Handle("/test", adminOnly(cmds.HandleTest))
b.bot.Handle("/purge", adminOnly(cmds.HandlePurge))
b.bot.Handle("/import", adminOnly(cmds.HandleImport))
b.bot.Handle(tele.OnDocument, adminOnly(cmds.HandleDocument))
b.bot.Handle(tele.OnText, adminOnly(cmds.HandleText))
}

365
internal/telegram/commands.go Normal file
View File

@@ -0,0 +1,365 @@
package telegram
import (
"context"
"encoding/json"
"fmt"
"io"
"net/http"
"net/url"
"strconv"
"strings"
"time"
"proxyrotator/internal/importer"
"proxyrotator/internal/model"
"proxyrotator/internal/store"
tele "gopkg.in/telebot.v3"
)
// Commands 命令处理器
type Commands struct {
store store.ProxyStore
scheduler *Scheduler
importer *importer.Importer
// 导入状态
importState map[int64]*importSession
}
type importSession struct {
Group string
Tags []string
}
// NewCommands 创建命令处理器
func NewCommands(store store.ProxyStore, scheduler *Scheduler) *Commands {
return &Commands{
store: store,
scheduler: scheduler,
importer: importer.NewImporter(),
importState: make(map[int64]*importSession),
}
}
// HandleStart /start 命令
func (c *Commands) HandleStart(ctx tele.Context) error {
return ctx.Send(`🚀 *ProxyRotator Bot*
欢迎使用代理池管理机器人！
使用 /help 查看可用命令`, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// HandleHelp /help 命令
func (c *Commands) HandleHelp(ctx tele.Context) error {
help := `📖 *可用命令*
*查询类*
/stats - 代理池统计（总数/存活/死亡/未知）
/groups - 分组统计
/get [n] - 获取 n 个可用代理（默认 5）
*操作类*
/import [group] - 导入代理（之后发送文本或文件）
/test [group] - 触发测活
/purge - 清理死代理
*其他*
/help - 显示帮助信息`
return ctx.Send(help, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// HandleStats /stats 命令
func (c *Commands) HandleStats(ctx tele.Context) error {
stats, err := c.store.GetStats(context.Background())
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 获取统计失败: %v", err))
}
alive := stats.ByStatus[model.StatusAlive]
dead := stats.ByStatus[model.StatusDead]
unknown := stats.ByStatus[model.StatusUnknown]
var alivePercent float64
if stats.Total > 0 {
alivePercent = float64(alive) / float64(stats.Total) * 100
}
msg := fmt.Sprintf(`📊 *代理池统计*
*总数:* %d
*存活:* %d (%.1f%%)
*死亡:* %d
*未知:* %d
*禁用:* %d
*平均延迟:* %d ms
*平均分数:* %.1f`,
stats.Total,
alive, alivePercent,
dead,
unknown,
stats.Disabled,
stats.AvgLatencyMs,
stats.AvgScore,
)
return ctx.Send(msg, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// HandleGroups /groups 命令
func (c *Commands) HandleGroups(ctx tele.Context) error {
stats, err := c.store.GetStats(context.Background())
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 获取统计失败: %v", err))
}
if len(stats.ByGroup) == 0 {
return ctx.Send("📁 暂无分组数据")
}
var sb strings.Builder
sb.WriteString("📁 *分组统计*\n\n")
for group, count := range stats.ByGroup {
sb.WriteString(fmt.Sprintf("• `%s`: %d\n", group, count))
}
return ctx.Send(sb.String(), &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// IPInfo ipinfo.io 返回结构
type IPInfo struct {
IP string `json:"ip"`
City string `json:"city"`
Region string `json:"region"`
Country string `json:"country"`
Org string `json:"org"`
}
// HandleGet /get [n] 命令
func (c *Commands) HandleGet(ctx tele.Context) error {
n := 1
args := ctx.Args()
if len(args) > 0 {
if parsed, err := strconv.Atoi(args[0]); err == nil && parsed > 0 {
n = parsed
}
}
if n > 20 {
n = 20
}
proxies, err := c.store.List(context.Background(), model.ProxyQuery{
StatusIn: []model.ProxyStatus{model.StatusAlive},
OnlyEnabled: true,
OrderBy: "random",
Limit: n,
})
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 获取代理失败: %v", err))
}
if len(proxies) == 0 {
return ctx.Send("😢 没有可用代理")
}
var sb strings.Builder
sb.WriteString(fmt.Sprintf("🔗 *可用代理 (%d)*\n\n", len(proxies)))
for _, p := range proxies {
var proxyURL string
if p.Username != "" {
proxyURL = fmt.Sprintf("%s://%s:%s@%s:%d", p.Protocol, p.Username, p.Password, p.Host, p.Port)
} else {
proxyURL = fmt.Sprintf("%s://%s:%d", p.Protocol, p.Host, p.Port)
}
// 获取 IP 位置信息
ipInfo := fetchIPInfo(proxyURL)
sb.WriteString(fmt.Sprintf("`%s`\n", proxyURL))
if ipInfo != nil {
location := fmt.Sprintf("%s, %s, %s", ipInfo.City, ipInfo.Region, ipInfo.Country)
sb.WriteString(fmt.Sprintf(" 📍 %s | %s\n", location, ipInfo.Org))
} else {
sb.WriteString(" 📍 位置获取失败\n")
}
sb.WriteString("\n")
}
return ctx.Send(sb.String(), &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// fetchIPInfo 通过代理获取 IP 信息
func fetchIPInfo(proxyURL string) *IPInfo {
proxy, err := url.Parse(proxyURL)
if err != nil {
return nil
}
client := &http.Client{
Transport: &http.Transport{
Proxy: http.ProxyURL(proxy),
},
Timeout: 10 * time.Second,
}
resp, err := client.Get("https://ipinfo.io/json")
if err != nil {
return nil
}
defer resp.Body.Close()
var info IPInfo
if err := json.NewDecoder(resp.Body).Decode(&info); err != nil {
return nil
}
return &info
}
// HandleTest /test [group] 命令
func (c *Commands) HandleTest(ctx tele.Context) error {
if c.scheduler == nil {
return ctx.Send("❌ 调度器未初始化")
}
group := ""
args := ctx.Args()
if len(args) > 0 {
group = args[0]
}
_ = ctx.Send("🔄 正在执行测活...")
err := c.scheduler.RunTestWithGroup(context.Background(), group)
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 测活失败: %v", err))
}
return ctx.Send("✅ 测活完成")
}
// HandlePurge /purge 命令
func (c *Commands) HandlePurge(ctx tele.Context) error {
deleted, err := c.store.DeleteMany(context.Background(), model.BulkDeleteRequest{
Status: model.StatusDead,
})
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 清理失败: %v", err))
}
return ctx.Send(fmt.Sprintf("🗑️ 已清理 %d 个死代理", deleted))
}
// HandleImport /import [group] 命令
func (c *Commands) HandleImport(ctx tele.Context) error {
group := "default"
args := ctx.Args()
if len(args) > 0 {
group = args[0]
}
userID := ctx.Sender().ID
c.importState[userID] = &importSession{
Group: group,
Tags: []string{"telegram"},
}
return ctx.Send(fmt.Sprintf(`📥 *导入模式已开启*
分组: `+"`%s`"+`
请发送代理列表（文本或文件），支持格式:
• host:port
• host:port:user:pass
• protocol://host:port
• protocol://user:pass@host:port
发送 /cancel 取消导入`, group), &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}
// HandleDocument 处理文件上传
func (c *Commands) HandleDocument(ctx tele.Context) error {
userID := ctx.Sender().ID
session, ok := c.importState[userID]
if !ok {
return nil // 不在导入模式，忽略
}
doc := ctx.Message().Document
if doc == nil {
return nil
}
// 下载文件
reader, err := ctx.Bot().File(&doc.File)
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 获取文件失败: %v", err))
}
defer reader.Close()
content, err := io.ReadAll(reader)
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 读取文件失败: %v", err))
}
return c.doImport(ctx, session, string(content))
}
// HandleText 处理文本消息（用于导入）
func (c *Commands) HandleText(ctx tele.Context) error {
userID := ctx.Sender().ID
session, ok := c.importState[userID]
if !ok {
return nil // 不在导入模式，忽略
}
text := ctx.Text()
if text == "/cancel" {
delete(c.importState, userID)
return ctx.Send("❌ 已取消导入")
}
// 检查是否像代理格式
if !strings.Contains(text, ":") {
return nil // 不像代理，忽略
}
return c.doImport(ctx, session, text)
}
// doImport 执行导入
func (c *Commands) doImport(ctx tele.Context, session *importSession, text string) error {
userID := ctx.Sender().ID
defer delete(c.importState, userID)
input := model.ImportInput{
Group: session.Group,
Tags: session.Tags,
}
proxies, invalid := c.importer.ParseText(context.Background(), input, text)
if len(proxies) == 0 {
return ctx.Send(fmt.Sprintf("❌ 未解析到有效代理\n无效行: %d", len(invalid)))
}
imported, duplicated, err := c.store.UpsertMany(context.Background(), proxies)
if err != nil {
return ctx.Send(fmt.Sprintf("❌ 导入失败: %v", err))
}
msg := fmt.Sprintf(`✅ *导入完成*
• 新增: %d
• 重复: %d
• 无效: %d
• 分组: `+"`%s`",
imported, duplicated, len(invalid), session.Group)
return ctx.Send(msg, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
}

83
internal/telegram/notifier.go Normal file
View File

@@ -0,0 +1,83 @@
package telegram
import (
"context"
"fmt"
"log/slog"
"strconv"
tele "gopkg.in/telebot.v3"
)
// Notifier 告警通知器
type Notifier struct {
bot *tele.Bot
chatID string
}
// NewNotifier 创建通知器
func NewNotifier(bot *tele.Bot, chatID string) *Notifier {
return &Notifier{
bot: bot,
chatID: chatID,
}
}
// SendAlert 发送告警
func (n *Notifier) SendAlert(ctx context.Context, alive, dead, total int, alivePercent float64) {
if n.chatID == "" {
slog.Warn("notify_chat_id not configured, skipping alert")
return
}
chatID, err := strconv.ParseInt(n.chatID, 10, 64)
if err != nil {
slog.Error("invalid chat_id", "chat_id", n.chatID, "error", err)
return
}
msg := fmt.Sprintf(`🚨 *代理池告警*
存活率低于阈值！
*统计:*
• 存活: %d (%.1f%%)
• 死亡: %d
• 总数: %d
请及时补充代理或检查网络状况。`, alive, alivePercent, dead, total)
chat, err := n.bot.ChatByID(chatID)
if err != nil {
slog.Error("failed to get chat", "chat_id", n.chatID, "error", err)
return
}
_, err = n.bot.Send(chat, msg, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
if err != nil {
slog.Error("failed to send alert", "error", err)
return
}
slog.Info("alert sent", "chat_id", n.chatID, "alive_percent", alivePercent)
}
// SendMessage 发送普通消息
func (n *Notifier) SendMessage(ctx context.Context, message string) error {
if n.chatID == "" {
return nil
}
chatID, err := strconv.ParseInt(n.chatID, 10, 64)
if err != nil {
return err
}
chat, err := n.bot.ChatByID(chatID)
if err != nil {
return err
}
_, err = n.bot.Send(chat, message, &tele.SendOptions{ParseMode: tele.ModeMarkdown})
return err
}

167
internal/telegram/scheduler.go Normal file
View File

@@ -0,0 +1,167 @@
package telegram
import (
"context"
"log/slog"
"sync"
"time"
"proxyrotator/internal/config"
"proxyrotator/internal/model"
"proxyrotator/internal/store"
"proxyrotator/internal/tester"
)
// Scheduler 定时测活调度器
type Scheduler struct {
mu sync.Mutex
store store.ProxyStore
notifier *Notifier
tester *tester.HTTPTester
cfg *config.Config
ticker *time.Ticker
stopChan chan struct{}
running bool
}
// NewScheduler 创建调度器
func NewScheduler(store store.ProxyStore, notifier *Notifier, cfg *config.Config) *Scheduler {
return &Scheduler{
store: store,
notifier: notifier,
tester: tester.NewHTTPTester(),
cfg: cfg,
stopChan: make(chan struct{}),
}
}
// Start 启动调度器
func (s *Scheduler) Start() {
s.mu.Lock()
defer s.mu.Unlock()
if s.running {
return
}
interval := time.Duration(s.cfg.TelegramTestIntervalMin) * time.Minute
if interval < 5*time.Minute {
interval = 5 * time.Minute
}
s.ticker = time.NewTicker(interval)
s.stopChan = make(chan struct{})
s.running = true
go s.loop()
slog.Info("telegram scheduler started", "interval", interval)
}
// Stop 停止调度器
func (s *Scheduler) Stop() {
s.mu.Lock()
defer s.mu.Unlock()
if !s.running {
return
}
if s.ticker != nil {
s.ticker.Stop()
}
close(s.stopChan)
s.running = false
slog.Info("telegram scheduler stopped")
}
// loop 调度循环
func (s *Scheduler) loop() {
for {
select {
case <-s.stopChan:
return
case <-s.ticker.C:
if err := s.RunTest(context.Background()); err != nil {
slog.Error("scheduled test failed", "error", err)
}
}
}
}
// RunTest 执行测活（所有分组）
func (s *Scheduler) RunTest(ctx context.Context) error {
return s.RunTestWithGroup(ctx, "")
}
// RunTestWithGroup 执行测活（指定分组）
func (s *Scheduler) RunTestWithGroup(ctx context.Context, group string) error {
slog.Info("running scheduled proxy test", "group", group)
// 获取待测试代理
query := model.ProxyQuery{
Group: group,
StatusIn: []model.ProxyStatus{model.StatusUnknown, model.StatusAlive},
OnlyEnabled: true,
Limit: 1000,
}
proxies, err := s.store.List(ctx, query)
if err != nil {
return err
}
if len(proxies) == 0 {
slog.Info("no proxies to test")
return nil
}
// 构建测试规格
spec := model.TestSpec{
URL: s.cfg.TelegramTestURL,
Method: "GET",
Timeout: time.Duration(s.cfg.TelegramTestTimeoutMs) * time.Millisecond,
}
// 执行测试
results := s.tester.TestBatch(ctx, proxies, spec, 50)
// 统计结果
alive, dead := 0, 0
for _, r := range results {
now := r.CheckedAt
if r.OK {
alive++
status := model.StatusAlive
_ = s.store.UpdateHealth(ctx, r.ProxyID, model.HealthPatch{
Status: &status,
ScoreDelta: 1,
SuccessInc: 1,
LatencyMs: &r.LatencyMs,
CheckedAt: &now,
})
} else {
dead++
status := model.StatusDead
_ = s.store.UpdateHealth(ctx, r.ProxyID, model.HealthPatch{
Status: &status,
ScoreDelta: -3,
FailInc: 1,
CheckedAt: &now,
})
}
}
slog.Info("scheduled test completed", "tested", len(results), "alive", alive, "dead", dead)
// 检查是否需要告警
total := len(results)
if total > 0 {
alivePercent := float64(alive) / float64(total) * 100
if alivePercent < float64(s.cfg.TelegramAlertThreshold) {
s.notifier.SendAlert(ctx, alive, dead, total, alivePercent)
}
}
return nil
}