av
c3acb50fcc
- авторизация бота ушла из NewBot в Run: повторяется в фоне с backoff от 30 секунд до 15 минут, пока не появится связь - отказ по токену (401/404) повторами не лечится — бот просто выключается, веб-часть продолжает работать - у клиента Telegram ограничен только dial (15 c): long polling и заливка вложений должны оставаться без общего таймаута
440 lines
12 KiB
Go
440 lines
12 KiB
Go
package telegram
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"log/slog"
|
||
"net"
|
||
"net/http"
|
||
"strconv"
|
||
"strings"
|
||
"time"
|
||
|
||
tgbotapi "github.com/go-telegram-bot-api/telegram-bot-api/v5"
|
||
|
||
"git.vakhrushev.me/av/remembos/internal/config"
|
||
"git.vakhrushev.me/av/remembos/internal/media"
|
||
"git.vakhrushev.me/av/remembos/internal/memory"
|
||
"git.vakhrushev.me/av/remembos/internal/memos"
|
||
"git.vakhrushev.me/av/remembos/internal/search"
|
||
)
|
||
|
||
// Интервалы повторной авторизации в Telegram: от 30 секунд с удвоением до 15 минут.
|
||
const (
|
||
connectRetryMin = 30 * time.Second
|
||
connectRetryMax = 15 * time.Minute
|
||
)
|
||
|
||
// Bot sends a daily memory via Telegram.
|
||
type Bot struct {
|
||
token string
|
||
api *tgbotapi.BotAPI // назначается в connect, до этого бот не работает
|
||
service *memory.Service
|
||
client *memos.Client
|
||
chatID int64
|
||
sendAt string // "HH:MM"
|
||
publicURL string
|
||
loc *time.Location
|
||
logger *slog.Logger
|
||
allowLoadMore bool
|
||
}
|
||
|
||
// NewBot creates a new Telegram bot. В сеть не ходит: авторизация происходит
|
||
// в Run, чтобы недоступный api.telegram.org не мешал старту приложения.
|
||
func NewBot(
|
||
cfg config.TelegramConfig,
|
||
service *memory.Service,
|
||
client *memos.Client,
|
||
memosURL, publicURL string,
|
||
allowLoadMore bool,
|
||
loc *time.Location,
|
||
logger *slog.Logger,
|
||
) *Bot {
|
||
pub := publicURL
|
||
if pub == "" {
|
||
pub = memosURL
|
||
}
|
||
pub = strings.TrimRight(pub, "/")
|
||
|
||
return &Bot{
|
||
token: cfg.Token,
|
||
service: service,
|
||
client: client,
|
||
chatID: cfg.ChatID,
|
||
sendAt: cfg.SendAt,
|
||
publicURL: pub,
|
||
loc: loc,
|
||
logger: logger,
|
||
allowLoadMore: allowLoadMore,
|
||
}
|
||
}
|
||
|
||
// Run authorizes the bot and starts the scheduling loop. It blocks until ctx is cancelled.
|
||
func (b *Bot) Run(ctx context.Context) {
|
||
if !b.connect(ctx) {
|
||
return
|
||
}
|
||
|
||
if b.allowLoadMore {
|
||
go b.listenForCommands(ctx)
|
||
}
|
||
|
||
for {
|
||
next := b.nextSendTime()
|
||
delay := time.Until(next)
|
||
b.logger.Info("next telegram send scheduled", "at", next.Format("2006-01-02 15:04:05"), "in", delay.Round(time.Second))
|
||
|
||
select {
|
||
case <-ctx.Done():
|
||
b.logger.Info("telegram bot stopped")
|
||
return
|
||
case <-time.After(delay):
|
||
b.sendDaily(ctx)
|
||
}
|
||
}
|
||
}
|
||
|
||
// connect авторизует бота, повторяя попытки до успеха или отмены ctx.
|
||
// Сеть до api.telegram.org может быть недоступна (у сервера бывает заблокирован
|
||
// исходящий трафик) — это не повод ронять приложение: веб-часть работает без бота,
|
||
// а бот подхватится сам, когда связь появится.
|
||
func (b *Bot) connect(ctx context.Context) bool {
|
||
delay := connectRetryMin
|
||
|
||
for attempt := 1; ; attempt++ {
|
||
api, err := tgbotapi.NewBotAPIWithClient(b.token, tgbotapi.APIEndpoint, newHTTPClient())
|
||
if err == nil {
|
||
b.api = api
|
||
b.logger.Info("telegram bot authorized", "username", api.Self.UserName, "attempt", attempt)
|
||
return true
|
||
}
|
||
|
||
// Отказ самого Telegram (неверный токен) повторами не лечится — выключаем бота.
|
||
var apiErr *tgbotapi.Error
|
||
if errors.As(err, &apiErr) && (apiErr.Code == http.StatusUnauthorized || apiErr.Code == http.StatusNotFound) {
|
||
b.logger.Error("telegram bot disabled: token rejected", "error", err)
|
||
return false
|
||
}
|
||
|
||
b.logger.Warn("telegram authorization failed, will retry",
|
||
// Duration в JSON-логе иначе печатается наносекундами.
|
||
"attempt", attempt, "retry_in", delay.String(), "error", err)
|
||
|
||
select {
|
||
case <-ctx.Done():
|
||
b.logger.Info("telegram bot stopped before authorization")
|
||
return false
|
||
case <-time.After(delay):
|
||
}
|
||
delay = min(delay*2, connectRetryMax)
|
||
}
|
||
}
|
||
|
||
// newHTTPClient — клиент для Telegram API. Общего таймаута намеренно нет:
|
||
// long polling ждёт до 60 секунд, а заливка вложений может идти долго.
|
||
// Ограничено только установление соединения, чтобы неудачная попытка
|
||
// авторизации не висела минутами на TCP-ретраях.
|
||
func newHTTPClient() *http.Client {
|
||
return &http.Client{
|
||
Transport: &http.Transport{
|
||
Proxy: http.ProxyFromEnvironment,
|
||
DialContext: (&net.Dialer{Timeout: 15 * time.Second, KeepAlive: 30 * time.Second}).DialContext,
|
||
TLSHandshakeTimeout: 15 * time.Second,
|
||
},
|
||
}
|
||
}
|
||
|
||
// nextSendTime returns the next occurrence of sendAt in the configured timezone.
|
||
func (b *Bot) nextSendTime() time.Time {
|
||
now := time.Now().In(b.loc)
|
||
|
||
parts := strings.SplitN(b.sendAt, ":", 2)
|
||
hour := 9
|
||
minute := 0
|
||
if len(parts) == 2 {
|
||
if h, err := strconv.Atoi(parts[0]); err == nil {
|
||
hour = h
|
||
}
|
||
if m, err := strconv.Atoi(parts[1]); err == nil {
|
||
minute = m
|
||
}
|
||
}
|
||
|
||
target := time.Date(now.Year(), now.Month(), now.Day(), hour, minute, 0, 0, b.loc)
|
||
if !target.After(now) {
|
||
target = target.AddDate(0, 0, 1)
|
||
}
|
||
return target
|
||
}
|
||
|
||
// sendDaily fetches today's memory and sends it.
|
||
func (b *Bot) sendDaily(ctx context.Context) {
|
||
b.logger.Info("sending daily memory via telegram")
|
||
|
||
mem, err := b.service.GetTodayMemory(ctx)
|
||
if err != nil {
|
||
b.logger.Error("failed to get today memory", "error", err)
|
||
return
|
||
}
|
||
|
||
b.sendMemory(ctx, mem)
|
||
}
|
||
|
||
// sendMemory formats and sends a memory via Telegram.
|
||
func (b *Bot) sendMemory(ctx context.Context, mem *search.Memory) {
|
||
if mem == nil {
|
||
b.logger.Info("no memory to send, skipping")
|
||
return
|
||
}
|
||
|
||
text := formatMemory(mem, b.publicURL)
|
||
imageAtts, videoAtts, audioAtts := mediaAttachments(mem.Memo)
|
||
|
||
var images []mediaFile
|
||
var skipped bool
|
||
if len(imageAtts) > 0 {
|
||
images, skipped = b.downloadAndCompressImages(ctx, imageAtts)
|
||
}
|
||
|
||
videos := b.downloadFiles(ctx, videoAtts)
|
||
audios := b.downloadFiles(ctx, audioAtts)
|
||
|
||
if skipped {
|
||
text += "\n\nПоказаны не все вложения"
|
||
}
|
||
|
||
if err := b.sendWithRetry(ctx, text, images, videos, audios); err != nil {
|
||
b.logger.Error("failed to send telegram message after retries", "error", err)
|
||
}
|
||
}
|
||
|
||
// listenForCommands polls for Telegram updates and handles /more commands.
|
||
func (b *Bot) listenForCommands(ctx context.Context) {
|
||
u := tgbotapi.NewUpdate(0)
|
||
u.Timeout = 60
|
||
|
||
updates := b.api.GetUpdatesChan(u)
|
||
|
||
for {
|
||
select {
|
||
case <-ctx.Done():
|
||
return
|
||
case update, ok := <-updates:
|
||
if !ok {
|
||
return
|
||
}
|
||
if update.Message == nil || !update.Message.IsCommand() {
|
||
continue
|
||
}
|
||
if update.Message.Chat.ID != b.chatID {
|
||
continue
|
||
}
|
||
if update.Message.Command() == "more" {
|
||
b.handleMore(ctx)
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// handleMore loads a new memory and sends it.
|
||
func (b *Bot) handleMore(ctx context.Context) {
|
||
b.logger.Info("handling /more command")
|
||
|
||
mem, err := b.service.LoadNewMemory(ctx)
|
||
if err != nil {
|
||
b.logger.Error("failed to load new memory", "error", err)
|
||
return
|
||
}
|
||
|
||
b.sendMemory(ctx, mem)
|
||
}
|
||
|
||
type mediaFile struct {
|
||
filename string
|
||
data []byte
|
||
}
|
||
|
||
// downloadAndCompressImages downloads image attachments and compresses them if needed.
|
||
// Returns downloaded files and whether any were skipped due to errors.
|
||
func (b *Bot) downloadAndCompressImages(ctx context.Context, attachments []memos.Attachment) ([]mediaFile, bool) {
|
||
files := make([]mediaFile, 0, len(attachments))
|
||
var skipped bool
|
||
for _, att := range attachments {
|
||
data, err := b.client.DownloadAttachment(ctx, att)
|
||
if err != nil {
|
||
b.logger.Warn("failed to download attachment, skipping", "name", att.Name, "error", err)
|
||
skipped = true
|
||
continue
|
||
}
|
||
|
||
data, filename, err := media.CompressImage(ctx, data, att.Filename)
|
||
if err != nil {
|
||
b.logger.Warn("failed to compress image, skipping", "name", att.Name, "error", err)
|
||
skipped = true
|
||
continue
|
||
}
|
||
|
||
files = append(files, mediaFile{filename: filename, data: data})
|
||
}
|
||
return files, skipped
|
||
}
|
||
|
||
// downloadFiles downloads attachments, skipping failures.
|
||
func (b *Bot) downloadFiles(ctx context.Context, attachments []memos.Attachment) []mediaFile {
|
||
files := make([]mediaFile, 0, len(attachments))
|
||
for _, att := range attachments {
|
||
data, err := b.client.DownloadAttachment(ctx, att)
|
||
if err != nil {
|
||
b.logger.Warn("failed to download attachment, skipping", "name", att.Name, "error", err)
|
||
continue
|
||
}
|
||
files = append(files, mediaFile{filename: att.Filename, data: data})
|
||
}
|
||
return files
|
||
}
|
||
|
||
// sendWithRetry attempts to send the message with up to 3 retries.
|
||
func (b *Bot) sendWithRetry(ctx context.Context, text string, images, videos, audios []mediaFile) error {
|
||
backoffs := []time.Duration{30 * time.Second, 60 * time.Second, 120 * time.Second}
|
||
var lastErr error
|
||
|
||
for attempt := range 3 {
|
||
lastErr = b.sendWithMedia(text, images, videos, audios)
|
||
if lastErr == nil {
|
||
return nil
|
||
}
|
||
b.logger.Warn("telegram send failed", "attempt", attempt+1, "error", lastErr)
|
||
|
||
if attempt < 2 {
|
||
select {
|
||
case <-ctx.Done():
|
||
return ctx.Err()
|
||
case <-time.After(backoffs[attempt]):
|
||
}
|
||
}
|
||
}
|
||
return lastErr
|
||
}
|
||
|
||
// sendWithMedia executes the actual Telegram API calls for text, images, videos, and audios.
|
||
func (b *Bot) sendWithMedia(text string, images, videos, audios []mediaFile) error {
|
||
hasMedia := len(images) > 0 || len(videos) > 0 || len(audios) > 0
|
||
|
||
// Send text
|
||
switch {
|
||
case !hasMedia:
|
||
return b.sendTextParts(splitText(text, maxMessageLen))
|
||
|
||
case len(text) <= maxCaptionLen && len(images) > 0:
|
||
// Short text — use as caption on image(s)
|
||
if len(images) == 1 {
|
||
if err := b.sendPhoto(images[0], text); err != nil {
|
||
return err
|
||
}
|
||
} else {
|
||
if err := b.sendMediaGroup(images, text); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
default:
|
||
// Long text or no images — send text first, then all media
|
||
if err := b.sendTextParts(splitText(text, maxMessageLen)); err != nil {
|
||
return err
|
||
}
|
||
// Send images without caption
|
||
if len(images) == 1 {
|
||
if err := b.sendPhoto(images[0], ""); err != nil {
|
||
return err
|
||
}
|
||
} else if len(images) > 1 {
|
||
if err := b.sendMediaGroup(images, ""); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
}
|
||
|
||
// Send videos one by one
|
||
for _, v := range videos {
|
||
if err := b.sendVideo(v); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
// Send audios one by one
|
||
for _, a := range audios {
|
||
if err := b.sendAudio(a); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
return nil
|
||
}
|
||
|
||
func (b *Bot) sendTextParts(parts []string) error {
|
||
for _, part := range parts {
|
||
if err := b.sendText(part); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func (b *Bot) sendText(text string) error {
|
||
msg := tgbotapi.NewMessage(b.chatID, text)
|
||
msg.ParseMode = tgbotapi.ModeHTML
|
||
msg.DisableWebPagePreview = true
|
||
_, err := b.api.Send(msg)
|
||
return err
|
||
}
|
||
|
||
func (b *Bot) sendPhoto(img mediaFile, caption string) error {
|
||
photo := tgbotapi.NewPhoto(b.chatID, tgbotapi.FileBytes{
|
||
Name: img.filename,
|
||
Bytes: img.data,
|
||
})
|
||
if caption != "" {
|
||
photo.Caption = caption
|
||
photo.ParseMode = tgbotapi.ModeHTML
|
||
}
|
||
_, err := b.api.Send(photo)
|
||
return err
|
||
}
|
||
|
||
func (b *Bot) sendVideo(v mediaFile) error {
|
||
video := tgbotapi.NewVideo(b.chatID, tgbotapi.FileBytes{
|
||
Name: v.filename,
|
||
Bytes: v.data,
|
||
})
|
||
_, err := b.api.Send(video)
|
||
return err
|
||
}
|
||
|
||
func (b *Bot) sendAudio(a mediaFile) error {
|
||
audio := tgbotapi.NewAudio(b.chatID, tgbotapi.FileBytes{
|
||
Name: a.filename,
|
||
Bytes: a.data,
|
||
})
|
||
_, err := b.api.Send(audio)
|
||
return err
|
||
}
|
||
|
||
func (b *Bot) sendMediaGroup(images []mediaFile, caption string) error {
|
||
media := make([]interface{}, len(images))
|
||
for i, img := range images {
|
||
photo := tgbotapi.NewInputMediaPhoto(tgbotapi.FileBytes{
|
||
Name: img.filename,
|
||
Bytes: img.data,
|
||
})
|
||
if i == 0 && caption != "" {
|
||
photo.Caption = caption
|
||
photo.ParseMode = tgbotapi.ModeHTML
|
||
}
|
||
media[i] = photo
|
||
}
|
||
|
||
mg := tgbotapi.NewMediaGroup(b.chatID, media)
|
||
_, err := b.api.SendMediaGroup(mg)
|
||
return err
|
||
}
|