relay.go
Rama main · réplica de lectura del repositorio original.
package main
import (
"context"
"fmt"
"log/slog"
"net/http"
"slices"
"strconv"
"strings"
"time"
"github.com/go-telegram/bot"
"github.com/go-telegram/bot/models"
"github.com/max-messenger/max-bot-api-client-go/v2/model"
)
// source is a Telegram channel together with the MAX channels its posts go to.
// Each source has its own queue: order matters within a channel only, and a
// long video upload in one channel must not hold back the others.
type source struct {
channel *models.ChatFullInfo
to []maxChat
changes map[int64]syncOptions
}
type post struct{ messages []*models.Message }
// postRef points a human to a post: a link for a public channel, a number otherwise.
func (s *source) postRef(id int) string {
if s.channel.Username == "" {
return fmt.Sprintf("«%s» №%d", s.channel.Title, id)
}
return "https://t.me/" + s.channel.Username + "/" + strconv.Itoa(id)
}
// relay moves posts of Telegram channels into MAX channels.
type relay struct {
tg *bot.Bot
max *maxClient
http *http.Client
adminID int64 // Telegram user to report failures to; 0 — nobody
fileLimit int64 // largest file Telegram hands out; 0 — no limit (local Bot API server)
tgAPI telegramAPI
maxMediaHTTP *http.Client
mediaDir string
maxFileLimit int64
maxPostLimit int64
mediaSlot chan struct{} // only one MAX media post occupies disk at a time
}
// uploads are the attachments of one post uploaded for one MAX channel.
type uploads struct {
visual []attachment // photos and videos: go together, like an album
files []attachment // files and audio: MAX allows only one per message
failed bool
}
type plannedMedia struct {
ID string
Kind string
}
type plannedMessage struct {
Text string
Entities []models.MessageEntity
Media []plannedMedia
Quiet bool
}
type maxPostPart struct {
text string
attachments []attachment
}
func (u *uploads) add(a attachment) {
if a.kind == model.AttachFile || a.kind == model.AttachAudio {
u.files = append(u.files, a)
} else {
u.visual = append(u.visual, a)
}
}
func (r *relay) forward(ctx context.Context, src *source, p *post, book *deliveryBook) {
msgs := slices.SortedFunc(slices.Values(p.messages), func(a, b *models.Message) int { return a.ID - b.ID })
ref := src.postRef(msgs[0].ID)
log := slog.With("channel", src.channel.Title, "post", msgs[0].ID)
needed := make([]bool, len(src.to))
for i, dst := range src.to {
var err error
needed[i], err = book.needsDelivery(ctx, channelKey{maxPlatform, dst.id})
if err != nil {
return
}
}
if !slices.Contains(needed, true) {
return
}
var (
files []media
lost bool // part of the post cannot reach MAX: readers get a link to the original
problems []string
)
text, preview := telegramPostText(msgs)
for _, m := range msgs {
if md, ok := mediaOf(m); ok {
files = append(files, md)
} else if unsupported(m) {
log.Warn("в MAX такого нет, часть поста пропущена")
lost = true
}
}
if len(text.units) == 0 && len(files) == 0 && !lost {
log.Debug("пересылать нечего") // service messages: a pin, a new title
return
}
// Each file is downloaded once and uploaded for every destination, so only
// one file is held at a time.
uploaded := make([]uploads, len(src.to))
for _, md := range files {
file, err := r.download(ctx, md)
if err != nil {
log.Error("вложение не скачано", "file", md.name, "err", err)
problems = append(problems, fmt.Sprintf("%s: %v", md.name, err))
lost = true
continue
}
// A closure, so the file is removed even when an upload panics.
func() {
defer file.remove()
md, err = md.prepared(file)
if err != nil {
log.Warn("вложение не подходит для MAX", "file", md.name, "err", err)
problems = append(problems, fmt.Sprintf("%s: %v", md.name, err))
lost = true
return
}
for i, dst := range src.to {
if !needed[i] {
continue
}
a, err := r.upload(ctx, md, file)
if err != nil {
log.Error("вложение не загружено в MAX", "file", md.name, "to", dst.title, "err", err)
problems = append(problems, fmt.Sprintf("%s → «%s»: %v", md.name, dst.title, err))
uploaded[i].failed = true
continue
}
uploaded[i].add(a)
}
}()
}
for i, dst := range src.to {
if !needed[i] {
continue
}
body := text.clone()
// A private channel's link would not open for MAX readers
if (lost || uploaded[i].failed) && src.channel.Username != "" {
body.append(linkText("Открыть пост в Telegram", ref))
}
parts := body.split(maxTextLimit)
var plan []plannedMessage
for _, msg := range maxPostMessages(parts, uploaded[i]) {
part := plannedMessage{Text: msg.text}
for _, a := range msg.attachments {
part.Media = append(part.Media, plannedMedia{a.sourceID, string(a.kind)})
}
plan = append(plan, part)
}
if err := book.checkPlan(ctx, channelKey{maxPlatform, dst.id}, struct {
Messages []plannedMessage
Preview bool
}{plan, preview}); err != nil {
log.Error("продолжение публикации остановлено", "to", dst.title, "err", err)
return
}
var n int
err := book.deliver(ctx, channelKey{maxPlatform, dst.id}, func(publish publish) (err error) {
n, err = r.send(ctx, dst, parts, uploaded[i], preview, publish)
return err
})
if err != nil {
log.Error("не отправлено в MAX", "to", dst.title, "err", err)
problems = append(problems, fmt.Sprintf("«%s»: %v", dst.title, err))
continue
}
if n > 0 {
log.Info("переслано", "to", dst.title, "messages", n)
}
}
if len(problems) > 0 && ctx.Err() == nil {
_ = r.alert(ctx, fmt.Sprintf("пост %s перенесён с ошибками:\n— %s", ref, strings.Join(problems, "\n— ")))
}
}
// send posts a message set to a MAX channel: attachments with the start of the
// text first, then the rest of the text, then the files one per message.
// MAX channels notify subscribers for every part. It returns how many
// messages went out.
func maxPostMessages(parts []string, u uploads) []maxPostPart {
first, files := u.visual, u.files
if len(first) == 0 && len(files) > 0 {
first, files = files[:1], files[1:]
}
if len(parts) == 0 && len(first) == 0 {
return nil
}
caption := ""
if len(parts) > 0 {
caption, parts = parts[0], parts[1:]
}
messages := []maxPostPart{{caption, first}}
for _, part := range parts {
messages = append(messages, maxPostPart{text: part})
}
for _, f := range files {
messages = append(messages, maxPostPart{attachments: []attachment{f}})
}
return messages
}
func (r *relay) send(ctx context.Context, dst maxChat, parts []string, u uploads, preview bool, publish publish) (int, error) {
messages := maxPostMessages(parts, u)
for i, msg := range messages {
kind := mirrorText
if len(msg.attachments) > 0 {
kind = mirrorMedia
if i == 0 {
kind = mirrorCaption
}
}
err := publish(ctx, func() ([]string, error) {
return retryPublish(ctx, func() ([]string, error) {
return r.max.send(ctx, dst.id, msg.text, msg.attachments, preview)
})
}, kind)
if err != nil {
return i, fmt.Errorf("сообщение %d из %d не отправлено: %w", i+1, len(messages), err)
}
}
return len(messages), nil
}
func (r *relay) download(ctx context.Context, md media) (tgFile, error) {
if err := md.validate(md.size); err != nil {
return tgFile{}, err
}
if r.fileLimit > 0 && md.size > r.fileLimit {
return tgFile{}, errTooBig
}
var file tgFile
err := retry(ctx, func() (err error) {
file, err = downloadFile(ctx, r.tg, r.http, md.fileID)
return err
})
return file, err
}
func (r *relay) upload(ctx context.Context, md media, file tgFile) (attachment, error) {
var token string
err := retry(ctx, func() (err error) {
token, err = r.max.upload(ctx, md.upload, md.name, file)
return err
})
return attachment{kind: md.attach, token: token, sourceID: md.fileID}, err
}
// alert writes to the admin in Telegram when TELEGRAM_ADMIN_ID is set.
func (r *relay) alert(ctx context.Context, text string) error {
if err := ctx.Err(); err != nil {
return err
}
if r.adminID == 0 {
return nil
}
if runes := []rune(text); len(runes) > 3000 {
text = string(runes[:3000]) + "…"
}
ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
defer cancel()
ctx = withPacingChannel(ctx, strconv.FormatInt(r.adminID, 10))
_, err := r.tg.SendMessage(ctx, &bot.SendMessageParams{
ChatID: r.adminID,
Text: "perepost: " + text,
LinkPreviewOptions: &models.LinkPreviewOptions{IsDisabled: bot.True()},
})
if err != nil {
slog.Warn("не удалось написать администратору", "err", err)
}
return err
}